Spaces:
Sleeping
Sleeping
File size: 5,346 Bytes
a746fba | 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 | """Tests for QualityEngine, DomainTermWhitelist, and quality schemas."""
import pytest
from app.core.quality_engine import QualityEngine, DomainTermWhitelist
from app.core.schemas import QualityReport, ARIResult, RePromptAttempt
# Verify textstat is installed
textstat = pytest.importorskip("textstat")
# ── QualityReport schema ──
def test_quality_report_defaults():
"""QualityReport has sensible defaults and validates correctly."""
report = QualityReport(
role="test", agent_output="Hello world", ari_before_any=5.0, final_ari=5.0
)
assert report.role == "test"
assert report.re_prompt_count == 0
assert report.final_disposition == "passed"
assert report.is_budget_met() # No dimensions = vacuously true
def test_quality_report_with_dimensions():
"""is_budget_met checks budget-named dimensions."""
report = QualityReport(
role="test",
agent_output="x",
ari_before_any=1,
final_ari=1,
dimensions=[
{
"name": "ari_budget_check",
"score": 16,
"threshold": 14,
"passed": False,
"details": "",
},
],
)
assert not report.is_budget_met()
def test_ari_result_passed():
"""ARIResult.passed correctly reflects score <= budget."""
r = ARIResult(raw_score=12, whitelist_score=11, budget=14, passed=True)
assert r.passed
r2 = ARIResult(raw_score=18, whitelist_score=17, budget=14, passed=False)
assert not r2.passed
def test_reprompt_attempt_stagnation():
"""RePromptAttempt.stagnant correctly identifies delta < 0.5."""
a = RePromptAttempt(
attempt_number=1,
strategy="simplify",
ari_before=15.0,
ari_after=15.3,
ari_delta=0.3,
stagnant=True,
)
assert a.stagnant
b = RePromptAttempt(
attempt_number=2,
strategy="restructure",
ari_before=15.0,
ari_after=13.2,
ari_delta=-1.8,
stagnant=False,
)
assert not b.stagnant
# ── DomainTermWhitelist ──
def test_whitelist_strips_known_terms():
"""Whitelisted terms are replaced with placeholders."""
wl = DomainTermWhitelist()
text = "Use dependency injection and bounded context patterns."
cleaned = wl.strip_known_terms(text)
assert "dependency injection" not in cleaned
assert "bounded context" not in cleaned
assert "word" in cleaned.lower() # placeholder
assert "patterns" in cleaned # non-whitelisted preserved
def test_whitelist_matched_terms():
"""get_matched_terms returns correct matches."""
wl = DomainTermWhitelist()
text = "Use microservices with event-driven architecture and ACID compliance."
matched = wl.get_matched_terms(text)
assert "microservices" in matched
assert "event-driven" in matched
assert len(matched) >= 3 # microservices, event-driven, ACID
def test_whitelist_multi_word_preserves_surrounding():
"""Multi-word replacement preserves surrounding text."""
wl = DomainTermWhitelist()
text = "The blue-green deployment strategy works."
cleaned = wl.strip_known_terms(text)
assert "blue-green deployment" not in cleaned
assert "strategy" in cleaned # surrounding text preserved
assert "works" in cleaned
# ── QualityEngine ──
def test_engine_score_output_returns_report():
"""score_output returns a valid QualityReport with expected fields."""
engine = QualityEngine()
report = engine.score_output(
"The system shall authenticate users via JWT tokens.", "product_owner"
)
assert isinstance(report, QualityReport)
assert report.role == "product_owner"
assert report.word_count > 0
assert report.sentence_count > 0
assert report.final_ari > 0
assert report.ari_result.budget == 12 # PO budget
def test_engine_score_output_async():
"""Async wrapper runs without error (basic smoke test)."""
import asyncio
engine = QualityEngine()
report = asyncio.run(
engine.score_output_async("Test output for async scoring.", "solution_architect")
)
assert isinstance(report, QualityReport)
assert report.final_ari > 0
def test_engine_role_specific_budgets():
"""Different roles have different budgets enforced."""
engine = QualityEngine()
long_text = (
"Asynchronous bidirectional synchronous replication infrastructure requires "
"containerization orchestration implementation across heterogeneous environments. "
"Microservices event-driven domain-driven architecture decomposition patterns."
)
po_report = engine.score_output(long_text, "product_owner")
sa_report = engine.score_output(long_text, "solution_architect")
# PO has tighter budget; SA has looser
assert po_report.ari_result.budget == 12
assert sa_report.ari_result.budget == 20
def test_engine_detect_stagnation():
"""Stagnation detected when delta < 0.5 between last two scores."""
engine = QualityEngine()
assert not engine.detect_stagnation([15.0]) # Only 1 score
assert not engine.detect_stagnation([15.0, 14.2]) # Delta = 0.8 >= 0.5
assert engine.detect_stagnation([15.0, 15.3]) # Delta = 0.3 < 0.5
assert engine.detect_stagnation([18.0, 18.1]) # Delta = 0.1 < 0.5
|