mirror of
https://github.com/dtzp555-max/aci-sim.git
synced 2026-07-19 09:46:33 +00:00
taodengandClaude Fable 5
7e9a175ce6
aci-sim is a stateful REST simulator of a 2-site Cisco ACI fabric (per-site APIC + ND/NDO management planes) — for testing ACI automation (Ansible cisco.aci / cisco.mso, aci-py, custom REST clients) and CI gates without real hardware. Highlights: APIC + NDO REST surface served from one in-memory MIT built from a declarative topology.yaml; port mode + sandbox (real per-device IPs on :443) run modes; LLDP/CDP neighbor visibility; NDO->APIC deploy mirror (multi-site templates materialize onto the target sites' APIC stores); real-APIC fvBD default attributes; file-backed state save/restore; an `aci-sim` CLI (validate/show/graph/run/new/init/lldp); and an 888-test suite. History squashed for the public release. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
621 lines
17 KiB
Python
621 lines
17 KiB
Python
"""Tests for topology/schema.py and topology/loader.py (Phase 2).
|
|
|
|
All tests load the default topology.yaml at the repo root. One negative test
|
|
builds a minimal bad topology inline to verify cross-reference validation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from pydantic import ValidationError
|
|
|
|
from aci_sim.topology.loader import load_topology
|
|
from aci_sim.topology.schema import Node, Topology
|
|
|
|
TOPOLOGY_YAML = Path(__file__).parent.parent / "topology.yaml"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def get_topo() -> Topology:
|
|
return load_topology(TOPOLOGY_YAML)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Positive tests against default topology.yaml
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDefaultTopologyLoad:
|
|
def test_has_two_sites(self) -> None:
|
|
topo = get_topo()
|
|
assert len(topo.sites) == 2
|
|
|
|
def test_site_names(self) -> None:
|
|
topo = get_topo()
|
|
names = [s.name for s in topo.sites]
|
|
assert "LAB1" in names
|
|
assert "LAB2" in names
|
|
|
|
def test_site_a_spine_count(self) -> None:
|
|
topo = get_topo()
|
|
site_a = topo.site_by_name("LAB1")
|
|
assert len(site_a.spine_nodes()) == 2
|
|
|
|
def test_site_a_leaf_count(self) -> None:
|
|
topo = get_topo()
|
|
site_a = topo.site_by_name("LAB1")
|
|
assert len(site_a.leaf_nodes()) == 2
|
|
|
|
def test_site_a_border_leaf_count(self) -> None:
|
|
topo = get_topo()
|
|
site_a = topo.site_by_name("LAB1")
|
|
assert len(site_a.border_leaves) == 2
|
|
|
|
def test_site_b_spine_count(self) -> None:
|
|
topo = get_topo()
|
|
site_b = topo.site_by_name("LAB2")
|
|
assert len(site_b.spine_nodes()) == 2
|
|
|
|
def test_site_b_leaf_count(self) -> None:
|
|
topo = get_topo()
|
|
site_b = topo.site_by_name("LAB2")
|
|
assert len(site_b.leaf_nodes()) == 2
|
|
|
|
def test_site_b_border_leaf_count(self) -> None:
|
|
topo = get_topo()
|
|
site_b = topo.site_by_name("LAB2")
|
|
assert len(site_b.border_leaves) == 2
|
|
|
|
|
|
class TestNodeIdExpansion:
|
|
"""Int-count → Node expansion produces deterministic, collision-free IDs."""
|
|
|
|
def test_site_a_spine_ids(self) -> None:
|
|
# site_index=0: spine_id = 201 + 0*200 + i → 201, 202
|
|
topo = get_topo()
|
|
ids = sorted(n.id for n in topo.site_by_name("LAB1").spine_nodes())
|
|
assert ids == [201, 202]
|
|
|
|
def test_site_a_leaf_ids(self) -> None:
|
|
# site_index=0: leaf_id = 101 + 0*200 + i → 101, 102
|
|
topo = get_topo()
|
|
ids = sorted(n.id for n in topo.site_by_name("LAB1").leaf_nodes())
|
|
assert ids == [101, 102]
|
|
|
|
def test_site_b_spine_ids(self) -> None:
|
|
# site_index=1: spine_id = 201 + 1*200 + i → 401, 402
|
|
topo = get_topo()
|
|
ids = sorted(n.id for n in topo.site_by_name("LAB2").spine_nodes())
|
|
assert ids == [401, 402]
|
|
|
|
def test_site_b_leaf_ids(self) -> None:
|
|
# site_index=1: leaf_id = 101 + 1*200 + i → 301, 302
|
|
topo = get_topo()
|
|
ids = sorted(n.id for n in topo.site_by_name("LAB2").leaf_nodes())
|
|
assert ids == [301, 302]
|
|
|
|
def test_no_collisions_across_sites(self) -> None:
|
|
topo = get_topo()
|
|
all_ids: list[int] = []
|
|
for site in topo.sites:
|
|
all_ids.extend(n.id for n in site.all_nodes())
|
|
|
|
duplicates = [nid for nid in set(all_ids) if all_ids.count(nid) > 1]
|
|
assert duplicates == [], f"Duplicate node IDs across sites: {duplicates}"
|
|
|
|
def test_spine_names_deterministic(self) -> None:
|
|
topo = get_topo()
|
|
site_a = topo.site_by_name("LAB1")
|
|
names = sorted(n.name for n in site_a.spine_nodes())
|
|
assert names == ["LAB1-ACI-SP01", "LAB1-ACI-SP02"]
|
|
|
|
def test_leaf_names_deterministic(self) -> None:
|
|
topo = get_topo()
|
|
site_b = topo.site_by_name("LAB2")
|
|
names = sorted(n.name for n in site_b.leaf_nodes())
|
|
assert names == ["LAB2-ACI-LF301", "LAB2-ACI-LF302"]
|
|
|
|
|
|
class TestStretchedTenant:
|
|
def test_stretched_tenant_exists(self) -> None:
|
|
topo = get_topo()
|
|
stretched = [t for t in topo.tenants if t.stretch]
|
|
assert len(stretched) >= 1, "No stretched tenant found in topology"
|
|
|
|
def test_stretched_tenant_in_both_sites(self) -> None:
|
|
topo = get_topo()
|
|
for tenant in topo.tenants:
|
|
if tenant.stretch:
|
|
assert "LAB1" in tenant.sites, (
|
|
f"Stretched tenant '{tenant.name}' missing LAB1"
|
|
)
|
|
assert "LAB2" in tenant.sites, (
|
|
f"Stretched tenant '{tenant.name}' missing LAB2"
|
|
)
|
|
|
|
def test_corp_tenant_vrfs_bds_aps(self) -> None:
|
|
topo = get_topo()
|
|
corp = next(t for t in topo.tenants if t.name == "MS-TN1")
|
|
assert len(corp.vrfs) >= 1
|
|
assert len(corp.bds) >= 1
|
|
assert len(corp.aps) >= 1
|
|
|
|
def test_corp_contracts_non_empty(self) -> None:
|
|
topo = get_topo()
|
|
corp = next(t for t in topo.tenants if t.name == "MS-TN1")
|
|
assert len(corp.contracts) >= 1
|
|
|
|
def test_corp_l3outs_per_site(self) -> None:
|
|
topo = get_topo()
|
|
corp = next(t for t in topo.tenants if t.name == "MS-TN1")
|
|
assert len(corp.l3outs) >= 2, "MS-TN1 tenant should have ≥1 L3Out per site"
|
|
|
|
|
|
class TestL3OutVrfResolves:
|
|
def test_l3out_vrfs_valid(self) -> None:
|
|
topo = get_topo()
|
|
for tenant in topo.tenants:
|
|
vrf_names = {v.name for v in tenant.vrfs}
|
|
for l3out in tenant.l3outs:
|
|
assert l3out.vrf in vrf_names, (
|
|
f"Tenant '{tenant.name}', L3Out '{l3out.name}' references "
|
|
f"VRF '{l3out.vrf}' not in {vrf_names}"
|
|
)
|
|
|
|
|
|
class TestHelpers:
|
|
def test_site_by_name_found(self) -> None:
|
|
topo = get_topo()
|
|
site = topo.site_by_name("LAB1")
|
|
assert site.name == "LAB1"
|
|
|
|
def test_site_by_name_missing_raises(self) -> None:
|
|
topo = get_topo()
|
|
with pytest.raises(KeyError):
|
|
topo.site_by_name("Nonexistent")
|
|
|
|
def test_other_site_from_a_returns_b(self) -> None:
|
|
topo = get_topo()
|
|
other = topo.other_site("LAB1")
|
|
assert other.name == "LAB2"
|
|
|
|
def test_other_site_from_b_returns_a(self) -> None:
|
|
topo = get_topo()
|
|
other = topo.other_site("LAB2")
|
|
assert other.name == "LAB1"
|
|
|
|
def test_other_site_gives_remote_asn(self) -> None:
|
|
"""other_site() should let builders look up the remote fabric ASN for ISN."""
|
|
topo = get_topo()
|
|
# LAB1's ISN peer is LAB2's ASN
|
|
remote_asn = topo.other_site("LAB1").asn
|
|
assert remote_asn == topo.site_by_name("LAB2").asn
|
|
|
|
def test_all_nodes_count(self) -> None:
|
|
topo = get_topo()
|
|
site_a = topo.site_by_name("LAB1")
|
|
# 2 spines + 2 leaves + 2 border leaves = 6
|
|
assert len(site_a.all_nodes()) == 6
|
|
|
|
|
|
class TestISN:
|
|
def test_isn_enabled(self) -> None:
|
|
topo = get_topo()
|
|
assert topo.isn.enabled is True
|
|
|
|
def test_isn_peer_mesh_full(self) -> None:
|
|
topo = get_topo()
|
|
assert topo.isn.peer_mesh == "full"
|
|
|
|
|
|
class TestFaultsAndEndpoints:
|
|
def test_fault_seeds_present(self) -> None:
|
|
topo = get_topo()
|
|
assert len(topo.faults.seed) >= 1
|
|
|
|
def test_health_defaults(self) -> None:
|
|
topo = get_topo()
|
|
assert topo.faults.health_defaults.node > 0
|
|
assert topo.faults.health_defaults.tenant > 0
|
|
|
|
def test_endpoints_per_epg(self) -> None:
|
|
topo = get_topo()
|
|
assert topo.endpoints.per_epg >= 1
|
|
|
|
|
|
class TestNewBuilderFields:
|
|
"""Fields added for the Phase 3/4 builders (version, l3 domains, addr pools)."""
|
|
|
|
def test_node_has_version_default(self) -> None:
|
|
topo = get_topo()
|
|
for node in topo.site_by_name("LAB1").all_nodes():
|
|
assert node.version, f"Node {node.id} missing version"
|
|
|
|
def test_explicit_node_version_override(self) -> None:
|
|
n = Node(id=201, name="spine-201", version="n9000-16.0(1)")
|
|
assert n.version == "n9000-16.0(1)"
|
|
|
|
def test_access_has_l3_domain(self) -> None:
|
|
topo = get_topo()
|
|
assert len(topo.access.l3_domains) >= 1
|
|
assert topo.access.l3_domains[0].name
|
|
|
|
def test_fabric_address_pools(self) -> None:
|
|
topo = get_topo()
|
|
assert topo.fabric.loopback_pool
|
|
assert topo.fabric.oob_pool
|
|
|
|
def test_l3out_site_association(self) -> None:
|
|
topo = get_topo()
|
|
corp = next(t for t in topo.tenants if t.name == "MS-TN1")
|
|
by_name = {lo.name: lo for lo in corp.l3outs}
|
|
assert by_name["l3o-bgp-Core_LAB1"].site == "LAB1"
|
|
assert by_name["l3o-bgp-Core_LAB2"].site == "LAB2"
|
|
|
|
def test_l3out_border_leaves_are_real(self) -> None:
|
|
topo = get_topo()
|
|
all_border_ids = {
|
|
bl.id for site in topo.sites for bl in site.border_leaves
|
|
}
|
|
for tenant in topo.tenants:
|
|
for l3out in tenant.l3outs:
|
|
for bl_id in l3out.border_leaves:
|
|
assert bl_id in all_border_ids
|
|
|
|
def test_fault_seed_nodes_are_real(self) -> None:
|
|
topo = get_topo()
|
|
real_ids = {n.id for site in topo.sites for n in site.all_nodes()}
|
|
for fseed in topo.faults.seed:
|
|
assert fseed.node in real_ids
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Negative test — bad cross-reference raises ValidationError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValidationFailures:
|
|
def _load_bad(self, yaml_text: str) -> None:
|
|
data = yaml.safe_load(yaml_text)
|
|
Topology.model_validate(data)
|
|
|
|
def test_epg_references_missing_bd_raises(self) -> None:
|
|
"""EPG.bd that does not exist in tenant.bds must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants:
|
|
- name: BadTenant
|
|
stretch: false
|
|
sites: [SiteA]
|
|
vrfs:
|
|
- name: vrf1
|
|
bds:
|
|
- name: real-bd
|
|
vrf: vrf1
|
|
subnets: []
|
|
aps:
|
|
- name: App
|
|
epgs:
|
|
- name: bad-epg
|
|
bd: nonexistent-bd
|
|
contracts: {}
|
|
contracts: []
|
|
l3outs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_bd_references_missing_vrf_raises(self) -> None:
|
|
"""BD.vrf that does not exist in tenant.vrfs must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants:
|
|
- name: BadTenant
|
|
stretch: false
|
|
sites: [SiteA]
|
|
vrfs:
|
|
- name: vrf1
|
|
bds:
|
|
- name: bd1
|
|
vrf: ghost-vrf
|
|
subnets: []
|
|
aps: []
|
|
contracts: []
|
|
l3outs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_stretched_tenant_with_one_site_raises(self) -> None:
|
|
"""stretch:true but only one site listed must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
- name: SiteB
|
|
id: "2"
|
|
apic_host: "127.0.0.1:8444"
|
|
asn: 65002
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: true
|
|
peer_mesh: full
|
|
tenants:
|
|
- name: BadStretch
|
|
stretch: true
|
|
sites: [SiteA]
|
|
vrfs:
|
|
- name: vrf1
|
|
bds: []
|
|
aps: []
|
|
contracts: []
|
|
l3outs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_tenant_references_unknown_site_raises(self) -> None:
|
|
"""tenant.sites that names a non-existent site must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants:
|
|
- name: BadTenant
|
|
stretch: false
|
|
sites: [SiteX]
|
|
vrfs: []
|
|
bds: []
|
|
aps: []
|
|
contracts: []
|
|
l3outs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_l3out_references_missing_vrf_raises(self) -> None:
|
|
"""L3Out.vrf that does not exist in tenant.vrfs must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
border_leaves:
|
|
- id: 103
|
|
name: leaf-103
|
|
vpc_domain: vpcdom-siteA
|
|
csw: {name: CSW-A, asn: 65100, peer_ip: 10.100.0.1, local_ip: 10.100.0.2, vlan: 100}
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants:
|
|
- name: BadTenant
|
|
stretch: false
|
|
sites: [SiteA]
|
|
vrfs:
|
|
- name: vrf1
|
|
bds: []
|
|
aps: []
|
|
contracts: []
|
|
l3outs:
|
|
- name: bad-l3out
|
|
vrf: ghost
|
|
border_leaves: [103]
|
|
ext_epgs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_isn_full_mesh_single_site_raises(self) -> None:
|
|
"""isn.peer_mesh='full' with fewer than 2 sites must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: true
|
|
peer_mesh: full
|
|
tenants: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_l3out_references_missing_border_leaf_raises(self) -> None:
|
|
"""L3Out.border_leaves id not present in any site's border_leaves must raise."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
border_leaves:
|
|
- id: 103
|
|
name: leaf-103
|
|
vpc_domain: vpcdom-siteA
|
|
csw: {name: CSW-A, asn: 65100, peer_ip: 10.100.0.1, local_ip: 10.100.0.2, vlan: 100}
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants:
|
|
- name: T
|
|
stretch: false
|
|
sites: [SiteA]
|
|
vrfs:
|
|
- name: vrf1
|
|
bds: []
|
|
aps: []
|
|
contracts: []
|
|
l3outs:
|
|
- name: l3o
|
|
vrf: vrf1
|
|
border_leaves: [999]
|
|
ext_epgs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_fault_seed_unknown_node_raises(self) -> None:
|
|
"""faults.seed[i].node that is not a real node id must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants: []
|
|
faults:
|
|
seed:
|
|
- severity: minor
|
|
code: "F0532"
|
|
node: 9999
|
|
descr: "fault on a non-existent node"
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_epg_references_missing_contract_raises(self) -> None:
|
|
"""EPG contract ref not present in tenant.contracts must raise ValidationError."""
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 1
|
|
leaves: 1
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants:
|
|
- name: T
|
|
stretch: false
|
|
sites: [SiteA]
|
|
vrfs:
|
|
- name: vrf1
|
|
bds:
|
|
- name: bd1
|
|
vrf: vrf1
|
|
subnets: []
|
|
aps:
|
|
- name: App
|
|
epgs:
|
|
- name: epg1
|
|
bd: bd1
|
|
contracts:
|
|
provide: [ghost-contract]
|
|
consume: []
|
|
contracts: []
|
|
l3outs: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|
|
|
|
def test_border_leaf_id_collision_raises(self) -> None:
|
|
"""A border-leaf id colliding with an auto-generated leaf id must raise."""
|
|
# SiteA auto leaves are 101,102; giving a border leaf id=101 collides.
|
|
bad_yaml = """
|
|
fabric:
|
|
name: bad-topo
|
|
sites:
|
|
- name: SiteA
|
|
id: "1"
|
|
apic_host: "127.0.0.1:8443"
|
|
asn: 65001
|
|
pod: 1
|
|
spines: 2
|
|
leaves: 2
|
|
border_leaves:
|
|
- id: 101
|
|
name: leaf-101-dup
|
|
vpc_domain: vpcdom-siteA
|
|
csw: {name: CSW-A, asn: 65100, peer_ip: 10.100.0.1, local_ip: 10.100.0.2, vlan: 100}
|
|
cabling: auto
|
|
isn:
|
|
enabled: false
|
|
tenants: []
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
self._load_bad(bad_yaml)
|