Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import re | |
| from collections import Counter | |
| from dataclasses import asdict | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| from .llm_runtime import call_structured_llm, is_llm_enabled | |
| from .models import AgentRuntimeConfig | |
| from .models import RequirementItem, ReviewReport, RunSession, TestCaseDesign | |
| def _split_requirements(requirements_text: str) -> list[str]: | |
| lines = [line.strip(" -*\t") for line in requirements_text.splitlines()] | |
| items = [line for line in lines if line] | |
| if items: | |
| return items | |
| sentences = [part.strip() for part in re.split(r"[.!?]\s+", requirements_text) if part.strip()] | |
| return sentences or [requirements_text.strip()] | |
| class RequirementsAnalystAgent: | |
| name: str = "Requirements Analyst" | |
| last_execution: dict[str, Any] = field(default_factory=dict) | |
| def analyze( | |
| self, | |
| requirements_text: str, | |
| feedback_messages: list[str] | None = None, | |
| runtime_config: AgentRuntimeConfig | None = None, | |
| run_session: RunSession | None = None, | |
| ) -> list[RequirementItem]: | |
| feedback_messages = feedback_messages or [] | |
| if is_llm_enabled(runtime_config): | |
| return self._analyze_with_llm(requirements_text, feedback_messages, runtime_config, run_session) | |
| items = [] | |
| for index, raw_text in enumerate(_split_requirements(requirements_text), start=1): | |
| requirement_id = f"REQ-{index:03d}" | |
| normalized = raw_text.strip().rstrip(".") | |
| criteria = self._extract_acceptance_criteria(normalized) | |
| assumptions = self._build_assumptions(normalized, criteria) | |
| criteria, assumptions = self._apply_feedback(requirement_id, criteria, assumptions, feedback_messages) | |
| items.append( | |
| RequirementItem( | |
| requirement_id=requirement_id, | |
| original_text=raw_text, | |
| normalized_text=normalized, | |
| priority=self._classify_priority(normalized), | |
| acceptance_criteria=criteria, | |
| assumptions=assumptions, | |
| ) | |
| ) | |
| items, validation_findings = self._standardize_requirement_items(items) | |
| self.last_execution = { | |
| "mode": "structured", | |
| "reasoning_source": "structured_trace", | |
| "notes": [], | |
| "llm_used": False, | |
| "validation_findings": validation_findings, | |
| "contract_version": "requirements.v1", | |
| } | |
| self._update_memory(run_session, items, feedback_messages, validation_findings) | |
| return items | |
| def _analyze_with_llm( | |
| self, | |
| requirements_text: str, | |
| feedback_messages: list[str], | |
| runtime_config: AgentRuntimeConfig, | |
| run_session: RunSession | None, | |
| ) -> list[RequirementItem]: | |
| system_prompt = ( | |
| f"You are {runtime_config.agent_name}.\n" | |
| f"{runtime_config.directives}\n" | |
| "Return only structured output. Extract only what the requirement text supports. " | |
| "Make uncertainty explicit. Keep requirement order stable." | |
| ) | |
| feedback_block = "\n".join(f"- {message}" for message in feedback_messages) or "- none" | |
| schema = { | |
| "type": "object", | |
| "additionalProperties": False, | |
| "properties": { | |
| "requirements": { | |
| "type": "array", | |
| "items": { | |
| "type": "object", | |
| "additionalProperties": False, | |
| "properties": { | |
| "original_text": {"type": "string"}, | |
| "normalized_text": {"type": "string"}, | |
| "priority": {"type": "string", "enum": ["high", "medium", "normal"]}, | |
| "acceptance_criteria": {"type": "array", "items": {"type": "string"}}, | |
| "assumptions": {"type": "array", "items": {"type": "string"}}, | |
| "trace_notes": {"type": "array", "items": {"type": "string"}}, | |
| }, | |
| "required": [ | |
| "original_text", | |
| "normalized_text", | |
| "priority", | |
| "acceptance_criteria", | |
| "assumptions", | |
| "trace_notes", | |
| ], | |
| }, | |
| }, | |
| "run_notes": {"type": "array", "items": {"type": "string"}}, | |
| }, | |
| "required": ["requirements", "run_notes"], | |
| } | |
| user_prompt = ( | |
| "Raw requirement text:\n" | |
| f"{requirements_text}\n\n" | |
| "Incoming upstream feedback:\n" | |
| f"{feedback_block}\n\n" | |
| "Important update rules:\n" | |
| "- Keep one requirement object for each original requirement statement.\n" | |
| "- Revise only the targeted requirement objects when feedback names specific requirement IDs.\n" | |
| "- Preserve untargeted requirements instead of dropping them.\n" | |
| "- Never return an empty requirements array when the source text contains requirement statements.\n\n" | |
| "Return one requirement object per requirement statement. " | |
| "Acceptance criteria must be observable and testable. " | |
| "Assumptions must stay explicit instead of being silently resolved." | |
| ) | |
| response, metadata = call_structured_llm( | |
| runtime_config=runtime_config, | |
| system_prompt=system_prompt, | |
| user_prompt=user_prompt, | |
| schema_name="requirements_analysis", | |
| schema=schema, | |
| ) | |
| items = [] | |
| raw_items = response.get("requirements") or [] | |
| for index, item in enumerate(raw_items, start=1): | |
| original_text = _clean_text(item.get("original_text")) | |
| normalized_text = _clean_text(item.get("normalized_text")) or original_text.strip().rstrip(".") | |
| priority = _clean_priority(item.get("priority")) | |
| acceptance_criteria = _clean_string_list(item.get("acceptance_criteria")) | |
| assumptions = _clean_string_list(item.get("assumptions")) | |
| items.append( | |
| RequirementItem( | |
| requirement_id=f"REQ-{index:03d}", | |
| original_text=original_text, | |
| normalized_text=normalized_text, | |
| priority=priority, | |
| acceptance_criteria=acceptance_criteria, | |
| assumptions=assumptions, | |
| ) | |
| ) | |
| items, validation_findings = self._standardize_requirement_items(items) | |
| items, validation_findings = self._recover_previous_requirements_if_needed( | |
| items, | |
| validation_findings, | |
| feedback_messages, | |
| run_session, | |
| ) | |
| self.last_execution = { | |
| "mode": "llm", | |
| "reasoning_source": "llm_structured_output", | |
| "notes": _clean_string_list(response.get("run_notes")) | |
| + _flatten_trace_notes(raw_items, "trace_notes"), | |
| "llm_used": True, | |
| "metadata": metadata, | |
| "validation_findings": validation_findings, | |
| "contract_version": "requirements.v1", | |
| } | |
| self._update_memory(run_session, items, feedback_messages, validation_findings) | |
| return items | |
| def _recover_previous_requirements_if_needed( | |
| self, | |
| items: list[RequirementItem], | |
| validation_findings: list[str], | |
| feedback_messages: list[str], | |
| run_session: RunSession | None, | |
| ) -> tuple[list[RequirementItem], list[str]]: | |
| if items or not feedback_messages or not run_session: | |
| return items, validation_findings | |
| previous_items = run_session.working_memory.read_shared("requirements", []) | |
| if not previous_items: | |
| return items, validation_findings | |
| recovered_items = [] | |
| for raw_item in previous_items: | |
| if not isinstance(raw_item, dict): | |
| continue | |
| recovered_items.append( | |
| RequirementItem( | |
| requirement_id=str(raw_item.get("requirement_id", "")).strip(), | |
| original_text=str(raw_item.get("original_text", "")).strip(), | |
| normalized_text=str(raw_item.get("normalized_text", "")).strip(), | |
| priority=_clean_priority(raw_item.get("priority")), | |
| acceptance_criteria=_clean_string_list(raw_item.get("acceptance_criteria")), | |
| assumptions=_clean_string_list(raw_item.get("assumptions")), | |
| ) | |
| ) | |
| recovered_items, recovered_findings = self._standardize_requirement_items(recovered_items) | |
| if not recovered_items: | |
| return items, validation_findings | |
| fallback_findings = [ | |
| *validation_findings, | |
| "Requirements Analyst returned no valid items after feedback, so the previous valid requirement set was preserved for this round.", | |
| ] | |
| for finding in recovered_findings: | |
| if finding not in fallback_findings: | |
| fallback_findings.append(finding) | |
| return recovered_items, fallback_findings | |
| def _update_memory( | |
| self, | |
| run_session: RunSession | None, | |
| items: list[RequirementItem], | |
| feedback_messages: list[str], | |
| validation_findings: list[str], | |
| ) -> None: | |
| if not run_session: | |
| return | |
| run_session.working_memory.write_shared( | |
| "requirements", | |
| [asdict(item) for item in items], | |
| author=self.name, | |
| ) | |
| run_session.working_memory.write_agent( | |
| "requirements_analyst", | |
| "last_feedback_messages", | |
| list(feedback_messages), | |
| ) | |
| run_session.working_memory.write_agent( | |
| "requirements_analyst", | |
| "validation_findings", | |
| list(validation_findings), | |
| ) | |
| run_session.working_memory.add_note( | |
| f"Requirements Analyst produced {len(items)} requirement item(s)." | |
| ) | |
| def _standardize_requirement_items( | |
| self, | |
| items: list[RequirementItem], | |
| ) -> tuple[list[RequirementItem], list[str]]: | |
| standardized: list[RequirementItem] = [] | |
| validation_findings: list[str] = [] | |
| for item in items: | |
| original_text = item.original_text.strip() | |
| normalized_text = item.normalized_text.strip() | |
| acceptance_criteria = [criterion.strip() for criterion in item.acceptance_criteria if criterion.strip()] | |
| assumptions = [assumption.strip() for assumption in item.assumptions if assumption.strip()] | |
| if not original_text: | |
| validation_findings.append(f"{item.requirement_id} was dropped because original_text was empty.") | |
| continue | |
| if not normalized_text: | |
| validation_findings.append(f"{item.requirement_id} was dropped because normalized_text was empty.") | |
| continue | |
| if not acceptance_criteria: | |
| validation_findings.append( | |
| f"{item.requirement_id} was dropped because acceptance_criteria was empty." | |
| ) | |
| continue | |
| standardized.append( | |
| RequirementItem( | |
| requirement_id=item.requirement_id, | |
| original_text=original_text, | |
| normalized_text=normalized_text, | |
| priority=item.priority, | |
| acceptance_criteria=acceptance_criteria, | |
| assumptions=assumptions, | |
| ) | |
| ) | |
| if not standardized: | |
| validation_findings.append( | |
| "Requirements Analyst did not produce any valid requirement items that satisfy the requirements.v1 contract." | |
| ) | |
| return standardized, validation_findings | |
| def _extract_acceptance_criteria(self, text: str) -> list[str]: | |
| lowered = text.lower() | |
| criteria = [f"The system shall satisfy the requirement: {text}"] | |
| if any(keyword in lowered for keyword in ("login", "logga in", "autentis", "sign in")): | |
| criteria.append("The user shall be authenticated when valid credentials are supplied.") | |
| criteria.append("Invalid credentials shall produce a clear error message.") | |
| if any(keyword in lowered for keyword in ("visa", "display", "list", "översikt", "dashboard")): | |
| criteria.append("Relevant information shall be presented without broken or empty states.") | |
| if any(keyword in lowered for keyword in ("skapa", "save", "spara", "registrera", "submit")): | |
| criteria.append("Input data shall be validated before it is stored.") | |
| criteria.append("A successful operation shall provide user confirmation.") | |
| if any(keyword in lowered for keyword in ("admin", "behörighet", "access", "roll")): | |
| criteria.append("Unauthorized access shall be blocked and logged.") | |
| return criteria | |
| def _build_assumptions(self, text: str, criteria: list[str]) -> list[str]: | |
| assumptions = [] | |
| if len(criteria) == 1: | |
| assumptions.append("The requirement likely needs more explicit acceptance criteria.") | |
| if not any(word in text.lower() for word in ("fel", "error", "ogiltig", "invalid")): | |
| assumptions.append("Error handling is not explicit and should be reviewed.") | |
| return assumptions | |
| def _classify_priority(self, text: str) -> str: | |
| lowered = text.lower() | |
| if any(keyword in lowered for keyword in ("måste", "must", "critical", "viktig")): | |
| return "high" | |
| if any(keyword in lowered for keyword in ("bör", "should", "nice")): | |
| return "medium" | |
| return "normal" | |
| def _apply_feedback( | |
| self, | |
| requirement_id: str, | |
| criteria: list[str], | |
| assumptions: list[str], | |
| feedback_messages: list[str], | |
| ) -> tuple[list[str], list[str]]: | |
| if not feedback_messages: | |
| return criteria, assumptions | |
| combined_feedback = " ".join(feedback_messages).lower() | |
| updated_criteria = list(criteria) | |
| updated_assumptions = list(assumptions) | |
| feedback_targets_requirement = requirement_id.lower() in combined_feedback or "requirement" in combined_feedback | |
| if feedback_targets_requirement and "acceptance criteria" in combined_feedback: | |
| clarification = "Observable success and failure outcomes shall be explicitly defined." | |
| if clarification not in updated_criteria: | |
| updated_criteria.append(clarification) | |
| updated_assumptions = [ | |
| assumption | |
| for assumption in updated_assumptions | |
| if "more explicit acceptance criteria" not in assumption.lower() | |
| ] | |
| if feedback_targets_requirement and any( | |
| phrase in combined_feedback | |
| for phrase in ("error handling", "negative path", "invalid credentials", "invalid input") | |
| ): | |
| clarification = "Negative and error flows shall be explicitly defined and testable." | |
| if clarification not in updated_criteria: | |
| updated_criteria.append(clarification) | |
| updated_assumptions = [ | |
| assumption | |
| for assumption in updated_assumptions | |
| if "error handling is not explicit" not in assumption.lower() | |
| ] | |
| return updated_criteria, updated_assumptions | |
| class TestDesignAgent: | |
| name: str = "Test Design Agent" | |
| last_execution: dict[str, Any] = field(default_factory=dict) | |
| def design( | |
| self, | |
| requirements: list[RequirementItem], | |
| feedback_messages: list[str] | None = None, | |
| runtime_config: AgentRuntimeConfig | None = None, | |
| run_session: RunSession | None = None, | |
| ) -> list[TestCaseDesign]: | |
| feedback_messages = feedback_messages or [] | |
| if is_llm_enabled(runtime_config): | |
| return self._design_with_llm(requirements, feedback_messages, runtime_config, run_session) | |
| designs = [] | |
| design_index = 1 | |
| for req in requirements: | |
| requirement_designs = self._build_structured_designs(req, feedback_messages, design_index) | |
| designs.extend(requirement_designs) | |
| design_index += len(requirement_designs) | |
| self.last_execution = { | |
| "mode": "structured", | |
| "reasoning_source": "structured_trace", | |
| "notes": [], | |
| "llm_used": False, | |
| "feedback_messages_to_requirements": self._feedback_messages_to_requirements(requirements), | |
| } | |
| self._update_memory(run_session, designs, feedback_messages, self.last_execution["feedback_messages_to_requirements"]) | |
| return designs | |
| def _design_with_llm( | |
| self, | |
| requirements: list[RequirementItem], | |
| feedback_messages: list[str], | |
| runtime_config: AgentRuntimeConfig, | |
| run_session: RunSession | None, | |
| ) -> list[TestCaseDesign]: | |
| requirements_payload = [ | |
| { | |
| "requirement_id": item.requirement_id, | |
| "normalized_text": item.normalized_text, | |
| "priority": item.priority, | |
| "acceptance_criteria": item.acceptance_criteria, | |
| "assumptions": item.assumptions, | |
| } | |
| for item in requirements | |
| ] | |
| feedback_block = "\n".join(f"- {message}" for message in feedback_messages) or "- none" | |
| system_prompt = ( | |
| f"You are {runtime_config.agent_name}.\n" | |
| f"{runtime_config.directives}\n" | |
| "Produce concrete planned test cases, not placeholders. " | |
| "Use action-oriented test titles. Keep traceability to requirement IDs." | |
| ) | |
| schema = { | |
| "type": "object", | |
| "additionalProperties": False, | |
| "properties": { | |
| "test_designs": { | |
| "type": "array", | |
| "items": { | |
| "type": "object", | |
| "additionalProperties": False, | |
| "properties": { | |
| "requirement_id": {"type": "string"}, | |
| "title": {"type": "string"}, | |
| "test_type": {"type": "string"}, | |
| "preconditions": {"type": "array", "items": {"type": "string"}}, | |
| "steps": {"type": "array", "items": {"type": "string"}}, | |
| "expected_results": {"type": "array", "items": {"type": "string"}}, | |
| "oracle": {"type": "string"}, | |
| "risks": {"type": "array", "items": {"type": "string"}}, | |
| "trace_notes": {"type": "array", "items": {"type": "string"}}, | |
| }, | |
| "required": [ | |
| "requirement_id", | |
| "title", | |
| "test_type", | |
| "preconditions", | |
| "steps", | |
| "expected_results", | |
| "oracle", | |
| "risks", | |
| "trace_notes", | |
| ], | |
| }, | |
| }, | |
| "feedback_messages_to_requirements": {"type": "array", "items": {"type": "string"}}, | |
| "run_notes": {"type": "array", "items": {"type": "string"}}, | |
| }, | |
| "required": ["test_designs", "feedback_messages_to_requirements", "run_notes"], | |
| } | |
| user_prompt = ( | |
| "Structured requirements:\n" | |
| f"{requirements_payload}\n\n" | |
| "Incoming upstream feedback:\n" | |
| f"{feedback_block}\n\n" | |
| "Create realistic QA design coverage, not a flat one-to-one mapping. " | |
| "A single requirement may need multiple planned test cases such as positive, negative, boundary, " | |
| "authorization, validation, or state-based checks. " | |
| "Keep every test case linked to exactly one primary requirement_id in this schema, but produce multiple " | |
| "test cases for the same requirement when the behavior clearly needs it. " | |
| "If a requirement is too unclear, keep the test design as strong as possible but also add targeted " | |
| "feedback messages to the requirements analyst." | |
| ) | |
| response, metadata = call_structured_llm( | |
| runtime_config=runtime_config, | |
| system_prompt=system_prompt, | |
| user_prompt=user_prompt, | |
| schema_name="test_design", | |
| schema=schema, | |
| ) | |
| raw_designs = response.get("test_designs") or [] | |
| designs = [] | |
| requirements_by_id = {item.requirement_id: item for item in requirements} | |
| for index, item in enumerate(raw_designs, start=1): | |
| requirement_id = _clean_text(item.get("requirement_id")) or f"REQ-{index:03d}" | |
| mapped_requirement = requirements_by_id.get(requirement_id) | |
| designs.append( | |
| TestCaseDesign( | |
| test_case_id=f"TC-{index:03d}", | |
| requirement_id=requirement_id, | |
| title=_clean_text(item.get("title")) or self._fallback_llm_title(mapped_requirement, requirement_id), | |
| test_type=_clean_text(item.get("test_type")) or "scenario", | |
| preconditions=_clean_string_list(item.get("preconditions")) or ["The application is available."], | |
| steps=_clean_string_list(item.get("steps")) or self._build_steps(mapped_requirement) if mapped_requirement else [], | |
| expected_results=_clean_string_list(item.get("expected_results")) or ( | |
| list(mapped_requirement.acceptance_criteria) if mapped_requirement else [] | |
| ), | |
| oracle=_clean_text(item.get("oracle")) | |
| or "All expected results are satisfied and no unauthorized side effects are observed.", | |
| risks=_clean_string_list(item.get("risks")), | |
| ) | |
| ) | |
| self.last_execution = { | |
| "mode": "llm", | |
| "reasoning_source": "llm_structured_output", | |
| "notes": _clean_string_list(response.get("run_notes")) | |
| + _flatten_trace_notes(raw_designs, "trace_notes"), | |
| "llm_used": True, | |
| "metadata": metadata, | |
| "feedback_messages_to_requirements": _clean_string_list( | |
| response.get("feedback_messages_to_requirements") | |
| ), | |
| } | |
| self._update_memory(run_session, designs, feedback_messages, self.last_execution["feedback_messages_to_requirements"]) | |
| return designs | |
| def _update_memory( | |
| self, | |
| run_session: RunSession | None, | |
| designs: list[TestCaseDesign], | |
| feedback_messages: list[str], | |
| feedback_to_requirements: list[str], | |
| ) -> None: | |
| if not run_session: | |
| return | |
| run_session.working_memory.write_shared( | |
| "test_designs", | |
| [asdict(item) for item in designs], | |
| author=self.name, | |
| ) | |
| run_session.working_memory.write_agent( | |
| "test_design", | |
| "last_feedback_messages", | |
| list(feedback_messages), | |
| ) | |
| run_session.working_memory.write_agent( | |
| "test_design", | |
| "feedback_messages_to_requirements", | |
| list(feedback_to_requirements), | |
| ) | |
| run_session.working_memory.add_note( | |
| f"Test Design Agent produced {len(designs)} design(s)." | |
| ) | |
| def _build_structured_designs( | |
| self, | |
| req: RequirementItem, | |
| feedback_messages: list[str], | |
| start_index: int, | |
| ) -> list[TestCaseDesign]: | |
| variants = [self._build_primary_variant(req)] | |
| if self._should_add_negative_design(req): | |
| variants.append(self._build_negative_variant(req)) | |
| if self._should_add_authorization_design(req): | |
| variants.append(self._build_authorization_variant(req)) | |
| seen_titles = set() | |
| deduplicated_variants = [] | |
| for variant in variants: | |
| title_key = variant["title"].strip().lower() | |
| if title_key in seen_titles: | |
| continue | |
| seen_titles.add(title_key) | |
| deduplicated_variants.append(variant) | |
| designs = [] | |
| for offset, variant in enumerate(deduplicated_variants, start=0): | |
| steps, expected_results, oracle = self._apply_feedback( | |
| req.requirement_id, | |
| variant["steps"], | |
| variant["expected_results"], | |
| feedback_messages, | |
| ) | |
| designs.append( | |
| TestCaseDesign( | |
| test_case_id=f"TC-{start_index + offset:03d}", | |
| requirement_id=req.requirement_id, | |
| title=variant["title"], | |
| test_type=variant["test_type"], | |
| preconditions=variant["preconditions"], | |
| steps=steps, | |
| expected_results=expected_results, | |
| oracle=oracle, | |
| risks=self._identify_risks(req, variant["risk_hints"]), | |
| ) | |
| ) | |
| return designs | |
| def _choose_test_type(self, req: RequirementItem) -> str: | |
| lowered = req.normalized_text.lower() | |
| if any(keyword in lowered for keyword in ("ui", "gui", "knapp", "formulär", "page", "sida")): | |
| return "gui/e2e" | |
| if any(keyword in lowered for keyword in ("api", "endpoint", "service")): | |
| return "api/integration" | |
| if any(keyword in lowered for keyword in ("beräkna", "validate", "regel", "business")): | |
| return "unit" | |
| return "scenario" | |
| def _build_steps(self, req: RequirementItem) -> list[str]: | |
| return [ | |
| f"Identify the functionality associated with {req.requirement_id}.", | |
| "Prepare a positive test case with valid data.", | |
| "Execute the primary flow.", | |
| "Verify the expected result against the acceptance criteria.", | |
| "Execute a negative or boundary case when applicable.", | |
| ] | |
| def _build_primary_variant(self, req: RequirementItem) -> dict[str, Any]: | |
| lowered = req.normalized_text.lower() | |
| if any(keyword in lowered for keyword in ("login", "logga in", "sign in", "authenticate", "autentis")): | |
| return { | |
| "title": "Successful sign-in with valid credentials", | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available.", "A registered user account exists."], | |
| "steps": [ | |
| "Open the sign-in view.", | |
| "Enter a valid email address and password.", | |
| "Submit the sign-in form.", | |
| ], | |
| "expected_results": [ | |
| "The user is authenticated successfully.", | |
| "The application redirects the user to the authorized landing page.", | |
| ], | |
| "risk_hints": [], | |
| } | |
| if any(keyword in lowered for keyword in ("register", "registrera", "new account", "konto", "signup", "sign up")): | |
| return { | |
| "title": "Register a new account with valid mandatory data", | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available.", "No existing account uses the test email address."], | |
| "steps": [ | |
| "Open the registration form.", | |
| "Enter valid mandatory account data.", | |
| "Submit the form.", | |
| ], | |
| "expected_results": [ | |
| "The account is created successfully.", | |
| "The user sees a clear confirmation or is signed in according to the product rule.", | |
| ], | |
| "risk_hints": [], | |
| } | |
| return { | |
| "title": self._build_title(req), | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available.", "Test data is initialized."], | |
| "steps": self._build_steps(req), | |
| "expected_results": [criterion for criterion in req.acceptance_criteria], | |
| "risk_hints": [], | |
| } | |
| def _build_negative_variant(self, req: RequirementItem) -> dict[str, Any]: | |
| lowered = req.normalized_text.lower() | |
| if any(keyword in lowered for keyword in ("login", "logga in", "sign in", "authenticate", "autentis")): | |
| return { | |
| "title": "Reject sign-in with invalid credentials", | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available.", "A registered user account exists."], | |
| "steps": [ | |
| "Open the sign-in view.", | |
| "Enter a valid email address and an invalid password.", | |
| "Submit the sign-in form.", | |
| ], | |
| "expected_results": [ | |
| "Authentication is rejected.", | |
| "A clear error message is shown to the user.", | |
| "No authenticated session is created.", | |
| ], | |
| "risk_hints": ["Negative-path coverage is required for this requirement."], | |
| } | |
| if any(keyword in lowered for keyword in ("register", "registrera", "new account", "konto", "signup", "sign up", "form")): | |
| return { | |
| "title": "Reject account registration with invalid or missing mandatory data", | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available."], | |
| "steps": [ | |
| "Open the registration form.", | |
| "Enter invalid or incomplete mandatory data.", | |
| "Submit the form.", | |
| ], | |
| "expected_results": [ | |
| "The account is not created.", | |
| "Validation feedback identifies the invalid or missing fields.", | |
| ], | |
| "risk_hints": ["Validation coverage is required for this requirement."], | |
| } | |
| return { | |
| "title": f"{self._build_title(req)} - negative path", | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available.", "Test data is initialized."], | |
| "steps": [ | |
| f"Identify the functionality associated with {req.requirement_id}.", | |
| "Prepare invalid, missing, or boundary input data.", | |
| "Execute the negative or failure flow.", | |
| "Observe how the system rejects or handles the input.", | |
| ], | |
| "expected_results": [ | |
| "The system rejects invalid or disallowed input safely.", | |
| "The failure is observable through validation, error messaging, or blocked state change.", | |
| ], | |
| "risk_hints": ["Negative-path coverage is required for this requirement."], | |
| } | |
| def _build_authorization_variant(self, req: RequirementItem) -> dict[str, Any]: | |
| return { | |
| "title": f"{self._build_title(req)} - unauthorized access is blocked", | |
| "test_type": self._choose_test_type(req), | |
| "preconditions": ["The application is available.", "A user without the required role exists."], | |
| "steps": [ | |
| "Authenticate as a user without the required role or permissions.", | |
| "Attempt to access the protected function or view.", | |
| ], | |
| "expected_results": [ | |
| "Access is denied.", | |
| "The user does not see or execute the protected behavior.", | |
| ], | |
| "risk_hints": ["Authorization coverage is required for this requirement."], | |
| } | |
| def _should_add_negative_design(self, req: RequirementItem) -> bool: | |
| lowered = " ".join( | |
| [req.normalized_text, *req.acceptance_criteria, *req.assumptions] | |
| ).lower() | |
| return any( | |
| keyword in lowered | |
| for keyword in ( | |
| "error", | |
| "invalid", | |
| "negative", | |
| "failure", | |
| "fel", | |
| "ogiltig", | |
| "login", | |
| "logga in", | |
| "sign in", | |
| "register", | |
| "registrera", | |
| "form", | |
| "validate", | |
| ) | |
| ) | |
| def _should_add_authorization_design(self, req: RequirementItem) -> bool: | |
| lowered = " ".join([req.normalized_text, *req.acceptance_criteria]).lower() | |
| return any(keyword in lowered for keyword in ("admin", "authorization", "authorisation", "role", "access", "behörighet")) | |
| def _build_title(self, req: RequirementItem) -> str: | |
| cleaned = req.normalized_text.strip() | |
| if cleaned: | |
| if len(cleaned) > 72: | |
| cleaned = cleaned[:69].rstrip() + "..." | |
| return cleaned | |
| return f"Test case for {req.requirement_id}" | |
| def _fallback_llm_title(self, req: RequirementItem | None, requirement_id: str) -> str: | |
| if not req: | |
| return f"Test case for {requirement_id}" | |
| text = req.normalized_text.strip().rstrip(".") | |
| lowered = text.lower() | |
| if any(keyword in lowered for keyword in ("sign in", "logga in", "login")): | |
| return "Successful sign-in with valid credentials" | |
| if any(keyword in lowered for keyword in ("error", "invalid", "ogiltig", "felmeddelande")): | |
| return "Display clear error for invalid credentials" | |
| if "admin" in lowered and any(keyword in lowered for keyword in ("overview", "översikt", "users", "användare")): | |
| return "Admin views registered users overview" | |
| if any(keyword in lowered for keyword in ("register", "registrera", "new account", "konto")): | |
| return "Register a new account through the form" | |
| return text if len(text) <= 72 else text[:69].rstrip() + "..." | |
| def _identify_risks(self, req: RequirementItem, extra_risks: list[str] | None = None) -> list[str]: | |
| risks = [] | |
| if req.assumptions: | |
| risks.append("The requirement contains assumptions that can distort test design.") | |
| if len(req.acceptance_criteria) < 2: | |
| risks.append("Limited specificity can reduce testability.") | |
| for risk in extra_risks or []: | |
| if risk not in risks: | |
| risks.append(risk) | |
| return risks | |
| def _feedback_messages_to_requirements(self, requirements: list[RequirementItem]) -> list[str]: | |
| feedback = [] | |
| for requirement in requirements: | |
| if len(requirement.acceptance_criteria) < 2: | |
| feedback.append( | |
| f"{requirement.requirement_id} needs clearer acceptance criteria before strong test design can continue." | |
| ) | |
| continue | |
| lowered_criteria = " ".join(requirement.acceptance_criteria).lower() | |
| if not any( | |
| phrase in lowered_criteria | |
| for phrase in ("error", "invalid", "negative", "failure") | |
| ): | |
| feedback.append( | |
| f"{requirement.requirement_id} needs explicit negative or error-path acceptance criteria before strong test design can continue." | |
| ) | |
| return feedback[:2] | |
| def _apply_feedback( | |
| self, | |
| requirement_id: str, | |
| steps: list[str], | |
| expected_results: list[str], | |
| feedback_messages: list[str], | |
| ) -> tuple[list[str], list[str], str]: | |
| combined_feedback = " ".join(feedback_messages).lower() | |
| targets_requirement = requirement_id.lower() in combined_feedback or not feedback_messages | |
| updated_steps = list(steps) | |
| updated_expected_results = list(expected_results) | |
| oracle = "All expected results are satisfied and no unauthorized side effects are observed." | |
| if feedback_messages and targets_requirement and any( | |
| phrase in combined_feedback | |
| for phrase in ("weak oracle", "more expected results", "negative scenarios", "negative case") | |
| ): | |
| extra_result = "Negative and boundary outcomes shall be verified explicitly against the acceptance criteria." | |
| if extra_result not in updated_expected_results: | |
| updated_expected_results.append(extra_result) | |
| extra_step = "Compare positive, negative, and boundary outcomes against explicit expected results." | |
| if extra_step not in updated_steps: | |
| updated_steps.append(extra_step) | |
| oracle = ( | |
| "Every positive, negative, and boundary expectation must be explicitly satisfied, " | |
| "and the system must not introduce unauthorized side effects." | |
| ) | |
| return updated_steps, updated_expected_results, oracle | |
| class ReviewAgent: | |
| name: str = "Review Agent" | |
| last_execution: dict[str, Any] = field(default_factory=dict) | |
| def review( | |
| self, | |
| requirements: list[RequirementItem], | |
| test_designs: list[TestCaseDesign], | |
| runtime_config: AgentRuntimeConfig | None = None, | |
| run_session: RunSession | None = None, | |
| ) -> ReviewReport: | |
| if is_llm_enabled(runtime_config): | |
| return self._review_with_llm(requirements, test_designs, runtime_config, run_session) | |
| findings = [] | |
| improvements = [] | |
| covered_requirements = {design.requirement_id for design in test_designs} | |
| design_counts = Counter(design.requirement_id for design in test_designs) | |
| coverage_ratio = len(covered_requirements) / len(requirements) if requirements else 0.0 | |
| for requirement in requirements: | |
| if requirement.requirement_id not in covered_requirements: | |
| findings.append(f"{requirement.requirement_id} is missing a designed test case.") | |
| improvements.append(f"Design at least one test for {requirement.requirement_id}.") | |
| continue | |
| if self._requires_multiple_test_designs(requirement) and design_counts[requirement.requirement_id] < 2: | |
| findings.append( | |
| f"{requirement.requirement_id} has only one designed test case even though the requirement implies multiple behaviors." | |
| ) | |
| improvements.append( | |
| f"Add at least one more targeted test case for {requirement.requirement_id}, such as a negative, validation, authorization, or boundary scenario." | |
| ) | |
| for design in test_designs: | |
| if len(design.expected_results) < 2: | |
| findings.append( | |
| f"{design.test_case_id} has a weak oracle definition and needs more expected results." | |
| ) | |
| improvements.append( | |
| f"Clarify acceptance criteria or add negative scenarios for {design.test_case_id}." | |
| ) | |
| for requirement in requirements: | |
| if requirement.assumptions: | |
| findings.append( | |
| f"{requirement.requirement_id} contains assumptions that should be confirmed before automation." | |
| ) | |
| improvements.append( | |
| f"Clarify requirement {requirement.requirement_id} to reduce misinterpretation." | |
| ) | |
| blocking_findings = any( | |
| "missing a designed test case" in finding | |
| or "has only one designed test case" in finding | |
| or "weak oracle definition" in finding | |
| for finding in findings | |
| ) | |
| approved = coverage_ratio == 1.0 and not blocking_findings and len(findings) <= max(1, len(requirements) // 2) | |
| if approved and not findings: | |
| findings.append("No critical issues were identified.") | |
| report = ReviewReport( | |
| approved=approved, | |
| coverage_ratio=round(coverage_ratio, 2), | |
| findings=findings, | |
| improvement_actions=improvements, | |
| ) | |
| self.last_execution = { | |
| "mode": "structured", | |
| "reasoning_source": "structured_trace", | |
| "notes": [], | |
| "llm_used": False, | |
| "routing_focus": self._infer_routing_focus(findings), | |
| } | |
| self._update_memory(run_session, report) | |
| return report | |
| def _review_with_llm( | |
| self, | |
| requirements: list[RequirementItem], | |
| test_designs: list[TestCaseDesign], | |
| runtime_config: AgentRuntimeConfig, | |
| run_session: RunSession | None, | |
| ) -> ReviewReport: | |
| requirements_payload = [ | |
| { | |
| "requirement_id": item.requirement_id, | |
| "normalized_text": item.normalized_text, | |
| "acceptance_criteria": item.acceptance_criteria, | |
| "assumptions": item.assumptions, | |
| } | |
| for item in requirements | |
| ] | |
| designs_payload = [ | |
| { | |
| "test_case_id": item.test_case_id, | |
| "requirement_id": item.requirement_id, | |
| "title": item.title, | |
| "test_type": item.test_type, | |
| "preconditions": item.preconditions, | |
| "steps": item.steps, | |
| "expected_results": item.expected_results, | |
| "oracle": item.oracle, | |
| "risks": item.risks, | |
| } | |
| for item in test_designs | |
| ] | |
| system_prompt = ( | |
| f"You are {runtime_config.agent_name}.\n" | |
| f"{runtime_config.directives}\n" | |
| "Review the planned test cases rigorously. " | |
| "Do not approve generic placeholders just because there is nominal coverage." | |
| ) | |
| schema = { | |
| "type": "object", | |
| "additionalProperties": False, | |
| "properties": { | |
| "approved": {"type": "boolean"}, | |
| "coverage_ratio": {"type": "number"}, | |
| "findings": {"type": "array", "items": {"type": "string"}}, | |
| "improvement_actions": {"type": "array", "items": {"type": "string"}}, | |
| "routing_focus": { | |
| "type": "array", | |
| "items": {"type": "string", "enum": ["requirements", "test_design"]}, | |
| }, | |
| "evaluation_notes": {"type": "array", "items": {"type": "string"}}, | |
| }, | |
| "required": [ | |
| "approved", | |
| "coverage_ratio", | |
| "findings", | |
| "improvement_actions", | |
| "routing_focus", | |
| "evaluation_notes", | |
| ], | |
| } | |
| user_prompt = ( | |
| "Requirements under review:\n" | |
| f"{requirements_payload}\n\n" | |
| "Planned test cases under review:\n" | |
| f"{designs_payload}\n\n" | |
| "Return concrete findings. Use routing_focus to indicate whether the next repair should go to " | |
| "the requirements analyst or the test design agent." | |
| ) | |
| response, metadata = call_structured_llm( | |
| runtime_config=runtime_config, | |
| system_prompt=system_prompt, | |
| user_prompt=user_prompt, | |
| schema_name="review_report", | |
| schema=schema, | |
| ) | |
| report = ReviewReport( | |
| approved=bool(response.get("approved")), | |
| coverage_ratio=round(_clean_ratio(response.get("coverage_ratio")), 2), | |
| findings=_clean_string_list(response.get("findings")), | |
| improvement_actions=_clean_string_list(response.get("improvement_actions")), | |
| ) | |
| if not report.approved and not report.findings: | |
| if report.coverage_ratio < 1.0: | |
| report.findings.append( | |
| "The review model rejected the run without concrete findings even though reported coverage was incomplete." | |
| ) | |
| else: | |
| report.findings.append( | |
| "The review model rejected the run without concrete findings. Treat this as incomplete review output." | |
| ) | |
| if not report.approved and not report.improvement_actions: | |
| if report.coverage_ratio < 1.0: | |
| report.improvement_actions.append( | |
| "Inspect missing or weak planned test coverage and rerun the review with a more reliable model." | |
| ) | |
| else: | |
| report.improvement_actions.append( | |
| "Rerun the review with another model or switch the Review Agent to Structured baseline." | |
| ) | |
| if report.approved and not report.findings: | |
| report.findings.append("No critical issues were identified.") | |
| self.last_execution = { | |
| "mode": "llm", | |
| "reasoning_source": "llm_structured_output", | |
| "notes": _clean_string_list(response.get("evaluation_notes")), | |
| "llm_used": True, | |
| "metadata": metadata, | |
| "routing_focus": _clean_string_list(response.get("routing_focus")) or self._infer_routing_focus(report.findings), | |
| } | |
| self._update_memory(run_session, report) | |
| return report | |
| def _update_memory(self, run_session: RunSession | None, report: ReviewReport) -> None: | |
| if not run_session: | |
| return | |
| run_session.working_memory.write_shared("review", asdict(report), author=self.name) | |
| run_session.working_memory.write_agent( | |
| "review", | |
| "routing_focus", | |
| list(self.last_execution.get("routing_focus") or []), | |
| ) | |
| run_session.working_memory.add_note( | |
| f"Review Agent recorded approved={report.approved} with {len(report.findings)} finding(s)." | |
| ) | |
| def _infer_routing_focus(self, findings: list[str]) -> list[str]: | |
| focus = [] | |
| combined_findings = " ".join(finding.lower() for finding in findings) | |
| if any( | |
| keyword in combined_findings | |
| for keyword in ( | |
| "assumption", | |
| "ambiguous requirement", | |
| "unclear requirement", | |
| "missing acceptance criteria", | |
| "clarify requirement", | |
| "requirements unclear", | |
| ) | |
| ): | |
| focus.append("requirements") | |
| if any( | |
| keyword in combined_findings | |
| for keyword in ( | |
| "weak oracle definition", | |
| "missing a designed test case", | |
| "has only one designed test case", | |
| "missing test case", | |
| "missing test coverage", | |
| "insufficient coverage", | |
| "weak oracle", | |
| "expected result", | |
| "negative scenario", | |
| "boundary", | |
| ) | |
| ): | |
| focus.append("test_design") | |
| return focus | |
| def _requires_multiple_test_designs(self, requirement: RequirementItem) -> bool: | |
| lowered = " ".join( | |
| [requirement.normalized_text, *requirement.acceptance_criteria, *requirement.assumptions] | |
| ).lower() | |
| return any( | |
| keyword in lowered | |
| for keyword in ( | |
| "error", | |
| "invalid", | |
| "negative", | |
| "failure", | |
| "fel", | |
| "ogiltig", | |
| "login", | |
| "logga in", | |
| "sign in", | |
| "register", | |
| "registrera", | |
| "admin", | |
| "access", | |
| "role", | |
| "authorization", | |
| "form", | |
| "validate", | |
| ) | |
| ) | |
| def _clean_text(value: Any) -> str: | |
| return str(value or "").strip() | |
| def _clean_string_list(value: Any) -> list[str]: | |
| if not isinstance(value, list): | |
| return [] | |
| return [str(item).strip() for item in value if str(item).strip()] | |
| def _clean_priority(value: Any) -> str: | |
| lowered = str(value or "").strip().lower() | |
| if lowered in {"high", "medium", "normal"}: | |
| return lowered | |
| return "normal" | |
| def _clean_ratio(value: Any) -> float: | |
| try: | |
| ratio = float(value) | |
| except (TypeError, ValueError): | |
| return 0.0 | |
| return min(1.0, max(0.0, ratio)) | |
| def _flatten_trace_notes(items: list[dict[str, Any]], key: str) -> list[str]: | |
| notes = [] | |
| for item in items: | |
| for note in _clean_string_list(item.get(key)): | |
| notes.append(note) | |
| return notes[:12] | |