|
b663b12…
|
lmata
|
1 |
"""Tests for KnowledgeIngester — manual knowledge curation.""" |
|
b663b12…
|
lmata
|
2 |
|
|
b663b12…
|
lmata
|
3 |
from unittest.mock import MagicMock |
|
b663b12…
|
lmata
|
4 |
|
|
b663b12…
|
lmata
|
5 |
import pytest |
|
b663b12…
|
lmata
|
6 |
|
|
b663b12…
|
lmata
|
7 |
from navegador.graph.schema import EdgeType, NodeLabel |
|
b663b12…
|
lmata
|
8 |
from navegador.ingestion.knowledge import KnowledgeIngester |
|
b663b12…
|
lmata
|
9 |
|
|
b663b12…
|
lmata
|
10 |
|
|
b663b12…
|
lmata
|
11 |
def _mock_store(): |
|
b663b12…
|
lmata
|
12 |
return MagicMock() |
|
b663b12…
|
lmata
|
13 |
|
|
b663b12…
|
lmata
|
14 |
|
|
b663b12…
|
lmata
|
15 |
class TestKnowledgeIngesterDomains: |
|
b663b12…
|
lmata
|
16 |
def test_add_domain(self): |
|
b663b12…
|
lmata
|
17 |
store = _mock_store() |
|
b663b12…
|
lmata
|
18 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
19 |
k.add_domain("auth", description="Authentication layer") |
|
b663b12…
|
lmata
|
20 |
store.create_node.assert_called_once_with( |
|
b663b12…
|
lmata
|
21 |
NodeLabel.Domain, {"name": "auth", "description": "Authentication layer"} |
|
b663b12…
|
lmata
|
22 |
) |
|
b663b12…
|
lmata
|
23 |
|
|
b663b12…
|
lmata
|
24 |
def test_add_domain_no_description(self): |
|
b663b12…
|
lmata
|
25 |
store = _mock_store() |
|
b663b12…
|
lmata
|
26 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
27 |
k.add_domain("billing") |
|
b663b12…
|
lmata
|
28 |
store.create_node.assert_called_once() |
|
b663b12…
|
lmata
|
29 |
|
|
b663b12…
|
lmata
|
30 |
|
|
b663b12…
|
lmata
|
31 |
class TestKnowledgeIngesterConcepts: |
|
b663b12…
|
lmata
|
32 |
def test_add_concept_minimal(self): |
|
b663b12…
|
lmata
|
33 |
store = _mock_store() |
|
b663b12…
|
lmata
|
34 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
35 |
k.add_concept("JWT") |
|
b663b12…
|
lmata
|
36 |
store.create_node.assert_called_once_with( |
|
b663b12…
|
lmata
|
37 |
NodeLabel.Concept, |
|
b663b12…
|
lmata
|
38 |
{"name": "JWT", "description": "", "domain": "", "status": "", |
|
b663b12…
|
lmata
|
39 |
"rules": "", "examples": "", "wiki_refs": ""} |
|
b663b12…
|
lmata
|
40 |
) |
|
b663b12…
|
lmata
|
41 |
|
|
b663b12…
|
lmata
|
42 |
def test_add_concept_with_domain_creates_edge(self): |
|
b663b12…
|
lmata
|
43 |
store = _mock_store() |
|
b663b12…
|
lmata
|
44 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
45 |
k.add_concept("JWT", domain="auth") |
|
b663b12…
|
lmata
|
46 |
# Should create Domain node + BELONGS_TO edge |
|
b663b12…
|
lmata
|
47 |
assert store.create_node.call_count == 2 |
|
b663b12…
|
lmata
|
48 |
store.create_edge.assert_called_once() |
|
b663b12…
|
lmata
|
49 |
edge_call = store.create_edge.call_args |
|
b663b12…
|
lmata
|
50 |
assert edge_call[0][2] == EdgeType.BELONGS_TO |
|
b663b12…
|
lmata
|
51 |
|
|
b663b12…
|
lmata
|
52 |
def test_relate_concepts(self): |
|
b663b12…
|
lmata
|
53 |
store = _mock_store() |
|
b663b12…
|
lmata
|
54 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
55 |
k.relate_concepts("JWT", "OAuth") |
|
b663b12…
|
lmata
|
56 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
57 |
NodeLabel.Concept, {"name": "JWT"}, |
|
b663b12…
|
lmata
|
58 |
EdgeType.RELATED_TO, |
|
b663b12…
|
lmata
|
59 |
NodeLabel.Concept, {"name": "OAuth"}, |
|
b663b12…
|
lmata
|
60 |
) |
|
b663b12…
|
lmata
|
61 |
|
|
b663b12…
|
lmata
|
62 |
|
|
b663b12…
|
lmata
|
63 |
class TestKnowledgeIngesterRules: |
|
b663b12…
|
lmata
|
64 |
def test_add_rule(self): |
|
b663b12…
|
lmata
|
65 |
store = _mock_store() |
|
b663b12…
|
lmata
|
66 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
67 |
k.add_rule("Tokens must expire", severity="critical") |
|
b663b12…
|
lmata
|
68 |
call_args = store.create_node.call_args[0] |
|
b663b12…
|
lmata
|
69 |
assert call_args[0] == NodeLabel.Rule |
|
b663b12…
|
lmata
|
70 |
assert call_args[1]["severity"] == "critical" |
|
b663b12…
|
lmata
|
71 |
|
|
b663b12…
|
lmata
|
72 |
def test_rule_governs(self): |
|
b663b12…
|
lmata
|
73 |
store = _mock_store() |
|
b663b12…
|
lmata
|
74 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
75 |
k.rule_governs("Tokens must expire", "JWT", NodeLabel.Concept) |
|
b663b12…
|
lmata
|
76 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
77 |
NodeLabel.Rule, {"name": "Tokens must expire"}, |
|
b663b12…
|
lmata
|
78 |
EdgeType.GOVERNS, |
|
b663b12…
|
lmata
|
79 |
NodeLabel.Concept, {"name": "JWT"}, |
|
b663b12…
|
lmata
|
80 |
) |
|
b663b12…
|
lmata
|
81 |
|
|
b663b12…
|
lmata
|
82 |
|
|
b663b12…
|
lmata
|
83 |
class TestKnowledgeIngesterDecisions: |
|
b663b12…
|
lmata
|
84 |
def test_add_decision(self): |
|
b663b12…
|
lmata
|
85 |
store = _mock_store() |
|
b663b12…
|
lmata
|
86 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
87 |
k.add_decision("Use JWT", status="accepted", rationale="Horizontal scaling") |
|
b663b12…
|
lmata
|
88 |
call_args = store.create_node.call_args[0] |
|
b663b12…
|
lmata
|
89 |
assert call_args[0] == NodeLabel.Decision |
|
b663b12…
|
lmata
|
90 |
assert call_args[1]["status"] == "accepted" |
|
b663b12…
|
lmata
|
91 |
assert call_args[1]["rationale"] == "Horizontal scaling" |
|
b663b12…
|
lmata
|
92 |
|
|
b663b12…
|
lmata
|
93 |
|
|
b663b12…
|
lmata
|
94 |
class TestKnowledgeIngesterPeople: |
|
b663b12…
|
lmata
|
95 |
def test_add_person(self): |
|
b663b12…
|
lmata
|
96 |
store = _mock_store() |
|
b663b12…
|
lmata
|
97 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
98 |
k.add_person("Alice", email="[email protected]", role="lead", team="auth") |
|
b663b12…
|
lmata
|
99 |
store.create_node.assert_called_once_with( |
|
b663b12…
|
lmata
|
100 |
NodeLabel.Person, |
|
b663b12…
|
lmata
|
101 |
{"name": "Alice", "email": "[email protected]", "role": "lead", "team": "auth"} |
|
b663b12…
|
lmata
|
102 |
) |
|
b663b12…
|
lmata
|
103 |
|
|
b663b12…
|
lmata
|
104 |
def test_assign(self): |
|
b663b12…
|
lmata
|
105 |
store = _mock_store() |
|
b663b12…
|
lmata
|
106 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
107 |
k.assign("validate_token", NodeLabel.Function, "Alice") |
|
b663b12…
|
lmata
|
108 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
109 |
NodeLabel.Function, {"name": "validate_token"}, |
|
b663b12…
|
lmata
|
110 |
EdgeType.ASSIGNED_TO, |
|
b663b12…
|
lmata
|
111 |
NodeLabel.Person, {"name": "Alice"}, |
|
b663b12…
|
lmata
|
112 |
) |
|
b663b12…
|
lmata
|
113 |
|
|
b663b12…
|
lmata
|
114 |
|
|
b663b12…
|
lmata
|
115 |
class TestKnowledgeIngesterWiki: |
|
b663b12…
|
lmata
|
116 |
def test_wiki_page(self): |
|
b663b12…
|
lmata
|
117 |
store = _mock_store() |
|
b663b12…
|
lmata
|
118 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
119 |
k.wiki_page("Auth Guide", url="https://example.com", source="github", content="# Auth") |
|
b663b12…
|
lmata
|
120 |
call_args = store.create_node.call_args[0] |
|
b663b12…
|
lmata
|
121 |
assert call_args[0] == NodeLabel.WikiPage |
|
b663b12…
|
lmata
|
122 |
assert call_args[1]["name"] == "Auth Guide" |
|
b663b12…
|
lmata
|
123 |
|
|
b663b12…
|
lmata
|
124 |
def test_wiki_documents(self): |
|
b663b12…
|
lmata
|
125 |
store = _mock_store() |
|
b663b12…
|
lmata
|
126 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
127 |
k.wiki_documents("Auth Guide", "JWT", {"name": "JWT"}, NodeLabel.Concept) |
|
b663b12…
|
lmata
|
128 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
129 |
NodeLabel.WikiPage, {"name": "Auth Guide"}, |
|
b663b12…
|
lmata
|
130 |
EdgeType.DOCUMENTS, |
|
b663b12…
|
lmata
|
131 |
NodeLabel.Concept, {"name": "JWT"}, |
|
b663b12…
|
lmata
|
132 |
) |
|
b663b12…
|
lmata
|
133 |
|
|
b663b12…
|
lmata
|
134 |
|
|
b663b12…
|
lmata
|
135 |
class TestKnowledgeIngesterAnnotate: |
|
b663b12…
|
lmata
|
136 |
def test_annotate_with_concept(self): |
|
b663b12…
|
lmata
|
137 |
store = _mock_store() |
|
b663b12…
|
lmata
|
138 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
139 |
k.annotate_code("validate_token", "Function", concept="JWT") |
|
b663b12…
|
lmata
|
140 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
141 |
NodeLabel.Concept, {"name": "JWT"}, |
|
b663b12…
|
lmata
|
142 |
EdgeType.ANNOTATES, |
|
b663b12…
|
lmata
|
143 |
NodeLabel.Function, {"name": "validate_token"}, |
|
b663b12…
|
lmata
|
144 |
) |
|
b663b12…
|
lmata
|
145 |
|
|
b663b12…
|
lmata
|
146 |
def test_annotate_with_rule(self): |
|
b663b12…
|
lmata
|
147 |
store = _mock_store() |
|
b663b12…
|
lmata
|
148 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
149 |
k.annotate_code("validate_token", "Function", rule="Tokens must expire") |
|
b663b12…
|
lmata
|
150 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
151 |
NodeLabel.Rule, {"name": "Tokens must expire"}, |
|
b663b12…
|
lmata
|
152 |
EdgeType.ANNOTATES, |
|
b663b12…
|
lmata
|
153 |
NodeLabel.Function, {"name": "validate_token"}, |
|
b663b12…
|
lmata
|
154 |
) |
|
b663b12…
|
lmata
|
155 |
|
|
b663b12…
|
lmata
|
156 |
def test_annotate_with_both(self): |
|
b663b12…
|
lmata
|
157 |
store = _mock_store() |
|
b663b12…
|
lmata
|
158 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
159 |
k.annotate_code("validate_token", "Function", concept="JWT", rule="Tokens must expire") |
|
b663b12…
|
lmata
|
160 |
assert store.create_edge.call_count == 2 |
|
b663b12…
|
lmata
|
161 |
|
|
b663b12…
|
lmata
|
162 |
def test_code_implements(self): |
|
b663b12…
|
lmata
|
163 |
store = _mock_store() |
|
b663b12…
|
lmata
|
164 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
165 |
k.code_implements("validate_token", "Function", "JWT") |
|
b663b12…
|
lmata
|
166 |
store.create_edge.assert_called_once_with( |
|
b663b12…
|
lmata
|
167 |
NodeLabel.Function, {"name": "validate_token"}, |
|
b663b12…
|
lmata
|
168 |
EdgeType.IMPLEMENTS, |
|
b663b12…
|
lmata
|
169 |
NodeLabel.Concept, {"name": "JWT"}, |
|
b663b12…
|
lmata
|
170 |
) |
|
b663b12…
|
lmata
|
171 |
|
|
b663b12…
|
lmata
|
172 |
def test_annotate_invalid_label_raises(self): |
|
b663b12…
|
lmata
|
173 |
store = _mock_store() |
|
b663b12…
|
lmata
|
174 |
k = KnowledgeIngester(store) |
|
b663b12…
|
lmata
|
175 |
with pytest.raises(ValueError): |
|
b663b12…
|
lmata
|
176 |
k.annotate_code("foo", "InvalidLabel", concept="Bar") |
|
7ae0080…
|
lmata
|
177 |
|
|
7ae0080…
|
lmata
|
178 |
|
|
7ae0080…
|
lmata
|
179 |
class TestKnowledgeIngesterRuleWithDomain: |
|
7ae0080…
|
lmata
|
180 |
def test_add_rule_with_domain_creates_link(self): |
|
7ae0080…
|
lmata
|
181 |
from navegador.graph.schema import EdgeType |
|
7ae0080…
|
lmata
|
182 |
store = _mock_store() |
|
7ae0080…
|
lmata
|
183 |
k = KnowledgeIngester(store) |
|
7ae0080…
|
lmata
|
184 |
k.add_rule("Tokens must expire", domain="auth", severity="critical") |
|
7ae0080…
|
lmata
|
185 |
# create_node called twice: once for rule, once for domain |
|
7ae0080…
|
lmata
|
186 |
assert store.create_node.call_count == 2 |
|
7ae0080…
|
lmata
|
187 |
# BELONGS_TO edge created |
|
7ae0080…
|
lmata
|
188 |
edge_calls = [c[0][2] for c in store.create_edge.call_args_list] |
|
7ae0080…
|
lmata
|
189 |
assert EdgeType.BELONGS_TO in edge_calls |
|
7ae0080…
|
lmata
|
190 |
|
|
7ae0080…
|
lmata
|
191 |
|
|
7ae0080…
|
lmata
|
192 |
class TestKnowledgeIngesterDecisionWithDomain: |
|
7ae0080…
|
lmata
|
193 |
def test_add_decision_with_domain_creates_link(self): |
|
7ae0080…
|
lmata
|
194 |
from navegador.graph.schema import EdgeType |
|
7ae0080…
|
lmata
|
195 |
store = _mock_store() |
|
7ae0080…
|
lmata
|
196 |
k = KnowledgeIngester(store) |
|
7ae0080…
|
lmata
|
197 |
k.add_decision("Use PostgreSQL", domain="infra", status="accepted") |
|
7ae0080…
|
lmata
|
198 |
assert store.create_node.call_count == 2 |
|
7ae0080…
|
lmata
|
199 |
edge_calls = [c[0][2] for c in store.create_edge.call_args_list] |
|
7ae0080…
|
lmata
|
200 |
assert EdgeType.BELONGS_TO in edge_calls |