fix(automations): 审计行不再显式插 TEXT id,PG identity 下不再丢审计
旧 _audit_row 显式插 uuid 当 audit_log.id:SQLite 下能跑, 生产 PG 的 id 是 BIGINT GENERATED BY DEFAULT AS IDENTITY, 插 TEXT 直接报错、审计行全丢(best-effort 吞掉后无声无息, #156 留下的已知局限)。修法:INSERT 一律不带 id,让两端 自增/identity 生成;表不存在时先按 PG 口径建、再按 SQLite 口径建、逐个重试——automations 按设计不 import 其它 cloud 模块,所以用先插、失败再按口径重试做方言探测。 单测:test_automation_run_due 4 项(含新增 PG-like 假连接 1 项:只接受不带 id 的 INSERT,旧实现在此应失败)。
This commit is contained in:
1 parent
0c458cb2ce
commit
8afcd3253e
2 files changed
+74
-12
No files matched your search
@@ -6,6 +6,7 @@ import time
|
||||
import unittest
|
||||
|
||||
from cloud.automations import (
|
||||
_audit_row,
|
||||
create_automation,
|
||||
ensure_automations_tables,
|
||||
list_automation_runs,
|
||||
@@ -48,6 +49,43 @@ def _set_quota(db_path, user_id, micros):
|
||||
conn.close()
|
||||
|
||||
|
||||
class AuditRowPgIdentityTestCase(unittest.TestCase):
|
||||
"""模拟 PG:id 是 BIGINT identity,显式插 TEXT uuid 会报错。
|
||||
|
||||
假连接只接受不带 id 的 INSERT;旧实现(带 TEXT id)在此
|
||||
假连接上应失败、新实现应成功且审计行落盘。
|
||||
"""
|
||||
|
||||
class _PgLikeConn:
|
||||
def __init__(self):
|
||||
self.rows = []
|
||||
self.committed = False
|
||||
|
||||
def execute(self, sql, params=()):
|
||||
if sql.strip().upper().startswith("INSERT INTO AUDIT_LOG(ID"):
|
||||
raise ValueError("column \"id\" is of type bigint")
|
||||
if sql.strip().upper().startswith("INSERT INTO AUDIT_LOG"):
|
||||
self.rows.append(params)
|
||||
return None
|
||||
raise ValueError("unexpected sql in fake PG conn")
|
||||
|
||||
def commit(self):
|
||||
self.committed = True
|
||||
|
||||
def rollback(self):
|
||||
pass
|
||||
|
||||
def test_insert_without_id_succeeds_on_pg_like_conn(self):
|
||||
conn = self._PgLikeConn()
|
||||
_audit_row(conn, "p1", "automation_run_created",
|
||||
"auto1 -> task1", 123.0)
|
||||
self.assertTrue(conn.committed)
|
||||
self.assertEqual(len(conn.rows), 1)
|
||||
# 参数里没有 uuid:第一位是 system,第二位是 project_id。
|
||||
self.assertEqual(conn.rows[0][0], "system")
|
||||
self.assertEqual(conn.rows[0][1], "p1")
|
||||
|
||||
|
||||
class RunDueAutomationTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
|
||||
Reference in new issue
Block a user