mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 02:00:46 +08:00
oliver
1ef43fdcd4
Seed a unique L2 World canvas under the UME World container, show the flat map only when nested regions have NEs, pack blocks without overlap, allow drag overrides on the world map while forbidding direct NE adds, and hide the world-map truncation banner. Co-authored-by: Cursor <cursoragent@cursor.com>
235 lines
8.5 KiB
Python
235 lines
8.5 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, 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_topology_to_fabric
|
|
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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|