mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 00:50:46 +08:00
Make biz compare template-driven with field rules, board UX, and import/export.
Operators pick metrics then Key/Compare/Display rules; seed built-ins only once; fail-first verdict board for cutover review. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
21b254120e
commit
d232044750
11 changed files with 1978 additions and 345 deletions
|
|
@ -5,6 +5,14 @@ from __future__ import annotations
|
|||
import unittest
|
||||
|
||||
from netx_api.biz_state.compare_engine import compare_rows, mapping_stats
|
||||
from netx_api.biz_state.compare_rules import (
|
||||
apply_row_filters,
|
||||
arp_dynamic_row_filters,
|
||||
effective_compare_fields,
|
||||
effective_display_fields,
|
||||
normalize_value,
|
||||
values_equal,
|
||||
)
|
||||
|
||||
|
||||
class CompareEngineTests(unittest.TestCase):
|
||||
|
|
@ -124,6 +132,159 @@ class CompareEngineTests(unittest.TestCase):
|
|||
self.assertEqual(len(out["diffs"]), 1)
|
||||
self.assertEqual(out["diffs"][0]["kind"], "unchanged")
|
||||
|
||||
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"],
|
||||
)
|
||||
# keys first, then display order extras, compare forced if missing
|
||||
self.assertEqual(explicit, ["ip", "iface", "vrf", "age", "mac"])
|
||||
|
||||
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_filter_inline_diffs_kind_and_kw(self) -> None:
|
||||
|
|
@ -152,5 +313,59 @@ class CompareDiffPagingTests(unittest.TestCase):
|
|||
self.assertTrue(any(d["kind"] == "changed" for d in hit2))
|
||||
|
||||
|
||||
class CompareSheetDefaultsTests(unittest.TestCase):
|
||||
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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue