feat: APIC-style write validation, query-target=children, contract shadow mirroring (v0.21.0)

Three fidelity gaps surfaced by one broken TN2 var file (missing firewall
block), all fixed:
- writes: reject malformed fvSubnet/l3extSubnet/vnsRedirectDest ip values
  with an APIC-style 400 before any store mutation (deletes exempt)
- query engine: support query-target=children (direct children, flat
  imdata, root excluded) — was silently returning empty
- deploy mirror: materialize vzFilter/vzEntry, vzBrCP/vzSubj (+ filter and
  service-graph subject bindings) and vnsAbsGraph shadows per target site;
  undeploy removes them

E2E-verified through the real pipeline (aci-py hidden push): a TN2 var
missing the fw block now fails with 'Invalid value "." for property ip of
vnsRedirectDest ... -> 400' and nothing lands; a complete var set builds
both MS tenants with contract shadows + sgt-FW graph binding on both sites.
Suite: 914 passed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
dtzp555-max
2026-07-07 18:11:59 +10:00
co-authored by Claude Fable 5
parent 7fdf31aca9
commit 867c05c9c5
7 changed files with 323 additions and 1 deletions
+59
View File
@@ -494,3 +494,62 @@ def test_app_level_undeploy_via_task_body():
assert resp.status_code == 200
assert apic_states["1"].store.get(f"uni/tn-{TENANT_NAME}/ap-AP1/epg-Web") is None
# ---------------------------------------------------------------------------
# Contract/filter/service-graph shadow mirroring (2026-07-07 gap: vzBrCP=0
# fabric-wide while mirrored EPGs carried fvRsProv/fvRsCons to those names)
# ---------------------------------------------------------------------------
def _firewall_template(state) -> None:
template = state.schema_details[SCHEMA_ID]["templates"][0]
template["filters"] = [{
"name": "flt-permit_ip_LAB0",
"entries": [{"name": "permit_ip", "etherType": "ip", "ipProtocol": "unspecified"}],
}]
template["serviceGraphs"] = [{"name": "sgt-FW_LAB0"}]
template["contracts"] = [{
"name": "con-Firewall_LAB0",
"scope": "context",
"filterRelationships": [
{"filterRef": f"/schemas/{SCHEMA_ID}/templates/{TEMPLATE_NAME}/filters/flt-permit_ip_LAB0"}
],
"serviceGraphRelationship": {
"serviceGraphRef": f"/schemas/{SCHEMA_ID}/templates/{TEMPLATE_NAME}/serviceGraphs/sgt-FW_LAB0"
},
}]
def test_mirror_materializes_contract_shadows() -> None:
state = _build_state()
_firewall_template(state)
apic = _FakeApicState()
mirror_template_to_sites(state, TEMPLATE_NAME, {"1": apic}, schema_id=SCHEMA_ID)
store = apic.store
tn = f"uni/tn-{TENANT_NAME}"
con_dn = f"{tn}/brc-con-Firewall_LAB0"
subj_dn = f"{con_dn}/subj-sub-Firewall_LAB0"
assert store.get(con_dn) is not None
assert store.get(con_dn).attrs["scope"] == "context"
assert store.get(subj_dn) is not None
filt_att = store.get(f"{subj_dn}/rssubjFiltAtt-flt-permit_ip_LAB0")
assert filt_att is not None and filt_att.attrs["tnVzFilterName"] == "flt-permit_ip_LAB0"
graph_att = store.get(f"{subj_dn}/rsSubjGraphAtt")
assert graph_att is not None and graph_att.attrs["tnVnsAbsGraphName"] == "sgt-FW_LAB0"
assert store.get(f"{tn}/flt-flt-permit_ip_LAB0") is not None
assert store.get(f"{tn}/flt-flt-permit_ip_LAB0/e-permit_ip") is not None
assert store.get(f"{tn}/AbsGraph-sgt-FW_LAB0") is not None
def test_undeploy_removes_contract_shadows() -> None:
state = _build_state()
_firewall_template(state)
apic = _FakeApicState()
mirror_template_to_sites(state, TEMPLATE_NAME, {"1": apic}, schema_id=SCHEMA_ID)
tn = f"uni/tn-{TENANT_NAME}"
assert apic.store.get(f"{tn}/brc-con-Firewall_LAB0") is not None
mirror_template_to_sites(state, TEMPLATE_NAME, {"1": apic}, schema_id=SCHEMA_ID, undeploy=True)
assert apic.store.get(f"{tn}/brc-con-Firewall_LAB0") is None
assert apic.store.get(f"{tn}/flt-flt-permit_ip_LAB0") is None
assert apic.store.get(f"{tn}/AbsGraph-sgt-FW_LAB0") is None
@@ -0,0 +1,97 @@
"""Malformed-property write validation (APIC-style 400) + query-target=children.
Both surfaced 2026-07-07 by a TN2 push whose var file was missing the
firewall block: the unguarded J2 template rendered ``fvSubnet ip=".1/24"``
and ``vnsRedirectDest ip="."`` and the sim ACCEPTED both (a real APIC 400s),
and the verification that should have caught it used ``query-target=children``
— which the sim silently treated as unsupported and answered with empty
imdata (a false negative)."""
from __future__ import annotations
import copy
import pytest
from fastapi.testclient import TestClient
from aci_sim.build.orchestrator import build_site
from aci_sim.rest_aci.app import ApicSiteState, make_apic_app
from aci_sim.topology.loader import load_topology
TOPO_PATH = "topology.yaml"
TENANT = "VALTEST-T1"
BD_DN = f"uni/tn-{TENANT}/BD-bd-Val1"
@pytest.fixture(scope="module")
def client():
topo = load_topology(TOPO_PATH)
site = topo.sites[0]
store = build_site(topo, site)
state = ApicSiteState(
name=site.name, site=site, topo=topo,
store=store, baseline=copy.deepcopy(store),
)
c = TestClient(make_apic_app(state))
resp = c.post(
"/api/aaaLogin.json",
json={"aaaUser": {"attributes": {"name": "admin", "pwd": "cisco"}}},
)
assert resp.status_code == 200
_post(c, f"uni/tn-{TENANT}", {"fvTenant": {"attributes": {"name": TENANT}}})
_post(c, BD_DN, {"fvBD": {"attributes": {"name": "bd-Val1", "dn": BD_DN}}})
return c
def _post(client, dn: str, body: dict):
return client.post(f"/api/mo/{dn}.json", json=body)
def test_malformed_fvsubnet_ip_rejected_like_real_apic(client):
dn = f"{BD_DN}/subnet-[.1/24]"
r = _post(client, dn, {"fvSubnet": {"attributes": {"ip": ".1/24", "dn": dn}}})
assert r.status_code == 400
err = r.json()["imdata"][0]["error"]["attributes"]
assert ".1/24" in err["text"]
q = client.get("/api/class/fvSubnet.json").json()
assert not any(
".1/24" in i["fvSubnet"]["attributes"]["dn"] for i in q["imdata"]
), "malformed subnet must not land in the MIT"
def test_malformed_vnsredirectdest_rejected(client):
pol_dn = f"uni/tn-{TENANT}/svcCont/svcRedirectPol-pbr-Val"
r = _post(client, pol_dn, {"vnsSvcRedirectPol": {"attributes": {"name": "pbr-Val", "dn": pol_dn}}})
assert r.status_code == 200
dest_dn = f"{pol_dn}/RedirectDest_ip-[.]"
r = _post(client, dest_dn, {"vnsRedirectDest": {"attributes": {"ip": ".", "dn": dest_dn}}})
assert r.status_code == 400
def test_valid_subnet_accepted_and_delete_skips_validation(client):
sub_dn = f"{BD_DN}/subnet-[10.99.1.1/24]"
r = _post(client, sub_dn, {"fvSubnet": {"attributes": {"ip": "10.99.1.1/24", "dn": sub_dn}}})
assert r.status_code == 200
# A delete carries no ip attribute — validation must not block cleanup.
r = _post(client, sub_dn, {"fvSubnet": {"attributes": {"dn": sub_dn, "status": "deleted"}}})
assert r.status_code == 200
def test_query_target_children_returns_direct_children_flat(client):
# Re-create the subnet so the BD has a child to find.
sub_dn = f"{BD_DN}/subnet-[10.99.1.1/24]"
_post(client, sub_dn, {"fvSubnet": {"attributes": {"ip": "10.99.1.1/24", "dn": sub_dn}}})
r = client.get(f"/api/node/mo/uni/tn-{TENANT}.json?query-target=children")
data = r.json()
classes = [next(iter(i)) for i in data["imdata"]]
assert "fvBD" in classes, f"direct child missing: {classes}"
assert "fvTenant" not in classes, "root must be excluded"
r2 = client.get(
f"/api/node/mo/uni/tn-{TENANT}.json?query-target=children&target-subtree-class=fvBD"
)
classes2 = [next(iter(i)) for i in r2.json()["imdata"]]
assert classes2 and set(classes2) == {"fvBD"}
# Grandchildren (the subnet) must NOT appear — children is one level only.
r3 = client.get(f"/api/node/mo/{BD_DN}.json?query-target=children")
classes3 = [next(iter(i)) for i in r3.json()["imdata"]]
assert "fvSubnet" in classes3