generated from coulomb/repo-seed
291 lines
10 KiB
Python
291 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from railiance_fabric.zone_view import ZoneDefinition, resolve_zones
|
|
|
|
|
|
def _node(node_id: str, **data: object) -> dict[str, dict[str, object]]:
|
|
return {"data": {"id": node_id, **data}}
|
|
|
|
|
|
def _edge(edge_id: str, source: str, target: str, edge_type: str, **data: object) -> dict[str, dict[str, object]]:
|
|
return {
|
|
"data": {
|
|
"id": edge_id,
|
|
"source": source,
|
|
"target": target,
|
|
"edgeType": edge_type,
|
|
**data,
|
|
}
|
|
}
|
|
|
|
|
|
def test_zone_definition_round_trips() -> None:
|
|
definition = ZoneDefinition.from_dict(
|
|
{
|
|
"id": "prod",
|
|
"label": "Production",
|
|
"membership": {
|
|
"rules": [
|
|
{"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
{"field": "kind", "op": "in", "value": ["service", "endpoint"]},
|
|
]
|
|
},
|
|
"attraction": {
|
|
"rules": [
|
|
{
|
|
"edge_type": "routes_to_service",
|
|
"direction": "out",
|
|
"depth": 2,
|
|
"node_filter": {"field": "kind", "op": "exists"},
|
|
}
|
|
]
|
|
},
|
|
"layout": {"algorithm": "cose", "options": {"animate": False}},
|
|
"presentation": {"height": 30, "color": "#be123c", "opacity": 0.2, "blur_below": True},
|
|
"collapse": {"enabled": True, "label": "Production Zone"},
|
|
}
|
|
)
|
|
|
|
assert definition.id == "prod"
|
|
assert definition.label == "Production"
|
|
assert definition.attraction_rules[0].edge_type == "routes_to_service"
|
|
assert definition.attraction_rules[0].depth == 2
|
|
assert definition.layout.algorithm == "cose"
|
|
assert definition.presentation.height == 30
|
|
assert definition.collapse.enabled is True
|
|
|
|
round_tripped = ZoneDefinition.from_dict(definition.to_dict())
|
|
|
|
assert round_tripped == definition
|
|
|
|
|
|
def test_resolver_assigns_seed_nodes_and_boundary_edges() -> None:
|
|
resolution = resolve_zones(
|
|
nodes=[
|
|
_node("svc.prod", deploymentEnvironment="prod"),
|
|
_node("svc.test", deploymentEnvironment="test"),
|
|
],
|
|
edges=[
|
|
_edge("edge.prod-test", "svc.prod", "svc.test", "routes_to_service"),
|
|
],
|
|
zone_definitions=[
|
|
{
|
|
"id": "prod",
|
|
"label": "prod",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
"presentation": {"height": 20},
|
|
},
|
|
{
|
|
"id": "test",
|
|
"label": "test",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "test"},
|
|
"presentation": {"height": 10},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert resolution.node_assignments["svc.prod"].zone_id == "prod"
|
|
assert resolution.node_assignments["svc.prod"].reason == "seed"
|
|
assert resolution.zone_by_id("prod").seed_node_ids == ("svc.prod",)
|
|
assert resolution.zone_by_id("test").seed_node_ids == ("svc.test",)
|
|
assert resolution.boundary_edges[0].edge_id == "edge.prod-test"
|
|
assert resolution.boundary_edges[0].source_zone_id == "prod"
|
|
assert resolution.boundary_edges[0].target_zone_id == "test"
|
|
assert "ZONE_EDGE_CROSSES_ZONE_BOUNDARY" in {
|
|
diagnostic.code for diagnostic in resolution.diagnostics
|
|
}
|
|
|
|
|
|
def test_resolver_ignores_context_only_edges_for_boundaries_and_attraction() -> None:
|
|
resolution = resolve_zones(
|
|
nodes=[
|
|
_node("repo", kind="Repository"),
|
|
_node("svc.prod", deploymentEnvironment="prod"),
|
|
_node("svc.context", kind="service"),
|
|
],
|
|
edges=[
|
|
_edge("edge.repo-prod", "repo", "svc.prod", "declares", displayOnly=True),
|
|
_edge("edge.prod-context", "svc.prod", "svc.context", "declares", displayOnly=True),
|
|
],
|
|
zone_definitions=[
|
|
{
|
|
"id": "prod",
|
|
"label": "prod",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
"attraction": {
|
|
"rules": [
|
|
{
|
|
"edge_type": "*",
|
|
"direction": "both",
|
|
"depth": 1,
|
|
}
|
|
]
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert resolution.zone_by_id("prod").boundary_edge_ids == ()
|
|
assert resolution.zone_by_id("prod").internal_edge_ids == ()
|
|
assert "svc.context" not in resolution.node_assignments
|
|
assert "ZONE_EDGE_CROSSES_ZONE_BOUNDARY" not in {
|
|
diagnostic.code for diagnostic in resolution.diagnostics
|
|
}
|
|
|
|
|
|
def test_resolver_attracts_nodes_by_edge_type_direction_and_depth() -> None:
|
|
resolution = resolve_zones(
|
|
nodes=[
|
|
_node("seed", deploymentEnvironment="prod", kind="service"),
|
|
_node("near", kind="endpoint"),
|
|
_node("far", kind="endpoint"),
|
|
],
|
|
edges=[
|
|
_edge("edge.seed-near", "seed", "near", "routes_to_service"),
|
|
_edge("edge.near-far", "near", "far", "routes_to_service"),
|
|
],
|
|
zone_definitions=[
|
|
{
|
|
"id": "prod",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
"attraction": {
|
|
"rules": [
|
|
{
|
|
"edge_type": "routes_to_service",
|
|
"direction": "out",
|
|
"depth": 1,
|
|
"node_filter": {"field": "kind", "op": "exists"},
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert resolution.node_assignments["near"].zone_id == "prod"
|
|
assert resolution.node_assignments["near"].reason == "attraction"
|
|
assert resolution.node_assignments["near"].depth == 1
|
|
assert "far" not in resolution.node_assignments
|
|
assert resolution.zone_by_id("prod").internal_edge_ids == ("edge.seed-near",)
|
|
assert resolution.zone_by_id("prod").boundary_edge_ids == ("edge.near-far",)
|
|
assert "ZONE_ATTRACTION_DEPTH_LIMIT_REACHED" in {
|
|
diagnostic.code for diagnostic in resolution.diagnostics
|
|
}
|
|
|
|
|
|
def test_resolver_keeps_seed_membership_over_attraction() -> None:
|
|
resolution = resolve_zones(
|
|
nodes=[
|
|
_node("seed", deploymentEnvironment="prod", kind="service"),
|
|
_node("tenant", deploymentEnvironment="tenant", kind="endpoint"),
|
|
],
|
|
edges=[
|
|
_edge("edge.seed-tenant", "seed", "tenant", "routes_to_service"),
|
|
],
|
|
zone_definitions=[
|
|
{
|
|
"id": "prod",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
"attraction": {
|
|
"rules": [
|
|
{
|
|
"edge_type": "routes_to_service",
|
|
"direction": "out",
|
|
"depth": 1,
|
|
}
|
|
]
|
|
},
|
|
"presentation": {"height": 100},
|
|
},
|
|
{
|
|
"id": "tenant",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "tenant"},
|
|
"presentation": {"height": 0},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert resolution.node_assignments["tenant"].zone_id == "tenant"
|
|
assert resolution.node_assignments["tenant"].reason == "seed"
|
|
assert "ZONE_SEED_OVERRIDES_ATTRACTION" in {
|
|
diagnostic.code for diagnostic in resolution.diagnostics
|
|
}
|
|
|
|
|
|
def test_resolver_uses_height_then_definition_order_for_overlapping_membership() -> None:
|
|
height_resolution = resolve_zones(
|
|
nodes=[_node("shared", deploymentEnvironment="prod", owner="lord")],
|
|
edges=[],
|
|
zone_definitions=[
|
|
{
|
|
"id": "lower",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
"presentation": {"height": 10},
|
|
},
|
|
{
|
|
"id": "higher",
|
|
"membership": {"field": "owner", "op": "equals", "value": "lord"},
|
|
"presentation": {"height": 20},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert height_resolution.node_assignments["shared"].zone_id == "higher"
|
|
assert "ZONE_NODE_SEEDED_BY_MULTIPLE_ZONES" in {
|
|
diagnostic.code for diagnostic in height_resolution.diagnostics
|
|
}
|
|
|
|
order_resolution = resolve_zones(
|
|
nodes=[_node("shared", deploymentEnvironment="prod", owner="lord")],
|
|
edges=[],
|
|
zone_definitions=[
|
|
{
|
|
"id": "first",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
"presentation": {"height": 10},
|
|
},
|
|
{
|
|
"id": "second",
|
|
"membership": {"field": "owner", "op": "equals", "value": "lord"},
|
|
"presentation": {"height": 10},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert order_resolution.node_assignments["shared"].zone_id == "first"
|
|
|
|
|
|
def test_resolver_serializes_resolution() -> None:
|
|
resolution = resolve_zones(
|
|
nodes=[_node("svc", deploymentEnvironment="prod")],
|
|
edges=[],
|
|
zone_definitions=[
|
|
{
|
|
"id": "prod",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
}
|
|
],
|
|
)
|
|
|
|
serialized = resolution.to_dict()
|
|
|
|
assert serialized["zones"][0]["id"] == "prod"
|
|
assert serialized["node_assignments"]["svc"]["zone_id"] == "prod"
|
|
assert serialized["node_assignments"]["svc"]["reason"] == "seed"
|
|
|
|
|
|
def test_resolver_reports_empty_zone_seed_set() -> None:
|
|
resolution = resolve_zones(
|
|
nodes=[_node("svc", deploymentEnvironment="dev")],
|
|
edges=[],
|
|
zone_definitions=[
|
|
{
|
|
"id": "prod",
|
|
"membership": {"field": "deploymentEnvironment", "op": "equals", "value": "prod"},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert resolution.zone_by_id("prod").node_ids == ()
|
|
assert "ZONE_EMPTY_SEED_SET" in {diagnostic.code for diagnostic in resolution.diagnostics}
|