"""集成 v2 证据与口径硬伤定向测试（本轨新增，5 项全覆盖）。 1. 证据重放绑定（cloud/evidence_binding + release 端点）： 原样重交被拒、正常证据通过、陈旧/未来被拒、只带其一 400、无绑定走旧路径并标注 absent；另钉住诚实边界： 绑定不鉴别内容（改 ci_status 后仍通过校验——防重盖 戳须签名或只读求证，不在本模块冒充）。 2. 无价 token 口径统一（cloud/quota.monthly_usage 为 server/worker/automations 三处共用 canonical 源）： actual 无价行计入、estimated 无价行不计入，并与 main 旧 SQL（计全部 NULL 行）做前后对比。 3. 台账钩子开箱（cloud/ledger_onboarding）：表缺时 not ready、表齐且钩子开时 ready、表齐但钩子关时 仍 not ready；DDL 清单引用常量本身，不复制文本。 4. 台账写失败 outbox（cloud/ledger_outbox + worker _record_ledger_attempt 接线）：失败进 outbox、重放 补写行、重放后 outbox 清空、二次重放 0 行；任务 侧返回值仍为 None（不挡任务，规则 7 不变）。 5. 只读核对器（cloud/gate_verification）：假 reader 下一致通过、谎报 CI/保护被 mismatch、无 reader / 无目标 / reader 失败一律 checked=False 且 verified 恒 False（skipped 绝不当 pass）；实时接线不可实现 的三点见 gate_verification 模块头，服务端 docstring 已同步该边界（本文件不测不存在的实时求证）。 """ import http.client import json import logging import os import sqlite3 import tempfile import threading import time import unittest from pathlib import Path try: from cloud import ledger_onboarding from cloud import ledger_outbox from cloud.attempt_hook import ensure_marker_table from cloud.attempt_ledger import PG_DDL as LEDGER_PG_DDL from cloud.attempt_ledger import record_attempt from cloud.attempt_hook import PG_MARKER_DDL as MARKER_PG_DDL from cloud.cost import CostStore from cloud.evidence_binding import EvidenceNonceStore from cloud.evidence_binding import bind_evidence from cloud.evidence_binding import binding_presence from cloud.evidence_binding import validate_bound_evidence from cloud.gate_verification import verify_gate_evidence from cloud.quota import decide as quota_decide from cloud.quota import monthly_usage except ImportError: import sys sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud import ledger_onboarding from cloud import ledger_outbox from cloud.attempt_hook import ensure_marker_table from cloud.attempt_ledger import PG_DDL as LEDGER_PG_DDL from cloud.attempt_ledger import record_attempt from cloud.attempt_hook import PG_MARKER_DDL as MARKER_PG_DDL from cloud.cost import CostStore from cloud.evidence_binding import EvidenceNonceStore from cloud.evidence_binding import bind_evidence from cloud.evidence_binding import binding_presence from cloud.evidence_binding import validate_bound_evidence from cloud.gate_verification import verify_gate_evidence from cloud.quota import decide as quota_decide from cloud.quota import monthly_usage def _hard_body(): return { "files_changed": ["apps/web/a.py"], "owned_dir": "apps/web", "branch_protection": "protected", "ci_status": "success", "ci_checks": [{"name": "ci", "status": "success"}], "required_checks": ["ci"], "contract_version": "v2", "contract_frozen": True, "coordinator_approved": True, "unresolved_review_comments": 0, } class EvidenceBindingTestCase(unittest.TestCase): def test_bind_and_validate_fresh(self): stamped = bind_evidence(_hard_body()) self.assertEqual(binding_presence(stamped), "bound") ok, reason = validate_bound_evidence(stamped) self.assertTrue(ok) self.assertIsNone(reason) def test_partial_binding_is_not_bound(self): body = _hard_body() body["evidence_ts"] = time.time() self.assertEqual(binding_presence(body), "partial") ok, reason = validate_bound_evidence(body) self.assertFalse(ok) self.assertEqual(reason, "evidence_binding_incomplete") def test_stale_and_future_rejected(self): now = time.time() old = bind_evidence(_hard_body(), now=now - 7200) ok, reason = validate_bound_evidence(old, now=now) self.assertFalse(ok) self.assertEqual(reason, "evidence_too_old") future = bind_evidence(_hard_body(), now=now + 3600) ok, reason = validate_bound_evidence(future, now=now) self.assertFalse(ok) self.assertEqual(reason, "evidence_from_future") def test_nonce_single_use_via_store(self): conn = sqlite3.connect(":memory:") store = EvidenceNonceStore(conn) stamped = bind_evidence(_hard_body(), nonce="nonce-1") self.assertTrue(store.consume("t1", stamped["evidence_nonce"], stamped["evidence_ts"])) conn.commit() # 原样重交：同一 nonce 第二次消费即重放。 self.assertFalse(store.consume("t1", stamped["evidence_nonce"], stamped["evidence_ts"])) def test_binding_does_not_authenticate_content(self): """诚实边界：绑定防原样重放，不防重盖戳/篡改内容。 盖戳后改 ci_status，校验仍通过——因为没有签名。 本项钉住该边界，防止后来人把 binding 误读成防篡改。 """ stamped = bind_evidence(_hard_body()) stamped["ci_status"] = "failure" ok, _reason = validate_bound_evidence(stamped) self.assertTrue(ok) class QuotaActualOnlyTestCase(unittest.TestCase): def test_estimated_unpriced_rows_do_not_trip_cap(self): tmp = tempfile.TemporaryDirectory() try: db_path = os.path.join(tmp.name, "control.db") store = CostStore(db_path) conn = sqlite3.connect(db_path) try: conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY," " created_by TEXT)") conn.execute("INSERT INTO tasks(id, created_by)" " VALUES(?, ?)", ("task-1", "user-1")) conn.commit() finally: conn.close() # actual 无价 100（60+40），estimated 无价 5000， # actual 有价 1 行（usd 非 NULL，不进无价计数）。 store.record("task-1", "proj", "terra-model", input_tokens=60, output_tokens=40, usd_micros=None, source="actual") store.record("task-1", "proj", "terra-model", input_tokens=5000, output_tokens=0, usd_micros=None, source="estimated") store.record("task-1", "proj", "gpt-priced", input_tokens=10, output_tokens=10, usd_micros=12345, source="actual") month = time.strftime("%Y-%m", time.gmtime()) conn = sqlite3.connect(db_path) try: used, unpriced = monthly_usage(conn, "user-1", month) # 前后对比：main 旧 SQL（计全部 NULL 行）会得 # 5100；actual-only 口径只得 100。 old_row = conn.execute( "SELECT COALESCE(SUM(c.input_tokens" " + c.output_tokens), 0) FROM cost_records c" " JOIN tasks t ON c.task_id = t.id" " WHERE t.created_by = ? AND c.day LIKE ?" " AND c.usd_micros IS NULL", ("user-1", month + "%")).fetchone() self.assertEqual(int(old_row[0]), 5100) self.assertEqual(unpriced, 100) self.assertEqual(used, 12345) # 双侧同一口径：CostStore 报表 unpriced 亦为 100。 report = store.report_by_task("task-1") self.assertEqual(report["unpriced_actual_tokens"], 100) # 上限判定：100 未超 1000 放行；5100（旧口径） # 会误拦——这正是统一前服务端与执行侧的分歧点。 self.assertIsNone(quota_decide(10000000, used, unpriced, 1000, month)) self.assertIsNotNone(quota_decide( 10000000, used, 5100, 1000, month)) finally: conn.close() finally: tmp.cleanup() class LedgerOnboardingTestCase(unittest.TestCase): def test_missing_tables_not_ready(self): conn = sqlite3.connect(":memory:") report = ledger_onboarding.ledger_setup_report( conn, hook_enabled=True) self.assertFalse(report["ready"]) self.assertFalse(report["tables_ready"]) self.assertIn("task_attempts table missing", report["missing"]) self.assertIn("task_attempt_origin_markers table missing", report["missing"]) def test_tables_ready_but_hook_off_not_ready(self): conn = sqlite3.connect(":memory:") record_attempt(conn, "task-1", "success", is_defect=False) conn.commit() ensure_marker_table(conn) conn.commit() report = ledger_onboarding.ledger_setup_report( conn, hook_enabled=False) self.assertTrue(report["tables_ready"]) self.assertFalse(report["ready"]) self.assertTrue(any("enable the hook" in s for s in report["next_steps"])) report_on = ledger_onboarding.ledger_setup_report( conn, hook_enabled=True) self.assertTrue(report_on["ready"]) def test_ddl_checklist_references_constants_not_copies(self): sources = ledger_onboarding.ledger_ddl_sources() self.assertEqual(sources["task_attempts_pg_ddl"], LEDGER_PG_DDL) self.assertEqual(sources["marker_pg_ddl"], MARKER_PG_DDL) self.assertIn("task_attempts", LEDGER_PG_DDL) self.assertIn("corrects_attempt_no", LEDGER_PG_DDL) self.assertIn("task_attempt_origin_markers", MARKER_PG_DDL) steps = [item["step"] for item in ledger_onboarding.LEDGER_DDL_CHECKLIST] self.assertEqual(steps, [1, 2, 3, 4, 5]) class LedgerOutboxTestCase(unittest.TestCase): def test_append_and_replay(self): tmp = tempfile.TemporaryDirectory() try: path = os.path.join(tmp.name, "failures.jsonl") self.assertTrue(ledger_outbox.append_failure( path, "task-1", "success", "boom")) records, bad = ledger_outbox.read_failures(path) self.assertEqual(len(records), 1) self.assertEqual(bad, 0) conn = sqlite3.connect(":memory:") stats = ledger_outbox.replay_failures(path, conn) conn.commit() self.assertEqual(stats["replayed"], 1) self.assertEqual(stats["remaining"], 0) rows = conn.execute( "SELECT outcome FROM task_attempts" " WHERE task_id = ?", ("task-1",)).fetchall() self.assertEqual([row[0] for row in rows], ["success"]) # 二次重放：outbox 已空，0 行，不会多写一行。 stats2 = ledger_outbox.replay_failures(path, conn) self.assertEqual(stats2["replayed"], 0) rows = conn.execute( "SELECT COUNT(*) FROM task_attempts" " WHERE task_id = ?", ("task-1",)).fetchone() self.assertEqual(rows[0], 1) finally: tmp.cleanup() def test_append_failure_never_raises(self): # 路径父级是个文件时 makedirs 失败：返回 False，不抛。 tmp = tempfile.TemporaryDirectory() try: blocker = os.path.join(tmp.name, "blocker") with open(blocker, "w") as handle: handle.write("x") bad_path = os.path.join(blocker, "failures.jsonl") self.assertFalse(ledger_outbox.append_failure( bad_path, "task-1", "success", "boom")) finally: tmp.cleanup() def test_worker_record_failure_goes_to_outbox(self): """worker 接线：_record_ledger_attempt 写失败时， 除日志外 outbox 里留一行，且返回 None（不挡任务）。""" import cloud.worker as worker_mod tmp = tempfile.TemporaryDirectory() try: db_path = os.path.join(tmp.name, "control.db") worker = worker_mod.Worker.__new__(worker_mod.Worker) worker.db_path = db_path worker.attempt_ledger_hook = True worker.logger = logging.getLogger("test-v2-evidence") original_connect = worker_mod._db_connect def _boom(_path): raise RuntimeError("db down") worker_mod._db_connect = _boom try: result = worker._record_ledger_attempt( "task-9", "success", True) finally: worker_mod._db_connect = original_connect self.assertIsNone(result) outbox = ledger_outbox.outbox_path_for(db_path) records, bad = ledger_outbox.read_failures(outbox) self.assertEqual(len(records), 1) self.assertEqual(records[0]["task_id"], "task-9") self.assertEqual(records[0]["outcome"], "success") self.assertEqual(bad, 0) finally: tmp.cleanup() class GateVerificationTestCase(unittest.TestCase): def _live(self, checks, protection): def reader(repo, ref): self.assertEqual(repo, "o/r") self.assertEqual(ref, "sha1") return {"ci_checks": checks, "branch_protection": protection} return reader def test_matching_evidence_verified(self): result = verify_gate_evidence( _hard_body(), reader=self._live( {"ci": "success"}, "protected"), repo="o/r", ref="sha1") self.assertTrue(result["checked"]) self.assertTrue(result["verified"]) self.assertEqual(result["mismatches"], []) def test_lied_ci_and_protection_mismatch(self): result = verify_gate_evidence( _hard_body(), reader=self._live( {"ci": "failure"}, "unprotected"), repo="o/r", ref="sha1") self.assertTrue(result["checked"]) self.assertFalse(result["verified"]) self.assertTrue(any("ci_not_green_live" in m for m in result["mismatches"])) self.assertTrue(any("branch_protection" in m for m in result["mismatches"])) def test_no_reader_is_not_a_pass(self): result = verify_gate_evidence(_hard_body(), reader=None, repo="o/r", ref="sha1") self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) self.assertEqual(result["reason"], "no_trusted_reader") def test_no_target_or_reader_failure_is_not_a_pass(self): result = verify_gate_evidence( _hard_body(), reader=self._live({}, "protected"), repo=None, ref=None) self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) def _boom(_repo, _ref): raise RuntimeError("github down") result = verify_gate_evidence(_hard_body(), reader=_boom, repo="o/r", ref="sha1") self.assertFalse(result["checked"]) self.assertFalse(result["verified"]) self.assertTrue(result["reason"].startswith("reader_failed")) class ReleaseEvidenceBindingServerTestCase(unittest.TestCase): def setUp(self): from cloud.auth import AuthStore from cloud.server import create_server self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") self._saved_env = { k: os.environ.get(k) for k in ( "CLOUD_MERGE_GATE_MODE", "CLOUD_BRANCH_PROTECTION", "CODEX_GLOBAL_ADMINS")} os.environ.pop("CLOUD_MERGE_GATE_MODE", None) os.environ.pop("CLOUD_BRANCH_PROTECTION", None) self.auth = AuthStore(self.db_path) self.owner_id, self.owner_key = self.auth.create_user("owner") os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id 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) status, task = self._req( "POST", "/tasks", {"project_id": self.project["id"], "prompt": "x"}, self.owner_key) self.assertEqual(status, 201) conn = sqlite3.connect(self.db_path) try: conn.execute( "UPDATE tasks SET status = 'completed' WHERE id = ?", (task["id"],)) conn.commit() finally: conn.close() status, _merged = self._req( "POST", "/tasks/%s/merge" % task["id"], _hard_body(), self.owner_key) self.assertEqual(status, 200) self.task_id = task["id"] def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join(timeout=5) for key, value in self._saved_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value try: self.auth.conn.close() except Exception: pass self.tmp.cleanup() def _req(self, method, path, body=None, key=None): conn = http.client.HTTPConnection( "127.0.0.1", self.port, timeout=10) hdrs = {} payload = None if body is not None: payload = json.dumps(body).encode("utf-8") hdrs["Content-Type"] = "application/json" if key is not None: hdrs["X-API-Key"] = key try: conn.request(method, path, body=payload, headers=hdrs) resp = conn.getresponse() data = resp.read() finally: conn.close() try: parsed = json.loads(data.decode("utf-8")) if data else None except ValueError: parsed = None return resp.status, parsed def test_replay_rejected_fresh_passes(self): bound = bind_evidence(_hard_body()) status, first = self._req( "POST", "/tasks/%s/release" % self.task_id, bound, self.owner_key) self.assertEqual(status, 200) self.assertEqual(first["evidence_binding"], "bound") # 原样重交同一份证据体：nonce 已消费，409 重放。 status, replay = self._req( "POST", "/tasks/%s/release" % self.task_id, bound, self.owner_key) self.assertEqual(status, 409) self.assertEqual(replay["release_gate"]["code"], "release_blocked_evidence_replayed") # 新 nonce 的正常证据仍通过（其余字段完全相同）。 fresh = bind_evidence(_hard_body()) status, second = self._req( "POST", "/tasks/%s/release" % self.task_id, fresh, self.owner_key) self.assertEqual(status, 200) self.assertEqual(second["evidence_binding"], "bound") def test_stale_partial_and_absent(self): stale = bind_evidence(_hard_body(), now=time.time() - 7200) status, resp = self._req( "POST", "/tasks/%s/release" % self.task_id, stale, self.owner_key) self.assertEqual(status, 409) self.assertEqual(resp["release_gate"]["code"], "release_blocked_evidence_stale") partial = _hard_body() partial["evidence_ts"] = time.time() status, _resp = self._req( "POST", "/tasks/%s/release" % self.task_id, partial, self.owner_key) self.assertEqual(status, 400) status, plain = self._req( "POST", "/tasks/%s/release" % self.task_id, _hard_body(), self.owner_key) self.assertEqual(status, 200) self.assertEqual(plain["evidence_binding"], "absent") if __name__ == "__main__": unittest.main()