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
0b52a561df
commit
72d16a95dc
2 files changed
+49
No files matched your search
@@ -23,6 +23,7 @@ approve_merge -> maintainer,manage_project -> owner(仅 owner)。
|
|||||||
|
|
||||||
import hashlib
|
import hashlib
|
||||||
import secrets
|
import secrets
|
||||||
|
import sqlite3
|
||||||
from cloud.db import connect as _db_connect
|
from cloud.db import connect as _db_connect
|
||||||
from cloud.ledger import append_entry as _ledger_append
|
from cloud.ledger import append_entry as _ledger_append
|
||||||
from cloud.ledger import ensure_schema as _ledger_ensure
|
from cloud.ledger import ensure_schema as _ledger_ensure
|
||||||
|
|||||||
@@ -153,6 +153,54 @@ class TestAuth(unittest.TestCase):
|
|||||||
self.assertFalse(
|
self.assertFalse(
|
||||||
self.store.check_permission(target_id, project_id, "create_task"))
|
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):
|
def test_audit_append_only(self):
|
||||||
project_id, owner_id, _v, _c, _m, _o = self._project_with_roles()
|
project_id, owner_id, _v, _c, _m, _o = self._project_with_roles()
|
||||||
self.assertFalse(hasattr(AuthStore, "delete_audit"))
|
self.assertFalse(hasattr(AuthStore, "delete_audit"))
|
||||||
|
|||||||
Reference in new issue
Block a user