#!/usr/bin/env python3
"""Tests for the retained AutoX case validator."""

from __future__ import annotations

import importlib.util
import json
import re
import tempfile
import unittest
from argparse import Namespace
from pathlib import Path


MODULE_PATH = Path(__file__).with_name("validate_case.py")
SPEC = importlib.util.spec_from_file_location("validate_case", MODULE_PATH)
assert SPEC and SPEC.loader
VALIDATE_CASE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(VALIDATE_CASE)


class ValidateCaseTest(unittest.TestCase):
    def write_case(self, action: str, level: str, advisory_gate: str | None,
                   validation_status: str = "not run") -> Path:
        workspace = Path(self.tempdir.name)
        report_dir = workspace / "report"
        report_dir.mkdir()
        result = {
            "rank": 1,
            "digest": "digest",
            "diagnosis_id": "diagnosis",
            "status": "completed",
            "recommended_action": action,
            "validation_level": level,
            "report_path": "report/report.md",
            "plan_validation_status": "passed" if level == "plan_verified" else "not_run",
            "runtime_evidence_status": "observed" if level == "observed" else "not_observed",
        }
        if level == "plan_verified":
            result["plan_validation"] = {
                "validation_source": "local_tidb",
                "target_tidb_version": "v8.5.4",
                "local_tidb_version": "v8.5.4",
                "source_commit": "0123456789abcdef",
                "reproduction_kind": (
                    "candidate"
                    if action in VALIDATE_CASE.PLAN_CHANGING_ACTIONS
                    else "baseline_recovered"
                ),
                "version_match": True,
                "schema_loaded": True,
                "stats_loaded": True,
                "full_sql_validated": True,
                "baseline_plan_captured": True,
                "candidate_plan_captured": True,
                "baseline_matches_expected_shape": (
                    action not in VALIDATE_CASE.PLAN_CHANGING_ACTIONS
                ),
                "syntax_accepted": True,
                "optimizer_selected_expected_path": True,
                "plan_shape_matches_diagnosis": True,
            }
            if action in VALIDATE_CASE.PLAN_CHANGING_ACTIONS:
                result["plan_validation"]["semantic_delta"] = {
                    "material": True,
                    "addresses_diagnosed_mechanism": True,
                    "changed_fields": ["access_conditions", "access_range"],
                    "before": "range: decided by [eq(test.t.a, 1)]",
                    "after": "range: [1 2,1 2], access:eq(a,1),eq(b,2)",
                    "summary": (
                        "The same operator tree uses an additional index access condition "
                        "and a narrower range."
                    ),
                }
        if level == "observed":
            result["runtime_observation"] = {
                "observation_source": "production_runtime",
                "runtime_evidence_observed": True,
                "diagnosis_supported": True,
                "evidence_reference": "report.md#observed-evidence",
            }
        if advisory_gate is not None:
            result["advisory_gate"] = advisory_gate
        (workspace / "result.json").write_text(
            json.dumps(result), encoding="utf-8"
        )
        if action in VALIDATE_CASE.PLAN_CHANGING_ACTIONS:
            conclusion = f"""Action:
{action}

Review-only SQL:
Review only. Not executed by AutoX.
```sql
CREATE INDEX idx_a ON t (a);
```"""
            candidate_plan = (
                "Covering IndexReader; candidate plan was not reproduced"
                if level in {"inferred", "observed"}
                else """id                    estRows  task       access object               operator info
IndexLookUp_20        1.00     root
├─IndexRangeScan_18   1.00     cop[tikv] table:t, index:idx_a(a,b) range:[1 2,1 2], access:eq(test.t.a, 1), eq(test.t.b, 2)
└─TableRowIDScan_19   1.00     cop[tikv] table:t                    keep order:false"""
            )
            plans = f"""\n\n## Plans Before & After

Plan before:
```text
id                    estRows  actRows  task       access object               operator info
IndexLookUp_10        1.00     1        root
├─IndexRangeScan_8    1.00     1        cop[tikv] table:t, index:idx_old(a)   range decided by [eq(test.t.a, 1)]
└─TableRowIDScan_9    1.00     1        cop[tikv] table:t                    keep order:false
```

Plan after:
```text
{candidate_plan}
```"""
        else:
            conclusion = f"""Action:
{action}"""
            plans = ""
        (report_dir / "report.md").write_text(
            f"""## Conclusion

{conclusion}{plans}

## Analysis

Why:
The candidate targets the diagnosed runtime mechanism, or no plan-changing action is justified by the observed plan.

Observed evidence:
- SQL digest: `digest`
The production plan performs a table-row fetch.

Inference:
The covering candidate targets that fetch.

Validation and risks:
- Validation status: {validation_status}
- Validation level: {level}
- Plan validation status: {result['plan_validation_status']}
- Runtime evidence status: {result['runtime_evidence_status']}
- Advisory gate: {advisory_gate or 'not applicable'}
- Selected candidate ID: idx_a
- Missing evidence or blocker: candidate plan was not reproduced
- Risk and next validation: validate plan shape before rollout
""",
            encoding="utf-8",
        )
        return workspace

    def setUp(self) -> None:
        self.tempdir = tempfile.TemporaryDirectory()

    def tearDown(self) -> None:
        self.tempdir.cleanup()

    def validate(self, workspace: Path, allow_legacy_report_name: bool = False,
                 allow_legacy_report_format: bool = False,
                 allow_legacy_validation_evidence: bool = False) -> list[str]:
        args = Namespace(
            workspace=str(workspace), rank=None, digest=None,
            diagnosis_id=None, json_output=False,
            allow_legacy_report_name=allow_legacy_report_name,
            allow_legacy_report_format=allow_legacy_report_format,
            allow_legacy_validation_evidence=allow_legacy_validation_evidence,
        )
        _, errors, _ = VALIDATE_CASE.validate(args)
        return errors

    def rename_report(self, workspace: Path, filename: str) -> None:
        canonical = workspace / "report" / "report.md"
        legacy = canonical.with_name(filename)
        canonical.rename(legacy)
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result["report_path"] = f"report/{filename}"
        result_path.write_text(json.dumps(result), encoding="utf-8")

    def test_inferred_index_with_passed_advisory_gate_is_valid(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        self.assertEqual([], self.validate(workspace))

    def test_inferred_index_without_passed_advisory_gate_is_invalid(self) -> None:
        workspace = self.write_case("Index first", "inferred", None)
        self.assertIn(
            "unverified Index first lacks passed advisory_gate",
            self.validate(workspace),
        )

    def test_observed_index_with_passed_advisory_gate_is_valid(self) -> None:
        workspace = self.write_case(
            "Index first", "observed", "passed",
            validation_status="observed in production runtime evidence",
        )
        self.assertEqual([], self.validate(workspace))

    def test_rejected_index_cannot_use_advisory_fallback(self) -> None:
        workspace = self.write_case(
            "Index first", "inferred", "passed", validation_status="rejected"
        )
        self.assertIn(
            "unverified Index first has invalid validation status",
            self.validate(workspace),
        )

    def test_inferred_binding_is_invalid(self) -> None:
        workspace = self.write_case("Binding first", "inferred", "passed")
        self.assertIn(
            "optimizer action is not plan_verified",
            self.validate(workspace),
        )

    def test_plan_verified_requires_local_validation_evidence(self) -> None:
        workspace = self.write_case("Binding first", "plan_verified", "passed")
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result.pop("plan_validation")
        result_path.write_text(json.dumps(result), encoding="utf-8")
        self.assertIn(
            "plan_verified result missing plan_validation evidence",
            self.validate(workspace),
        )

    def test_plan_verified_rejects_production_static_source(self) -> None:
        workspace = self.write_case("Binding first", "plan_verified", "passed")
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result["plan_validation"]["validation_source"] = "production_static_explain"
        result_path.write_text(json.dumps(result), encoding="utf-8")
        self.assertIn(
            "plan_validation source must be local_tidb",
            self.validate(workspace),
        )

    def test_plan_verified_requires_material_semantic_delta(self) -> None:
        workspace = self.write_case("Binding first", "plan_verified", "passed")
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result["plan_validation"].pop("semantic_delta")
        result_path.write_text(json.dumps(result), encoding="utf-8")
        self.assertIn(
            "plan-changing plan_validation missing semantic_delta",
            self.validate(workspace),
        )

    def test_estimates_alone_are_not_a_material_semantic_delta(self) -> None:
        workspace = self.write_case("Binding first", "plan_verified", "passed")
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result["plan_validation"]["semantic_delta"]["changed_fields"] = [
            "estimated_rows",
            "estimated_cost",
        ]
        result_path.write_text(json.dumps(result), encoding="utf-8")
        self.assertIn(
            "plan_validation semantic_delta lacks a material plan field",
            self.validate(workspace),
        )

    def test_same_operator_tree_with_better_access_conditions_is_valid(self) -> None:
        workspace = self.write_case("Binding first", "plan_verified", "passed")
        self.assertEqual([], self.validate(workspace))

    def test_plan_verified_accepts_natural_baseline_recovery(self) -> None:
        workspace = self.write_case(
            "No optimizer action", "plan_verified", None,
            validation_status="plan already recovered",
        )
        self.assertEqual([], self.validate(workspace))

    def test_observed_does_not_imply_plan_verified(self) -> None:
        workspace = self.write_case(
            "Investigate non-optimizer bottleneck", "observed", None,
            validation_status="observed in production runtime evidence",
        )
        result = json.loads((workspace / "result.json").read_text(encoding="utf-8"))
        self.assertEqual("not_run", result["plan_validation_status"])
        self.assertEqual([], self.validate(workspace))

    def test_observed_requires_runtime_evidence(self) -> None:
        workspace = self.write_case(
            "Investigate non-optimizer bottleneck", "observed", None,
            validation_status="observed in production runtime evidence",
        )
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result["runtime_observation"]["observation_source"] = "historical_plan"
        result_path.write_text(json.dumps(result), encoding="utf-8")
        self.assertIn(
            "runtime_observation source must be production_runtime",
            self.validate(workspace),
        )

    def test_legacy_validation_evidence_requires_explicit_flag(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        result_path = workspace / "result.json"
        result = json.loads(result_path.read_text(encoding="utf-8"))
        result.pop("plan_validation_status")
        result.pop("runtime_evidence_status")
        result_path.write_text(json.dumps(result), encoding="utf-8")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = re.sub(
            r"^- (?:Plan validation|Runtime evidence) status:.*\n",
            "",
            text,
            flags=re.MULTILINE,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "invalid or missing plan_validation_status",
            self.validate(workspace),
        )
        self.assertEqual(
            [],
            self.validate(workspace, allow_legacy_validation_evidence=True),
        )

    def test_noncanonical_report_name_is_invalid(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        self.rename_report(workspace, "final_report.md")
        self.assertIn(
            "report_path must resolve to report/report.md",
            self.validate(workspace),
        )

    def test_legacy_report_name_requires_explicit_compatibility_flag(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        self.rename_report(workspace, "final-report.md")
        self.assertEqual(
            [],
            self.validate(workspace, allow_legacy_report_name=True),
        )

    def test_no_action_with_compact_reason_is_valid(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        self.assertEqual([], self.validate(workspace))

    def test_no_action_without_reason_is_invalid(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = text.replace(
            "\nWhy:\nThe candidate targets the diagnosed runtime mechanism, or no "
            "plan-changing action is justified by the observed plan.",
            "",
            1,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "Analysis must contain only Why, Observed evidence, Inference, and Validation and risks",
            self.validate(workspace),
        )

    def test_report_requires_digest(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8").replace(
            "- SQL digest: `digest`\n", "", 1
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "Observed evidence must start with exactly one SQL digest bullet",
            self.validate(workspace),
        )

    def test_report_digest_must_match_result(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8").replace(
            "- SQL digest: `digest`", "- SQL digest: `different`", 1
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn("report SQL digest does not match result.json", self.validate(workspace))

    def test_report_must_be_english_only(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        report = workspace / "report" / "report.md"
        report.write_text(
            report.read_text(encoding="utf-8").replace(
                "The covering candidate targets that fetch.", "该候选索引避免回表。", 1
            ),
            encoding="utf-8",
        )
        self.assertIn("customer-facing report must be English only", self.validate(workspace))

    def test_no_action_conclusion_rejects_plan_fields(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = text.replace(
            "No optimizer action\n\n## Analysis",
            "No optimizer action\n\nPlan before:\nIndexLookUp\n\n## Analysis",
            1,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "non-plan-changing Conclusion must contain only Action",
            self.validate(workspace),
        )

    def test_plan_action_requires_before_and_after_plans(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = text.replace(
            "\nPlan after:\n```text\n"
            "Covering IndexReader; candidate plan was not reproduced\n```\n",
            "\n",
            1,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "Plans Before & After must contain only complete Plan before and Plan after blocks",
            self.validate(workspace),
        )

    def test_plan_before_rejects_decoded_plan_json(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = re.sub(
            r"(?<=Plan before:\n```text\n).*?(?=\n```\n\nPlan after:)",
            '{"main":{"name":"IndexLookUp_10","children":[]}}',
            text,
            count=1,
            flags=re.DOTALL,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "Plan before must be a rendered TiDB EXPLAIN-style plan, not JSON or prose",
            self.validate(workspace),
        )

    def test_plan_before_rejects_prose_summary(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        report = workspace / "report" / "report.md"
        text = re.sub(
            r"(?<=Plan before:\n```text\n).*?(?=\n```\n\nPlan after:)",
            "IndexLookUp with table-row fetch",
            report.read_text(encoding="utf-8"),
            count=1,
            flags=re.DOTALL,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "Plan before must be a rendered TiDB EXPLAIN-style plan, not JSON or prose",
            self.validate(workspace),
        )

    def test_plan_action_requires_review_sql_in_conclusion(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        sql_start = text.index("\nReview-only SQL:")
        plans_start = text.index("\n## Plans Before & After")
        review_sql = text[sql_start:plans_start]
        text = text[:sql_start] + text[plans_start:]
        text = text.replace("\n## Analysis\n", f"\n## Analysis\n{review_sql}\n", 1)
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "plan-changing Conclusion must contain only Action and concrete review-only SQL",
            self.validate(workspace),
        )

    def test_plan_action_rejects_review_sql_none(self) -> None:
        workspace = self.write_case("Binding first", "plan_verified", "passed")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = text.replace("CREATE INDEX idx_a ON t (a);", "none", 1)
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "plan-changing Conclusion must contain concrete review-only SQL",
            self.validate(workspace),
        )

    def test_no_action_rejects_plans_in_analysis(self) -> None:
        workspace = self.write_case("No optimizer action", "inferred", None)
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = text.replace(
            "Observed evidence:\n",
            "Observed evidence:\nPlan before: IndexLookUp\n",
            1,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "non-plan-changing report must omit before/after plans",
            self.validate(workspace),
        )

    def test_conclusion_rejects_verbose_plan_metadata(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        text = text.replace(
            "Plan before:\n```text",
            "Plan before:\n- Source: slow log decoded_plan\n```text",
            1,
        )
        report.write_text(text, encoding="utf-8")
        self.assertIn(
            "Plans Before & After must contain only complete Plan before and Plan after blocks",
            self.validate(workspace),
        )

    def test_legacy_report_format_requires_explicit_compatibility_flag(self) -> None:
        workspace = self.write_case("Index first", "inferred", "passed")
        report = workspace / "report" / "report.md"
        text = report.read_text(encoding="utf-8")
        analysis_start = text.index("\n## Analysis")
        legacy = """## Conclusion

Action:
Index first

Plan before:
```text
IndexLookUp with table-row fetch
```

Plan after:
```text
Covering IndexReader; candidate plan was not reproduced
```

Why:
The candidate removes the diagnosed table-row fetch.
""" + text[analysis_start:].replace(
            "## Analysis\n\nWhy:",
            "## Analysis\n\nReview-only SQL:\nReview only. Not executed by AutoX.\n"
            "```sql\nCREATE INDEX idx_a ON t (a);\n```\n\nWhy:",
            1,
        )
        report.write_text(legacy, encoding="utf-8")
        self.assertTrue(
            any(
                error.startswith("invalid top-level report headings")
                for error in self.validate(workspace)
            )
        )
        self.assertEqual(
            [],
            self.validate(workspace, allow_legacy_report_format=True),
        )


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