From b089392668f66262af2388c962300a2a183530d1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 16 Jun 2026 19:01:06 +0000 Subject: [PATCH 1/6] Add post-run LLM result synthesis (--synthesize) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After all testers complete, an optional LLM call (opt-in via --synthesize) reads the full deduplicated findings list and produces a SynthesisResult: executive summary, prioritised recommendations, root-cause clusters, and false-positive candidates. The synthesis is appended to Markdown and JSON reports and stored on TestSession.synthesis for downstream consumers. - qa_agent/synthesizer.py: new module; serialize session → LLM → parse JSON - qa_agent/models.py: RootCauseCluster + SynthesisResult dataclasses; TestSession.synthesis field; to_dict() includes synthesis - qa_agent/config.py: TestConfig.synthesize_results flag (default False) - qa_agent/agent.py: _synthesize_results() called post-run when flag set - qa_agent/reporters/markdown.py: ## AI Analysis section when synthesis present - qa_agent/reporters/json_reporter.py: synthesis key in JSON output - qa_agent/cli.py: --synthesize flag - tests/test_synthesizer.py: 35 unit tests covering serialization, parsing, LLM error handling, and reporter integration https://claude.ai/code/session_019zZ6wHiVa4mYMLi6e8KJcE --- qa_agent/agent.py | 22 ++ qa_agent/cli.py | 10 + qa_agent/config.py | 3 + qa_agent/models.py | 36 +++ qa_agent/reporters/json_reporter.py | 1 + qa_agent/reporters/markdown.py | 55 ++++- qa_agent/synthesizer.py | 165 +++++++++++++ tests/test_synthesizer.py | 368 ++++++++++++++++++++++++++++ 8 files changed, 659 insertions(+), 1 deletion(-) create mode 100644 qa_agent/synthesizer.py create mode 100644 tests/test_synthesizer.py diff --git a/qa_agent/agent.py b/qa_agent/agent.py index 3d0f23b..ec4da19 100644 --- a/qa_agent/agent.py +++ b/qa_agent/agent.py @@ -170,6 +170,10 @@ def run(self) -> TestSession: self.session.end_time = datetime.now() + # Post-run LLM synthesis (opt-in) + if self.config.synthesize_results: + self._synthesize_results() + # Generate reports self._generate_reports() @@ -259,6 +263,24 @@ def _apply_test_plan(self): if self.test_plan.notes: self.console.print_progress(f"Notes: {self.test_plan.notes}") + def _synthesize_results(self): + """Call the LLM post-run to produce a narrative synthesis of findings.""" + from .synthesizer import synthesize + + self.console.print_progress("Synthesizing results with AI...") + result = synthesize( + session=self.session, + provider=self.config.llm_provider, + model=self.config.ai_model, + ) + if result: + self.session.synthesis = result + self.console.print_progress("AI synthesis complete.") + else: + self.console.print_progress( + "Warning: AI synthesis failed — continuing without it." + ) + def _factory(self): """Return the playwright context-manager factory (real or injected mock).""" return self._playwright_factory if self._playwright_factory is not None else sync_playwright diff --git a/qa_agent/cli.py b/qa_agent/cli.py index eb8be1a..157235a 100644 --- a/qa_agent/cli.py +++ b/qa_agent/cli.py @@ -309,6 +309,15 @@ def main(): action="store_true", help="Bypass the test plan cache and always call the AI. Only valid with --instructions or --instructions-file.", ) + parser.add_argument( + "--synthesize", + action="store_true", + help=( + "After the run, call the LLM once to produce a narrative synthesis: " + "executive summary, prioritised recommendations, root-cause clusters, " + "and likely false positives. Appended to Markdown and JSON reports." + ), + ) args = parser.parse_args() # Validate: --no-cache requires instructions @@ -422,6 +431,7 @@ def main(): llm_provider=LLMProvider(args.llm_provider), ai_model=args.ai_model or None, use_plan_cache=not args.no_cache, + synthesize_results=args.synthesize, workers=args.workers, rate_limit=args.rate_limit, invocation_context="cli", diff --git a/qa_agent/config.py b/qa_agent/config.py index 63874e8..c56f8ce 100644 --- a/qa_agent/config.py +++ b/qa_agent/config.py @@ -101,6 +101,9 @@ class TestConfig: ai_model: str | None = None # None → use the provider's default model use_plan_cache: bool = True # Cache generated test plans to avoid redundant API calls + # Post-run LLM synthesis: narrative analysis of findings after the run completes + synthesize_results: bool = False + # Invocation context — used to tailor diagnostic hints invocation_context: Literal["cli", "web"] | None = None diff --git a/qa_agent/models.py b/qa_agent/models.py index 7102aff..2b8b327 100644 --- a/qa_agent/models.py +++ b/qa_agent/models.py @@ -178,6 +178,40 @@ def _normalize_url(url: str) -> str: return normalised +@dataclass +class RootCauseCluster: + """A cluster of related findings sharing a common root cause.""" + label: str + finding_titles: list[str] + root_cause: str + suggested_fix: str + + +@dataclass +class SynthesisResult: + """LLM-generated narrative analysis produced after a test run.""" + executive_summary: str + priority_recommendations: list[str] = field(default_factory=list) + root_cause_clusters: list[RootCauseCluster] = field(default_factory=list) + false_positive_candidates: list[str] = field(default_factory=list) + + def to_dict(self) -> dict: + return { + "executive_summary": self.executive_summary, + "priority_recommendations": self.priority_recommendations, + "root_cause_clusters": [ + { + "label": c.label, + "finding_titles": c.finding_titles, + "root_cause": c.root_cause, + "suggested_fix": c.suggested_fix, + } + for c in self.root_cause_clusters + ], + "false_positive_candidates": self.false_positive_candidates, + } + + @dataclass class TestSession: """Complete test session results.""" @@ -190,6 +224,7 @@ class TestSession: findings_by_severity: dict[str, int] = field(default_factory=dict) findings_by_category: dict[str, int] = field(default_factory=dict) recording_path: str | None = None + synthesis: "SynthesisResult | None" = None def add_page_analysis(self, page: PageAnalysis): """Add page analysis and update totals.""" @@ -272,4 +307,5 @@ def to_dict(self) -> dict: "recording_path": self.recording_path, "status": self.status, "findings": [f.to_dict() for f in self.get_deduplicated_findings()], + "synthesis": self.synthesis.to_dict() if self.synthesis else None, } diff --git a/qa_agent/reporters/json_reporter.py b/qa_agent/reporters/json_reporter.py index 79d58bd..d7384f2 100644 --- a/qa_agent/reporters/json_reporter.py +++ b/qa_agent/reporters/json_reporter.py @@ -72,6 +72,7 @@ def _build_report(self, session: "TestSession") -> dict: for page in session.pages_tested ], "findings": [finding.to_dict() for finding in session.get_deduplicated_findings()], + "synthesis": session.synthesis.to_dict() if session.synthesis else None, } def get_json_string(self, session: "TestSession") -> str: diff --git a/qa_agent/reporters/markdown.py b/qa_agent/reporters/markdown.py index 85bff02..001f5ac 100644 --- a/qa_agent/reporters/markdown.py +++ b/qa_agent/reporters/markdown.py @@ -8,7 +8,7 @@ from .base import BaseReporter if TYPE_CHECKING: - from ..models import Finding, TestSession + from ..models import Finding, SynthesisResult, TestSession class MarkdownReporter(BaseReporter): @@ -80,6 +80,10 @@ def _build_report(self, session: "TestSession") -> str: lines.append(f"| {emoji} {cat_name} | {count} |") lines.append("") + # AI synthesis section + if session.synthesis: + lines.extend(self._format_synthesis(session.synthesis)) + # Test reliability warnings plan_warnings = session.config_summary.get("plan_warnings", []) if plan_warnings: @@ -171,6 +175,55 @@ def _build_report(self, session: "TestSession") -> str: return "\n".join(lines) + def _format_synthesis(self, synthesis: "SynthesisResult") -> list[str]: + """Format the AI synthesis block.""" + lines: list[str] = [] + + lines.append("## AI Analysis") + lines.append("") + lines.append("> *Generated by LLM after the test run. Treat as advisory — verify before acting.*") + lines.append("") + + lines.append("### Executive Summary") + lines.append("") + lines.append(synthesis.executive_summary) + lines.append("") + + if synthesis.priority_recommendations: + lines.append("### Priority Recommendations") + lines.append("") + for rec in synthesis.priority_recommendations: + lines.append(f"1. {rec}") + lines.append("") + + if synthesis.root_cause_clusters: + lines.append("### Root Cause Clusters") + lines.append("") + for cluster in synthesis.root_cause_clusters: + lines.append(f"#### {cluster.label}") + lines.append("") + lines.append(f"**Root cause:** {cluster.root_cause}") + lines.append("") + lines.append(f"**Suggested fix:** {cluster.suggested_fix}") + lines.append("") + lines.append("**Related findings:**") + for title in cluster.finding_titles: + lines.append(f"- {title}") + lines.append("") + + if synthesis.false_positive_candidates: + lines.append("### Possible False Positives") + lines.append("") + lines.append( + "> These findings may be test artifacts rather than real bugs — review before filing." + ) + lines.append("") + for candidate in synthesis.false_positive_candidates: + lines.append(f"- {candidate}") + lines.append("") + + return lines + def _escape_html_tags(self, text: str) -> str: """Wrap HTML tags in backticks so they render correctly in markdown previews. diff --git a/qa_agent/synthesizer.py b/qa_agent/synthesizer.py new file mode 100644 index 0000000..e761ffa --- /dev/null +++ b/qa_agent/synthesizer.py @@ -0,0 +1,165 @@ +"""Post-run result synthesis via LLM. + +After all testers complete, this module calls the LLM once with the full +findings list and produces a ``SynthesisResult`` — a narrative analysis that +clusters related issues, identifies root causes, prioritises fixes, and flags +likely false positives. + +The call is non-fatal: if the LLM is unavailable or returns unparseable output, +a warning is logged and ``None`` is returned so the rest of reporting continues. +""" + +import json +import logging + +from .llm_client import LLMError, LLMProvider, create_llm_client +from .models import RootCauseCluster, SynthesisResult, TestSession + +logger = logging.getLogger(__name__) + +_SYSTEM_PROMPT = """You are a senior QA lead reviewing automated test results for a web application. + +You will receive a JSON object describing a completed QA test session, including: +- Session metadata (pages tested, duration, URLs) +- Findings grouped by severity and category +- The full list of deduplicated findings with descriptions + +Your job is to produce a concise, developer-facing synthesis of the results. + +Return ONLY valid JSON matching this exact schema — no markdown, no commentary: + +{ + "executive_summary": "<2-3 sentence narrative: overall health, most important themes, what the developer should read first>", + "priority_recommendations": [ + "", + ... + ], + "root_cause_clusters": [ + { + "label": "", + "finding_titles": ["", ...], + "root_cause": "", + "suggested_fix": "" + }, + ... + ], + "false_positive_candidates": [ + "", + ... + ] +} + +Guidelines: +- executive_summary: be direct. If there are critical issues, lead with them. +- priority_recommendations: 3-6 items max. Focus on highest-impact fixes. +- root_cause_clusters: only create a cluster when 2+ findings genuinely share a root cause. + Skip this field (empty array) if all findings are independent. +- false_positive_candidates: conservative — only flag something if there is a clear signal + it is a test artifact (e.g. a navigation finding on a page that requires auth, + a console error that is a known third-party script, etc.). Empty array is fine. +- Keep language terse and actionable. Avoid padding and hedging.""" + + +def _serialize_session(session: TestSession) -> str: + """Produce a compact JSON payload for the LLM from the completed session.""" + findings = session.get_deduplicated_findings() + duration = None + if session.end_time: + duration = round((session.end_time - session.start_time).total_seconds(), 1) + + payload = { + "pages_tested": len(session.pages_tested), + "duration_seconds": duration, + "urls": session.config_summary.get("urls", []), + "mode": session.config_summary.get("mode"), + "instructions": session.config_summary.get("instructions"), + "total_findings": session.total_findings, + "unique_findings": len(findings), + "findings_by_severity": session.findings_by_severity, + "findings_by_category": session.findings_by_category, + "findings": [ + { + "title": f.title, + "severity": f.severity.value, + "category": f.category.value, + "description": f.description, + "url": f.url, + "affected_urls_count": len(f.affected_urls), + "element_selector": f.element_selector, + "expected_behavior": f.expected_behavior, + "actual_behavior": f.actual_behavior, + } + for f in findings + ], + } + return json.dumps(payload, default=str) + + +def _parse_synthesis(text: str) -> SynthesisResult: + """Parse LLM JSON response into a SynthesisResult.""" + # Strip markdown fences if the model wrapped it despite instructions + stripped = text.strip() + if stripped.startswith("```"): + lines = stripped.splitlines() + stripped = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:]) + + data = json.loads(stripped) + + clusters = [ + RootCauseCluster( + label=c.get("label", ""), + finding_titles=c.get("finding_titles", []), + root_cause=c.get("root_cause", ""), + suggested_fix=c.get("suggested_fix", ""), + ) + for c in data.get("root_cause_clusters", []) + ] + + return SynthesisResult( + executive_summary=data.get("executive_summary", ""), + priority_recommendations=data.get("priority_recommendations", []), + root_cause_clusters=clusters, + false_positive_candidates=data.get("false_positive_candidates", []), + ) + + +def synthesize( + session: TestSession, + provider: LLMProvider = LLMProvider.ANTHROPIC, + model: str | None = None, + api_key: str | None = None, + timeout: int = 60, +) -> SynthesisResult | None: + """Call the LLM to synthesize the completed session results. + + Returns ``None`` on any failure so callers can treat synthesis as optional. + """ + if not session.pages_tested: + return None + + findings = session.get_deduplicated_findings() + if not findings: + return None + + try: + client = create_llm_client(provider=provider, model=model, api_key=api_key) + user_message = ( + "Here is the completed QA test session data:\n\n" + + _serialize_session(session) + + "\n\nProvide your synthesis as JSON." + ) + response = client.complete( + system=_SYSTEM_PROMPT, + user=user_message, + max_tokens=2048, + timeout=timeout, + ) + return _parse_synthesis(response.text) + except LLMError as e: + logger.warning("Result synthesis LLM call failed (%s): %s", e.status_code, e) + except (json.JSONDecodeError, KeyError, TypeError) as e: + logger.warning("Failed to parse synthesis response: %s", e) + except Exception as e: + logger.warning("Unexpected error during result synthesis: %s", e) + + return None diff --git a/tests/test_synthesizer.py b/tests/test_synthesizer.py new file mode 100644 index 0000000..4192afd --- /dev/null +++ b/tests/test_synthesizer.py @@ -0,0 +1,368 @@ +"""Tests for qa_agent/synthesizer.py and synthesis integration.""" + +from __future__ import annotations + +import json +from datetime import datetime +from unittest.mock import MagicMock, patch + +import pytest + +from qa_agent.models import ( + RootCauseCluster, + SynthesisResult, + TestSession, +) +from qa_agent.synthesizer import _parse_synthesis, _serialize_session, synthesize +from tests.conftest import make_finding, make_session, make_session_with_findings + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_synthesis(**kwargs) -> SynthesisResult: + defaults = dict( + executive_summary="Two critical accessibility issues require immediate attention.", + priority_recommendations=["Fix ARIA labels", "Add focus indicators"], + root_cause_clusters=[ + RootCauseCluster( + label="Missing ARIA labels", + finding_titles=["No alt text on logo"], + root_cause="Icon-only buttons lack accessible names", + suggested_fix="Add aria-label to all icon buttons", + ) + ], + false_positive_candidates=["Console error on third-party script"], + ) + defaults.update(kwargs) + return SynthesisResult(**defaults) + + +_VALID_LLM_RESPONSE = json.dumps({ + "executive_summary": "Three issues found, one critical.", + "priority_recommendations": ["Fix focus trap", "Add ARIA labels"], + "root_cause_clusters": [ + { + "label": "Focus management", + "finding_titles": ["Keyboard trap in modal"], + "root_cause": "Modal does not implement focus trap correctly", + "suggested_fix": "Use a focus trap library or manually manage tabIndex", + } + ], + "false_positive_candidates": ["Network error on beacon endpoint"], +}) + + +# --------------------------------------------------------------------------- +# _serialize_session +# --------------------------------------------------------------------------- + +class TestSerializeSession: + def test_produces_valid_json(self): + session = make_session_with_findings() + payload = _serialize_session(session) + data = json.loads(payload) + assert "findings" in data + assert "pages_tested" in data + assert "total_findings" in data + + def test_finding_fields_present(self): + session = make_session_with_findings() + data = json.loads(_serialize_session(session)) + finding = data["findings"][0] + assert "title" in finding + assert "severity" in finding + assert "category" in finding + assert "description" in finding + + def test_duration_computed(self): + session = make_session_with_findings() + data = json.loads(_serialize_session(session)) + assert data["duration_seconds"] == pytest.approx(300.0) + + def test_no_end_time_produces_none_duration(self): + session = TestSession( + session_id="x", + start_time=datetime(2024, 1, 1), + config_summary={"urls": []}, + ) + data = json.loads(_serialize_session(session)) + assert data["duration_seconds"] is None + + def test_empty_findings_list(self): + session = make_session() + data = json.loads(_serialize_session(session)) + assert data["findings"] == [] + + def test_uses_deduplicated_findings(self): + """Duplicates across pages should be collapsed before sending to LLM.""" + f1 = make_finding(title="Focus issue", url="https://example.com/a") + f2 = make_finding(title="Focus issue", url="https://example.com/b") + from tests.conftest import make_page_analysis + session = TestSession( + session_id="dup", + start_time=datetime(2024, 1, 1), + config_summary={"urls": []}, + ) + session.add_page_analysis(make_page_analysis(findings=[f1])) + session.add_page_analysis(make_page_analysis(url="https://example.com/b", findings=[f2])) + data = json.loads(_serialize_session(session)) + assert data["unique_findings"] == 1 + assert len(data["findings"]) == 1 + + +# --------------------------------------------------------------------------- +# _parse_synthesis +# --------------------------------------------------------------------------- + +class TestParseSynthesis: + def test_parses_valid_response(self): + result = _parse_synthesis(_VALID_LLM_RESPONSE) + assert isinstance(result, SynthesisResult) + assert result.executive_summary == "Three issues found, one critical." + assert result.priority_recommendations == ["Fix focus trap", "Add ARIA labels"] + assert len(result.root_cause_clusters) == 1 + cluster = result.root_cause_clusters[0] + assert cluster.label == "Focus management" + assert cluster.finding_titles == ["Keyboard trap in modal"] + assert result.false_positive_candidates == ["Network error on beacon endpoint"] + + def test_strips_markdown_fences(self): + wrapped = f"```json\n{_VALID_LLM_RESPONSE}\n```" + result = _parse_synthesis(wrapped) + assert result.executive_summary == "Three issues found, one critical." + + def test_strips_markdown_fences_no_language(self): + wrapped = f"```\n{_VALID_LLM_RESPONSE}\n```" + result = _parse_synthesis(wrapped) + assert result.executive_summary == "Three issues found, one critical." + + def test_empty_arrays_allowed(self): + payload = json.dumps({ + "executive_summary": "All good.", + "priority_recommendations": [], + "root_cause_clusters": [], + "false_positive_candidates": [], + }) + result = _parse_synthesis(payload) + assert result.priority_recommendations == [] + assert result.root_cause_clusters == [] + + def test_missing_optional_fields_default_to_empty(self): + payload = json.dumps({"executive_summary": "Brief."}) + result = _parse_synthesis(payload) + assert result.priority_recommendations == [] + assert result.root_cause_clusters == [] + assert result.false_positive_candidates == [] + + def test_invalid_json_raises(self): + with pytest.raises(json.JSONDecodeError): + _parse_synthesis("not json at all") + + +# --------------------------------------------------------------------------- +# synthesize() — with mocked LLM client +# --------------------------------------------------------------------------- + +class TestSynthesizeFunction: + def _mock_client(self, response_text: str): + from qa_agent.llm_client import LLMProvider, LLMResponse + client = MagicMock() + client.complete.return_value = LLMResponse( + text=response_text, + provider=LLMProvider.ANTHROPIC, + model="claude-sonnet-4-6", + ) + return client + + def test_returns_synthesis_result_on_success(self): + session = make_session_with_findings() + client = self._mock_client(_VALID_LLM_RESPONSE) + with patch("qa_agent.synthesizer.create_llm_client", return_value=client): + result = synthesize(session) + assert isinstance(result, SynthesisResult) + assert result.executive_summary == "Three issues found, one critical." + + def test_returns_none_on_empty_session(self): + session = TestSession( + session_id="empty", + start_time=datetime(2024, 1, 1), + config_summary={}, + ) + result = synthesize(session) + assert result is None + + def test_returns_none_on_no_findings(self): + session = make_session() # session with page but no findings + result = synthesize(session) + assert result is None + + def test_returns_none_on_llm_error(self): + from qa_agent.llm_client import LLMError + session = make_session_with_findings() + with patch("qa_agent.synthesizer.create_llm_client") as mock_factory: + mock_factory.return_value.complete.side_effect = LLMError("API error", status_code=500) + result = synthesize(session) + assert result is None + + def test_returns_none_on_json_parse_error(self): + session = make_session_with_findings() + client = self._mock_client("not valid json {{{") + with patch("qa_agent.synthesizer.create_llm_client", return_value=client): + result = synthesize(session) + assert result is None + + def test_passes_provider_and_model_to_client_factory(self): + from qa_agent.llm_client import LLMProvider + session = make_session_with_findings() + client = self._mock_client(_VALID_LLM_RESPONSE) + with patch("qa_agent.synthesizer.create_llm_client", return_value=client) as mock_factory: + synthesize(session, provider=LLMProvider.OPENAI, model="gpt-4o") + mock_factory.assert_called_once_with( + provider=LLMProvider.OPENAI, + model="gpt-4o", + api_key=None, + ) + + def test_passes_api_key_to_client_factory(self): + from qa_agent.llm_client import LLMProvider + session = make_session_with_findings() + client = self._mock_client(_VALID_LLM_RESPONSE) + with patch("qa_agent.synthesizer.create_llm_client", return_value=client) as mock_factory: + synthesize(session, api_key="sk-test") + mock_factory.assert_called_once_with( + provider=LLMProvider.ANTHROPIC, + model=None, + api_key="sk-test", + ) + + +# --------------------------------------------------------------------------- +# SynthesisResult model +# --------------------------------------------------------------------------- + +class TestSynthesisResultModel: + def test_to_dict_structure(self): + synthesis = _make_synthesis() + d = synthesis.to_dict() + assert "executive_summary" in d + assert "priority_recommendations" in d + assert "root_cause_clusters" in d + assert "false_positive_candidates" in d + + def test_to_dict_cluster_fields(self): + synthesis = _make_synthesis() + cluster = synthesis.to_dict()["root_cause_clusters"][0] + assert cluster["label"] == "Missing ARIA labels" + assert "finding_titles" in cluster + assert "root_cause" in cluster + assert "suggested_fix" in cluster + + def test_session_synthesis_field_defaults_to_none(self): + session = make_session() + assert session.synthesis is None + + def test_session_to_dict_includes_synthesis_none(self): + session = make_session() + d = session.to_dict() + assert d["synthesis"] is None + + def test_session_to_dict_includes_synthesis_when_set(self): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + d = session.to_dict() + assert d["synthesis"] is not None + assert d["synthesis"]["executive_summary"].startswith("Two critical") + + +# --------------------------------------------------------------------------- +# Reporter integration — Markdown +# --------------------------------------------------------------------------- + +class TestMarkdownSynthesisSection: + def _report_content(self, session: TestSession, tmp_path) -> str: + from qa_agent.reporters.markdown import MarkdownReporter + reporter = MarkdownReporter(str(tmp_path)) + filepath = reporter.generate(session) + return open(filepath).read() + + def test_synthesis_section_absent_when_none(self, tmp_path): + session = make_session_with_findings() + content = self._report_content(session, tmp_path) + assert "## AI Analysis" not in content + + def test_synthesis_section_present_when_set(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + content = self._report_content(session, tmp_path) + assert "## AI Analysis" in content + + def test_executive_summary_in_report(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + content = self._report_content(session, tmp_path) + assert "Two critical accessibility issues" in content + + def test_priority_recommendations_in_report(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + content = self._report_content(session, tmp_path) + assert "Fix ARIA labels" in content + + def test_root_cause_cluster_in_report(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + content = self._report_content(session, tmp_path) + assert "Missing ARIA labels" in content + assert "Icon-only buttons" in content + + def test_false_positive_candidates_in_report(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + content = self._report_content(session, tmp_path) + assert "Possible False Positives" in content + assert "Console error on third-party script" in content + + def test_no_false_positive_section_when_empty(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis(false_positive_candidates=[]) + content = self._report_content(session, tmp_path) + assert "Possible False Positives" not in content + + def test_no_cluster_section_when_empty(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis(root_cause_clusters=[]) + content = self._report_content(session, tmp_path) + assert "Root Cause Clusters" not in content + + +# --------------------------------------------------------------------------- +# Reporter integration — JSON +# --------------------------------------------------------------------------- + +class TestJSONSynthesisSection: + def _report_data(self, session: TestSession, tmp_path) -> dict: + from qa_agent.reporters.json_reporter import JSONReporter + reporter = JSONReporter(str(tmp_path)) + filepath = reporter.generate(session) + return json.loads(open(filepath).read()) + + def test_synthesis_null_when_none(self, tmp_path): + session = make_session_with_findings() + data = self._report_data(session, tmp_path) + assert data["synthesis"] is None + + def test_synthesis_populated_when_set(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + data = self._report_data(session, tmp_path) + assert data["synthesis"] is not None + assert data["synthesis"]["executive_summary"] == "Two critical accessibility issues require immediate attention." + + def test_synthesis_clusters_serialised(self, tmp_path): + session = make_session_with_findings() + session.synthesis = _make_synthesis() + data = self._report_data(session, tmp_path) + clusters = data["synthesis"]["root_cause_clusters"] + assert len(clusters) == 1 + assert clusters[0]["label"] == "Missing ARIA labels" From 25a12550f42be442ef27de41463e2280b64f4389 Mon Sep 17 00:00:00 2001 From: openhands <openhands@all-hands.dev> Date: Wed, 17 Jun 2026 01:31:27 +0000 Subject: [PATCH 2/6] =?UTF-8?q?Rename=20synthesize/synthesis=20=E2=86=92?= =?UTF-8?q?=20summarize/summary=20throughout=20codebase?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Option A implementation: Complete terminology change for clarity. Breaking changes: - CLI: --synthesize → --summarize - Config: synthesize_results → generate_summary - Models: SynthesisResult → SummaryResult - Session field: synthesis → summary - JSON output: 'synthesis' → 'ai_summary' Files renamed: - qa_agent/synthesizer.py → qa_agent/summarizer.py - tests/test_synthesizer.py → tests/test_summarizer.py All 35 tests passing. Co-authored-by: openhands <openhands@all-hands.dev> --- qa_agent/agent.py | 22 +-- qa_agent/cli.py | 6 +- qa_agent/config.py | 4 +- qa_agent/models.py | 6 +- qa_agent/reporters/json_reporter.py | 2 +- qa_agent/reporters/markdown.py | 26 ++-- qa_agent/{synthesizer.py => summarizer.py} | 32 ++--- ...test_synthesizer.py => test_summarizer.py} | 128 +++++++++--------- 8 files changed, 113 insertions(+), 113 deletions(-) rename qa_agent/{synthesizer.py => summarizer.py} (85%) rename tests/{test_synthesizer.py => test_summarizer.py} (77%) diff --git a/qa_agent/agent.py b/qa_agent/agent.py index ec4da19..3bb7f4a 100644 --- a/qa_agent/agent.py +++ b/qa_agent/agent.py @@ -170,9 +170,9 @@ def run(self) -> TestSession: self.session.end_time = datetime.now() - # Post-run LLM synthesis (opt-in) - if self.config.synthesize_results: - self._synthesize_results() + # Post-run LLM summary (opt-in) + if self.config.generate_summary: + self._generate_summary() # Generate reports self._generate_reports() @@ -263,22 +263,22 @@ def _apply_test_plan(self): if self.test_plan.notes: self.console.print_progress(f"Notes: {self.test_plan.notes}") - def _synthesize_results(self): - """Call the LLM post-run to produce a narrative synthesis of findings.""" - from .synthesizer import synthesize + def _generate_summary(self): + """Call the LLM post-run to produce a narrative summary of findings.""" + from .summarizer import generate_summary - self.console.print_progress("Synthesizing results with AI...") - result = synthesize( + self.console.print_progress("Generating results summary with AI...") + result = generate_summary( session=self.session, provider=self.config.llm_provider, model=self.config.ai_model, ) if result: - self.session.synthesis = result - self.console.print_progress("AI synthesis complete.") + self.session.summary = result + self.console.print_progress("AI summary complete.") else: self.console.print_progress( - "Warning: AI synthesis failed — continuing without it." + "Warning: AI summary generation failed — continuing without it." ) def _factory(self): diff --git a/qa_agent/cli.py b/qa_agent/cli.py index 157235a..e948456 100644 --- a/qa_agent/cli.py +++ b/qa_agent/cli.py @@ -310,10 +310,10 @@ def main(): help="Bypass the test plan cache and always call the AI. Only valid with --instructions or --instructions-file.", ) parser.add_argument( - "--synthesize", + "--summarize", action="store_true", help=( - "After the run, call the LLM once to produce a narrative synthesis: " + "After the run, call the LLM once to produce a narrative summary: " "executive summary, prioritised recommendations, root-cause clusters, " "and likely false positives. Appended to Markdown and JSON reports." ), @@ -431,7 +431,7 @@ def main(): llm_provider=LLMProvider(args.llm_provider), ai_model=args.ai_model or None, use_plan_cache=not args.no_cache, - synthesize_results=args.synthesize, + generate_summary=args.summarize, workers=args.workers, rate_limit=args.rate_limit, invocation_context="cli", diff --git a/qa_agent/config.py b/qa_agent/config.py index c56f8ce..6cfb3d2 100644 --- a/qa_agent/config.py +++ b/qa_agent/config.py @@ -101,8 +101,8 @@ class TestConfig: ai_model: str | None = None # None → use the provider's default model use_plan_cache: bool = True # Cache generated test plans to avoid redundant API calls - # Post-run LLM synthesis: narrative analysis of findings after the run completes - synthesize_results: bool = False + # Post-run LLM summary: narrative analysis of findings after the run completes + generate_summary: bool = False # Invocation context — used to tailor diagnostic hints invocation_context: Literal["cli", "web"] | None = None diff --git a/qa_agent/models.py b/qa_agent/models.py index 2b8b327..99ae454 100644 --- a/qa_agent/models.py +++ b/qa_agent/models.py @@ -188,7 +188,7 @@ class RootCauseCluster: @dataclass -class SynthesisResult: +class SummaryResult: """LLM-generated narrative analysis produced after a test run.""" executive_summary: str priority_recommendations: list[str] = field(default_factory=list) @@ -224,7 +224,7 @@ class TestSession: findings_by_severity: dict[str, int] = field(default_factory=dict) findings_by_category: dict[str, int] = field(default_factory=dict) recording_path: str | None = None - synthesis: "SynthesisResult | None" = None + summary: "SummaryResult | None" = None def add_page_analysis(self, page: PageAnalysis): """Add page analysis and update totals.""" @@ -307,5 +307,5 @@ def to_dict(self) -> dict: "recording_path": self.recording_path, "status": self.status, "findings": [f.to_dict() for f in self.get_deduplicated_findings()], - "synthesis": self.synthesis.to_dict() if self.synthesis else None, + "summary": self.summary.to_dict() if self.summary else None, } diff --git a/qa_agent/reporters/json_reporter.py b/qa_agent/reporters/json_reporter.py index d7384f2..1592fa9 100644 --- a/qa_agent/reporters/json_reporter.py +++ b/qa_agent/reporters/json_reporter.py @@ -72,7 +72,7 @@ def _build_report(self, session: "TestSession") -> dict: for page in session.pages_tested ], "findings": [finding.to_dict() for finding in session.get_deduplicated_findings()], - "synthesis": session.synthesis.to_dict() if session.synthesis else None, + "ai_summary": session.summary.to_dict() if session.summary else None, } def get_json_string(self, session: "TestSession") -> str: diff --git a/qa_agent/reporters/markdown.py b/qa_agent/reporters/markdown.py index 001f5ac..17270c7 100644 --- a/qa_agent/reporters/markdown.py +++ b/qa_agent/reporters/markdown.py @@ -8,7 +8,7 @@ from .base import BaseReporter if TYPE_CHECKING: - from ..models import Finding, SynthesisResult, TestSession + from ..models import Finding, SummaryResult, TestSession class MarkdownReporter(BaseReporter): @@ -80,9 +80,9 @@ def _build_report(self, session: "TestSession") -> str: lines.append(f"| {emoji} {cat_name} | {count} |") lines.append("") - # AI synthesis section - if session.synthesis: - lines.extend(self._format_synthesis(session.synthesis)) + # AI summary section + if session.summary: + lines.extend(self._format_summary(session.summary)) # Test reliability warnings plan_warnings = session.config_summary.get("plan_warnings", []) @@ -175,8 +175,8 @@ def _build_report(self, session: "TestSession") -> str: return "\n".join(lines) - def _format_synthesis(self, synthesis: "SynthesisResult") -> list[str]: - """Format the AI synthesis block.""" + def _format_summary(self, summary: "SummaryResult") -> list[str]: + """Format the AI summary block.""" lines: list[str] = [] lines.append("## AI Analysis") @@ -186,20 +186,20 @@ def _format_synthesis(self, synthesis: "SynthesisResult") -> list[str]: lines.append("### Executive Summary") lines.append("") - lines.append(synthesis.executive_summary) + lines.append(summary.executive_summary) lines.append("") - if synthesis.priority_recommendations: + if summary.priority_recommendations: lines.append("### Priority Recommendations") lines.append("") - for rec in synthesis.priority_recommendations: + for rec in summary.priority_recommendations: lines.append(f"1. {rec}") lines.append("") - if synthesis.root_cause_clusters: + if summary.root_cause_clusters: lines.append("### Root Cause Clusters") lines.append("") - for cluster in synthesis.root_cause_clusters: + for cluster in summary.root_cause_clusters: lines.append(f"#### {cluster.label}") lines.append("") lines.append(f"**Root cause:** {cluster.root_cause}") @@ -211,14 +211,14 @@ def _format_synthesis(self, synthesis: "SynthesisResult") -> list[str]: lines.append(f"- {title}") lines.append("") - if synthesis.false_positive_candidates: + if summary.false_positive_candidates: lines.append("### Possible False Positives") lines.append("") lines.append( "> These findings may be test artifacts rather than real bugs — review before filing." ) lines.append("") - for candidate in synthesis.false_positive_candidates: + for candidate in summary.false_positive_candidates: lines.append(f"- {candidate}") lines.append("") diff --git a/qa_agent/synthesizer.py b/qa_agent/summarizer.py similarity index 85% rename from qa_agent/synthesizer.py rename to qa_agent/summarizer.py index e761ffa..bea8776 100644 --- a/qa_agent/synthesizer.py +++ b/qa_agent/summarizer.py @@ -1,7 +1,7 @@ -"""Post-run result synthesis via LLM. +"""Post-run result summary via LLM. After all testers complete, this module calls the LLM once with the full -findings list and produces a ``SynthesisResult`` — a narrative analysis that +findings list and produces a ``SummaryResult`` — a narrative analysis that clusters related issues, identifies root causes, prioritises fixes, and flags likely false positives. @@ -13,7 +13,7 @@ import logging from .llm_client import LLMError, LLMProvider, create_llm_client -from .models import RootCauseCluster, SynthesisResult, TestSession +from .models import RootCauseCluster, SummaryResult, TestSession logger = logging.getLogger(__name__) @@ -24,7 +24,7 @@ - Findings grouped by severity and category - The full list of deduplicated findings with descriptions -Your job is to produce a concise, developer-facing synthesis of the results. +Your job is to produce a concise, developer-facing summary of the results. Return ONLY valid JSON matching this exact schema — no markdown, no commentary: @@ -95,8 +95,8 @@ def _serialize_session(session: TestSession) -> str: return json.dumps(payload, default=str) -def _parse_synthesis(text: str) -> SynthesisResult: - """Parse LLM JSON response into a SynthesisResult.""" +def _parse_summary(text: str) -> SummaryResult: + """Parse LLM JSON response into a SummaryResult.""" # Strip markdown fences if the model wrapped it despite instructions stripped = text.strip() if stripped.startswith("```"): @@ -115,7 +115,7 @@ def _parse_synthesis(text: str) -> SynthesisResult: for c in data.get("root_cause_clusters", []) ] - return SynthesisResult( + return SummaryResult( executive_summary=data.get("executive_summary", ""), priority_recommendations=data.get("priority_recommendations", []), root_cause_clusters=clusters, @@ -123,16 +123,16 @@ def _parse_synthesis(text: str) -> SynthesisResult: ) -def synthesize( +def generate_summary( session: TestSession, provider: LLMProvider = LLMProvider.ANTHROPIC, model: str | None = None, api_key: str | None = None, timeout: int = 60, -) -> SynthesisResult | None: - """Call the LLM to synthesize the completed session results. +) -> SummaryResult | None: + """Call the LLM to generate a summary of the completed session results. - Returns ``None`` on any failure so callers can treat synthesis as optional. + Returns ``None`` on any failure so callers can treat summary generation as optional. """ if not session.pages_tested: return None @@ -146,7 +146,7 @@ def synthesize( user_message = ( "Here is the completed QA test session data:\n\n" + _serialize_session(session) - + "\n\nProvide your synthesis as JSON." + + "\n\nProvide your summary as JSON." ) response = client.complete( system=_SYSTEM_PROMPT, @@ -154,12 +154,12 @@ def synthesize( max_tokens=2048, timeout=timeout, ) - return _parse_synthesis(response.text) + return _parse_summary(response.text) except LLMError as e: - logger.warning("Result synthesis LLM call failed (%s): %s", e.status_code, e) + logger.warning("Result summary LLM call failed (%s): %s", e.status_code, e) except (json.JSONDecodeError, KeyError, TypeError) as e: - logger.warning("Failed to parse synthesis response: %s", e) + logger.warning("Failed to parse summary response: %s", e) except Exception as e: - logger.warning("Unexpected error during result synthesis: %s", e) + logger.warning("Unexpected error during result summary generation: %s", e) return None diff --git a/tests/test_synthesizer.py b/tests/test_summarizer.py similarity index 77% rename from tests/test_synthesizer.py rename to tests/test_summarizer.py index 4192afd..c610368 100644 --- a/tests/test_synthesizer.py +++ b/tests/test_summarizer.py @@ -1,4 +1,4 @@ -"""Tests for qa_agent/synthesizer.py and synthesis integration.""" +"""Tests for qa_agent/summarizer.py and summary integration.""" from __future__ import annotations @@ -10,17 +10,17 @@ from qa_agent.models import ( RootCauseCluster, - SynthesisResult, + SummaryResult, TestSession, ) -from qa_agent.synthesizer import _parse_synthesis, _serialize_session, synthesize +from qa_agent.summarizer import _parse_summary, _serialize_session, generate_summary from tests.conftest import make_finding, make_session, make_session_with_findings # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- -def _make_synthesis(**kwargs) -> SynthesisResult: +def _make_summary(**kwargs) -> SummaryResult: defaults = dict( executive_summary="Two critical accessibility issues require immediate attention.", priority_recommendations=["Fix ARIA labels", "Add focus indicators"], @@ -35,7 +35,7 @@ def _make_synthesis(**kwargs) -> SynthesisResult: false_positive_candidates=["Console error on third-party script"], ) defaults.update(kwargs) - return SynthesisResult(**defaults) + return SummaryResult(**defaults) _VALID_LLM_RESPONSE = json.dumps({ @@ -112,13 +112,13 @@ def test_uses_deduplicated_findings(self): # --------------------------------------------------------------------------- -# _parse_synthesis +# _parse_summary # --------------------------------------------------------------------------- -class TestParseSynthesis: +class TestParseSummary: def test_parses_valid_response(self): - result = _parse_synthesis(_VALID_LLM_RESPONSE) - assert isinstance(result, SynthesisResult) + result = _parse_summary(_VALID_LLM_RESPONSE) + assert isinstance(result, SummaryResult) assert result.executive_summary == "Three issues found, one critical." assert result.priority_recommendations == ["Fix focus trap", "Add ARIA labels"] assert len(result.root_cause_clusters) == 1 @@ -129,12 +129,12 @@ def test_parses_valid_response(self): def test_strips_markdown_fences(self): wrapped = f"```json\n{_VALID_LLM_RESPONSE}\n```" - result = _parse_synthesis(wrapped) + result = _parse_summary(wrapped) assert result.executive_summary == "Three issues found, one critical." def test_strips_markdown_fences_no_language(self): wrapped = f"```\n{_VALID_LLM_RESPONSE}\n```" - result = _parse_synthesis(wrapped) + result = _parse_summary(wrapped) assert result.executive_summary == "Three issues found, one critical." def test_empty_arrays_allowed(self): @@ -144,27 +144,27 @@ def test_empty_arrays_allowed(self): "root_cause_clusters": [], "false_positive_candidates": [], }) - result = _parse_synthesis(payload) + result = _parse_summary(payload) assert result.priority_recommendations == [] assert result.root_cause_clusters == [] def test_missing_optional_fields_default_to_empty(self): payload = json.dumps({"executive_summary": "Brief."}) - result = _parse_synthesis(payload) + result = _parse_summary(payload) assert result.priority_recommendations == [] assert result.root_cause_clusters == [] assert result.false_positive_candidates == [] def test_invalid_json_raises(self): with pytest.raises(json.JSONDecodeError): - _parse_synthesis("not json at all") + _parse_summary("not json at all") # --------------------------------------------------------------------------- -# synthesize() — with mocked LLM client +# generate_summary() — with mocked LLM client # --------------------------------------------------------------------------- -class TestSynthesizeFunction: +class TestGenerateSummaryFunction: def _mock_client(self, response_text: str): from qa_agent.llm_client import LLMProvider, LLMResponse client = MagicMock() @@ -175,12 +175,12 @@ def _mock_client(self, response_text: str): ) return client - def test_returns_synthesis_result_on_success(self): + def test_returns_summary_result_on_success(self): session = make_session_with_findings() client = self._mock_client(_VALID_LLM_RESPONSE) - with patch("qa_agent.synthesizer.create_llm_client", return_value=client): - result = synthesize(session) - assert isinstance(result, SynthesisResult) + with patch("qa_agent.summarizer.create_llm_client", return_value=client): + result = generate_summary(session) + assert isinstance(result, SummaryResult) assert result.executive_summary == "Three issues found, one critical." def test_returns_none_on_empty_session(self): @@ -189,35 +189,35 @@ def test_returns_none_on_empty_session(self): start_time=datetime(2024, 1, 1), config_summary={}, ) - result = synthesize(session) + result = generate_summary(session) assert result is None def test_returns_none_on_no_findings(self): session = make_session() # session with page but no findings - result = synthesize(session) + result = generate_summary(session) assert result is None def test_returns_none_on_llm_error(self): from qa_agent.llm_client import LLMError session = make_session_with_findings() - with patch("qa_agent.synthesizer.create_llm_client") as mock_factory: + with patch("qa_agent.summarizer.create_llm_client") as mock_factory: mock_factory.return_value.complete.side_effect = LLMError("API error", status_code=500) - result = synthesize(session) + result = generate_summary(session) assert result is None def test_returns_none_on_json_parse_error(self): session = make_session_with_findings() client = self._mock_client("not valid json {{{") - with patch("qa_agent.synthesizer.create_llm_client", return_value=client): - result = synthesize(session) + with patch("qa_agent.summarizer.create_llm_client", return_value=client): + result = generate_summary(session) assert result is None def test_passes_provider_and_model_to_client_factory(self): from qa_agent.llm_client import LLMProvider session = make_session_with_findings() client = self._mock_client(_VALID_LLM_RESPONSE) - with patch("qa_agent.synthesizer.create_llm_client", return_value=client) as mock_factory: - synthesize(session, provider=LLMProvider.OPENAI, model="gpt-4o") + with patch("qa_agent.summarizer.create_llm_client", return_value=client) as mock_factory: + generate_summary(session, provider=LLMProvider.OPENAI, model="gpt-4o") mock_factory.assert_called_once_with( provider=LLMProvider.OPENAI, model="gpt-4o", @@ -228,8 +228,8 @@ def test_passes_api_key_to_client_factory(self): from qa_agent.llm_client import LLMProvider session = make_session_with_findings() client = self._mock_client(_VALID_LLM_RESPONSE) - with patch("qa_agent.synthesizer.create_llm_client", return_value=client) as mock_factory: - synthesize(session, api_key="sk-test") + with patch("qa_agent.summarizer.create_llm_client", return_value=client) as mock_factory: + generate_summary(session, api_key="sk-test") mock_factory.assert_called_once_with( provider=LLMProvider.ANTHROPIC, model=None, @@ -238,100 +238,100 @@ def test_passes_api_key_to_client_factory(self): # --------------------------------------------------------------------------- -# SynthesisResult model +# SummaryResult model # --------------------------------------------------------------------------- -class TestSynthesisResultModel: +class TestSummaryResultModel: def test_to_dict_structure(self): - synthesis = _make_synthesis() - d = synthesis.to_dict() + summary = _make_summary() + d = summary.to_dict() assert "executive_summary" in d assert "priority_recommendations" in d assert "root_cause_clusters" in d assert "false_positive_candidates" in d def test_to_dict_cluster_fields(self): - synthesis = _make_synthesis() - cluster = synthesis.to_dict()["root_cause_clusters"][0] + summary = _make_summary() + cluster = summary.to_dict()["root_cause_clusters"][0] assert cluster["label"] == "Missing ARIA labels" assert "finding_titles" in cluster assert "root_cause" in cluster assert "suggested_fix" in cluster - def test_session_synthesis_field_defaults_to_none(self): + def test_session_summary_field_defaults_to_none(self): session = make_session() - assert session.synthesis is None + assert session.summary is None - def test_session_to_dict_includes_synthesis_none(self): + def test_session_to_dict_includes_summary_none(self): session = make_session() d = session.to_dict() - assert d["synthesis"] is None + assert d["summary"] is None - def test_session_to_dict_includes_synthesis_when_set(self): + def test_session_to_dict_includes_summary_when_set(self): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() d = session.to_dict() - assert d["synthesis"] is not None - assert d["synthesis"]["executive_summary"].startswith("Two critical") + assert d["summary"] is not None + assert d["summary"]["executive_summary"].startswith("Two critical") # --------------------------------------------------------------------------- # Reporter integration — Markdown # --------------------------------------------------------------------------- -class TestMarkdownSynthesisSection: +class TestMarkdownSummarySection: def _report_content(self, session: TestSession, tmp_path) -> str: from qa_agent.reporters.markdown import MarkdownReporter reporter = MarkdownReporter(str(tmp_path)) filepath = reporter.generate(session) return open(filepath).read() - def test_synthesis_section_absent_when_none(self, tmp_path): + def test_summary_section_absent_when_none(self, tmp_path): session = make_session_with_findings() content = self._report_content(session, tmp_path) assert "## AI Analysis" not in content - def test_synthesis_section_present_when_set(self, tmp_path): + def test_summary_section_present_when_set(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() content = self._report_content(session, tmp_path) assert "## AI Analysis" in content def test_executive_summary_in_report(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() content = self._report_content(session, tmp_path) assert "Two critical accessibility issues" in content def test_priority_recommendations_in_report(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() content = self._report_content(session, tmp_path) assert "Fix ARIA labels" in content def test_root_cause_cluster_in_report(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() content = self._report_content(session, tmp_path) assert "Missing ARIA labels" in content assert "Icon-only buttons" in content def test_false_positive_candidates_in_report(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() content = self._report_content(session, tmp_path) assert "Possible False Positives" in content assert "Console error on third-party script" in content def test_no_false_positive_section_when_empty(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis(false_positive_candidates=[]) + session.summary = _make_summary(false_positive_candidates=[]) content = self._report_content(session, tmp_path) assert "Possible False Positives" not in content def test_no_cluster_section_when_empty(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis(root_cause_clusters=[]) + session.summary = _make_summary(root_cause_clusters=[]) content = self._report_content(session, tmp_path) assert "Root Cause Clusters" not in content @@ -340,29 +340,29 @@ def test_no_cluster_section_when_empty(self, tmp_path): # Reporter integration — JSON # --------------------------------------------------------------------------- -class TestJSONSynthesisSection: +class TestJSONSummarySection: def _report_data(self, session: TestSession, tmp_path) -> dict: from qa_agent.reporters.json_reporter import JSONReporter reporter = JSONReporter(str(tmp_path)) filepath = reporter.generate(session) return json.loads(open(filepath).read()) - def test_synthesis_null_when_none(self, tmp_path): + def test_summary_null_when_none(self, tmp_path): session = make_session_with_findings() data = self._report_data(session, tmp_path) - assert data["synthesis"] is None + assert data["ai_summary"] is None - def test_synthesis_populated_when_set(self, tmp_path): + def test_summary_populated_when_set(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() data = self._report_data(session, tmp_path) - assert data["synthesis"] is not None - assert data["synthesis"]["executive_summary"] == "Two critical accessibility issues require immediate attention." + assert data["ai_summary"] is not None + assert data["ai_summary"]["executive_summary"] == "Two critical accessibility issues require immediate attention." - def test_synthesis_clusters_serialised(self, tmp_path): + def test_summary_clusters_serialised(self, tmp_path): session = make_session_with_findings() - session.synthesis = _make_synthesis() + session.summary = _make_summary() data = self._report_data(session, tmp_path) - clusters = data["synthesis"]["root_cause_clusters"] + clusters = data["ai_summary"]["root_cause_clusters"] assert len(clusters) == 1 assert clusters[0]["label"] == "Missing ARIA labels" From 21d8681afe6eda88d2baba8c955693ad3f8e2c7b Mon Sep 17 00:00:00 2001 From: openhands <openhands@all-hands.dev> Date: Wed, 17 Jun 2026 01:35:51 +0000 Subject: [PATCH 3/6] Fix mypy errors in test_summarizer.py - Add type: ignore[arg-type] for _make_summary helper to handle **kwargs unpacking - Change _report_data return type from dict to dict[str, Any] - Add cast() for json.loads() return value to satisfy mypy type checking - Import typing.Any and typing.cast for proper type annotations All 4 mypy errors resolved. Tests still pass (35/35). Co-authored-by: openhands <openhands@all-hands.dev> --- tests/test_summarizer.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/tests/test_summarizer.py b/tests/test_summarizer.py index c610368..dc55eaa 100644 --- a/tests/test_summarizer.py +++ b/tests/test_summarizer.py @@ -4,6 +4,7 @@ import json from datetime import datetime +from typing import Any, cast from unittest.mock import MagicMock, patch import pytest @@ -21,10 +22,10 @@ # --------------------------------------------------------------------------- def _make_summary(**kwargs) -> SummaryResult: - defaults = dict( - executive_summary="Two critical accessibility issues require immediate attention.", - priority_recommendations=["Fix ARIA labels", "Add focus indicators"], - root_cause_clusters=[ + defaults: dict = { + "executive_summary": "Two critical accessibility issues require immediate attention.", + "priority_recommendations": ["Fix ARIA labels", "Add focus indicators"], + "root_cause_clusters": [ RootCauseCluster( label="Missing ARIA labels", finding_titles=["No alt text on logo"], @@ -32,10 +33,10 @@ def _make_summary(**kwargs) -> SummaryResult: suggested_fix="Add aria-label to all icon buttons", ) ], - false_positive_candidates=["Console error on third-party script"], - ) + "false_positive_candidates": ["Console error on third-party script"], + } defaults.update(kwargs) - return SummaryResult(**defaults) + return SummaryResult(**defaults) # type: ignore[arg-type] _VALID_LLM_RESPONSE = json.dumps({ @@ -341,11 +342,11 @@ def test_no_cluster_section_when_empty(self, tmp_path): # --------------------------------------------------------------------------- class TestJSONSummarySection: - def _report_data(self, session: TestSession, tmp_path) -> dict: + def _report_data(self, session: TestSession, tmp_path) -> dict[str, Any]: from qa_agent.reporters.json_reporter import JSONReporter reporter = JSONReporter(str(tmp_path)) filepath = reporter.generate(session) - return json.loads(open(filepath).read()) + return cast(dict[str, Any], json.loads(open(filepath).read())) def test_summary_null_when_none(self, tmp_path): session = make_session_with_findings() From 57ee45c39e56e78439ad28d33db3ed196278938d Mon Sep 17 00:00:00 2001 From: Bill Richards <richards.bill0@gmail.com> Date: Tue, 16 Jun 2026 21:45:19 -0400 Subject: [PATCH 4/6] Fix mypy --check-untyped-defs errors across source and tests Adds Optional/dict type annotations where mypy's stricter untyped-def checking surfaced real gaps, plus assert guards in agent.py matching its existing Optional-narrowing pattern. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --- qa_agent/agent.py | 13 +++++++++++-- qa_agent/config.py | 2 +- qa_agent/testers/accessibility.py | 3 ++- qa_agent/testers/errors.py | 5 +++-- qa_agent/testers/keyboard.py | 6 ++++-- qa_agent/testers/mouse.py | 3 ++- tests/test_agent.py | 10 +++++----- tests/test_batch.py | 4 +++- tests/test_concurrency.py | 2 +- tests/test_config.py | 2 +- tests/web/test_server.py | 5 +++-- 11 files changed, 36 insertions(+), 19 deletions(-) diff --git a/qa_agent/agent.py b/qa_agent/agent.py index 3bb7f4a..d47c18a 100644 --- a/qa_agent/agent.py +++ b/qa_agent/agent.py @@ -190,6 +190,9 @@ def _generate_test_plan(self): from .llm_client import LLMError from .plan_cache import PlanCache + assert self.session is not None + assert self.config.instructions is not None + cache = PlanCache() if self.config.use_plan_cache else None cache_key = PlanCache.make_key(self.config.instructions, self.config.urls) if cache else None @@ -236,6 +239,7 @@ def _generate_test_plan(self): def _apply_test_plan(self): """Print the test plan summary and enqueue any suggested URLs.""" + assert self.test_plan is not None self.console.print_progress(f"Test plan: {self.test_plan.summary}") if self.test_plan.focus_areas: self.console.print_progress( @@ -267,6 +271,7 @@ def _generate_summary(self): """Call the LLM post-run to produce a narrative summary of findings.""" from .summarizer import generate_summary + assert self.session is not None self.console.print_progress("Generating results summary with AI...") result = generate_summary( session=self.session, @@ -448,6 +453,7 @@ def _run_explore_mode(self): # Discover new links if current_depth < self.config.max_depth: + assert self.page is not None new_links = self._discover_links(self.page, url) for link in new_links: new_url = link['href'] @@ -466,6 +472,7 @@ def _run_concurrent(self): is performed once on a bootstrap context and exported as a ``storage_state`` dict that seeds every worker context. """ + assert self.session is not None storage_state = self._bootstrap_auth() if self.config.mode == TestMode.FOCUSED: @@ -874,13 +881,14 @@ def _take_screenshot(self, page: Page, name: str) -> str | None: def _cleanup(self): """Clean up browser resources.""" - if self.config.recording.enabled and self.context: + if self.config.recording.enabled and self.context and self.page: # Get video path try: video = self.page.video if video: video_path = video.path() - self.session.recording_path = video_path + assert self.session is not None + self.session.recording_path = str(video_path) except Exception: pass @@ -891,6 +899,7 @@ def _cleanup(self): def _generate_reports(self): """Generate all configured reports.""" + assert self.session is not None for reporter in self.reporters: if isinstance(reporter, ConsoleReporter): reporter.generate(self.session) diff --git a/qa_agent/config.py b/qa_agent/config.py index 6cfb3d2..9cce86b 100644 --- a/qa_agent/config.py +++ b/qa_agent/config.py @@ -30,7 +30,7 @@ class AuthConfig: username_selector: str | None = None # Selector for username field password_selector: str | None = None # Selector for password field submit_selector: str | None = None # Selector for submit button - cookies: dict | None = None # Pre-set cookies for authentication + cookies: dict | list[dict] | None = None # Pre-set cookies for authentication headers: dict | None = None # Custom headers (e.g., Bearer token) diff --git a/qa_agent/testers/accessibility.py b/qa_agent/testers/accessibility.py index 50dc64b..7519b29 100644 --- a/qa_agent/testers/accessibility.py +++ b/qa_agent/testers/accessibility.py @@ -1,6 +1,7 @@ """Accessibility testing module.""" import logging +from typing import Any from playwright.sync_api import Page @@ -39,7 +40,7 @@ def _test_images_alt_text(self): images = self.page.locator('img:visible') count = images.count() - issues = { + issues: dict[str, Any] = { "missing_alt": [], "empty_decorative": 0, "suspicious_alt": [], diff --git a/qa_agent/testers/errors.py b/qa_agent/testers/errors.py index 7e5fc6d..19e907a 100644 --- a/qa_agent/testers/errors.py +++ b/qa_agent/testers/errors.py @@ -1,6 +1,7 @@ """Error detection module for console and network errors.""" from datetime import datetime +from typing import Any from playwright.sync_api import ConsoleMessage, Page, Request, Response @@ -81,7 +82,7 @@ def _analyze_console_errors(self): if len(errors) > 0: # Group similar errors - error_groups = {} + error_groups: dict[str, list[dict[str, Any]]] = {} for error in errors: key = error["text"][:100] if key not in error_groups: @@ -145,7 +146,7 @@ def _analyze_network_errors(self): return # Group by status code - status_groups = {} + status_groups: dict[Any, list[dict[str, Any]]] = {} for error in self.network_errors: status = error.get("status", "failed") if status not in status_groups: diff --git a/qa_agent/testers/keyboard.py b/qa_agent/testers/keyboard.py index d29ce5f..f65a5ee 100644 --- a/qa_agent/testers/keyboard.py +++ b/qa_agent/testers/keyboard.py @@ -1,6 +1,7 @@ """Keyboard navigation and input testing.""" import logging +from typing import Any from playwright.sync_api import Page @@ -40,7 +41,7 @@ def _test_tab_navigation(self): self.page.evaluate("document.body.focus()") self.page.keyboard.press("Tab") - visited_elements = [] + visited_elements: list[dict[str, Any]] = [] max_tabs = 100 # Prevent infinite loops tabs_pressed = 0 @@ -236,7 +237,8 @@ def _test_enter_activation(self): if not is_focused: tag = element.evaluate("el => el.tagName.toLowerCase()") - text = element.text_content()[:30] if element.text_content() else "" + content = element.text_content() + text = content[:30] if content else "" self.findings.append(Finding( title="Interactive element not focusable", description=f"{tag} element cannot receive focus", diff --git a/qa_agent/testers/mouse.py b/qa_agent/testers/mouse.py index 73eb812..5b30199 100644 --- a/qa_agent/testers/mouse.py +++ b/qa_agent/testers/mouse.py @@ -191,7 +191,8 @@ def _test_double_click(self): after_state = self.page.content()[:500] if initial_state == after_state: - text = element.text_content()[:30] if element.text_content() else "" + content = element.text_content() + text = content[:30] if content else "" self.findings.append(Finding( title="Double-click handler has no effect", description="Element has ondblclick but double-clicking produces no visible change", diff --git a/tests/test_agent.py b/tests/test_agent.py index 7148471..6aa7afb 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -3,7 +3,7 @@ from __future__ import annotations import threading -from typing import Literal +from typing import Any, Literal from unittest.mock import MagicMock, patch import pytest @@ -273,7 +273,7 @@ def mock_accessibility_run(self): call_count[0] += 1 return [finding] if call_count[0] == 1 else [] - patchers = [ + patchers: list[Any] = [ patch("qa_agent.agent.KeyboardTester.run", return_value=[]), patch("qa_agent.agent.MouseTester.run", return_value=[]), patch("qa_agent.agent.FormTester.run", return_value=[]), @@ -335,7 +335,7 @@ def test_tester_exception_does_not_abort_run(self): def raising_run(self): raise RuntimeError("tester exploded") - patchers = [ + patchers: list[Any] = [ patch("qa_agent.agent.KeyboardTester.run", raising_run), patch("qa_agent.agent.MouseTester.run", return_value=[]), patch("qa_agent.agent.FormTester.run", return_value=[]), @@ -574,7 +574,7 @@ class TestQAAgentParallel: def _patch_testers(self, accessibility_run=None): from unittest.mock import patch as _patch - targets = { + targets: dict[str, Any] = { "qa_agent.agent.KeyboardTester.run": [], "qa_agent.agent.MouseTester.run": [], "qa_agent.agent.FormTester.run": [], @@ -583,7 +583,7 @@ def _patch_testers(self, accessibility_run=None): "qa_agent.agent.ErrorDetector.attach_listeners": None, "qa_agent.agent.ErrorDetector.get_summary": {}, } - patchers = [] + patchers: list[Any] = [] for target, retval in targets.items(): if target == "qa_agent.agent.AccessibilityTester.run" and accessibility_run: patchers.append(_patch(target, accessibility_run)) diff --git a/tests/test_batch.py b/tests/test_batch.py index a886a6d..1742c8c 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -27,7 +27,9 @@ def fake_run(self): with BatchRunner(pool_size=4) as runner: cfgs = [_cfg(f"https://example.com/{i}") for i in range(5)] results = runner.run_all(cfgs) - assert [r.session_id for r in results] == [f"https://example.com/{i}" for i in range(5)] + assert [r.session_id for r in results if isinstance(r, TestSession)] == [ + f"https://example.com/{i}" for i in range(5) + ] def test_per_job_exception_isolated(self, monkeypatch): def fake_run(self): diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index c66aa5d..9be153d 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -37,7 +37,7 @@ def test_dedup_on_seed_and_add(self): def test_depth_limit_blocks_deeper_links(self): f = Frontier(max_pages=10, max_depth=1) f.seed(["root"]) - url, depth = f.claim() + url, depth = f.claim() # type: ignore[misc] assert depth == 0 f.add_links(["child"], parent_depth=0) # depth 1 — allowed f.add_links(["grandchild"], parent_depth=1) # depth 2 — rejected diff --git a/tests/test_config.py b/tests/test_config.py index 3cf323e..69dbe02 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -139,7 +139,7 @@ def test_value_above_max_is_clamped(self): assert cfg.rate_limit == cfg.RATE_LIMIT_MAX def test_non_numeric_falls_back_to_default(self): - assert TestConfig(rate_limit="not-a-number").rate_limit == 3.0 + assert TestConfig(rate_limit="not-a-number").rate_limit == 3.0 # type: ignore[arg-type] class TestTestMode: diff --git a/tests/web/test_server.py b/tests/web/test_server.py index f045477..58241a6 100644 --- a/tests/web/test_server.py +++ b/tests/web/test_server.py @@ -80,6 +80,7 @@ def test_auth_cookies_malformed_json_string_set_to_none(self): "urls": ["https://example.com"], "auth": {"cookies": "not json {{{"}, }) + assert config.auth is not None assert config.auth.cookies is None def test_screenshot_on_interaction_requires_enabled(self): @@ -370,8 +371,8 @@ def test_unknown_session_returns_404(self, client, tmp_output): class TestQueueWriter: def _writer(self): - q = queue.Queue() - events = [] + q: queue.Queue = queue.Queue() + events: list[dict] = [] return _QueueWriter(q, events), q, events def test_log_line_emits_log_event(self): From 323f2dc8a7bca59db90b00efad974cf501c6961f Mon Sep 17 00:00:00 2001 From: Bill Richards <richards.bill0@gmail.com> Date: Tue, 16 Jun 2026 21:51:01 -0400 Subject: [PATCH 5/6] Fix Windows UnicodeDecodeError in test_summarizer.py file reads open() without encoding= falls back to the platform default (cp1252 on Windows), which chokes on the UTF-8 em-dashes in generated markdown/JSON reports. Pass encoding="utf-8" explicitly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --- tests/test_summarizer.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_summarizer.py b/tests/test_summarizer.py index dc55eaa..30390fc 100644 --- a/tests/test_summarizer.py +++ b/tests/test_summarizer.py @@ -285,7 +285,7 @@ def _report_content(self, session: TestSession, tmp_path) -> str: from qa_agent.reporters.markdown import MarkdownReporter reporter = MarkdownReporter(str(tmp_path)) filepath = reporter.generate(session) - return open(filepath).read() + return open(filepath, encoding="utf-8").read() def test_summary_section_absent_when_none(self, tmp_path): session = make_session_with_findings() @@ -346,7 +346,7 @@ def _report_data(self, session: TestSession, tmp_path) -> dict[str, Any]: from qa_agent.reporters.json_reporter import JSONReporter reporter = JSONReporter(str(tmp_path)) filepath = reporter.generate(session) - return cast(dict[str, Any], json.loads(open(filepath).read())) + return cast(dict[str, Any], json.loads(open(filepath, encoding="utf-8").read())) def test_summary_null_when_none(self, tmp_path): session = make_session_with_findings() From 62d8db88fe719fec7e40a8185a8645149619604e Mon Sep 17 00:00:00 2001 From: Bill Richards <richards.bill0@gmail.com> Date: Tue, 16 Jun 2026 21:55:31 -0400 Subject: [PATCH 6/6] Add explicit utf-8 encoding to remaining unencoded file reads open()/read_text() without encoding= falls back to the platform default (cp1252 on Windows), which breaks on non-ASCII bytes. Fixes the auth-file and cookies-file readers in cli.py (real-world JSON configs may contain non-ASCII), and the server.py source read in test_packaging.py (the file contains em dashes/bullets). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --- qa_agent/cli.py | 4 ++-- tests/test_packaging.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/qa_agent/cli.py b/qa_agent/cli.py index e948456..3fce648 100644 --- a/qa_agent/cli.py +++ b/qa_agent/cli.py @@ -23,7 +23,7 @@ def parse_auth_config(auth_str: str | None, auth_file: str | None) -> AuthConfig """Parse authentication configuration from string or file.""" if auth_file: try: - with open(auth_file) as f: + with open(auth_file, encoding="utf-8") as f: auth_data = json.load(f) return AuthConfig(**auth_data) except Exception as e: @@ -363,7 +363,7 @@ def main(): # Handle cookies file if args.cookies: try: - with open(args.cookies) as f: + with open(args.cookies, encoding="utf-8") as f: cookies = json.load(f) if auth_config: auth_config.cookies = cookies diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 12da5ec..bb49e5d 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -235,7 +235,7 @@ def test_pyproject_declares_package_data_for_web(self): def test_flask_app_references_correct_template_dir(self): """Flask app in server.py must resolve templates relative to the web package.""" - server_src = (REPO_ROOT / "qa_agent" / "web" / "server.py").read_text() + server_src = (REPO_ROOT / "qa_agent" / "web" / "server.py").read_text(encoding="utf-8") # Flask(name) uses the module's directory — templates must live there assert "Flask(__name__" in server_src, ( "Flask app is not created with __name__; template resolution may be wrong"