mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 05:30:46 +08:00
293 lines
10 KiB
Python
293 lines
10 KiB
Python
"""Unit tests for UME port normalize + fabric apply provenance."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
import netx_api.models # noqa: F401
|
|
from netx_api.db import Base
|
|
from netx_api.models import (
|
|
TopoFabricEdge,
|
|
TopoFabricNode,
|
|
UmeInventoryNE,
|
|
UmeSyncJob,
|
|
UmeTopoLink,
|
|
UmeTopoNode,
|
|
)
|
|
from netx_api.ume_port_normalize import (
|
|
extract_ifnames_from_user_label,
|
|
port_keys_compatible,
|
|
port_suffix_from_tp_ref,
|
|
resolve_link_ifnames,
|
|
)
|
|
from netx_api.ume_topology_apply import (
|
|
apply_ume_dock_to_fabric_if_needed,
|
|
apply_ume_topology_to_fabric,
|
|
ume_topology_apply_gap,
|
|
)
|
|
from netx_api.topology_fabric_links import upsert_fabric_edge
|
|
|
|
|
|
class UmePortNormalizeTests(unittest.TestCase):
|
|
def test_label_xxvgei_pair(self):
|
|
label = (
|
|
"MDN-UJSB-EN1-Z20HS-SMGD[1/1/0]_xxvgei-1/1/0/32(10G:to_MDN-UJSR-EN1-Z680H_"
|
|
"xgei-1/2/0/1:via_OTB)_MDN-UJSR"
|
|
)
|
|
tokens = extract_ifnames_from_user_label(label)
|
|
self.assertEqual(tokens[0], "xxvgei-1/1/0/32")
|
|
self.assertEqual(tokens[1], "xgei-1/2/0/1")
|
|
a, z = resolve_link_ifnames(
|
|
a_end_tp_ref="ME{a},EQ={/r=0/sh=1/sl=1},PTP={/p=1_32}",
|
|
z_end_tp_ref="ME{b},EQ={/r=0/sh=1/sl=2},PTP={/p=1_1}",
|
|
user_label=label,
|
|
)
|
|
self.assertEqual(a, "xxvgei-1/1/0/32")
|
|
self.assertEqual(z, "xgei-1/2/0/1")
|
|
|
|
def test_eq_ptp_suffix(self):
|
|
self.assertEqual(
|
|
port_suffix_from_tp_ref("ME{x},EQ={/r=0/sh=1/sl=1},PTP={/p=1_28}"),
|
|
"1/1/0/28",
|
|
)
|
|
self.assertEqual(
|
|
port_suffix_from_tp_ref("ME{x},EQ={/r=0/sh=0/sl=0/ssl=1},PTP={/p=1_4}"),
|
|
"0/0/1/4",
|
|
)
|
|
|
|
def test_eth_colon_fallback(self):
|
|
label = "JBI-SIMK-EN1-Z20HS-[1/1/0]_ETH:28_JBI-OTOB-EN1-Z20HS-[1/1/0]_ETH:28"
|
|
a, z = resolve_link_ifnames(
|
|
a_end_tp_ref="ME{a},EQ={/r=0/sh=1/sl=1},PTP={/p=1_28}",
|
|
z_end_tp_ref="ME{b},EQ={/r=0/sh=1/sl=1},PTP={/p=1_28}",
|
|
user_label=label,
|
|
)
|
|
self.assertEqual(a, "1/1/0/28")
|
|
self.assertEqual(z, "1/1/0/28")
|
|
|
|
def test_port_compatible(self):
|
|
self.assertTrue(port_keys_compatible("xxvgei-1/1/0/32", "1/1/0/32"))
|
|
self.assertTrue(port_keys_compatible("xxvgei-1/1/0/32", "xxvgei-1/1/0/32"))
|
|
self.assertFalse(port_keys_compatible("xxvgei-1/1/0/32", "xxvgei-1/1/0/28"))
|
|
|
|
|
|
class UmeFabricApplyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
|
TestingSessionLocal = sessionmaker(
|
|
bind=engine, autoflush=False, autocommit=False, expire_on_commit=False
|
|
)
|
|
Base.metadata.create_all(bind=engine)
|
|
self.db = TestingSessionLocal()
|
|
self.now = datetime(2026, 8, 6, 12, 0, 0)
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
|
|
def _seed_ume(self):
|
|
self.db.add(
|
|
UmeInventoryNE(
|
|
ne_id="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
ne_name="ME{a}",
|
|
user_label="NE-A",
|
|
host_name="NE-A",
|
|
ip_address="10.0.0.1",
|
|
)
|
|
)
|
|
self.db.add(
|
|
UmeInventoryNE(
|
|
ne_id="bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
|
|
ne_name="ME{b}",
|
|
user_label="NE-B",
|
|
host_name="NE-B",
|
|
ip_address="10.0.0.2",
|
|
)
|
|
)
|
|
self.db.add(
|
|
UmeTopoNode(
|
|
node_id="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
name="MD=ZTE;ME=aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
node_type="TOPO_NODE_ME",
|
|
user_label="NE-A",
|
|
parent_node="sbn-1",
|
|
x_pos=100,
|
|
y_pos=200,
|
|
ume_ne_id="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
first_seen_at=self.now,
|
|
last_seen_at=self.now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
UmeTopoNode(
|
|
node_id="bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
|
|
name="MD=ZTE;ME=bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
|
|
node_type="TOPO_NODE_ME",
|
|
user_label="NE-B",
|
|
parent_node="sbn-1",
|
|
x_pos=300,
|
|
y_pos=400,
|
|
ume_ne_id="bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
|
|
first_seen_at=self.now,
|
|
last_seen_at=self.now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
UmeTopoLink(
|
|
link_id="link-1",
|
|
name="TL{…}",
|
|
user_label="NE-A_xxvgei-1/1/0/32_NE-B_xxvgei-1/1/0/28",
|
|
a_end_tp_ref="ME{aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa},EQ={/r=0/sh=1/sl=1},PTP={/p=1_32}",
|
|
z_end_tp_ref="ME{bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb},EQ={/r=0/sh=1/sl=1},PTP={/p=1_28}",
|
|
a_ume_ne_id="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
z_ume_ne_id="bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
|
|
a_ptp="/p=1_32",
|
|
z_ptp="/p=1_28",
|
|
a_ifname="xxvgei-1/1/0/32",
|
|
z_ifname="xxvgei-1/1/0/28",
|
|
first_seen_at=self.now,
|
|
last_seen_at=self.now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
def test_apply_sets_coords_and_ume_edge(self):
|
|
self._seed_ume()
|
|
stats = apply_ume_topology_to_fabric(self.db)
|
|
self.assertEqual(stats["nodes_coords"], 2)
|
|
self.assertGreaterEqual(stats["edges_upserted"] + stats["edges_merged"], 1)
|
|
a = (
|
|
self.db.query(TopoFabricNode)
|
|
.filter(TopoFabricNode.ume_ne_id == "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
|
.one()
|
|
)
|
|
self.assertEqual((a.attrs or {}).get("ume_local_x"), 100.0)
|
|
self.assertEqual((a.attrs or {}).get("ume_local_y"), 200.0)
|
|
# Flat packing: single SBN slot → local relative to cluster min.
|
|
self.assertEqual(a.world_x, 0.0)
|
|
self.assertEqual(a.world_y, 0.0)
|
|
edges = self.db.query(TopoFabricEdge).all()
|
|
self.assertEqual(len(edges), 1)
|
|
self.assertIn("ume", (edges[0].attrs or {}).get("sources", []))
|
|
|
|
def test_lldp_edge_survives_ume_clear(self):
|
|
self._seed_ume()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
a = (
|
|
self.db.query(TopoFabricNode)
|
|
.filter(TopoFabricNode.ume_ne_id == "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
|
.one()
|
|
)
|
|
b = (
|
|
self.db.query(TopoFabricNode)
|
|
.filter(TopoFabricNode.ume_ne_id == "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")
|
|
.one()
|
|
)
|
|
edge = self.db.query(TopoFabricEdge).one()
|
|
# Simulate LLDP also seeing it
|
|
edge.source = "lldp"
|
|
edge.attrs = {"sources": ["ume", "lldp"]}
|
|
self.db.commit()
|
|
|
|
# Remove UME link → apply should keep edge (lldp remains)
|
|
self.db.query(UmeTopoLink).delete()
|
|
self.db.commit()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
edge2 = self.db.query(TopoFabricEdge).one()
|
|
self.assertEqual(edge2.status, "active")
|
|
self.assertIn("lldp", (edge2.attrs or {}).get("sources", []))
|
|
self.assertNotIn("ume", (edge2.attrs or {}).get("sources", []))
|
|
|
|
def test_pure_ume_becomes_missing_when_gone(self):
|
|
self._seed_ume()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
self.db.query(UmeTopoLink).delete()
|
|
self.db.commit()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
edge = self.db.query(TopoFabricEdge).one()
|
|
self.assertEqual(edge.status, "missing")
|
|
|
|
def test_manual_edge_never_cleared(self):
|
|
self._seed_ume()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
a = (
|
|
self.db.query(TopoFabricNode)
|
|
.filter(TopoFabricNode.ume_ne_id == "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
|
.one()
|
|
)
|
|
b = (
|
|
self.db.query(TopoFabricNode)
|
|
.filter(TopoFabricNode.ume_ne_id == "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")
|
|
.one()
|
|
)
|
|
upsert_fabric_edge(
|
|
self.db,
|
|
a_node_id=a.id,
|
|
b_node_id=b.id,
|
|
a_port="gi0/1",
|
|
b_port="gi0/2",
|
|
source="manual",
|
|
)
|
|
self.db.commit()
|
|
self.db.query(UmeTopoLink).delete()
|
|
self.db.commit()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
manuals = (
|
|
self.db.query(TopoFabricEdge).filter(TopoFabricEdge.source == "manual").all()
|
|
)
|
|
self.assertEqual(len(manuals), 1)
|
|
self.assertEqual(manuals[0].status, "active")
|
|
|
|
def test_gap_needs_apply_when_dock_only(self):
|
|
self._seed_ume()
|
|
self.db.commit()
|
|
gap = ume_topology_apply_gap(self.db)
|
|
self.assertGreater(gap["dock_me_count"], 0)
|
|
self.assertEqual(gap["fabric_ume_count"], 0)
|
|
self.assertTrue(gap["needs_apply"])
|
|
|
|
def test_gap_needs_apply_on_partial_job(self):
|
|
self._seed_ume()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
self.db.add(
|
|
UmeSyncJob(
|
|
domain="topology",
|
|
status="partial",
|
|
trigger_mode="manual",
|
|
started_at=self.now,
|
|
ended_at=self.now,
|
|
error_message="dock_ok_fabric_apply_failed:boom",
|
|
)
|
|
)
|
|
self.db.commit()
|
|
gap = ume_topology_apply_gap(self.db)
|
|
self.assertTrue(gap["partial_apply"])
|
|
self.assertTrue(gap["needs_apply"])
|
|
|
|
def test_apply_if_needed_skips_when_healthy(self):
|
|
self._seed_ume()
|
|
apply_ume_topology_to_fabric(self.db)
|
|
# World folder may still be missing → force a clean gap check after ensure.
|
|
from netx_api.ume_topology_world import ensure_ume_world_and_sbn_folders
|
|
|
|
ensure_ume_world_and_sbn_folders(self.db)
|
|
self.db.add(
|
|
UmeSyncJob(
|
|
domain="topology",
|
|
status="done",
|
|
trigger_mode="manual",
|
|
started_at=self.now,
|
|
ended_at=self.now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
gap = ume_topology_apply_gap(self.db)
|
|
self.assertFalse(gap["needs_apply"])
|
|
self.assertIsNone(apply_ume_dock_to_fabric_if_needed(self.db, reason="test"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|