code-to10 v6 A4: fix AuthStore.audit NameError (import sqlite3) + retry regression test
Pre-fix any first-attempt audit failure raised NameError at auth.py:424 (sqlite3 never imported), aborting the 5-attempt retry loop on attempt 1 and masking the original error. The 424-line coverage mystery is also explained: the only full-suite hit is test_child_model.test_coordinator_inherits_child_defaults via worker best-effort _audit swallowing an OperationalError (database is locked); 422-423/426-427 stay uncovered because rollback succeeds and NameError aborts before either raise. Bubble audit (see report): no merge/release gate swallows the NameError and proceeds, so this stays a downgrade, not a production blocker. (cherry picked from commit 039c6adbac3b120b5b9a1dcbafeff6a91407ce70)
This commit is contained in:
1 parent
328ffdb7c6
commit
5db89d1b06
2 files changed
+49
No files matched your search
@@ -23,6 +23,7 @@ approve_merge -> maintainer,manage_project -> owner(仅 owner)。
|
||||
|
||||
import hashlib
|
||||
import secrets
|
||||
import sqlite3
|
||||
from cloud.db import connect as _db_connect
|
||||
from cloud.ledger import append_entry as _ledger_append
|
||||
from cloud.ledger import ensure_schema as _ledger_ensure
|
||||
|
||||
@@ -153,6 +153,54 @@ class TestAuth(unittest.TestCase):
|
||||
self.assertFalse(
|
||||
self.store.check_permission(target_id, project_id, "create_task"))
|
||||
|
||||
def test_audit_retries_integrity_error_and_reraises_others(self):
|
||||
"""A4 回归:audit() 重试分支的 sqlite3 引用必须存在。
|
||||
|
||||
首调 IntegrityError 后重试成功:审计对恰好一行(audit_log
|
||||
1 行 + audit_ledger 1 行),不是 0 行也不是多行;非完整
|
||||
性错误原样抛回(RuntimeError,不是 NameError 掩盖原错)。
|
||||
背景:auth.py 曾全文件无 import sqlite3,audit() 的
|
||||
except 分支在 424 行求 isinstance(exc, sqlite3.IntegrityError)
|
||||
时直接 NameError,任何首调失败都走不到重试或原错抛回。
|
||||
"""
|
||||
import cloud.auth as auth_mod
|
||||
|
||||
original = auth_mod.audit_on_conn
|
||||
calls = {"n": 0}
|
||||
|
||||
def _flaky(conn, *args, **kwargs):
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise sqlite3.IntegrityError(
|
||||
"UNIQUE constraint failed: audit_ledger.entry_hash")
|
||||
return original(conn, *args, **kwargs)
|
||||
|
||||
auth_mod.audit_on_conn = _flaky
|
||||
try:
|
||||
self.store.audit("u1", "p1", "act", "d")
|
||||
finally:
|
||||
auth_mod.audit_on_conn = original
|
||||
self.assertEqual(calls["n"], 2)
|
||||
self.assertEqual(len(self.store.list_audit("p1")), 1)
|
||||
ledger_rows = self.store.conn.execute(
|
||||
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
||||
self.assertEqual(ledger_rows, 1)
|
||||
|
||||
def _broken(conn, *args, **kwargs):
|
||||
raise RuntimeError("ledger boom")
|
||||
|
||||
auth_mod.audit_on_conn = _broken
|
||||
try:
|
||||
with self.assertRaises(RuntimeError) as caught:
|
||||
self.store.audit("u1", "p1", "act", "d2")
|
||||
finally:
|
||||
auth_mod.audit_on_conn = original
|
||||
self.assertEqual(str(caught.exception), "ledger boom")
|
||||
self.assertEqual(len(self.store.list_audit("p1")), 1)
|
||||
ledger_rows = self.store.conn.execute(
|
||||
"SELECT COUNT(*) FROM audit_ledger").fetchone()[0]
|
||||
self.assertEqual(ledger_rows, 1)
|
||||
|
||||
def test_audit_append_only(self):
|
||||
project_id, owner_id, _v, _c, _m, _o = self._project_with_roles()
|
||||
self.assertFalse(hasattr(AuthStore, "delete_audit"))
|
||||
|
||||
Reference in new issue
Block a user