"""Unit tests for biz_state compare engine."""

from __future__ import annotations

import copy
import unittest

from netx_api.biz_state.compare_engine import compare_rows, mapping_stats, stratify_take
from netx_api.biz_state.compare_service import _kind_allows
from netx_api.biz_state.compare_rules import (
    apply_row_filters,
    arp_dynamic_row_filters,
    effective_compare_fields,
    effective_display_fields,
    eval_leaf_filter,
    normalize_value,
    values_equal,
)


class CompareEngineTests(unittest.TestCase):
    def test_basic_added_removed_changed(self) -> None:
        before = [
            {"local_if": "gei-0/1", "remote_sys": "A", "remote_if": "x1", "remote_ip": "1.1.1.1"},
            {"local_if": "gei-0/2", "remote_sys": "B", "remote_if": "y1", "remote_ip": "2.2.2.2"},
        ]
        after = [
            {"local_if": "gei-0/1", "remote_sys": "A", "remote_if": "x1", "remote_ip": "1.1.1.9"},
            {"local_if": "gei-0/3", "remote_sys": "C", "remote_if": "z1", "remote_ip": ""},
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_sys", "remote_if", "remote_ip"],
            port_map={},
        )
        s = out["summary"]
        self.assertEqual(s["changed"], 1)
        self.assertEqual(s["removed"], 1)
        self.assertEqual(s["added"], 1)
        kinds = {d["kind"] for d in out["diffs"]}
        self.assertEqual(kinds, {"changed", "removed", "added"})

    def test_port_map_rewrites_before_key(self) -> None:
        before = [
            {"local_if": "old-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": ""},
        ]
        after = [
            {"local_if": "new-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": ""},
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_sys", "remote_if", "remote_ip"],
            port_map={"old-1": "new-1"},
        )
        self.assertEqual(out["summary"]["unchanged"], 1)
        self.assertEqual(out["summary"]["added"], 0)
        self.assertEqual(out["summary"]["removed"], 0)

    def test_mapping_stats(self) -> None:
        stats = mapping_stats(
            before_rows=[{"local_if": "a"}, {"local_if": "b"}],
            after_rows=[{"local_if": "x"}, {"local_if": "y"}],
            iface_fields=["local_if"],
            port_map={"a": "x", "missing": "y"},
        )
        self.assertIn("a", stats["hit_before"])
        self.assertIn("missing", stats["miss_before"])
        self.assertFalse(stats["ok"])

    def test_presence_only_empty_compare(self) -> None:
        """Empty compare_fields → only entry set matters; value diffs ignored."""
        before = [
            {"local_if": "gei-0/1", "remote_sys": "A", "remote_if": "x1", "remote_ip": "1.1.1.1"},
            {"local_if": "gei-0/2", "remote_sys": "B", "remote_if": "y1", "remote_ip": "2.2.2.2"},
        ]
        after = [
            {"local_if": "gei-0/1", "remote_sys": "A", "remote_if": "x1", "remote_ip": "9.9.9.9"},
            {"local_if": "gei-0/3", "remote_sys": "C", "remote_if": "z1", "remote_ip": ""},
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=[],
            port_map={},
        )
        s = out["summary"]
        self.assertEqual(s["changed"], 0)
        self.assertEqual(s["unchanged"], 1)
        self.assertEqual(s["removed"], 1)
        self.assertEqual(s["added"], 1)
        kinds = {d["kind"] for d in out["diffs"]}
        # Default: unchanged counted but not listed (million-row safe)
        self.assertNotIn("unchanged", kinds)
        out_full = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=[],
            port_map={},
            include_unchanged=True,
        )
        self.assertIn("unchanged", {d["kind"] for d in out_full["diffs"]})

    def test_empty_port_map_ignores_iface_in_key(self) -> None:
        """No port map → ignore local_if when matching (same neighbor, renamed port)."""
        before = [
            {"local_if": "old-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": "1.1.1.1"},
        ]
        after = [
            {"local_if": "new-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": "1.1.1.1"},
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_ip"],
            port_map={},
        )
        self.assertEqual(out["summary"]["unchanged"], 1)
        self.assertEqual(out["summary"]["added"], 0)
        self.assertEqual(out["summary"]["removed"], 0)
        self.assertTrue(out["mapping_stats"].get("ignore_port_changes"))

    def test_empty_port_map_keeps_iface_when_required_for_uniqueness(self) -> None:
        """OSPF/VRRP-style: same neighbor_id on many interfaces must not collapse."""
        before = [
            {"process_id": "1", "neighbor_id": "2.2.2.2", "interface": "sg1", "address": "10.0.0.1"},
            {"process_id": "1", "neighbor_id": "2.2.2.2", "interface": "sg2", "address": "10.0.0.2"},
        ]
        after = copy.deepcopy(before)
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["process_id", "neighbor_id", "interface"],
            iface_fields=["interface"],
            compare_fields=["address"],
            port_map={},
        )
        self.assertEqual(out["summary"]["unchanged"], 2)
        self.assertEqual(out["summary"]["changed"], 0)
        self.assertIn("interface", out["summary"]["match_key_fields"])
        self.assertFalse(out["mapping_stats"].get("ignore_port_changes"))

    def test_unchanged_rows_are_listed(self) -> None:
        before = [{"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "1"}]
        after = [{"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "1"}]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_ip"],
            port_map={},
            include_unchanged=True,
        )
        self.assertEqual(out["summary"]["unchanged"], 1)
        self.assertEqual(len(out["diffs"]), 1)
        self.assertEqual(out["diffs"][0]["kind"], "unchanged")
        slim = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_ip"],
            port_map={},
        )
        self.assertEqual(slim["summary"]["unchanged"], 1)
        self.assertEqual(slim["diffs"], [])

    def test_unchanged_sample_limit_and_compact(self) -> None:
        before = [
            {"k": str(i), "v": "1", "_netx": {"row_id": f"b{i}"}} for i in range(5)
        ]
        after = [
            {"k": str(i), "v": "1", "_netx": {"row_id": f"a{i}"}} for i in range(5)
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["k"],
            iface_fields=[],
            compare_fields=["v"],
            port_map={},
            include_unchanged=True,
            unchanged_limit=2,
            compact_unchanged=True,
        )
        self.assertEqual(out["summary"]["unchanged"], 5)
        self.assertEqual(out["summary"]["unchanged_listed"], 2)
        self.assertTrue(out["summary"]["unchanged_truncated"])
        self.assertTrue(out["summary"]["unchanged_compact"])
        self.assertEqual(out["summary"]["unchanged_sample_mode"], "stratified")
        self.assertEqual(len(out["diffs"]), 2)
        self.assertEqual(out["diffs"][0]["before"], {})
        self.assertEqual(out["diffs"][0]["after"], {})
        self.assertEqual(out["diffs"][0]["before_row_id"], "b0")
        self.assertEqual(out["diffs"][0]["after_row_id"], "a0")

    def test_stratify_take_round_robin(self) -> None:
        items = [("a", i) for i in range(5)] + [("b", i) for i in range(5)]
        got = stratify_take(items, 4, key_fn=lambda x: x[0])
        self.assertEqual([x[0] for x in got], ["a", "b", "a", "b"])

    def test_kind_allows_tabs(self) -> None:
        self.assertTrue(_kind_allows("all", "unchanged"))
        self.assertTrue(_kind_allows("diff", "removed"))
        self.assertTrue(_kind_allows("diff", "changed"))
        self.assertFalse(_kind_allows("diff", "added"))
        self.assertFalse(_kind_allows("diff", "unchanged"))
        self.assertTrue(_kind_allows("unchanged", "unchanged"))
        self.assertFalse(_kind_allows("unchanged", "removed"))
        self.assertTrue(_kind_allows("removed", "removed"))
        self.assertTrue(_kind_allows("changed", "changed"))

    def test_unchanged_sample_stratified_across_neighbors(self) -> None:
        """Sample must cover multiple neighbors, not only the first command's rows."""
        before = []
        after = []
        for n_i, neigh in enumerate(("1.1.1.1", "2.2.2.2", "3.3.3.3")):
            for j in range(10):
                net = f"10.{n_i}.{j}.0/24"
                before.append(
                    {
                        "neighbor": neigh,
                        "direction": "in",
                        "network": net,
                        "v": "1",
                        "_netx": {"row_id": f"b-{neigh}-{j}"},
                    }
                )
                after.append(
                    {
                        "neighbor": neigh,
                        "direction": "in",
                        "network": net,
                        "v": "1",
                        "_netx": {"row_id": f"a-{neigh}-{j}"},
                    }
                )
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["neighbor", "direction", "network"],
            iface_fields=[],
            compare_fields=["v"],
            port_map={},
            include_unchanged=True,
            unchanged_limit=6,
            compact_unchanged=True,
        )
        self.assertEqual(out["summary"]["unchanged"], 30)
        self.assertEqual(out["summary"]["unchanged_listed"], 6)
        keys = [d["key"]["neighbor"] for d in out["diffs"]]
        self.assertEqual(set(keys), {"1.1.1.1", "2.2.2.2", "3.3.3.3"})

    def test_mac_normalize_via_field_rules(self) -> None:
        before = [{"ip": "1.1.1.1", "mac": "00:11:22:33:44:55", "iface": "gei-0/1"}]
        after = [{"ip": "1.1.1.1", "mac": "0011.2233.4455", "iface": "gei-0/1"}]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["ip", "iface"],
            iface_fields=["iface"],
            compare_fields=["mac"],
            port_map={},
            field_rules=[{"field": "mac", "normalize": "mac"}],
        )
        self.assertEqual(out["summary"]["unchanged"], 1)
        self.assertEqual(out["summary"]["changed"], 0)

    def test_numeric_tolerance(self) -> None:
        before = [{"id": "1", "cnt": "100"}]
        after = [{"id": "1", "cnt": "102"}]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["id"],
            iface_fields=[],
            compare_fields=["cnt"],
            port_map={},
            field_rules=[{"field": "cnt", "compare": "numeric", "tolerance": 5}],
        )
        self.assertEqual(out["summary"]["unchanged"], 1)
        out2 = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["id"],
            iface_fields=[],
            compare_fields=["cnt"],
            port_map={},
            field_rules=[{"field": "cnt", "compare": "numeric", "tolerance": 0}],
        )
        self.assertEqual(out2["summary"]["changed"], 1)

    def test_percent_tolerance(self) -> None:
        before = [{"id": "1", "networks": "100"}]
        after_ok = [{"id": "1", "networks": "104"}]  # 4% < 5%
        after_bad = [{"id": "1", "networks": "106"}]  # 6% > 5%
        rule = [{"field": "networks", "compare": "percent", "tolerance": 5}]
        ok = compare_rows(
            before_rows=before,
            after_rows=after_ok,
            key_fields=["id"],
            iface_fields=[],
            compare_fields=["networks"],
            port_map={},
            field_rules=rule,
        )
        self.assertEqual(ok["summary"]["unchanged"], 1)
        bad = compare_rows(
            before_rows=before,
            after_rows=after_bad,
            key_fields=["id"],
            iface_fields=[],
            compare_fields=["networks"],
            port_map={},
            field_rules=rule,
        )
        self.assertEqual(bad["summary"]["changed"], 1)

    def test_percent_zero_baseline(self) -> None:
        self.assertTrue(
            values_equal("0", "0", rule={"compare": "percent", "tolerance": 5})
        )
        self.assertFalse(
            values_equal("0", "1", rule={"compare": "percent", "tolerance": 5})
        )


class CompareRulesTests(unittest.TestCase):
    def test_arp_dynamic_row_filters(self) -> None:
        rows = [
            {"ip": "1.1.1.1", "entry_type": "dynamic", "age": "00:01:02"},
            {"ip": "1.1.1.2", "entry_type": "static", "age": "H"},
            {"ip": "1.1.1.3", "entry_type": "", "age": "01:02:03"},
            {"ip": "1.1.1.4", "entry_type": "", "age": "I"},
        ]
        kept = apply_row_filters(rows, arp_dynamic_row_filters())
        ips = {r["ip"] for r in kept}
        self.assertEqual(ips, {"1.1.1.1", "1.1.1.3"})

    def test_effective_compare_ignore(self) -> None:
        fields = effective_compare_fields(
            ["mac", "vlan", "age"],
            [{"field": "age", "compare": "ignore"}, {"field": "vlan", "ignore": True}],
        )
        self.assertEqual(fields, ["mac"])

    def test_normalize_mac(self) -> None:
        self.assertEqual(normalize_value("00-11-22-33-44-55", "mac"), "001122334455")
        self.assertTrue(values_equal("Aa", "aa", rule={"normalize": "lower"}))

    def test_explain_diff_percent(self) -> None:
        from netx_api.biz_state.compare_rules import explain_diff

        reason = explain_diff(
            "100", "108", rule={"compare": "percent", "tolerance": 5}
        )
        self.assertIn("pct", reason)
        self.assertIn("8", reason)

    def test_changed_includes_reason(self) -> None:
        before = [{"id": "1", "networks": "100"}]
        after = [{"id": "1", "networks": "110"}]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["id"],
            iface_fields=[],
            compare_fields=["networks"],
            port_map={},
            field_rules=[{"field": "networks", "compare": "percent", "tolerance": 5}],
        )
        self.assertEqual(out["summary"]["changed"], 1)
        ch = out["diffs"][0]["changes"]["networks"]
        self.assertIn("reason", ch)
        self.assertIn("pct", ch["reason"])

    def test_effective_display_legacy_and_explicit(self) -> None:
        legacy = effective_display_fields(
            key_fields=["ip", "iface"],
            compare_fields=["mac"],
            display_fields=None,
        )
        self.assertEqual(legacy, ["ip", "iface", "mac"])
        explicit = effective_display_fields(
            key_fields=["ip", "iface"],
            compare_fields=["mac"],
            display_fields=["vrf", "age", "ip"],
        )
        # Key → Compare → Display-only (compare forced even if not ticked)
        self.assertEqual(explicit, ["ip", "iface", "mac", "vrf", "age"])

    def test_display_only_field_does_not_change(self) -> None:
        """Context column in display but not compare → value drift ignored."""
        before = [{"ip": "1.1.1.1", "mac": "aabb", "vrf": "A"}]
        after = [{"ip": "1.1.1.1", "mac": "aabb", "vrf": "B"}]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["ip"],
            iface_fields=[],
            compare_fields=["mac"],
            port_map={},
        )
        self.assertEqual(out["summary"]["unchanged"], 1)
        self.assertEqual(out["summary"]["changed"], 0)


class CompareDiffPagingTests(unittest.TestCase):
    def test_resolve_unchanged_policy(self) -> None:
        from netx_api.biz_state.compare_service import resolve_unchanged_policy

        small = resolve_unchanged_policy("auto", before_n=100, after_n=100)
        self.assertTrue(small["include"])
        self.assertIsNone(small["limit"])
        self.assertFalse(small["compact"])
        large = resolve_unchanged_policy("auto", before_n=1_500_000, after_n=1_500_000)
        self.assertTrue(large["include"])
        self.assertEqual(large["limit"], 5000)
        self.assertTrue(large["compact"])
        self.assertFalse(resolve_unchanged_policy("never", before_n=10, after_n=10)["include"])
        keys = resolve_unchanged_policy("keys", before_n=1_500_000, after_n=1_500_000)
        self.assertTrue(keys["include"])
        self.assertIsNone(keys["limit"])
        self.assertTrue(keys["compact"])

    def test_hydrate_diff_rows_fills_sides(self) -> None:
        from netx_api.biz_state.compare_service import _hydrate_diff_rows

        class _FakeDb:
            pass

        items = [
            {
                "kind": "unchanged",
                "key": {"k": "1"},
                "before": {},
                "after": {},
                "mapped_before": {},
                "changes": {},
                "before_row_id": "b1",
                "after_row_id": "a1",
            }
        ]
        # Patch loader
        import netx_api.biz_state.compare_service as cs

        orig = cs._metric_rows_by_ids
        try:
            cs._metric_rows_by_ids = lambda _db, ids: {  # type: ignore[assignment]
                "b1": {"k": "1", "v": "pre"},
                "a1": {"k": "1", "v": "post"},
            }
            out = _hydrate_diff_rows(_FakeDb(), items)
        finally:
            cs._metric_rows_by_ids = orig
        self.assertEqual(out[0]["before"]["v"], "pre")
        self.assertEqual(out[0]["after"]["v"], "post")
        self.assertEqual(out[0]["mapped_before"]["v"], "pre")

    def test_filter_inline_diffs_kind_and_kw(self) -> None:
        from netx_api.biz_state.compare_service import _filter_inline_diffs

        diffs = [
            {"kind": "added", "key": {"p": "1"}, "before": {}, "after": {"p": "1"}, "changes": {}},
            {"kind": "removed", "key": {"p": "2"}, "before": {"p": "2"}, "after": {}, "changes": {}},
            {"kind": "unchanged", "key": {"p": "3"}, "before": {"p": "3"}, "after": {"p": "3"}, "changes": {}},
            {
                "kind": "changed",
                "key": {"p": "4"},
                "before": {"x": "a"},
                "after": {"x": "b"},
                "changes": {"x": {"before": "a", "after": "b"}},
            },
        ]
        only_diff = _filter_inline_diffs(diffs, kind="diff", kw="")
        self.assertEqual({d["kind"] for d in only_diff}, {"removed", "changed"})
        only_added = _filter_inline_diffs(diffs, kind="added", kw="")
        self.assertEqual(len(only_added), 1)
        hit = _filter_inline_diffs(diffs, kind="all", kw='"p":"4"')
        self.assertEqual(len(hit), 1)
        self.assertEqual(hit[0]["kind"], "changed")
        hit2 = _filter_inline_diffs(diffs, kind="all", kw="changed")
        self.assertTrue(any(d["kind"] == "changed" for d in hit2))


class CompareSheetDefaultsTests(unittest.TestCase):
    def test_contains_filter_op(self) -> None:
        row = {"af": "IPv4,IPv6"}
        self.assertTrue(eval_leaf_filter(row, {"field": "af", "op": "contains", "value": "IPv4"}))
        self.assertTrue(eval_leaf_filter(row, {"field": "af", "op": "contains", "value": "ipv6"}))
        self.assertFalse(eval_leaf_filter(row, {"field": "af", "op": "contains", "value": "vpn"}))

    def test_zte_default_splits_bgp_and_isis(self) -> None:
        from netx_api.biz_state.compare_service import _default_zte_status_sheets, sheet_key

        sheets = _default_zte_status_sheets()
        ids = [sheet_key(s) for s in sheets]
        self.assertIn("bgp_peer.ipv4", ids)
        self.assertIn("bgp_peer.vpnv4", ids)
        self.assertIn("bgp_peer.vpnv6", ids)
        self.assertIn("isis_adjacency.ipv4", ids)
        self.assertIn("isis_adjacency.ipv6", ids)
        # Same source metric may appear multiple times (afi splits)
        self.assertEqual(sum(1 for s in sheets if s["metric_id"] == "bgp_peer"), 6)
        self.assertGreaterEqual(sum(1 for s in sheets if s["metric_id"] == "bgp_route"), 4)
        self.assertIn("bgp_peer.evpn", ids)
        self.assertIn("bgp_peer.vpls", ids)
        self.assertIn("vrrp.ipv4", ids)
        self.assertIn("lldp_neighbor", ids)
        # Packaged IOH default: filtered bgp_route ipv4 + percent pfx_rcd
        route4 = next(s for s in sheets if sheet_key(s) == "bgp_route")
        self.assertEqual(route4["metric_id"], "bgp_route")
        self.assertTrue(
            any(
                f.get("field") == "afi" and f.get("value") == "ipv4"
                for f in (route4.get("row_filters") or [])
            )
        )
        self.assertTrue(
            any(
                r.get("field") == "pfx_rcd" and r.get("compare") == "percent"
                for r in (route4.get("field_rules") or [])
            )
        )
        route_vpn = next(s for s in sheets if sheet_key(s) == "bgp_route.vpnv4")
        self.assertTrue(
            any(
                f.get("field") == "afi" and f.get("value") == "vpnv4"
                for f in (route_vpn.get("row_filters") or [])
            )
        )
        vpnv4 = next(s for s in sheets if sheet_key(s) == "bgp_peer.vpnv4")
        self.assertEqual(
            vpnv4["row_filters"],
            [{"field": "afi", "op": "eq", "value": "vpnv4"}],
        )
        isis4 = next(s for s in sheets if sheet_key(s) == "isis_adjacency.ipv4")
        self.assertEqual(isis4["row_filters"][0]["op"], "contains")

    def test_ordered_same_key_pairing_2v2(self) -> None:
        """Same key, two rows each: zip by order (not first-wins + duplicate)."""
        before = [
            {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "1"},
            {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "9"},
        ]
        after = [
            {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "1"},
            {"local_if": "a", "remote_sys": "X", "remote_if": "1", "remote_ip": "2"},
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_ip"],
            port_map={"a": "a"},
        )
        self.assertEqual(out["summary"]["before_count"], 2)
        self.assertEqual(out["summary"]["after_count"], 2)
        self.assertEqual(out["summary"]["duplicate"], 0)
        self.assertEqual(out["summary"]["duplicate_keys_before"], 1)
        self.assertEqual(out["summary"]["duplicate_keys_after"], 1)
        self.assertIn("a|X|1", out["summary"]["duplicate_key_list"])
        kinds = [d["kind"] for d in out["diffs"]]
        self.assertNotIn("duplicate", kinds)
        # Pair 0: 1==1 unchanged; pair 1: 9!=2 changed
        self.assertEqual(out["summary"]["unchanged"], 1)
        self.assertEqual(out["summary"]["changed"], 1)
        self.assertEqual(out["summary"]["added"], 0)
        self.assertEqual(out["summary"]["removed"], 0)

    def test_ordered_multipath_5_vs_2(self) -> None:
        """5 before / 2 after same key → 2 compared + 3 removed failures."""
        key = {"local_if": "a", "remote_sys": "X", "remote_if": "1"}
        before = [{**key, "remote_ip": str(i)} for i in range(5)]
        after = [{**key, "remote_ip": "0"}, {**key, "remote_ip": "1"}]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_ip"],
            port_map={"a": "a"},
        )
        self.assertEqual(out["summary"]["before_count"], 5)
        self.assertEqual(out["summary"]["after_count"], 2)
        self.assertEqual(out["summary"]["removed"], 3)
        self.assertEqual(out["summary"]["added"], 0)
        self.assertEqual(
            out["summary"]["unchanged"] + out["summary"]["changed"],
            2,
        )
        self.assertEqual(out["summary"]["unchanged"], 2)
        self.assertEqual(out["summary"]["changed"], 0)
        self.assertEqual(out["summary"]["duplicate"], 0)
        kinds = [d["kind"] for d in out["diffs"]]
        self.assertEqual(kinds.count("removed"), 3)
        self.assertNotIn("duplicate", kinds)

    def test_ignore_port_changes_false_keeps_iface(self) -> None:
        before = [
            {"local_if": "old-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": "1.1.1.1"},
        ]
        after = [
            {"local_if": "new-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": "1.1.1.1"},
        ]
        out = compare_rows(
            before_rows=before,
            after_rows=after,
            key_fields=["local_if", "remote_sys", "remote_if"],
            iface_fields=["local_if"],
            compare_fields=["remote_ip"],
            port_map={},
            ignore_port_changes=False,
        )
        self.assertEqual(out["summary"]["removed"], 1)
        self.assertEqual(out["summary"]["added"], 1)
        self.assertFalse(out["mapping_stats"].get("ignore_port_changes"))

    def test_normalize_allows_duplicate_metric_with_distinct_sheet_id(self) -> None:
        from netx_api.biz_state.compare_service import _normalize_sheet, sheet_key

        a = _normalize_sheet(
            {
                "sheet_id": "bgp_peer.vpnv4",
                "title": "BGP VPNv4",
                "metric_id": "bgp_peer",
                "key_fields": ["afi", "neighbor"],
                "compare_fields": ["state"],
                "row_filters": [{"field": "afi", "op": "eq", "value": "vpnv4"}],
            }
        )
        b = _normalize_sheet(
            {
                "sheet_id": "bgp_peer.ipv4",
                "title": "BGP IPv4",
                "metric_id": "bgp_peer",
                "key_fields": ["afi", "neighbor"],
                "compare_fields": ["state"],
                "row_filters": [{"field": "afi", "op": "eq", "value": "ipv4"}],
            }
        )
        self.assertIsNotNone(a)
        self.assertIsNotNone(b)
        self.assertEqual(a["metric_id"], b["metric_id"])
        self.assertNotEqual(sheet_key(a), sheet_key(b))

    def test_template_in_roundtrip_keeps_split_sheet_ids(self) -> None:
        """Export → TemplateIn → create must preserve 拆表 sheet_id (ISIS/BGP/VRRP)."""
        from netx_api.biz_state.compare_service import (
            _default_zte_status_sheets,
            _parse_metrics_body,
            sheet_key,
        )
        from netx_api.biz_state_router import TemplateIn

        sheets = _default_zte_status_sheets()
        body = TemplateIn(
            name="ZTE status roundtrip",
            metrics=sheets,
        ).model_dump()
        parsed = _parse_metrics_body(body)
        self.assertEqual(len(parsed), len(sheets))
        ids = [sheet_key(s) for s in parsed]
        self.assertIn("isis_adjacency.ipv4", ids)
        self.assertIn("isis_adjacency.ipv6", ids)
        self.assertEqual(sum(1 for s in parsed if s["metric_id"] == "isis_adjacency"), 2)
        self.assertEqual(sum(1 for s in parsed if s["metric_id"] == "bgp_peer"), 6)
        isis4 = next(s for s in parsed if sheet_key(s) == "isis_adjacency.ipv4")
        self.assertTrue(str(isis4.get("title") or "").strip())

    def test_filter_enabled_sheets_empty_means_all(self) -> None:
        from netx_api.biz_state.compare_service import _filter_enabled_sheets, sheet_key

        sheets = [
            {"sheet_id": "isis_adjacency.ipv4", "metric_id": "isis_adjacency"},
            {"sheet_id": "isis_adjacency.ipv6", "metric_id": "isis_adjacency"},
            {"sheet_id": "arp", "metric_id": "arp"},
        ]
        self.assertEqual(len(_filter_enabled_sheets(sheets, None)), 3)
        self.assertEqual(len(_filter_enabled_sheets(sheets, [])), 3)
        only = _filter_enabled_sheets(sheets, ["isis_adjacency.ipv4", "arp"])
        self.assertEqual([sheet_key(s) for s in only], ["isis_adjacency.ipv4", "arp"])
        self.assertEqual(_filter_enabled_sheets(sheets, ["missing"]), [])

    def test_arp_default_sheet_has_row_filters(self) -> None:
        from netx_api.biz_state.compare_service import _default_sheet_for_metric

        sheet = _default_sheet_for_metric("arp")
        self.assertTrue(sheet.get("row_filters"))
        self.assertEqual(sheet["metric_id"], "arp")
        self.assertIn("display_fields", sheet)
        # Context columns present without being compare-only
        for ctx in ("vrf", "entry_type", "age"):
            if ctx in sheet["display_fields"]:
                self.assertTrue(
                    ctx in sheet["display_fields"]
                    and (ctx in sheet["key_fields"] or ctx not in sheet["compare_fields"] or True)
                )

    def test_load_metric_rows_has_no_arp_hardcode(self) -> None:
        import inspect

        from netx_api.biz_state import compare_service as mod

        src = inspect.getsource(mod._load_metric_rows)
        self.assertNotIn("is_valid_arp_age", src)
        self.assertNotIn('metric_id == "arp"', src)

    def test_sheet_csv_uses_display_fields(self) -> None:
        from netx_api.biz_state.compare_service import _sheet_csv

        sheet = {
            "key_fields": ["ip"],
            "compare_fields": ["mac"],
            "display_fields": ["ip", "vrf", "mac"],
            "diffs": [
                {
                    "kind": "unchanged",
                    "key": {"ip": "1.1.1.1"},
                    "before": {"ip": "1.1.1.1", "mac": "aa", "vrf": "v1"},
                    "after": {"ip": "1.1.1.1", "mac": "aa", "vrf": "v1"},
                    "mapped_before": {"ip": "1.1.1.1", "mac": "aa", "vrf": "v1"},
                    "changes": {},
                }
            ],
        }
        csv = _sheet_csv(sheet)
        header = csv.splitlines()[0]
        self.assertIn("kind", header)
        self.assertIn("ip", header)
        self.assertIn("vrf", header)
        self.assertIn("mac__pre", header)
        self.assertIn("mac__post", header)
        # display-only vrf is single column, not pre/post
        self.assertNotIn("vrf__pre", header)


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