test_evolution.py

import hashlib
import json
import tempfile
import unittest
from pathlib import Path

from evolution import (
    STABLE,
    _load_gate,
    _replay_case,
    classify_risk,
    diagnose,
    generate_candidate,
    generate_rejected_control,
    release_manifest,
    validate_candidate,
)


ROOT = Path(__file__).parent


def _load(name):
    return json.loads((ROOT / name).read_text(encoding="utf-8"))


class DiagnosisTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.trajectories = _load("failure_trajectories.json")
        cls.diagnosis = diagnose(cls.trajectories)

    def test_support_threshold_creates_change_request(self):
        self.assertTrue(self.diagnosis["change_required"])
        cluster_ids = {p["cluster_id"] for p in self.diagnosis["patterns"]}
        self.assertEqual(
            {"unconfirmed_delete_file", "unconfirmed_force_push", "unconfirmed_destructive_sql"},
            cluster_ids,
        )
        for pattern in self.diagnosis["patterns"]:
            self.assertGreaterEqual(pattern["cross_trajectory_support"], 2)

    def test_below_threshold_cluster_is_ignored(self):
        # rm -rf 审计轨迹只有一条，低于支持门槛，不形成失败簇
        cluster_ids = {p["cluster_id"] for p in self.diagnosis["patterns"]}
        self.assertNotIn("unconfirmed_dangerous_shell", cluster_ids)

    def test_confirmed_operation_and_low_risk_feedback_excluded(self):
        source_ids = self.diagnosis["source_case_ids"]
        self.assertNotIn("traj-2026-0725-confirmed-delete", source_ids)
        self.assertNotIn("traj-2026-0724-thumb-write", source_ids)

    def test_signals_cover_all_three_sources(self):
        signals = {s for p in self.diagnosis["patterns"] for s in p["signals"]}
        self.assertEqual({"user_correction", "user_thumbs_down", "post_hoc_audit"}, signals)

    def test_root_cause_targets_dispatch_layer(self):
        self.assertEqual("stable/tool_dispatcher.py", self.diagnosis["target"])
        self.assertEqual("tool_dispatch_confirmation_gate", self.diagnosis["target_component"])

    def test_no_failure_means_no_change_request(self):
        result = diagnose([
            {"id": "ok-1", "signal": "post_hoc_audit", "outcome": "ok",
             "tool_calls": [{"tool": "delete_file", "args": {"path": "a"}, "user_confirmed": True}]},
        ])
        self.assertFalse(result["change_required"])
        self.assertIsNone(result["target"])


class RiskClassifierTest(unittest.TestCase):
    def test_high_risk_calls_are_flagged(self):
        high_risk = [
            ("delete_file", {"path": "reports/2026-Q1-draft.docx"}),
            ("delete_file", {"path": "tmp/cache-0417.tmp"}),
            ("git_push", {"remote": "origin", "branch": "main", "force": True}),
            ("sql_query", {"query": "DROP TABLE users"}),
            ("sql_query", {"query": "TRUNCATE TABLE orders"}),
            ("sql_query", {"query": "DELETE FROM orders"}),
            ("run_shell", {"command": "rm -rf build/"}),
        ]
        for tool, args in high_risk:
            kind, reason = classify_risk(tool, args)
            self.assertIsNotNone(kind, f"{tool} {args} 应判为高风险")
            self.assertTrue(reason)

    def test_low_risk_calls_pass_through(self):
        low_risk = [
            ("read_file", {"path": "a.md"}),
            ("write_file", {"path": "a.md", "content": "x"}),
            ("git_push", {"remote": "origin", "branch": "main", "force": False}),
            ("sql_query", {"query": "SELECT * FROM users"}),
            ("sql_query", {"query": "DELETE FROM users WHERE id = 2"}),
            ("run_shell", {"command": "ls -la"}),
        ]
        for tool, args in low_risk:
            kind, _ = classify_risk(tool, args)
            self.assertIsNone(kind, f"{tool} {args} 应判为低风险")


class CandidateGateTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.trajectories = _load("failure_trajectories.json")
        cls.boundary = _load("boundary_cases.json")
        cls.retention = _load("retention_cases.json")
        cls.stable = (ROOT / "stable" / "tool_dispatcher.py").read_text(encoding="utf-8")
        cls.diagnosis = diagnose(cls.trajectories)
        cls.candidate = generate_candidate(cls.stable, cls.diagnosis)
        cls.gate = _load_gate(cls.candidate["source"])

    def test_candidate_passes_all_gates(self):
        checks = validate_candidate(self.candidate["source"], self.boundary, self.retention)
        self.assertTrue(all(checks.values()), checks)

    def test_candidate_classifier_matches_reference(self):
        self.assertTrue(self.gate["requires_confirmation"]("delete_file", {"path": "x"}))
        self.assertTrue(self.gate["requires_confirmation"]("git_push", {"force": True}))
        self.assertTrue(self.gate["requires_confirmation"]("sql_query", {"query": "DROP TABLE t"}))
        self.assertTrue(self.gate["requires_confirmation"]("sql_query", {"query": "DELETE FROM t"}))
        self.assertFalse(self.gate["requires_confirmation"]("read_file", {"path": "x"}))
        self.assertFalse(self.gate["requires_confirmation"]("sql_query", {"query": "DELETE FROM t WHERE id=1"}))
        self.assertFalse(self.gate["requires_confirmation"]("git_push", {"force": False}))

    def test_token_is_single_use_and_bound_to_operation(self):
        env = STABLE.default_env()
        calls = []

        def execute(name, call_args):
            calls.append(name)
            return STABLE.dispatch(name, call_args, env=env)

        gate = self.gate
        token = gate["issue_confirmation"]("delete_file", {"path": "tmp/cache-0417.tmp"})
        first = gate["dispatch"]("delete_file", {"path": "tmp/cache-0417.tmp"},
                                 execute=execute, confirm_token=token)
        self.assertEqual("executed", first["status"])
        # 同一 token 第二次使用：必须拒绝且不得执行
        second = gate["dispatch"]("delete_file", {"path": "tmp/cache-0417.tmp"},
                                  execute=execute, confirm_token=token)
        self.assertEqual("rejected", second["status"])
        # 同一 token 换操作：同样拒绝
        third = gate["dispatch"]("delete_file", {"path": "notes/todo.md"},
                                 execute=execute, confirm_token=token)
        self.assertEqual("rejected", third["status"])
        self.assertEqual(1, len(calls))
    def test_generate_synthetic_perturbations_creates_edge_cases(self):
        from evolution import generate_synthetic_perturbations
        sample = [{"id": "t1", "tool_name": "delete_file", "args": {"path": "/tmp/a"}}]
        perturbed = generate_synthetic_perturbations(sample)
        self.assertEqual(3, len(perturbed))
        self.assertIsNone(perturbed[0]["args"])
        self.assertEqual("  ", perturbed[1]["tool_name"])
        self.assertIsInstance(perturbed[2]["args"], list)

    def test_suspended_call_never_reaches_executor(self):
        case = self.boundary[0]  # 第六章实验 6-5 的"高风险删除前确认"场景
        passed, detail = _replay_case(self.gate, case)
        self.assertTrue(passed, detail)

    def test_release_accepts_good_candidate(self):
        checks = validate_candidate(self.candidate["source"], self.boundary, self.retention)
        manifest = release_manifest(self.stable, self.candidate, self.diagnosis, checks)
        self.assertEqual("release_to_canary", manifest["decision"])
        self.assertEqual(manifest["stable_sha256"], manifest["rollback_sha256"])
        self.assertTrue(manifest["failure_cluster"])
        self.assertTrue(manifest["source_trajectories"])
        self.assertTrue(manifest["integration_diff"])

    def test_release_rejects_when_any_gate_fails(self):
        checks = {"static_compile": True, "security_scan": True, "gate_contract": True,
                  "boundary_replay": False, "retention_replay": True, "confirmation_single_use": True}
        manifest = release_manifest(self.stable, self.candidate, self.diagnosis, checks)
        self.assertEqual("reject_candidate", manifest["decision"])
        self.assertIn("boundary_replay", manifest["failed_checks"])

    def test_rejected_control_fails_boundary_replay(self):
        rejected = generate_rejected_control(self.stable, self.diagnosis)
        checks = validate_candidate(rejected["source"], self.boundary, self.retention)
        self.assertFalse(checks["boundary_replay"])
        manifest = release_manifest(self.stable, rejected, self.diagnosis, checks)
        self.assertEqual("reject_candidate", manifest["decision"])
        self.assertTrue(manifest["rejection_reason"])

    def test_degenerate_candidate_is_rejected_without_crashing(self):
        for source in ("", "\n", "# 只有注释\n"):
            checks = validate_candidate(source, self.boundary, self.retention)
            self.assertFalse(all(checks.values()))
        checks = validate_candidate("\ud800", self.boundary, self.retention)
        self.assertFalse(all(checks.values()))

    def test_unsafe_candidate_is_rejected_before_execution(self):
        with tempfile.TemporaryDirectory() as directory:
            marker = Path(directory) / "marker"
            unsafe = (
                "import os\n"
                f"os.system('touch {marker}')\n"
                "def requires_confirmation(tool_name, args=None):\n    return False\n"
                "def issue_confirmation(tool_name, args=None):\n    return 'x'\n"
                "def dispatch(tool_name, args=None, *, execute, confirm_token=None):\n"
                "    return {'status': 'executed', 'result': execute(tool_name, args)}\n"
            )
            checks = validate_candidate(unsafe, self.boundary, self.retention)
            self.assertTrue(checks["static_compile"])
            self.assertFalse(checks["security_scan"])
            self.assertFalse(marker.exists())

    def test_stable_hash_unchanged_through_pipeline(self):
        stable_path = ROOT / "stable" / "tool_dispatcher.py"
        before = hashlib.sha256(stable_path.read_bytes()).hexdigest()
        generate_candidate(self.stable, self.diagnosis)
        generate_rejected_control(self.stable, self.diagnosis)
        validate_candidate(self.candidate["source"], self.boundary, self.retention)
        after = hashlib.sha256(stable_path.read_bytes()).hexdigest()
        self.assertEqual(before, after)


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