File size: 5,776 Bytes
50f4235
e382248
50f4235
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e382248
50f4235
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
"""
Tests for ConsistencyValidator - cross-agent contradiction detection.
"""

import pytest

from app.core.consistency_validator import ConsistencyValidator
from app.core.kg_schemas import (
    Component,
    Requirement,
)
from app.core.kg_service import KGService, reset_kg_service


@pytest.fixture(autouse=True)
def reset_service():
    """Reset KGService singleton between tests."""
    reset_kg_service()
    yield


@pytest.fixture
def empty_graph_dict() -> dict:
    return {"nodes": [], "edges": [], "kg_schema_version": 2}


@pytest.fixture
def validator() -> ConsistencyValidator:
    return ConsistencyValidator()


def test_no_contradictions_empty_graph(validator):
    """Empty knowledge graph returns no contradictions."""
    state = {"knowledge_graph": {}}
    assert validator.validate_all(state) == []


def test_no_contradictions_single_entity(validator, empty_graph_dict):
    """Single entity produces no contradictions."""
    svc = KGService()
    state: dict = {"knowledge_graph": {}}
    g = empty_graph_dict
    g = svc.add_entity(
        g, Requirement(id="req-1", role="product_owner", content="Test requirement")
    )
    state["knowledge_graph"] = svc.set_role_graph(state, "product_owner", g)

    result = validator.validate_all(state)
    assert result == []


def test_detects_property_conflict(validator, empty_graph_dict):
    """SA and DA write Component with same name but different technology."""
    svc = KGService()
    state: dict = {"knowledge_graph": {}}

    # SA writes Component(name="AuthService", technology="FastAPI")
    g_sa = empty_graph_dict
    g_sa = svc.add_entity(
        g_sa,
        Component(
            id="comp-1",
            role="solution_architect",
            name="AuthService",
            technology="FastAPI",
            content="Auth service component",
        ),
    )
    state["knowledge_graph"] = svc.set_role_graph(state, "solution_architect", g_sa)

    # DA writes Component(name="AuthService", technology="Express.js")
    g_da = empty_graph_dict
    g_da = svc.add_entity(
        g_da,
        Component(
            id="comp-2",
            role="data_architect",
            name="AuthService",
            technology="Express.js",
            content="Auth service data layer",
        ),
    )
    state["knowledge_graph"] = svc.set_role_graph(state, "data_architect", g_da)

    result = validator.validate_all(state)
    assert len(result) >= 1
    assert any(c["type"] == "property_conflict" for c in result)
    found = [c for c in result if c["type"] == "property_conflict"]
    conflict_detail = found[0].get("detail", "")
    assert "AuthService" in conflict_detail


def test_detects_missing_dependency(validator, empty_graph_dict):
    """SA writes Component referencing PaymentService with no entity for it."""
    svc = KGService()
    state: dict = {"knowledge_graph": {}}

    g_sa = empty_graph_dict
    g_sa = svc.add_entity(
        g_sa,
        Component(
            id="comp-1",
            role="solution_architect",
            name="OrderService",
            technology="FastAPI",
            content="Order service depends on PaymentService for processing",
        ),
    )
    state["knowledge_graph"] = svc.set_role_graph(state, "solution_architect", g_sa)

    result = validator.validate_all(state)
    assert len(result) >= 1
    assert any(c["type"] == "missing_dependency" for c in result)
    found = [c for c in result if c["type"] == "missing_dependency"]
    assert "PaymentService" in found[0].get("detail", "")


def test_no_false_positive_for_same_role(validator, empty_graph_dict):
    """Two Requirements from same role are NOT contradictions."""
    svc = KGService()
    state: dict = {"knowledge_graph": {}}

    g_po = empty_graph_dict
    g_po = svc.add_entity(
        g_po,
        Requirement(id="req-1", role="product_owner", content="Requirement one"),
    )
    g_po = svc.add_entity(
        g_po,
        Requirement(id="req-2", role="product_owner", content="Requirement two"),
    )
    state["knowledge_graph"] = svc.set_role_graph(state, "product_owner", g_po)

    result = validator.validate_all(state)
    # Same-role entities should not produce contradictions
    for c in result:
        roles = c.get("roles", [])
        if len(roles) == 1 and "product_owner" in roles:
            pytest.fail(f"Same-role contradiction detected: {c}")
    # This is fine - at minimum no property_conflicts should appear
    property_conflicts = [c for c in result if c["type"] == "property_conflict"]
    assert property_conflicts == []


def test_format_warnings_section(validator):
    """Verify formatted output contains expected content."""
    contradictions = [
        {
            "type": "property_conflict",
            "entity_ids": ["comp-1", "comp-2"],
            "roles": ["solution_architect", "data_architect"],
            "detail": "Conflicting properties for Component:AuthService",
            "severity": "high",
            "conflicts": [
                {
                    "property": "Technology stack",
                    "values": {
                        "solution_architect": "FastAPI",
                        "data_architect": "Express.js",
                    },
                }
            ],
        }
    ]

    output = validator.format_warnings_section(contradictions)
    assert "Consistency Warnings" in output
    assert "AuthService" in output
    assert "FastAPI" in output
    assert "Express.js" in output
    assert "solution_architect" in output
    assert "data_architect" in output
    assert "high" in output


def test_format_warnings_section_empty(validator):
    """Empty contradictions produce empty string."""
    assert validator.format_warnings_section([]) == ""