"""审计第 2 批（I-1、I-2、I-5）单测：可见性三件。 运行（在仓库根目录下）： python3 -m unittest discover -s cloud/tests -p "test_audit_batch2.py" -v 说明：只断言本批新增的可见性行为，不碰 I-3、I-4、I-6～I-14、 N-1～N-12。I-1 断在报告层（不只断解析返回 None）；I-2 断在 任务级审计与影子报告字段层；I-5 四处各一例失败注入。 不跑真实模型回合、不联网。 """ import http.client import json import os import sqlite3 import sys import tempfile import threading import time import unittest from pathlib import Path from types import SimpleNamespace try: from cloud.auth import AuthStore from cloud.models import load_routing_policy from cloud.models import parse_routing_policy from cloud.routing_shadow import evaluate_shadow from cloud.sandbox import DockerExecutor from cloud.server import create_server from cloud.server import init_db from cloud.worker import Worker from cloud.worker import _CoordGateLookupFailed except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.auth import AuthStore from cloud.models import load_routing_policy from cloud.models import parse_routing_policy from cloud.routing_shadow import evaluate_shadow from cloud.sandbox import DockerExecutor from cloud.server import create_server from cloud.server import init_db from cloud.worker import Worker from cloud.worker import _CoordGateLookupFailed def _completed_runner(cmd): return SimpleNamespace(returncode=0, stdout="{}", stderr="") def _insert_task(db_path, task_id, status="queued", parent=None, part=None, error=None, audit_loop=0, parts=None, project_id="p1"): now = time.time() conn = sqlite3.connect(db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," " created_at, updated_at, last_heartbeat, audit_loop," " parent_task_id, coordinator_part, result_text, error_text," " paused_by_user, interrupted_reason, coordinator_parts," " started_at)" " VALUES(?, ?, 'do', ?, NULL, '[]', 0, 'tester'," " ?, ?, NULL, ?, ?, ?, NULL, ?, 0, NULL, ?, ?)", (task_id, project_id, status, now, now, audit_loop, parent, part, error, json.dumps(parts) if parts is not None else None, now)) conn.commit() finally: conn.close() def _audit_rows(db_path, project_id="p1"): store = AuthStore(db_path) try: return [(e["action"], e["detail"]) for e in store.list_audit( project_id)] finally: store.conn.close() def _task_row(db_path, task_id): conn = sqlite3.connect(db_path) try: row = conn.execute( "SELECT status, audit_loop_packet FROM tasks WHERE id = ?", (task_id,)).fetchone() return row finally: conn.close() class DroppedRowsTestCase(unittest.TestCase): def test_buckets_and_identity_at_report_level(self): # 6 行输入：2 个有效样本 + 4 种丢行原因各 1 行。 rows = [ {"detail": "t1 tier=plan source=shadow applied=False", "created_at": 1.0, "id": 1}, {"detail": "t2 tier=goal source=policy applied=True", "created_at": 2.0, "id": 2}, {"detail": "this line does not parse at all", "created_at": 3.0, "id": 3}, {"detail": "t3 tier=weird source=shadow applied=True", "created_at": 4.0, "id": 4}, {"detail": "t4 tier=plan source=explicit applied=False", "created_at": 5.0, "id": 5}, {"detail": "t1 tier=plan source=shadow applied=False", "created_at": 6.0, "id": 6}, ] tasks = {"t1": {"status": "completed"}, "t2": {"status": "completed"}} report = evaluate_shadow(rows, tasks, {}) self.assertEqual(report["dropped_rows"], { "unparseable": 1, "unknown_tier": 1, "explicit": 1, "duplicate": 1}) self.assertEqual(report["dropped_total"], 4) self.assertEqual(report["audit_rows_total"], 6) self.assertEqual(report["routed_task_n"], 2) # 行恒等式：总行数 = 样本数 + 丢行总数。 self.assertEqual( report["audit_rows_total"], report["routed_task_n"] + report["dropped_total"]) # 显式行同时计 per-tier explicit_n（不是丢失）。 self.assertEqual(report["tiers"]["plan"]["explicit_n"], 1) self.assertEqual(report["tiers"]["plan"]["routed_n"], 1) class ShadowEndpointPolicyTestCase(unittest.TestCase): def setUp(self): self._old_json = os.environ.get("CLOUD_ROUTING_POLICY_JSON") self._old_file = os.environ.get("CLOUD_ROUTING_POLICY_FILE") self._old_mode = os.environ.get("CLOUD_ROUTING_MODE") # 策略含一个无效档模式的档（plan）与一个有效档（goal）： # 端点报告里必须出现丢弃档 plan，而不是只剩 goal。 os.environ["CLOUD_ROUTING_POLICY_JSON"] = json.dumps({ "mode": "shadow", "tiers": {"plan": {"model": "m-plan", "mode": "bogus"}, "goal": "m-goal"}}) os.environ.pop("CLOUD_ROUTING_POLICY_FILE", None) os.environ.pop("CLOUD_ROUTING_MODE", None) self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.auth = AuthStore(self.db_path) _owner_id, self.owner_key = self.auth.create_user("owner") self.server = create_server(self.db_path, "127.0.0.1", 0) self.port = self.server.server_address[1] self.thread = threading.Thread( target=self.server.serve_forever, daemon=True) self.thread.start() status, self.project = self._req( "POST", "/projects", {"name": "demo"}, self.owner_key) self.assertEqual(status, 201, self.project) def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() for key, old in (("CLOUD_ROUTING_POLICY_JSON", self._old_json), ("CLOUD_ROUTING_POLICY_FILE", self._old_file), ("CLOUD_ROUTING_MODE", self._old_mode)): if old is None: os.environ.pop(key, None) else: os.environ[key] = old def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) headers = {"Content-Type": "application/json"} if key: headers["X-API-Key"] = key conn.request(method, path, json.dumps(body) if body is not None else None, headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) except ValueError: data = None return resp.status, data def test_endpoint_reports_dropped_tiers_and_fingerprint(self): status, report = self._req( "GET", "/routing-shadow?project_id=" + self.project["id"], key=self.owner_key) self.assertEqual(status, 200, report) policy = report["routing_policy"] self.assertEqual(policy["status"], "partial") self.assertEqual(policy["dropped_tiers"], ["plan"]) self.assertEqual(len(policy["fingerprint"]), 64) # 无样本时丢行计数仍在报告里（不是只有样本才有）。 self.assertEqual(report["audit_rows_total"], 0) self.assertEqual(report["dropped_total"], 0) class PolicyDropWarningTestCase(unittest.TestCase): def test_invalid_mode_warns_with_tier_name_and_keeps_others(self): with self.assertLogs("cloud.models", level="WARNING") as logs: policy = parse_routing_policy({ "mode": "shadow", "tiers": {"plan": {"model": "m-plan", "mode": "bogus"}, "goal": "m-goal"}}) self.assertIsNotNone(policy) self.assertNotIn("plan", policy.tiers) self.assertIn("goal", policy.tiers) self.assertEqual(policy.dropped_tiers, ["plan"]) self.assertTrue(any("plan" in m for m in logs.output)) def test_invalid_requires_drops_tier_with_warning(self): with self.assertLogs("cloud.models", level="WARNING") as logs: policy = parse_routing_policy({ "mode": "shadow", "tiers": {"plan": {"model": "m-plan", "requires": {"bogus_key": True}}, "goal": "m-goal"}}) self.assertIsNotNone(policy) self.assertNotIn("plan", policy.tiers) self.assertIn("goal", policy.tiers) self.assertEqual(policy.dropped_tiers, ["plan"]) self.assertTrue(any("plan" in m for m in logs.output)) class PartialFailureTaskAuditTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) _insert_task(self.db_path, "P1") def tearDown(self): self.tmp.cleanup() def test_partial_failure_writes_task_audit(self): # 文件来源失败 + JSON 来源可用 = 部分失败：策略可用， # 但必须在任务级审计里可见（与全失败同款 action）。 env = {"XCM_API_KEY": "test-placeholder", "CLOUD_ROUTING_POLICY_FILE": "/nonexistent/policy.json", "CLOUD_ROUTING_POLICY_JSON": json.dumps({ "mode": "shadow", "tiers": {"default": "m1"}})} worker = Worker(self.db_path, "/fake/host.py", runner=_completed_runner, env=env, max_retries=0, retry_backoff_seconds=0) self.assertFalse(worker.routing_policy_load_failed) self.assertTrue(worker.routing_policy_partial_failure) self.assertTrue(worker.routing_policy.partial_failures) result = worker.run_one() self.assertEqual(result["status"], "completed") actions = _audit_rows(self.db_path) partial = [d for a, d in actions if a == "worker_routing_policy_load_failed"] self.assertTrue(partial) self.assertIn("failure=partial", partial[0]) self.assertIn("policy.json", partial[0]) class CoordGateLookupFailureTestCase(unittest.TestCase): def setUp(self): self._old = os.environ.get("CODEX_PUSH_ALLOW_PRIVATE_TARGETS") os.environ["CODEX_PUSH_ALLOW_PRIVATE_TARGETS"] = "1" self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.env = {"XCM_API_KEY": "test-placeholder"} self.audits = [] self.worker = Worker(self.db_path, "/fake/host.py", runner=_completed_runner, env=self.env, max_retries=0, retry_backoff_seconds=0) self.worker._audit = lambda p, a, d="": self.audits.append( (p, a, d)) conn = sqlite3.connect(self.db_path) try: self.worker._ensure_result_columns(conn) conn.commit() finally: conn.close() def tearDown(self): if self._old is None: os.environ.pop("CODEX_PUSH_ALLOW_PRIVATE_TARGETS", None) else: os.environ[CODEX_PUSH_ALLOW_PRIVATE_TARGETS] = self._old self.tmp.cleanup() def test_direct_lookup_raises_instead_of_empty_set(self): # init_db 不建 audit_log 表（生产里由 AuthStore 建）： # 查询失败时旧代码返回空集，新代码必须抛错——空集 # 会被合流读成“没有门拒绝”而把门拒绝推回。 with self.assertRaises(_CoordGateLookupFailed): self.worker._coord_gate_rejected_child_ids("p1", ["C1"]) def test_merge_escalates_without_requeue_or_round(self): # 父任务审计开启、子任务执行失败；门拒绝审计表缺失 # 导致查询失败时，合流必须升级、不重排、不进推回。 # 注意：这里故意不建 audit_log 表——缺表即查询失败。 _insert_task(self.db_path, "P1", status="running", audit_loop=1, parts=[{"name": "a", "prompt": "do a"}]) _insert_task(self.db_path, "C1", status="failed", parent="P1", part="a", error="boom") out = self.worker._coord_audit_loop_merge( {"id": "P1", "project_id": "p1", "audit_loop": 1}, [{"name": "a", "prompt": "do a"}], [{"id": "C1", "name": "a"}]) self.assertEqual(out["decision"], "escalate") self.assertIn("查询失败", out["error"]) self.assertTrue(out.get("gate_lookup_failed")) # 子任务未被重排回 queued，仍是 failed。 self.assertEqual(_task_row(self.db_path, "C1")[0], "failed") # 升级审计里点名了查询失败。 self.assertTrue(any( a == "worker_coord_audit_loop_escalated" and "gate_lookup_failed" in d for _p, a, d in self.audits)) class PersistedModelLookupFailureTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) def tearDown(self): self.tmp.cleanup() def test_lookup_failure_audits_and_still_dispatches(self): # _db_connect 桩成抛错：复核查不到依据时返回 False # （放行），但必须在审计里点名依据缺失，不能只有日志。 # _audit 走 AuthStore 独立连接，不受该桩影响。 import cloud.worker as worker_mod real_connect = worker_mod._db_connect def _boom(_path): raise RuntimeError("db down for lookup") worker_mod._db_connect = _boom try: worker = Worker(self.db_path, "/fake/host.py", runner=_completed_runner, env={"XCM_API_KEY": "test-placeholder"}, max_retries=0, retry_backoff_seconds=0) self.assertFalse(worker._is_persisted_model( "T1", "p1", "m1")) finally: worker_mod._db_connect = real_connect actions = _audit_rows(self.db_path) found = [d for a, d in actions if a == "worker_persisted_model_lookup_failed"] self.assertTrue(found) self.assertIn("T1", found[0]) class RequiredToolsCheckFailureTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) def tearDown(self): self.tmp.cleanup() def test_check_failure_audits_and_still_allows(self): # 租户目录读取失败时返回 [] 放行不变，但必须在审计 # 里点名依据缺失，不能只有 warning 日志。 worker = Worker(self.db_path, "/fake/host.py", runner=_completed_runner, env={"XCM_API_KEY": "test-placeholder"}, max_retries=0, retry_backoff_seconds=0) worker.tenant_isolation = True worker.executor = DockerExecutor.__new__(DockerExecutor) def _boom(_project_id): raise RuntimeError("tenant home unreadable") worker._tenant_dirs = _boom result = worker._missing_required_tools( {"id": "T1", "project_id": "p1", "agent_tools": ["websearch"]}) self.assertEqual(result, []) actions = _audit_rows(self.db_path) found = [d for a, d in actions if a == "worker_required_tools_check_failed"] self.assertTrue(found) self.assertIn("T1", found[0]) class AuditLoopEvaluateFailureTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) def tearDown(self): self.tmp.cleanup() def test_evaluate_failure_still_escalates(self): # 回归断言：审计循环评估失败仍 fail-closed 到 escalate # （I-5 策略表里评估失败→升级这一路本来就有，不动）。 # 非法门结果值让纯函数 evaluate 抛 ValueError，经 # audit_loop_evaluate 捕获后应为 escalate。 worker = Worker(self.db_path, "/fake/host.py", runner=_completed_runner, env={"XCM_API_KEY": "test-placeholder"}, max_retries=0, retry_backoff_seconds=0) out = worker.audit_loop_evaluate( {"id": "T1", "project_id": "p1", "audit_loop": 1}, {"child:a": "not-a-bool"}) self.assertEqual(out["decision"], "escalate") self.assertEqual(out["reason"], "evaluate_failed") if __name__ == "__main__": unittest.main()