"""Lane-contract conformance: every lane, five checks, positive and negative.

Parameterized (subTest) over lane_contract.LANE_TABLE. For each lane the

valid synthetic rows (mirroring the lane's real emitters) must pass all five

checks, and a crafted violation must fail each check that applies — negative

tests prove the checks have teeth. The meta-test guarantees a new lane cannot

ship without a LANE_TABLE entry.

"""

from __future__ import annotations

import copy

import sys

from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

import unittest

from oakbench import lane_contract, remotes, rows as oakbench_rows

from oakbench.lane_contract import (

    DE_FACTO_LANES,

    LANE_TABLE,

    catalog_operations,

    check_catalog_membership,

    check_measurement_source,

    check_row_contract,

    check_skip_coherence,

    check_transport_purity,

    expected_lanes,

)

def sample_rows(lane: str) -> list[dict]:

    """Fresh deep copy so one subTest's mutation never leaks into another."""

    return copy.deepcopy(LANE_TABLE[lane].sample_rows_builder())

def catalog_errors(contract: lane_contract.LaneContract, rows: list[dict]) -> list[str]:

    catalog = catalog_operations(contract.catalog_path) if contract.catalog_path else None

    return check_catalog_membership(

        rows,

        catalog,

        exempt_ops=contract.catalog_exempt_ops,

        exempt_prefixes=contract.catalog_exempt_prefixes,

        exempt_suffixes=contract.catalog_exempt_suffixes,

    )

class LaneTableMetaTest(unittest.TestCase):

    """Adding a lane without a contract entry fails CI with a clear message."""

    def test_every_lane_has_a_contract_entry(self) -> None:

        missing = sorted(expected_lanes() - set(LANE_TABLE))

        self.assertEqual(

            missing,

            [],

            "New lane(s) without a LANE_TABLE entry in scripts/oakbench/lane_contract.py: "

            f"{missing}. Add one LaneContract per lane (see docs/new-lane-template.md) "

            "so row contracts, skip coherence, transport purity, measurement_source, "

            "and catalog membership stay machine-checked.",

        )

    def test_no_orphan_contract_entries(self) -> None:

        orphans = sorted(set(LANE_TABLE) - expected_lanes())

        self.assertEqual(

            orphans,

            [],

            f"LANE_TABLE entries for unknown lanes: {orphans}. Either add the lane to "

            "rows.LANES, list it in lane_contract.DE_FACTO_LANES, or delete the entry.",

        )

    def test_rows_lanes_all_validatable(self) -> None:

        # Every rows.py lane delegates to validate_rows under its own key.

        for lane in oakbench_rows.LANES:

            with self.subTest(lane=lane):

                self.assertEqual(LANE_TABLE[lane].validate_lane, lane)

    def test_de_facto_lanes_documented(self) -> None:

        self.assertEqual(set(DE_FACTO_LANES), {"netshape", "external"})

        # netshape opts into the core key contract (netshape_bench.write_rows).

        self.assertEqual(LANE_TABLE["netshape"].validate_lane, "core")

        # external (tperf) rows are not bench rows; universal keys only.

        self.assertIsNone(LANE_TABLE["external"].validate_lane)

        self.assertTrue(LANE_TABLE["external"].universal_keys)

class ValidRowsPassAllChecksTest(unittest.TestCase):

    def test_all_lanes(self) -> None:

        for lane, contract in LANE_TABLE.items():

            rows = sample_rows(lane)

            with self.subTest(lane=lane, check="row_contract"):

                self.assertEqual(check_row_contract(rows, contract), [])

            with self.subTest(lane=lane, check="skip_coherence"):

                self.assertEqual(check_skip_coherence(rows), [])

            with self.subTest(lane=lane, check="transport_purity"):

                self.assertEqual(check_transport_purity(rows), [])

            with self.subTest(lane=lane, check="measurement_source"):

                self.assertEqual(

                    check_measurement_source(

                        rows, contract.allowed_sources, required=contract.source_required

                    ),

                    [],

                )

            with self.subTest(lane=lane, check="catalog_membership"):

                self.assertEqual(catalog_errors(contract, rows), [])

class RowContractNegativeTest(unittest.TestCase):

    def test_missing_required_key_fails(self) -> None:

        for lane, contract in LANE_TABLE.items():

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                if contract.validate_lane is not None:

                    victim = oakbench_rows.LANE_REQUIRED_KEYS[contract.validate_lane][0]

                else:

                    victim = contract.universal_keys[0]

                del rows[0][victim]

                errors = check_row_contract(rows, contract)

                self.assertTrue(errors, f"deleting {victim!r} must fail the row contract")

                self.assertIn(victim, "\n".join(errors))

    def test_non_numeric_elapsed_fails(self) -> None:

        for lane, contract in LANE_TABLE.items():

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                rows[0]["elapsed_ms"] = "fast"

                self.assertTrue(check_row_contract(rows, contract))

class SkipCoherenceNegativeTest(unittest.TestCase):

    def test_77_without_reason_fails(self) -> None:

        for lane in LANE_TABLE:

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                bad = copy.deepcopy(rows[0])

                bad["returncode"] = oakbench_rows.SKIP_RETURNCODE

                bad.pop("skipped", None)

                bad.pop("skip_reason", None)

                errors = check_skip_coherence(rows + [bad])

                self.assertTrue(any("skipped: true" in e for e in errors))

                self.assertTrue(any("skip_reason" in e for e in errors))

    def test_skipped_without_77_fails(self) -> None:

        for lane in LANE_TABLE:

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                bad = copy.deepcopy(rows[0])

                bad["returncode"] = 0

                bad["skipped"] = True

                bad["skip_reason"] = "half-marked skip"

                errors = check_skip_coherence(rows + [bad])

                self.assertTrue(any("requires returncode 77" in e for e in errors))

class TransportPurityNegativeTest(unittest.TestCase):

    def test_mixed_transports_same_subject_scenario_operation_fail(self) -> None:

        for lane in LANE_TABLE:

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                a = copy.deepcopy(rows[0])

                b = copy.deepcopy(rows[0])

                a.update(

                    remote_repo="x",

                    remote_transport=remotes.TRANSPORT_NETWORK,

                    remote_server="oak.space",

                )

                a.pop("netshape_profile", None)

                b.update(

                    remote_repo="y",

                    remote_transport=remotes.TRANSPORT_LOCAL_FILE,

                    remote_server="local",

                )

                b.pop("netshape_profile", None)

                errors = check_transport_purity(rows + [a, b])

                self.assertTrue(any("transport mixing" in e for e in errors))

    def test_same_operation_different_subjects_is_legal(self) -> None:

        # Side-by-side files may hold one operation at different transports

        # for different subjects (publish_gate precedent).

        rows = sample_rows("core")

        a = {**copy.deepcopy(rows[0]), "subject": "git", "remote_repo": "x",

             "remote_transport": remotes.TRANSPORT_LOCAL_FILE, "remote_server": "local"}

        b = {**copy.deepcopy(rows[0]), "subject": "oak", "remote_repo": "y",

             "remote_transport": remotes.TRANSPORT_NETWORK, "remote_server": "oak.space"}

        self.assertEqual(check_transport_purity(rows + [a, b]), [])

    def test_remote_row_without_transport_fails(self) -> None:

        rows = sample_rows("core")

        bad = copy.deepcopy(rows[0])

        bad["remote_repo"] = "oak/bench-tmp"

        bad.pop("remote_transport", None)

        errors = check_transport_purity(rows + [bad])

        self.assertTrue(any("without remote_transport" in e for e in errors))

    def test_unknown_transport_fails(self) -> None:

        rows = sample_rows("core")

        bad = copy.deepcopy(rows[0])

        bad.update(remote_repo="x", remote_transport="carrier_pigeon", remote_server="loft")

        errors = check_transport_purity(rows + [bad])

        self.assertTrue(any("unknown transport" in e for e in errors))

    def test_shaped_row_without_profile_fails(self) -> None:

        rows = sample_rows("netshape")

        bad = copy.deepcopy(rows[0])

        bad["netshape_profile"] = ""

        errors = check_transport_purity([bad])

        self.assertTrue(any("netshape_profile" in e for e in errors))

class MeasurementSourceNegativeTest(unittest.TestCase):

    def test_off_list_source_fails_or_lane_is_exempt(self) -> None:

        for lane, contract in LANE_TABLE.items():

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                rows[0]["measurement_source"] = "made_up_source"

                errors = check_measurement_source(

                    rows, contract.allowed_sources, required=contract.source_required

                )

                if contract.allowed_sources is None:

                    # Encoded reality: agent/contention nest their sources in

                    # metric blocks; mount/netshape carry none. The top-level

                    # check is exempt for them, by the table, not by accident.

                    self.assertEqual(errors, [])

                else:

                    self.assertTrue(any("made_up_source" in e for e in errors))

    def test_required_source_missing_fails(self) -> None:

        contract = LANE_TABLE["external"]

        rows = sample_rows("external")

        del rows[0]["measurement_source"]

        errors = check_measurement_source(rows, contract.allowed_sources, required=True)

        self.assertTrue(any("missing required measurement_source" in e for e in errors))

class CatalogMembershipNegativeTest(unittest.TestCase):

    def test_off_catalog_operation_fails_or_lane_is_exempt(self) -> None:

        for lane, contract in LANE_TABLE.items():

            with self.subTest(lane=lane):

                rows = sample_rows(lane)

                bad = copy.deepcopy(rows[0])

                bad["operation"] = "made.up.operation"

                errors = catalog_errors(contract, rows + [bad])

                if contract.catalog_path is None:

                    # Encoded reality: workflow/contention/agent/task-loop/

                    # netshape/external declare no operation catalog today.

                    self.assertEqual(errors, [])

                else:

                    self.assertTrue(any("made.up.operation" in e for e in errors))

    def test_catalogs_parse_and_are_nonempty(self) -> None:

        for lane, contract in LANE_TABLE.items():

            if contract.catalog_path is None:

                continue

            with self.subTest(lane=lane):

                ops = catalog_operations(contract.catalog_path)

                self.assertTrue(ops, f"{contract.catalog_path} parsed to an empty catalog")

                self.assertTrue(all("." in op or op.isidentifier() for op in ops), ops)

    def test_core_catalog_contains_known_operations(self) -> None:

        ops = catalog_operations(LANE_TABLE["core"].catalog_path)

        for op in ("repo.init", "status.clean", "merge.conflict"):

            self.assertIn(op, ops)

    def test_exemptions_are_documented_reality_not_wildcards(self) -> None:

        # The core exemption families mirror bench.py's own diagnostic-row

        # exclusions (average_by_subject); keep them in sync deliberately.

        core = LANE_TABLE["core"]

        self.assertIn("proc.spawn", core.catalog_exempt_ops)

        self.assertIn("remote.", core.catalog_exempt_prefixes)

        mount = LANE_TABLE["mount"]

        self.assertIn("acquire.", mount.catalog_exempt_prefixes)

class SampleRowsMirrorRealEmittersTest(unittest.TestCase):

    """Spot-check the synthetic rows against the real emitters' fixed points

    so the table cannot drift into testing an imaginary suite."""

    def test_core_sample_passes_rows_py_directly(self) -> None:

        self.assertEqual(oakbench_rows.validate_rows(sample_rows("core"), "core"), [])

    def test_netshape_sample_carries_shaped_pipe_identity(self) -> None:

        for row in sample_rows("netshape"):

            self.assertEqual(row["remote_transport"], remotes.TRANSPORT_NETWORK_SHAPED)

            self.assertTrue(row["netshape_profile"])

            self.assertTrue(row["remote_server"])

    def test_netshape_shaped_rows_compare_legally_same_pipe_only(self) -> None:

        rows = sample_rows("netshape")

        self.assertTrue(remotes.shaped_comparison_legal(rows[0], rows[1]))

        other_pipe = {**rows[1], "netshape_profile": "wan-80ms"}

        self.assertFalse(remotes.shaped_comparison_legal(rows[0], other_pipe))

    def test_external_sample_matches_tperf_runner_source_label(self) -> None:

        import tperf_runner

        self.assertEqual(

            LANE_TABLE["external"].allowed_sources,

            frozenset({tperf_runner.MEASUREMENT_SOURCE}),

        )

        for row in sample_rows("external"):

            self.assertEqual(row["measurement_source"], tperf_runner.MEASUREMENT_SOURCE)

    def test_skip_rows_present_in_lanes_that_emit_them(self) -> None:

        for lane in ("core", "workflow", "mount", "platform", "netshape"):

            with self.subTest(lane=lane):

                skips = [r for r in sample_rows(lane) if r.get("skipped")]

                self.assertTrue(skips, f"{lane} sample must include a skip row")

                for row in skips:

                    self.assertEqual(row["returncode"], oakbench_rows.SKIP_RETURNCODE)

                    self.assertTrue(str(row["skip_reason"]).strip())

if __name__ == "__main__":

    unittest.main()