"""RAG quality evaluation: content coverage, keyword hit rate, retrieval quality.""" from pathlib import Path from collections import Counter from app.core.rag import CORPUS_DIR from app.core.schemas import TeamRole # ── Corpus file structure check ─────────────────────────────────────── ROLES_WITH_RAG = [ TeamRole.PRODUCT_OWNER, TeamRole.BUSINESS_ANALYST, TeamRole.SOLUTION_ARCHITECT, TeamRole.DATA_ARCHITECT, TeamRole.SECURITY_ANALYST, TeamRole.UX_DESIGNER, TeamRole.API_DESIGNER, TeamRole.QA_STRATEGIST, TeamRole.DEVOPS_ARCHITECT, TeamRole.TECHNICAL_WRITER, ] ROLE_DIR_MAP: dict[TeamRole, str] = { TeamRole.PRODUCT_OWNER: "product_owner", TeamRole.BUSINESS_ANALYST: "business_analyst", TeamRole.SOLUTION_ARCHITECT: "solution_architect", TeamRole.DATA_ARCHITECT: "data_architect", TeamRole.SECURITY_ANALYST: "security_analyst", TeamRole.UX_DESIGNER: "ux_designer", TeamRole.API_DESIGNER: "api_designer", TeamRole.QA_STRATEGIST: "qa_strategist", TeamRole.DEVOPS_ARCHITECT: "devops_architect", TeamRole.TECHNICAL_WRITER: "technical_writer", } MIN_FILES_PER_ROLE = 3 MIN_LINES_PER_ROLE = 60 # ── Keyword coverage per role ───────────────────────────────────────── # Topics that each role's corpus should cover, based on the role's domain ROLE_TOPICS: dict[TeamRole, list[str]] = { TeamRole.PRODUCT_OWNER: [ "backlog", "prioritiz", "stakeholder", "value", "moscow", "mvp", "scope", "kano", ], TeamRole.BUSINESS_ANALYST: [ "requirement", "elicitation", "use case", "interview", "workshop", "stakeholder", "user story", ], TeamRole.SOLUTION_ARCHITECT: [ "architecture", "trade-off", "scal", "pattern", "case study", "decomposition", "latency", "caching", ], TeamRole.DATA_ARCHITECT: [ "data model", "schema", "storage", "sql", "nosql", "normalization", "partition", "index", ], TeamRole.SECURITY_ANALYST: [ "threat model", "compliance", "authentication", "oauth", "encryption", "owasp", "vulnerability", ], TeamRole.UX_DESIGNER: [ "navigation", "accessibility", "wcag", "feedback", "interaction", "touch target", "empty state", ], TeamRole.API_DESIGNER: [ "versioning", "error handling", "rest", "status code", "idempotent", "rate limit", "pagination", ], TeamRole.QA_STRATEGIST: [ "test strategy", "unit test", "integration", "e2e", "quality metric", "coverage", "regression", "flaky", ], TeamRole.DEVOPS_ARCHITECT: [ "ci/cd", "deployment", "docker", "kubernetes", "monitoring", "rollback", "infrastructure", ], TeamRole.TECHNICAL_WRITER: [ "documentation", "api doc", "style guide", "voice", "user guide", "readme", "changelog", ], } def _load_role_texts(role: TeamRole) -> list[str]: """Load all text from a role's corpus directory.""" dir_name = ROLE_DIR_MAP.get(role, role.value) role_dir = CORPUS_DIR / dir_name if not role_dir.is_dir(): return [] texts: list[str] = [] for path in sorted(role_dir.iterdir()): if path.suffix in {".txt", ".md", ".yaml", ".yml"}: texts.append(path.read_text(encoding="utf-8")) return texts def _keyword_overlap(query: str, document: str) -> float: """Measure what fraction of query words appear in the document.""" query_words = set(query.lower().split()) doc_words = set(document.lower().split()) if not query_words: return 0.0 return len(query_words & doc_words) / len(query_words) def _hit_rate(results: list[bool]) -> float: return sum(results) / len(results) if results else 0.0 # ── Tests ───────────────────────────────────────────────────────────── class TestCorpusStructure: """Validate basic corpus structure for all roles.""" def test_corpus_dir_exists(self) -> None: assert CORPUS_DIR.is_dir(), f"Corpus dir not found: {CORPUS_DIR}" def test_all_roles_have_directories(self) -> None: missing: list[str] = [] for role, dir_name in ROLE_DIR_MAP.items(): role_dir = CORPUS_DIR / dir_name if not role_dir.is_dir(): missing.append(f"{role.value} ({dir_name})") assert not missing, f"Missing corpus directories: {missing}" def test_no_orphan_directories(self) -> None: expected_dirs = {d for d in ROLE_DIR_MAP.values()} actual_dirs = {d.name for d in CORPUS_DIR.iterdir() if d.is_dir()} orphans = actual_dirs - expected_dirs assert not orphans, f"Orphaned corpus directories: {orphans}" def test_each_role_meets_min_file_count(self) -> None: failures: list[str] = [] for role in ROLES_WITH_RAG: texts = _load_role_texts(role) if len(texts) < MIN_FILES_PER_ROLE: failures.append(f"{role.value}: {len(texts)} files (min {MIN_FILES_PER_ROLE})") assert not failures, f"Role file count below minimum:\n" + "\n".join(failures) def test_each_role_meets_min_line_count(self) -> None: failures: list[str] = [] for role in ROLES_WITH_RAG: texts = _load_role_texts(role) total_lines = sum(t.count("\n") + 1 for t in texts) if total_lines < MIN_LINES_PER_ROLE: failures.append(f"{role.value}: {total_lines} lines (min {MIN_LINES_PER_ROLE})") assert not failures, f"Role line count below minimum:\n" + "\n".join(failures) class TestTopicCoverage: """Validate that each role's corpus covers expected topics.""" def test_all_roles_cover_expected_topics(self) -> None: failures: list[str] = [] for role in ROLES_WITH_RAG: texts = _load_role_texts(role) combined = "\n".join(texts).lower() missing_topics: list[str] = [] for topic in ROLE_TOPICS.get(role, []): if topic.lower() not in combined: missing_topics.append(topic) if missing_topics: failures.append(f"{role.value}: missing topics: {missing_topics}") assert not failures, "\n" + "\n".join(failures) def test_topic_coverage_percentage(self) -> None: """Measure and report topic coverage percentage per role.""" report: list[str] = [] for role in ROLES_WITH_RAG: texts = _load_role_texts(role) combined = "\n".join(texts).lower() topics = ROLE_TOPICS.get(role, []) covered = sum(1 for t in topics if t.lower() in combined) pct = (covered / len(topics) * 100) if topics else 0 report.append(f" {role.value}: {covered}/{len(topics)} ({pct:.0f}%)") print("Topic coverage by role:\n" + "\n".join(report)) class TestKeywordHitRate: """Simple keyword-based retrieval quality eval (proxy for semantic similarity).""" QUERIES: dict[TeamRole, list[tuple[str, str]]] = { TeamRole.PRODUCT_OWNER: [ ("how to prioritize backlog items using moscow method", "backlog_prioritization"), ("how to make value-driven product decisions", "value_driven_decisions"), ], TeamRole.BUSINESS_ANALYST: [ ("techniques for gathering requirements from stakeholders", "requirements_elicitation"), ("how to write well-structured use cases", "use_case_writing"), ], TeamRole.SOLUTION_ARCHITECT: [ ("how to evaluate architectural tradeoffs and patterns", "patterns_and_tradeoffs"), ("case study for scaling an application", "case_study_scaling"), ], TeamRole.DATA_ARCHITECT: [ ("patterns for data modeling and schema design", "data_modeling_patterns"), ("deciding between sql and nosql storage", "storage_decisions"), ], TeamRole.SECURITY_ANALYST: [ ("threat modeling approaches and techniques", "threat_modeling"), ("compliance frameworks for security", "compliance_frameworks"), ], TeamRole.API_DESIGNER: [ ("strategies for versioning rest apis", "api_versioning_strategies"), ("error handling patterns for api responses", "error_handling_patterns"), ], TeamRole.UX_DESIGNER: [ ("ui interaction patterns for web applications", "interaction_patterns"), ("wcag accessibility standards and best practices", "accessibility_standards"), ], TeamRole.QA_STRATEGIST: [ ("test strategy patterns and the testing pyramid", "test_strategy_patterns"), ("quality metrics for measuring test effectiveness", "quality_metrics"), ], TeamRole.DEVOPS_ARCHITECT: [ ("ci cd best practices for continuous deployment", "cicd_best_practices"), ("deployment patterns for production systems", "deployment_patterns"), ], TeamRole.TECHNICAL_WRITER: [ ("types of documentation and when to use them", "documentation_types"), ("style and tone guidelines for technical writing", "style_and_tone"), ], } def test_keyword_hit_rate_meets_threshold(self) -> None: """For each query, check that the most relevant corpus doc has non-trivial keyword overlap with the query.""" failures: list[str] = [] all_queries = 0 hits = 0 for role in ROLES_WITH_RAG: texts = _load_role_texts(role) if not texts: continue queries = self.QUERIES.get(role, []) for query, expected_substring in queries: all_queries += 1 scores = [_keyword_overlap(query, t) for t in texts] best_score = max(scores) if scores else 0.0 # Find which doc best completed the query - check if expected file contributed expected_hit = any(expected_substring in t[:200].lower() for t in texts) if best_score >= 0.1 or expected_hit: hits += 1 else: failures.append( f"{role.value}: query '{query[:50]}...' " f"best overlap {best_score:.2f}" ) hit_pct = _hit_rate([True] * hits + [False] * (all_queries - hits)) * 100 print(f"Keyword hit rate: {hits}/{all_queries} ({hit_pct:.0f}%)") if hit_pct < 70: failures.insert(0, f"Overall hit rate {hit_pct:.0f}% < 70% threshold") assert not failures, "\n" + "\n".join(failures) class TestRoleScopedRetrieval: """Verify that role-scoped retrieval returns more relevant docs than global.""" QUERIES: dict[TeamRole, str] = { TeamRole.PRODUCT_OWNER: "how to prioritize features in the backlog", TeamRole.SECURITY_ANALYST: "threat modeling and security compliance", TeamRole.DEVOPS_ARCHITECT: "ci cd deployment pipeline best practices", } def test_role_scoped_outperforms_global(self) -> None: """Role-scoped context should have higher keyword overlap than global.""" failures: list[str] = [] for role, query in self.QUERIES.items(): role_texts = _load_role_texts(role) if not role_texts: continue role_combined = "\n".join(role_texts) role_score = _keyword_overlap(query, role_combined) # Global: combine all roles' text and measure all_texts: list[str] = [] for r in ROLES_WITH_RAG: if r != role: all_texts.extend(_load_role_texts(r)) global_combined = "\n".join(all_texts) global_score = _keyword_overlap(query, global_combined) if role_score <= global_score: failures.append( f"{role.value}: role score {role_score:.2f} <= global {global_score:.2f}" ) assert not failures, "\n".join(failures)