|
| 1 | +"""Flow view edge case tests — degenerate graphs, boundary conditions.""" |
| 2 | + |
| 3 | +import pytest |
| 4 | + |
| 5 | +from code_intelligence.flow.engine import FlowEngine, AVAILABLE_VIEWS |
| 6 | +from code_intelligence.flow.models import FlowDiagram |
| 7 | +from code_intelligence.flow.renderer import render_mermaid, render_json, render_html |
| 8 | +from code_intelligence.graph.store import GraphStore |
| 9 | +from code_intelligence.models.graph import GraphNode, GraphEdge, NodeKind, EdgeKind |
| 10 | + |
| 11 | + |
| 12 | +class TestEmptyGraph: |
| 13 | + """All views should handle an empty graph gracefully.""" |
| 14 | + |
| 15 | + def test_overview_empty(self): |
| 16 | + d = FlowEngine(GraphStore()).generate("overview") |
| 17 | + assert isinstance(d, FlowDiagram) |
| 18 | + assert d.view == "overview" |
| 19 | + assert len(d.all_nodes()) == 0 |
| 20 | + |
| 21 | + def test_all_views_empty(self): |
| 22 | + engine = FlowEngine(GraphStore()) |
| 23 | + for view in AVAILABLE_VIEWS: |
| 24 | + d = engine.generate(view) |
| 25 | + assert isinstance(d, FlowDiagram) |
| 26 | + assert d.view == view |
| 27 | + |
| 28 | + def test_render_mermaid_empty(self): |
| 29 | + d = FlowEngine(GraphStore()).generate("overview") |
| 30 | + mmd = render_mermaid(d) |
| 31 | + assert "graph" in mmd |
| 32 | + assert isinstance(mmd, str) |
| 33 | + |
| 34 | + def test_render_json_empty(self): |
| 35 | + d = FlowEngine(GraphStore()).generate("overview") |
| 36 | + j = render_json(d) |
| 37 | + import json |
| 38 | + data = json.loads(j) |
| 39 | + assert data["view"] == "overview" |
| 40 | + |
| 41 | + def test_render_html_empty(self): |
| 42 | + html = FlowEngine(GraphStore()).render_interactive() |
| 43 | + assert "<!DOCTYPE html>" in html |
| 44 | + |
| 45 | + |
| 46 | +class TestSingleNode: |
| 47 | + """Graph with exactly 1 node.""" |
| 48 | + |
| 49 | + @pytest.fixture |
| 50 | + def single_endpoint_store(self): |
| 51 | + s = GraphStore() |
| 52 | + s.add_node(GraphNode(id="ep1", kind=NodeKind.ENDPOINT, label="GET /health")) |
| 53 | + return s |
| 54 | + |
| 55 | + def test_overview_single_endpoint(self, single_endpoint_store): |
| 56 | + d = FlowEngine(single_endpoint_store).generate("overview") |
| 57 | + assert len(d.all_nodes()) >= 1 |
| 58 | + assert d.stats.get("endpoints", 0) == 1 |
| 59 | + |
| 60 | + def test_auth_single_endpoint_unprotected(self, single_endpoint_store): |
| 61 | + d = FlowEngine(single_endpoint_store).generate("auth") |
| 62 | + # Should show 1 unprotected endpoint |
| 63 | + unprotected = [n for n in d.all_nodes() if n.style == "danger"] |
| 64 | + assert len(unprotected) >= 1 |
| 65 | + |
| 66 | + def test_runtime_single_endpoint(self, single_endpoint_store): |
| 67 | + d = FlowEngine(single_endpoint_store).generate("runtime") |
| 68 | + assert isinstance(d, FlowDiagram) |
| 69 | + |
| 70 | + |
| 71 | +class TestInfraOnly: |
| 72 | + """Graph with only infrastructure nodes (no app code).""" |
| 73 | + |
| 74 | + @pytest.fixture |
| 75 | + def infra_store(self): |
| 76 | + s = GraphStore() |
| 77 | + for i in range(5): |
| 78 | + s.add_node(GraphNode(id=f"k8s:default/deploy-{i}", kind=NodeKind.INFRA_RESOURCE, |
| 79 | + label=f"Deployment {i}", properties={"kind": "Deployment"})) |
| 80 | + s.add_node(GraphNode(id="k8s:default/svc-0", kind=NodeKind.INFRA_RESOURCE, |
| 81 | + label="Service 0", properties={"kind": "Service"})) |
| 82 | + s.add_edge(GraphEdge(source="k8s:default/svc-0", target="k8s:default/deploy-0", kind=EdgeKind.CONNECTS_TO)) |
| 83 | + return s |
| 84 | + |
| 85 | + def test_overview_infra_only(self, infra_store): |
| 86 | + d = FlowEngine(infra_store).generate("overview") |
| 87 | + # Should have infra subgraph, no app subgraph |
| 88 | + infra_sgs = [sg for sg in d.subgraphs if "infra" in sg.id.lower() or "deploy" in sg.label.lower()] |
| 89 | + assert len(infra_sgs) >= 1 |
| 90 | + |
| 91 | + def test_deploy_view_infra(self, infra_store): |
| 92 | + d = FlowEngine(infra_store).generate("deploy") |
| 93 | + assert len(d.all_nodes()) >= 1 |
| 94 | + |
| 95 | + def test_runtime_empty_for_infra(self, infra_store): |
| 96 | + d = FlowEngine(infra_store).generate("runtime") |
| 97 | + # Runtime view should still work (may be empty or minimal) |
| 98 | + assert isinstance(d, FlowDiagram) |
| 99 | + |
| 100 | + |
| 101 | +class TestAuthOnly: |
| 102 | + """Graph with guards but no endpoints.""" |
| 103 | + |
| 104 | + @pytest.fixture |
| 105 | + def guards_no_endpoints(self): |
| 106 | + s = GraphStore() |
| 107 | + s.add_node(GraphNode(id="g1", kind=NodeKind.GUARD, label="JwtGuard", properties={"auth_type": "jwt"})) |
| 108 | + s.add_node(GraphNode(id="g2", kind=NodeKind.GUARD, label="RoleGuard", properties={"auth_type": "rbac"})) |
| 109 | + return s |
| 110 | + |
| 111 | + def test_auth_view_guards_only(self, guards_no_endpoints): |
| 112 | + d = FlowEngine(guards_no_endpoints).generate("auth") |
| 113 | + guard_nodes = [n for n in d.all_nodes() if n.kind == "guard"] |
| 114 | + assert len(guard_nodes) >= 1 |
| 115 | + # No endpoint subgraph since there are no endpoints |
| 116 | + assert d.stats.get("coverage_pct", 0) == 0 |
| 117 | + |
| 118 | + |
| 119 | +class TestCIOnly: |
| 120 | + """Graph with CI nodes but nothing else.""" |
| 121 | + |
| 122 | + @pytest.fixture |
| 123 | + def ci_store(self): |
| 124 | + s = GraphStore() |
| 125 | + s.add_node(GraphNode(id="gha:ci.yml", kind=NodeKind.MODULE, label="CI Workflow")) |
| 126 | + s.add_node(GraphNode(id="gha:ci.yml:job:build", kind=NodeKind.METHOD, label="build")) |
| 127 | + s.add_node(GraphNode(id="gha:ci.yml:job:test", kind=NodeKind.METHOD, label="test")) |
| 128 | + s.add_edge(GraphEdge(source="gha:ci.yml", target="gha:ci.yml:job:build", kind=EdgeKind.CONTAINS)) |
| 129 | + s.add_edge(GraphEdge(source="gha:ci.yml", target="gha:ci.yml:job:test", kind=EdgeKind.CONTAINS)) |
| 130 | + s.add_edge(GraphEdge(source="gha:ci.yml:job:test", target="gha:ci.yml:job:build", kind=EdgeKind.DEPENDS_ON)) |
| 131 | + return s |
| 132 | + |
| 133 | + def test_overview_ci_only(self, ci_store): |
| 134 | + d = FlowEngine(ci_store).generate("overview") |
| 135 | + ci_sgs = [sg for sg in d.subgraphs if sg.drill_down_view == "ci"] |
| 136 | + assert len(ci_sgs) >= 1 |
| 137 | + |
| 138 | + def test_ci_view_shows_jobs(self, ci_store): |
| 139 | + d = FlowEngine(ci_store).generate("ci") |
| 140 | + assert len(d.all_nodes()) >= 2 # At least the 2 jobs |
| 141 | + |
| 142 | + |
| 143 | +class TestLargeGraph: |
| 144 | + """Graph with thousands of nodes — flow views should still be small.""" |
| 145 | + |
| 146 | + @pytest.fixture |
| 147 | + def large_store(self): |
| 148 | + s = GraphStore() |
| 149 | + for i in range(5000): |
| 150 | + s.add_node(GraphNode(id=f"method_{i}", kind=NodeKind.METHOD, label=f"method{i}")) |
| 151 | + for i in range(100): |
| 152 | + s.add_node(GraphNode(id=f"ep_{i}", kind=NodeKind.ENDPOINT, label=f"GET /api/{i}")) |
| 153 | + s.add_node(GraphNode(id=f"ent_{i}", kind=NodeKind.ENTITY, label=f"Entity{i}")) |
| 154 | + return s |
| 155 | + |
| 156 | + def test_overview_max_nodes(self, large_store): |
| 157 | + d = FlowEngine(large_store).generate("overview") |
| 158 | + assert len(d.all_nodes()) <= 30 # Views should collapse, not explode |
| 159 | + |
| 160 | + def test_runtime_max_nodes(self, large_store): |
| 161 | + d = FlowEngine(large_store).generate("runtime") |
| 162 | + assert len(d.all_nodes()) <= 30 |
| 163 | + |
| 164 | + |
| 165 | +class TestRenderEdgeCases: |
| 166 | + """Renderer edge cases.""" |
| 167 | + |
| 168 | + def test_mermaid_special_chars_in_labels(self): |
| 169 | + s = GraphStore() |
| 170 | + s.add_node(GraphNode(id="n1", kind=NodeKind.ENDPOINT, label='GET /users?name="foo"&age=<30>')) |
| 171 | + d = FlowEngine(s).generate("overview") |
| 172 | + mmd = render_mermaid(d) |
| 173 | + assert "&" not in mmd or "&#" in mmd # Should be escaped |
| 174 | + |
| 175 | + def test_html_with_all_views(self): |
| 176 | + s = GraphStore() |
| 177 | + s.add_node(GraphNode(id="ep1", kind=NodeKind.ENDPOINT, label="GET /")) |
| 178 | + s.add_node(GraphNode(id="g1", kind=NodeKind.GUARD, label="Auth")) |
| 179 | + s.add_edge(GraphEdge(source="g1", target="ep1", kind=EdgeKind.PROTECTS)) |
| 180 | + html = FlowEngine(s).render_interactive() |
| 181 | + # Should contain data for all 5 views |
| 182 | + for view in AVAILABLE_VIEWS: |
| 183 | + assert view in html |
| 184 | + |
| 185 | + |
| 186 | +class TestDeterminism: |
| 187 | + """All views must be deterministic.""" |
| 188 | + |
| 189 | + def test_all_views_deterministic(self): |
| 190 | + s = GraphStore() |
| 191 | + for i in range(50): |
| 192 | + s.add_node(GraphNode(id=f"n{i}", kind=NodeKind.METHOD, label=f"m{i}")) |
| 193 | + for i in range(10): |
| 194 | + s.add_node(GraphNode(id=f"ep{i}", kind=NodeKind.ENDPOINT, label=f"E{i}")) |
| 195 | + s.add_node(GraphNode(id=f"g{i}", kind=NodeKind.GUARD, label=f"G{i}", properties={"auth_type": "jwt"})) |
| 196 | + |
| 197 | + engine = FlowEngine(s) |
| 198 | + for view in AVAILABLE_VIEWS: |
| 199 | + d1 = engine.generate(view) |
| 200 | + d2 = engine.generate(view) |
| 201 | + assert render_mermaid(d1) == render_mermaid(d2), f"Non-deterministic: {view}" |
0 commit comments