Merge pull request #161 from openqj/feat-automation-audit-pg

fix(automations): 审计行不再显式插 TEXT id，PG identity 下不再丢审计

cloud/tests/test_automation_run_due.py

@@ -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()