hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 00:43:17 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2
netx/tests/test_biz_migration_evaluate.py
oliver 88b7558a48 Tighten cutover verdicts and scope iface metrics by port mapping. 
Mapped interfaces are the expect set and must appear after cutover; unmapped same-name down/up is anomaly. Success presets no longer treat bare unchanged as green.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-19 01:10:13 +08:00

1115 lines
40 KiB
Python
Raw Blame History

"""Unit tests for cutover migration evaluate helpers (port status focus)."""
from __future__ import annotations
import unittest
from netx_api.biz_migration.evaluate import (
classify_status,
dual_verdict,
evaluate_metric_dual,
expect_keys_for_metric,
parse_expect_set,
port_sheet_def,
port_status_label,
side_verdict,
_remap_key_str,
)
PORT_OVERRIDE = {
"metric_id": "interface_brief",
"status_fields": ["admin", "phy", "prot"],
"down_values": ["down"],
"up_values": ["up"],
"success": [{"old": ["removed", "down"], "new": ["added", "up", "unchanged"]}],
}
class ParseExpectSetTests(unittest.TestCase):
def test_ports_go_to_interface_brief(self):
got = parse_expect_set({"ports": ["gei-1", "gei-2", ""]})
self.assertEqual(got["interface_brief"], {"gei-1", "gei-2"})
self.assertEqual(got["_ports"], {"gei-1", "gei-2"})
def test_items_multi_metric(self):
got = parse_expect_set(
{
"items": [
{"metric_id": "bgp_peer", "keys": ["AS1", "1.1.1.1"]},
{"metric_id": "arp", "key": "10.0.0.1"},
]
}
)
self.assertEqual(got["bgp_peer"], {"AS1|1.1.1.1"})
self.assertEqual(got["arp"], {"10.0.0.1"})
def test_multi_select_each_key_is_separate(self):
"""UI must send one item per selected row, not a flat keys list."""
got = parse_expect_set(
{
"items": [
{"metric_id": "bgp_peer", "key": "1.1.1.1"},
{"metric_id": "bgp_peer", "key": "2.2.2.2"},
{"metric_id": "arp", "key": "10.0.0.1"},
]
}
)
self.assertEqual(got["bgp_peer"], {"1.1.1.1", "2.2.2.2"})
self.assertEqual(got["arp"], {"10.0.0.1"})
def test_flat_keys_list_is_one_composite(self):
got = parse_expect_set(
{"items": [{"metric_id": "bgp_peer", "keys": ["1.1.1.1", "2.2.2.2"]}]}
)
self.assertEqual(got["bgp_peer"], {"1.1.1.1|2.2.2.2"})
def test_nested_keys_list(self):
got = parse_expect_set(
{
"items": [
{"metric_id": "isis_adjacency", "keys": [["p1", "gei-1", "sys1"], ["p1", "gei-2", "sys2"]]},
{"metric_id": "arp", "key": ["10.0.0.1", "vrf1"]},
]
}
)
self.assertEqual(got["isis_adjacency"], {"p1|gei-1|sys1", "p1|gei-2|sys2"})
self.assertEqual(got["arp"], {"10.0.0.1|vrf1"})
def test_ports_only_apply_to_interface_brief(self):
expect = parse_expect_set({"ports": ["gei-1"]})
self.assertEqual(
expect_keys_for_metric(expect, metric_id="interface_brief", iface_fields=["interface"]),
{"gei-1"},
)
self.assertEqual(
expect_keys_for_metric(expect, metric_id="arp", iface_fields=["interface"]),
set(),
)
class PortSheetTests(unittest.TestCase):
def test_port_sheet_fields(self):
s = port_sheet_def()
self.assertEqual(s["metric_id"], "interface_brief")
self.assertEqual(s["key_fields"], ["interface"])
self.assertEqual(s["compare_fields"], ["admin", "phy", "prot"])
def test_status_label(self):
self.assertEqual(
port_status_label({"admin": "up", "phy": "up", "prot": "up"}),
"up/up/up",
)
self.assertEqual(port_status_label({}), "—")
class ClassifyStatusTests(unittest.TestCase):
def test_up_down(self):
self.assertEqual(
classify_status({"admin": "up", "phy": "up", "prot": "up"}, PORT_OVERRIDE),
"up",
)
self.assertEqual(
classify_status({"admin": "up", "phy": "down", "prot": "up"}, PORT_OVERRIDE),
"down",
)
self.assertEqual(classify_status({}, PORT_OVERRIDE), "none")
self.assertEqual(classify_status({"admin": "up"}, None), "none")
class VerdictTests(unittest.TestCase):
def test_side_anomaly_gone(self):
self.assertEqual(
side_verdict(kind="removed", in_expect=False, window_active=True),
("anomaly_gone", "red"),
)
self.assertEqual(
side_verdict(kind="removed", in_expect=True, window_active=True),
("expected_gone", "yellow"),
)
def test_dual_migrated(self):
self.assertEqual(
dual_verdict(old_kind="removed", new_kind="added", in_expect=True, window_active=True),
("migrated", "green"),
)
def test_dual_lost(self):
self.assertEqual(
dual_verdict(old_kind="removed", new_kind="", in_expect=True, window_active=False),
("lost", "red"),
)
def test_acceptance_unfinished_is_red(self):
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="",
in_expect=True,
window_active=False,
acceptance=True,
),
("unfinished", "red"),
)
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="",
in_expect=True,
window_active=True,
acceptance=False,
),
("migrating", "yellow"),
)
def test_dual_down_up_via_success_patterns(self):
self.assertEqual(
dual_verdict(
old_kind="changed",
new_kind="changed",
in_expect=True,
window_active=True,
old_status="down",
new_status="up",
success_patterns=PORT_OVERRIDE["success"],
),
("migrated", "green"),
)
def test_out_of_expect_ignore(self):
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="",
in_expect=False,
window_active=False,
out_of_expect="ignore",
),
("not_involved", "gray"),
)
def test_out_of_expect_warn_is_yellow(self):
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="unchanged",
in_expect=False,
window_active=False,
out_of_expect="warn",
),
("anomaly", "yellow"),
)
def test_out_of_expect_new_side_only_removed(self):
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="removed",
in_expect=False,
window_active=False,
out_of_expect="strict",
),
("anomaly", "red"),
)
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="removed",
in_expect=False,
window_active=False,
out_of_expect="warn",
),
("anomaly", "yellow"),
)
def test_rules_mode_skips_auto_migrate(self):
"""With success patterns configured, bare kind migrate is not invented."""
verdict, color = dual_verdict(
old_kind="removed",
new_kind="added",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [
[
{"type": "presence", "value": "added"},
{"type": "value", "field": "state", "op": "eq", "value": "up"},
]
],
}
],
new_row={"state": "down"},
)
self.assertNotEqual(verdict, "migrated")
self.assertEqual(color, "yellow")
def test_rules_mode_skips_catch_all_anomaly(self):
"""With anomaly list present, mid-cutover stays migrating not invented anomaly."""
self.assertEqual(
dual_verdict(
old_kind="changed",
new_kind="unchanged",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
anomaly_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "removed"}]],
}
],
),
("migrating", "yellow"),
)
def test_field_token_success(self):
self.assertEqual(
dual_verdict(
old_kind="changed",
new_kind="changed",
in_expect=True,
window_active=True,
old_status="other",
new_status="other",
success_patterns=[
{"old": ["field:state:idle"], "new": ["field:state:established"]}
],
old_row={"state": "Idle"},
new_row={"state": "Established"},
sheet_override={"status_fields": ["state"]},
),
("migrated", "green"),
)
def test_structured_field_conds_all_mode(self):
self.assertEqual(
dual_verdict(
old_kind="changed",
new_kind="changed",
in_expect=True,
window_active=True,
old_status="other",
new_status="other",
success_patterns=[
{
"old_mode": "all",
"new_mode": "all",
"old_conds": [
{"type": "kind", "value": "changed"},
{"type": "field", "field": "state", "op": "in", "value": ["idle", "active"]},
],
"new_conds": [
{"type": "kind", "value": "changed"},
{"type": "field", "field": "state", "op": "eq", "value": "established"},
],
}
],
old_row={"state": "Idle"},
new_row={"state": "Established"},
),
("migrated", "green"),
)
def test_structured_conds_fail_when_all_incomplete(self):
verdict, _color = dual_verdict(
old_kind="unchanged",
new_kind="changed",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_mode": "all",
"new_mode": "any",
"old_conds": [
{"type": "kind", "value": "unchanged"},
{"type": "field", "field": "state", "op": "eq", "value": "idle"},
],
"new_conds": [{"type": "kind", "value": "changed"}],
}
],
old_row={"state": "Active"},
new_row={"state": "Established"},
)
self.assertNotEqual(verdict, "migrated")
def test_groups_or_of_and(self):
"""Within a group = AND; between groups = OR."""
patterns = [
{
"old_groups": [
[{"type": "presence", "value": "removed"}],
[
{"type": "presence", "value": "changed"},
{"type": "value", "field": "state", "op": "in", "value": ["idle", "active"]},
],
],
"new_groups": [
[{"type": "value", "field": "state", "op": "eq", "value": "established"}],
],
}
]
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="changed",
in_expect=True,
window_active=True,
success_patterns=patterns,
old_row={},
new_row={"state": "Established"},
),
("migrated", "green"),
)
self.assertEqual(
dual_verdict(
old_kind="changed",
new_kind="changed",
in_expect=True,
window_active=True,
success_patterns=patterns,
old_row={"state": "Idle"},
new_row={"state": "Established"},
),
("migrated", "green"),
)
verdict, _color = dual_verdict(
old_kind="changed",
new_kind="changed",
in_expect=True,
window_active=True,
success_patterns=patterns,
old_row={"state": "Established"},
new_row={"state": "Established"},
)
self.assertNotEqual(verdict, "migrated")
def test_value_changed_vs_baseline(self):
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="changed",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "unchanged"}]],
"new_groups": [
[{"type": "value", "field": "state", "op": "changed"}]
],
}
],
old_row={"state": "Idle"},
new_row={"state": "Established"},
old_base={"state": "Idle"},
new_base={"state": "Idle"},
),
("migrated", "green"),
)
def test_screenshot_style_success(self):
"""Old removed OR state!=up; new added AND state=up."""
patterns = [
{
"old_groups": [
[{"type": "presence", "value": "removed"}],
[{"type": "value", "field": "state", "op": "ne", "value": "up"}],
],
"new_groups": [
[
{"type": "presence", "value": "added"},
{"type": "value", "field": "state", "op": "eq", "value": "up"},
]
],
}
]
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="added",
in_expect=True,
window_active=True,
success_patterns=patterns,
old_row={},
new_row={"state": "up"},
),
("migrated", "green"),
)
def test_anomaly_both_gone(self):
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="removed",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
anomaly_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "removed"}]],
}
],
),
("anomaly", "red"),
)
def test_anomaly_new_side_dont_care_old(self):
"""Side-only anomaly is deferred while window is active (avoid false red)."""
anomaly = [
{
"old_groups": [],
"new_groups": [
[{"type": "presence", "value": "removed"}],
[
{
"type": "value",
"field": "state",
"op": "ne",
"value": "up",
}
],
],
}
]
success = [
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
]
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="removed",
in_expect=True,
window_active=True,
acceptance=False,
success_patterns=success,
anomaly_patterns=anomaly,
new_row={},
),
("migrating", "yellow"),
)
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="removed",
in_expect=True,
window_active=False,
acceptance=True,
success_patterns=success,
anomaly_patterns=anomaly,
new_row={},
),
("anomaly", "red"),
)
def test_success_beats_anomaly(self):
success = [
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
]
anomaly = [
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
]
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="added",
in_expect=True,
window_active=True,
success_patterns=success,
anomaly_patterns=anomaly,
),
("migrated", "green"),
)
class EvaluateMetricDualTests(unittest.TestCase):
def test_port_migration_happy_path(self):
old_base = [{"interface": "gei-old", "admin": "up", "phy": "up", "prot": "up"}]
old_cur: list[dict] = []
new_cur = [{"interface": "gei-new", "admin": "up", "phy": "up", "prot": "up"}]
expect = parse_expect_set({"ports": ["gei-old"]})
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={"gei-old": "gei-new"},
expect=expect,
window_active=True,
sheet_override=PORT_OVERRIDE,
)
self.assertEqual(out["progress_total"], 1)
migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
self.assertTrue(migrated, out["rows"])
self.assertEqual(migrated[0]["color"], "green")
self.assertEqual(migrated[0]["old_status"], "gone")
self.assertEqual(migrated[0]["new_status"], "up/up/up")
self.assertEqual(out["progress_ok"], 1)
def test_port_down_to_up_is_migrated(self):
old_base = [{"interface": "gei-old", "admin": "up", "phy": "up", "prot": "up"}]
old_cur = [{"interface": "gei-old", "admin": "down", "phy": "down", "prot": "down"}]
new_cur = [{"interface": "gei-new", "admin": "up", "phy": "up", "prot": "up"}]
expect = parse_expect_set({"ports": ["gei-old"]})
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={"gei-old": "gei-new"},
expect=expect,
window_active=True,
sheet_override=PORT_OVERRIDE,
)
migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
self.assertTrue(migrated, out["rows"])
self.assertEqual(out["progress_ok"], 1)
def test_bgp_presence_only_migrated(self):
"""Metrics without status_fields use kind-only dual (presence)."""
old_base = [{"peer": "1.1.1.1", "state": "Established"}]
old_cur: list[dict] = []
new_cur = [{"peer": "1.1.1.1", "state": "Established"}]
expect = parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]})
out = evaluate_metric_dual(
metric_id="bgp_peer",
key_fields=["peer"],
iface_fields=[],
compare_fields=[],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=True,
sheet_override={},
)
migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
self.assertTrue(migrated, out["rows"])
def test_arp_presence_migrated(self):
old_base = [{"ip": "10.0.0.1", "mac": "aaaa.bbbb.cccc"}]
old_cur: list[dict] = []
new_cur = [{"ip": "10.0.0.1", "mac": "aaaa.bbbb.cccc"}]
expect = parse_expect_set({"items": [{"metric_id": "arp", "key": "10.0.0.1"}]})
out = evaluate_metric_dual(
metric_id="arp",
key_fields=["ip"],
iface_fields=[],
compare_fields=[],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=True,
sheet_override={
"metric_id": "arp",
"success": [{"old": ["removed"], "new": ["added", "unchanged"]}],
},
)
migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
self.assertTrue(migrated, out["rows"])
def test_bgp_idle_to_established(self):
bgp_ov = {
"metric_id": "bgp_peer",
"status_fields": ["state"],
"down_values": ["idle", "active", "connect", "down"],
"up_values": ["established"],
"success": [{"old": ["removed", "down"], "new": ["added", "up", "unchanged"]}],
}
old_base = [{"peer": "1.1.1.1", "state": "Established"}]
old_cur = [{"peer": "1.1.1.1", "state": "Idle"}]
new_base = [{"peer": "1.1.1.1", "state": "Idle"}]
new_cur = [{"peer": "1.1.1.1", "state": "Established"}]
expect = parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]})
out = evaluate_metric_dual(
metric_id="bgp_peer",
key_fields=["peer"],
iface_fields=[],
compare_fields=["state"],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=new_base,
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=True,
sheet_override=bgp_ov,
)
migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
self.assertTrue(migrated, out["rows"])
def test_out_of_expect_ignore_no_red(self):
old_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
old_cur: list[dict] = []
new_cur = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
expect = parse_expect_set({"ports": []})
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=False,
out_of_expect="ignore",
)
reds = [r for r in out["rows"] if r["color"] == "red"]
self.assertFalse(reds, out["rows"])
involved = [r for r in out["rows"] if r["verdict"] != "not_involved"]
self.assertFalse(involved, out["rows"])
def test_unexpected_loss_is_anomaly(self):
old_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
old_cur: list[dict] = []
new_cur = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
expect = parse_expect_set({"ports": []})
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=False,
)
reds = [r for r in out["rows"] if r["color"] == "red"]
self.assertTrue(reds, out["rows"])
def test_new_baseline_missing_flag(self):
old_base = [{"peer": "1.1.1.1", "state": "Established"}]
new_cur = [{"peer": "1.1.1.1", "state": "Established"}]
expect = parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]})
out = evaluate_metric_dual(
metric_id="bgp_peer",
key_fields=["peer"],
iface_fields=[],
compare_fields=["state"],
old_baseline_rows=old_base,
old_current_rows=old_base,
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=True,
sheet_override={},
)
self.assertTrue(out["new_baseline_missing"])
self.assertEqual(out["new_baseline_mode"], "empty")
# With empty new before, current rows are all "added"
kinds = {r["new_kind"] for r in out["rows"] if r.get("in_expect")}
self.assertIn("added", kinds)
def test_new_baseline_port_mapped_not_missing(self):
old_base = [{"interface": "gei-old", "admin": "up", "phy": "up", "prot": "up"}]
new_cur = [{"interface": "gei-new", "admin": "up", "phy": "up", "prot": "up"}]
expect = parse_expect_set({"ports": ["gei-old"]})
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=[],
new_baseline_rows=None,
new_current_rows=new_cur,
port_map={"gei-old": "gei-new"},
expect=expect,
window_active=True,
sheet_override={},
)
self.assertFalse(out["new_baseline_missing"])
self.assertEqual(out["new_baseline_mode"], "port_mapped")
def test_empty_provided_new_baseline_falls_back_or_missing(self):
old_base = [{"peer": "1.1.1.1", "state": "Established"}]
new_cur = [{"peer": "1.1.1.1", "state": "Established"}]
expect = parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]})
out = evaluate_metric_dual(
metric_id="bgp_peer",
key_fields=["peer"],
iface_fields=[],
compare_fields=["state"],
old_baseline_rows=old_base,
old_current_rows=[],
new_baseline_rows=[],
new_current_rows=new_cur,
port_map={},
expect=expect,
window_active=True,
sheet_override={},
)
self.assertTrue(out["new_baseline_missing"])
self.assertEqual(out["new_baseline_mode"], "empty")
# Still reports presence as added vs empty before — callers should heed the flag.
kinds = {r["new_kind"] for r in out["rows"] if r.get("in_expect")}
self.assertIn("added", kinds)
def test_value_ne_false_when_removed(self):
verdict, color = dual_verdict(
old_kind="removed",
new_kind="unchanged",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [
[{"type": "value", "field": "state", "op": "ne", "value": "up"}]
],
"new_groups": [[{"type": "presence", "value": "unchanged"}]],
}
],
old_row={},
new_row={"state": "up"},
)
self.assertNotEqual(verdict, "migrated")
def test_composite_key_port_remap(self):
mapped = _remap_key_str(
"gei-old|vrf1",
key_fields=["interface", "vrf"],
iface_fields=["interface"],
port_map={"gei-old": "gei-new"},
)
self.assertEqual(mapped, "gei-new|vrf1")
def test_anomaly_not_gated_by_field_tokens_on_removed(self):
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="removed",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
anomaly_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "removed"}]],
}
],
sheet_override={
"field_tokens": [{"side": "old", "field": "state", "in": ["up"]}],
},
old_row={},
new_row={},
),
("anomaly", "red"),
)
def test_side_only_anomaly_deferred_during_window(self):
"""Mid-cutover: new-side gone alone must not red while window is active."""
side_only = [
{
"old_groups": [],
"new_groups": [[{"type": "presence", "value": "removed"}]],
}
]
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="removed",
in_expect=True,
window_active=True,
acceptance=False,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
anomaly_patterns=side_only,
),
("migrating", "yellow"),
)
# Acceptance / window closed → side-only fires
self.assertEqual(
dual_verdict(
old_kind="unchanged",
new_kind="removed",
in_expect=True,
window_active=False,
acceptance=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
anomaly_patterns=side_only,
),
("anomaly", "red"),
)
def test_both_gone_anomaly_still_fires_during_window(self):
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="removed",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
anomaly_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "removed"}]],
}
],
),
("anomaly", "red"),
)
def test_bare_unchanged_not_success_with_tight_preset(self):
from netx_api.biz_migration.monitor_templates import preset_override_for_metric
ov = preset_override_for_metric("interface_brief")
# Old removed + new still present but down → not green
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="unchanged",
in_expect=True,
window_active=True,
old_status="none",
new_status="down",
success_patterns=ov["success"],
anomaly_patterns=ov["anomaly"],
old_row={},
new_row={"admin": "down", "phy": "down"},
),
("migrating", "yellow"),
)
# Old removed + new up → green
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="unchanged",
in_expect=True,
window_active=True,
old_status="none",
new_status="up",
success_patterns=ov["success"],
anomaly_patterns=ov["anomaly"],
old_row={},
new_row={"admin": "up", "phy": "up"},
),
("migrated", "green"),
)
def test_arp_preset_requires_added_not_bare_unchanged(self):
from netx_api.biz_migration.monitor_templates import preset_override_for_metric
ov = preset_override_for_metric("arp")
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="unchanged",
in_expect=True,
window_active=True,
success_patterns=ov["success"],
anomaly_patterns=ov["anomaly"],
),
("migrating", "yellow"),
)
self.assertEqual(
dual_verdict(
old_kind="removed",
new_kind="added",
in_expect=True,
window_active=True,
success_patterns=ov["success"],
anomaly_patterns=ov["anomaly"],
),
("migrated", "green"),
)
def test_rule_hit_labels(self):
from netx_api.biz_migration.evaluate import dual_verdict_ex
v, c, hit = dual_verdict_ex(
old_kind="removed",
new_kind="added",
in_expect=True,
window_active=True,
success_patterns=[
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
)
self.assertEqual((v, c), ("migrated", "green"))
self.assertTrue(hit.startswith("success#"))
class PortMapScopeTests(unittest.TestCase):
def test_mapping_is_expect_for_iface_sheet(self):
old_base = [
{"interface": "gei-old", "admin": "up", "phy": "up", "prot": "up"},
{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"},
]
new_base = [
{"interface": "gei-new", "admin": "up", "phy": "up", "prot": "up"},
{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"},
]
new_cur = [
{"interface": "gei-new", "admin": "up", "phy": "up", "prot": "up"},
{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"},
]
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=[
{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"},
],
new_baseline_rows=new_base,
new_current_rows=new_cur,
port_map={"gei-old": "gei-new"},
expect=parse_expect_set({}),
window_active=True,
sheet_override=PORT_OVERRIDE,
)
self.assertEqual(out["progress_total"], 1)
keys = {r["key_str"] for r in out["rows"]}
self.assertIn("gei-old", keys)
self.assertNotIn("gei-keep", keys)
def test_unmapped_same_iface_old_down_new_up_is_anomaly(self):
old_base = [
{"interface": "gei-old", "admin": "up", "phy": "up", "prot": "up"},
{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"},
]
old_cur = [{"interface": "gei-keep", "admin": "down", "phy": "down", "prot": "down"}]
new_base = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
new_cur = [{"interface": "gei-keep", "admin": "up", "phy": "up", "prot": "up"}]
out = evaluate_metric_dual(
metric_id="interface_brief",
key_fields=["interface"],
iface_fields=["interface"],
compare_fields=["admin", "phy", "prot"],
old_baseline_rows=old_base,
old_current_rows=old_cur,
new_baseline_rows=new_base,
new_current_rows=new_cur,
port_map={"gei-old": "gei-new"},
expect=parse_expect_set({}),
window_active=True,
sheet_override=PORT_OVERRIDE,
)
keep = [r for r in out["rows"] if r["key_str"] == "gei-keep"]
self.assertEqual(len(keep), 1, out["rows"])
self.assertEqual(keep[0]["verdict"], "anomaly")
self.assertEqual(keep[0]["rule_hit"], "unmapped_same_iface")
self.assertFalse(keep[0]["in_expect"])
def test_arp_filtered_by_iface_mapping(self):
old_base = [
{"ip": "10.0.0.1", "interface": "gei-old"},
{"ip": "10.0.0.2", "interface": "gei-keep"},
]
new_cur = [{"ip": "10.0.0.1", "interface": "gei-new"}]
out = evaluate_metric_dual(
metric_id="arp",
key_fields=["ip", "interface"],
iface_fields=["interface"],
compare_fields=[],
old_baseline_rows=old_base,
old_current_rows=[],
new_baseline_rows=[{"ip": "9.9.9.9", "interface": "gei-other"}],
new_current_rows=new_cur,
port_map={"gei-old": "gei-new"},
expect=parse_expect_set({}),
window_active=True,
sheet_override={
"success": [
{
"old_groups": [[{"type": "presence", "value": "removed"}]],
"new_groups": [[{"type": "presence", "value": "added"}]],
}
],
},
)
keys = {r["key_str"] for r in out["rows"]}
self.assertIn("10.0.0.1|gei-old", keys)
self.assertNotIn("10.0.0.2|gei-keep", keys)
migrated = [r for r in out["rows"] if r["verdict"] == "migrated"]
self.assertTrue(migrated, out["rows"])
self.assertEqual(migrated[0]["new_key_str"], "10.0.0.1|gei-new")
def test_bgp_still_uses_expect_picker(self):
out = evaluate_metric_dual(
metric_id="bgp_peer",
key_fields=["peer"],
iface_fields=["interface"],
compare_fields=["state"],
old_baseline_rows=[{"peer": "1.1.1.1", "interface": "gei-old", "state": "Established"}],
old_current_rows=[],
new_baseline_rows=[{"peer": "1.1.1.1", "interface": "gei-new", "state": "Idle"}],
new_current_rows=[{"peer": "1.1.1.1", "interface": "gei-new", "state": "Established"}],
port_map={"gei-keep": "gei-other"},
expect=parse_expect_set({"items": [{"metric_id": "bgp_peer", "key": "1.1.1.1"}]}),
window_active=True,
sheet_override={
"success": [{"old": ["removed"], "new": ["changed", "up"]}],
"status_fields": ["state"],
"up_values": ["established"],
"down_values": ["idle"],
},
)
self.assertTrue(any(r["key_str"] == "1.1.1.1" and r["in_expect"] for r in out["rows"]))
if __name__ == "__main__":
unittest.main()
Reference in a new issue View git blame Copy permalink