mini-openclaw-v2/backend/test_improvements.py

"""T11 改进治理测试：对照、泄漏、独立审核与发布门禁。

覆盖 AC-15/16/34/37/41（BR-15/16/40、E-11/24）。
"""

import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.testclient import TestClient

from api import prescreen as api
from contracts.improvement import ContrastMetrics, ImprovementCandidate
from services import assistant_store as store
from services import improvements
from services.project_assistant import ProjectAssistantService

HEADERS = {"X-Xingbang-Demo-Role": "manager"}
PROJECT = "DEMO-B"
BASELINE = {"id": "assistant.current", "version": "1.0.0"}
CANDIDATE = {"id": "prescreen.v2", "version": "0.2.0"}


def sample(sample_id, *, split="dev", observed=True, baseline_pass=True,
           candidate_pass=True, predicted="2026-09-29T01:00:00+00:00",
           revealed="2026-09-29T03:00:00+00:00", source_in_dev=False):
    return {"id": sample_id, "split": split, "observed": observed,
            "baseline_pass": baseline_pass, "candidate_pass": candidate_pass,
            "predicted_at": predicted, "label_revealed_at": revealed,
            "source_in_dev": source_in_dev}


def contrast(samples, *, independent=True, reviewer="reviewer-1"):
    return improvements.evaluate_contrast(
        samples, BASELINE, CANDIDATE, dataset_ref="fixtures/cases/manifest.json",
        split_ref="split.v1", reviewer_id=reviewer, independent=independent)


class ContrastTests(unittest.TestCase):
    def test_labels_read_before_the_lock_void_the_evaluation(self):
        with self.assertRaises(improvements.LeakageDetected):
            contrast([sample("S-1", predicted="2026-09-29T05:00:00+00:00")])

    def test_holdout_source_used_in_dev_is_contaminated(self):
        report = contrast([sample("S-1"), sample("S-2", split="holdout", source_in_dev=True)])
        self.assertEqual(report["contaminated"], ["S-2"])
        self.assertEqual(report["sample_size"], 1, "泄漏样本作废，不参与指标")

    def test_metrics_keep_numerator_denominator_and_unobserved_apart(self):
        report = contrast([sample("S-1"), sample("S-2", observed=False),
                           sample("S-3", baseline_pass=False, candidate_pass=True)])
        baseline = report["baseline_metrics"][0]
        candidate = report["candidate_metrics"][0]
        self.assertEqual((baseline["numerator"], baseline["observable_denominator"],
                          baseline["unobserved"]), (1, 2, 1))
        self.assertEqual((candidate["numerator"], candidate["observable_denominator"]), (2, 2))
        self.assertEqual(baseline["coverage"], 0.6667)
        self.assertEqual(report["verdict"], "improved")

    def test_zero_denominator_is_not_reported_as_zero_percent(self):
        report = contrast([sample("S-1", observed=False)])
        self.assertIsNone(report["candidate_metrics"][0]["rate"])
        self.assertIsNone(report["candidate_metrics"][0]["coverage"])
        self.assertEqual(report["verdict"], "inconclusive")

    def test_non_independent_evaluation_cannot_claim_improvement(self):
        report = contrast([sample("S-1")], independent=False)
        self.assertEqual(report["verdict"], "inconclusive")

    def test_contract_refuses_rate_without_a_denominator(self):
        with self.assertRaises(ValueError):
            ContrastMetrics(label="高预测组", numerator=0, observable_denominator=0,
                            unobserved=3, rate=0.0)

    def test_metrics_cannot_have_more_passes_than_samples(self):
        with self.assertRaises(ValueError):
            ContrastMetrics(label="高预测组", numerator=3, observable_denominator=2,
                            unobserved=0, rate=1.0)


class ImprovementLifecycleTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
        self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3")
        self.published: list[dict] = []

    def tearDown(self):
        self.temp.cleanup()

    def publish(self, body):
        ref = {"id": f"card-{len(self.published) + 1}", "version": "2.0.0"}
        self.published.append(ref)
        return ref

    def propose(self, actor="manager-1"):
        with store.transaction(self.service.db_path) as db:
            return improvements.propose(db, PROJECT, {
                "target": "knowledge", "target_ref": {"id": "card-1", "version": "1.0.0"},
                "proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
                "change_summary": "把回款核验的经验扩到制造业设备直租",
                "source_run_ids": ["run-1"]}, actor, "demo")

    def transition(self, candidate, action, actor="reviewer-9", **over):
        with store.transaction(self.service.db_path) as db:
            return improvements.transition(db, candidate["id"], action, actor,
                                           over.pop("expected_revision", None), **over)

    def test_proposal_without_sources_is_rejected(self):
        with self.assertRaises(ValueError):
            with store.transaction(self.service.db_path) as db:
                improvements.propose(db, PROJECT, {
                    "target": "knowledge", "target_ref": {"id": "c", "version": "1"},
                    "proposed_revision_ref": {"id": "c", "version": "2"},
                    "change_summary": "改一下"}, "manager-1", "demo")

    def test_release_requires_contrast_evidence(self):
        candidate = self.propose()
        with self.assertRaises(improvements.NotValidated) as ctx:
            self.transition(candidate, "release", released_ref={"id": "x", "version": "1"})
        self.assertIn("未通过审核", str(ctx.exception))

    def test_improved_candidate_can_be_approved_and_released(self):
        candidate = self.propose()
        report = contrast([sample("S-1", candidate_pass=True),
                           sample("S-2", baseline_pass=False, candidate_pass=True)])
        evaluated = self.transition(candidate, "evaluate", contrast=report)
        self.assertEqual(evaluated["status"], "ready")
        self.transition(evaluated, "submit")
        approved = self.transition(evaluated, "approve", actor="reviewer-9")
        self.assertEqual(approved["status"], "approved")
        released = self.transition(approved, "release", publisher=self.publish)
        self.assertEqual(released["status"], "released")
        self.assertEqual(released["released_ref"], {"id": "card-1", "version": "2.0.0"})
        self.assertEqual(len(self.published), 1)

    def test_self_approval_is_refused(self):
        candidate = self.propose(actor="manager-1")
        report = contrast([sample("S-1")])
        self.transition(candidate, "evaluate", contrast=report)
        self.transition(candidate, "submit")
        with self.assertRaises(PermissionError):
            self.transition(candidate, "approve", actor="manager-1")

    def test_regressed_candidate_cannot_be_released(self):
        candidate = self.propose()
        report = contrast([sample("S-1", baseline_pass=True, candidate_pass=False)])
        self.assertEqual(report["verdict"], "regressed")
        evaluated = self.transition(candidate, "evaluate", contrast=report)
        self.assertEqual(evaluated["status"], "evaluating")
        with self.assertRaises(improvements.NotValidated):
            self.transition(evaluated, "release", released_ref={"id": "x", "version": "1"})

    def test_contaminated_evaluation_does_not_advance_the_candidate(self):
        candidate = self.propose()
        report = contrast([sample("S-1"), sample("S-2", split="holdout", source_in_dev=True)])
        evaluated = self.transition(candidate, "evaluate", contrast=report)
        self.assertEqual(evaluated["status"], "evaluating")
        self.assertIn("泄漏", evaluated["rejection_reason"])

    def test_release_is_idempotent(self):
        candidate = self.propose()
        report = contrast([sample("S-1")])
        evaluated = self.transition(candidate, "evaluate", contrast=report)
        self.transition(evaluated, "submit")
        approved = self.transition(evaluated, "approve")
        released = self.transition(approved, "release", publisher=self.publish)
        again = self.transition(released, "release", publisher=self.publish)
        self.assertEqual(again, released, "重复发布不得新建版本")
        self.assertEqual(len(self.published), 1)

    def test_withdraw_only_applies_to_released_versions(self):
        candidate = self.propose()
        with self.assertRaises(improvements.NotValidated):
            self.transition(candidate, "withdraw")

    def test_contract_refuses_released_without_a_version_reference(self):
        body = {"id": "imp-1", "project_id": PROJECT, "data_domain": "demo", "revision": 1,
                "target": "knowledge", "target_ref": {"id": "c", "version": "1"},
                "change_summary": "改", "proposed_revision_ref": {"id": "c", "version": "2"},
                "status": "released", "reviewer_id": "r", "proposed_by": "m",
                "created_at": "2026-09-30T01:00:00+00:00"}
        with self.assertRaises(ValueError):
            ImprovementCandidate(**body)


class ImprovementApiTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
        self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3")
        self.patches = [
            patch.object(api, "_service", lambda: self.service),
            patch.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0",
                                    "XINGBANG_LOCAL_MODEL_URL": ""}),
        ]
        for item in self.patches:
            item.start()
        with store.transaction(self.service.db_path) as db:
            store.put(db, "project", {"id": PROJECT, "data_domain": "demo", "revision": 1})
        app = FastAPI()
        app.add_exception_handler(RequestValidationError, api.validation_handler)
        app.include_router(api.router)
        self.client = TestClient(app)
        self.base = "/api/xingbang/risk-agent/project-assistant/prescreen"

    def tearDown(self):
        self.client.close()
        for item in reversed(self.patches):
            item.stop()
        self.temp.cleanup()

    def test_propose_then_blocked_release(self):
        created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
                                   json={"target_ref": {"id": "card-1", "version": "1.0.0"},
                                         "proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
                                         "change_summary": "扩适用范围",
                                         "source_run_ids": ["run-1"]}, headers=HEADERS)
        self.assertEqual(created.status_code, 201, created.text)
        candidate = created.json()
        self.assertEqual(candidate["status"], "draft")

        # release 不再接受调用方自报的版本号（BR-51 / AC-57）；未过门禁在此先命中
        blocked = self.client.post(
            f"{self.base}/improvements/{candidate['id']}/transitions",
            json={"action": "release"}, headers=HEADERS)
        self.assertEqual(blocked.status_code, 409)
        self.assertEqual(blocked.json()["error"]["code"], "IMPROVEMENT_NOT_VALIDATED")

    def test_client_supplied_released_ref_is_rejected(self):
        """发布版本必须由服务端从真实发布结果产生，不接受调用方提交。"""
        created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
                                   json={"target_ref": {"id": "card-1", "version": "1.0.0"},
                                         "proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
                                         "change_summary": "扩适用范围",
                                         "source_run_ids": ["run-1"]}, headers=HEADERS).json()
        response = self.client.post(
            f"{self.base}/improvements/{created['id']}/transitions",
            json={"action": "release",
                  "released_ref": {"id": "card-1", "version": "2.0.0"}}, headers=HEADERS)
        self.assertEqual(response.status_code, 422, response.text)
        self.assertEqual(response.json()["error"]["code"], "INVALID_REQUEST")
        # 自报的版本号不得让候选变成"已发布"
        stored = self.client.get(f"{self.base}/projects/{PROJECT}/improvements",
                                 headers=HEADERS).json()["items"][0]
        self.assertEqual(stored["status"], "draft")
        self.assertIsNone(stored["released_ref"])

    def test_leakage_is_reported_as_e11(self):
        created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
                                   json={"target_ref": {"id": "card-1", "version": "1.0.0"},
                                         "proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
                                         "change_summary": "扩适用范围",
                                         "outcome_ids": ["outcome-1"]}, headers=HEADERS).json()
        # 泄漏在服务端判定：预测锁定前已读到标签的样本使整次评测作废
        leaked = self.client.post(
            f"{self.base}/improvements/{created['id']}/transitions",
            json={"action": "evaluate", "evaluation": {
                "baseline_ref": BASELINE, "candidate_ref": CANDIDATE,
                "dataset_ref": "fixtures/cases/manifest.json", "split_ref": "split.v1",
                "independent": True,
                "samples": [sample("S-1", predicted="2026-09-29T05:00:00+00:00")]}},
            headers=HEADERS)
        self.assertEqual(leaked.status_code, 409, leaked.text)
        self.assertEqual(leaked.json()["error"]["code"], "EVALUATION_CONTAMINATED")

        clean = self.client.post(
            f"{self.base}/improvements/{created['id']}/transitions",
            json={"action": "evaluate", "evaluation": {
                "baseline_ref": BASELINE, "candidate_ref": CANDIDATE,
                "dataset_ref": "fixtures/cases/manifest.json", "split_ref": "split.v1",
                "independent": True, "samples": [sample("S-1")]}},
            headers=HEADERS)
        self.assertEqual(clean.status_code, 200, clean.text)
        self.assertEqual(clean.json()["status"], "ready")

    def test_evaluation_without_samples_is_rejected(self):
        created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
                                   json={"target_ref": {"id": "card-1", "version": "1.0.0"},
                                         "proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
                                         "change_summary": "扩适用范围",
                                         "source_run_ids": ["run-1"]}, headers=HEADERS).json()
        response = self.client.post(
            f"{self.base}/improvements/{created['id']}/transitions",
            json={"action": "evaluate", "contrast": {}}, headers=HEADERS)
        self.assertEqual(response.status_code, 422, response.text)
        self.assertEqual(response.json()["error"]["code"], "INVALID_REQUEST")

    def test_unknown_action_is_rejected(self):
        created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
                                   json={"target_ref": {"id": "card-1", "version": "1.0.0"},
                                         "proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
                                         "change_summary": "扩适用范围",
                                         "source_run_ids": ["run-1"]}, headers=HEADERS).json()
        response = self.client.post(
            f"{self.base}/improvements/{created['id']}/transitions",
            json={"action": "promote"}, headers=HEADERS)
        self.assertEqual(response.status_code, 422)
        self.assertEqual(response.json()["error"]["code"], "INVALID_REQUEST")


if __name__ == "__main__":
    unittest.main()