codex-app-server/cloud/tests/test_v2_evidence.py

"""集成 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()