orchestrata_ultimate.agents.volks.qa._testgen
Reference › orchestrata_ultimate › agents › volks › qa › _testgen
_testgen provides TestGenMixin for the QA volk (volk): extracting generated tests, enforcing the no-mocks rule, and running pytest. _extract_and_write_tests takes the LLM reply, pulls out each test file's code block (_extract_code_block) and finalises it with _finalize_test_code: dedent, syntax check and mock validation by _validate_no_mocks. A file with mock violations is rewritten by _retry_without_mocks, and _mock_gate rewrites or deletes workspace test files that still use mocks. _import_gate deletes test files whose imports fail validation. Import repair helpers (_rewrite_relative_imports, _rewrite_root_relative_imports, _rewrite_package_relative_imports, _ensure_init_files) make the workspace importable. _run_tests runs pytest through a command from _build_pytest_command (coverage, ignored directories, target files) with the environment from _build_test_env, and returns a pass flag and the output.
At a glance
| Source | src/orchestrata_ultimate/agents/volks/qa/_testgen.py |
| Lines of code | 511 |
| Classes | 1 |
| Functions | 0 |
| Imports (internal) | orchestrata_ultimate.agents.volks.qa._host, orchestrata_ultimate.core.base.traces, orchestrata_ultimate.core.infra_config |
| Imported by | orchestrata_ultimate.agents.volks.qa.volk |
| Tests that import it | tests/agents/volks/qa/test_testgen.py |
Class diagram
Dependency diagram
API
Classes
TestGenMixin
Bases: _MixinBase · Line 25
Test extraction, mock-validation, and pytest execution for U:Qa.
| Method | Signature | Summary |
|---|---|---|
_extract_and_write_tests |
async _extract_and_write_tests(llm_response: str) -> list[str] |
Extract test files from LLM response and write to workspace. |
_extract_code_block staticmethod |
_extract_code_block(lines: list) -> list |
Extract lines between python and fences from a block. |
_finalize_test_code |
async _finalize_test_code(filename: str, code_lines: list) -> str | None |
Dedent, syntax-check and mock-validate a test code block. |
_validate_no_mocks |
_validate_no_mocks(code: str) -> list[tuple[int, str]] |
Check code for mock/stub usage. |
_retry_without_mocks |
async _retry_without_mocks(filename: str, original_code: str) -> str | None |
G157-1: Rewrite a test file that contains mock violations. |
_rewrite_relative_imports |
_rewrite_relative_imports() |
Rewrite relative imports to absolute imports in all .py files. |
_rewrite_root_relative_imports |
_rewrite_root_relative_imports() -> int |
G115: Strip the dot prefix from relative imports in root files. |
_rewrite_package_relative_imports |
_rewrite_package_relative_imports() -> int |
G113: Rewrite relative imports inside package dirs with pkg prefix. |
_ensure_init_files |
_ensure_init_files() |
Add init.py to subdirs with .py files so imports work. |
_run_tests |
async _run_tests() -> tuple[bool, str] |
Run pytest on the workspace test files. |
_mock_gate |
async _mock_gate() |
G-MOCK: rewrite (or delete) workspace test files that use mocks. |
_import_gate |
_import_gate(venv_python: str) -> None |
G-77-10: delete workspace test files that fail import validation. |
_build_pytest_command |
_build_pytest_command(venv_python: str) -> list[str] |
Build the pytest argv: coverage, ignored dirs, and target files. |
_build_test_env |
_build_test_env() -> dict |
Build the pytest subprocess env: forced Redis vars + PYTHONPATH. |
Source
# mypy: ignore-errors
"""Test extraction, mock-validation, and pytest execution for U:Qa."""
import logging
import os
import re
import subprocess
from typing import TYPE_CHECKING
from orchestrata_ultimate.agents.volks.qa._host import _QaHost
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.infra_config import REDIS_HOST, REDIS_PORT
_logger = logging.getLogger(__name__)
# W6.4b: typing-only host contract so pyright can see across mixin
# composition (see _host.py). No runtime effect — TestGenMixin still
# inherits from plain ``object`` outside TYPE_CHECKING.
if TYPE_CHECKING:
_MixinBase = _QaHost
else:
_MixinBase = object
class TestGenMixin(_MixinBase):
"""Test extraction, mock-validation, and pytest execution for U:Qa."""
@traces(
spec="031",
story="US-031-004",
reqs=["REQ-031-051"],
)
async def _extract_and_write_tests(self, llm_response: str) -> list[str]:
"""Extract test files from LLM response and write to workspace."""
test_files = []
parts = llm_response.split("TEST_FILE:")
for part in parts[1:]: # Skip everything before first TEST_FILE:
lines = part.strip().split("\n")
if not lines:
continue
filename = lines[0].strip()
if not filename:
continue
code_lines = self._extract_code_block(lines[1:])
if not code_lines:
continue
code = await self._finalize_test_code(filename, code_lines)
if code is None:
continue
filepath = os.path.join(self.workspace, filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, "w", encoding="utf-8") as f:
f.write(code)
test_files.append(filename)
print(f" [U:Qa] Wrote {filename}")
return test_files
@staticmethod
@traces(spec="031", story="US-031-004", reqs=["REQ-031-051"])
def _extract_code_block(lines: list) -> list:
"""Extract lines between ```python and ``` fences from a block."""
code_lines = []
in_code = False
for line in lines:
if line.strip().startswith("```python"):
in_code = True
continue
elif line.strip() == "```" and in_code:
in_code = False
continue
elif in_code:
code_lines.append(line)
return code_lines
@traces(spec="031", story="US-031-004", reqs=["REQ-031-052", "REQ-031-053"])
async def _finalize_test_code(self, filename: str, code_lines: list) -> str | None:
"""Dedent, syntax-check and mock-validate a test code block.
Returns the finalized code, or None when the file should be
skipped (syntax error, or mock retry failed).
"""
code = "\n".join(code_lines)
# G178-1: Dedent code that has uniform leading whitespace
# (Kimi K2.5 sometimes indents entire code blocks)
import textwrap
code = textwrap.dedent(code)
# G64: syntax check before writing — skip truncated files
try:
compile(code, filename, "exec")
except SyntaxError as e:
print(f" [U:Qa] SKIP {filename} — syntax error: {e}")
return None
# G-MOCK: reject test files that use mocking
mock_violations = self._validate_no_mocks(code)
if mock_violations:
for ln, desc in mock_violations[:5]:
print(f" [U:Qa] MOCK violation L{ln}: {desc}")
# G157-1: Retry once — ask LLM to rewrite without mocks
code = await self._retry_without_mocks(filename, code)
if code is None:
print(f" [U:Qa] REJECTED {filename} — mock retry failed")
return None
# Re-check syntax after rewrite
try:
compile(code, filename, "exec")
except SyntaxError:
print(
f" [U:Qa] REJECTED {filename} — "
f"syntax error after mock retry"
)
return None
return code
@traces(
spec="031",
story="US-031-004",
reqs=["REQ-031-052"],
)
def _validate_no_mocks(self, code: str) -> list[tuple[int, str]]:
"""Check code for mock/stub usage. Returns list of (line_num, description)."""
violations = []
for line_num, line in enumerate(code.split("\n"), 1):
for pattern, description in self._MOCK_PATTERNS:
if re.search(pattern, line):
violations.append((line_num, description))
break # One violation per line is enough
return violations
@traces(
spec="031",
story="US-031-004",
reqs=["REQ-031-053"],
)
async def _retry_without_mocks(
self,
filename: str,
original_code: str,
) -> str | None:
"""G157-1: Rewrite a test file that contains mock violations.
Sends the rejected code back to DeepSeek with explicit
instructions to replace all mock usage with real objects.
Args:
filename: Test file name.
original_code: Code that failed mock validation.
Returns:
Rewritten code without mocks, or None if retry fails.
"""
print(
f" [U:Qa] G157-1: retrying {filename} without mocks...",
flush=True,
)
prompt = (
f"Rewrite this test file to remove ALL mock usage.\n\n"
f"RULES:\n"
f"- NEVER import unittest.mock, MagicMock, patch, "
f"AsyncMock, or Mock\n"
f"- NEVER create fake/stub/dummy classes\n"
f"- Use REAL objects connected to real Redis\n"
f"- Use pytest fixtures for setup/teardown\n"
f"- Keep the same test functions and assertions\n\n"
f"Original file ({filename}):\n"
f"```python\n{original_code}\n```\n\n"
f"Output ONLY the rewritten Python code, no explanation."
)
try:
# W4.5 (backlog #14): route through the base-agent seam so this
# rewrite call gets memory recall (SPEC-010) folded in. _call_llm
# stamps volk="U:Qa" (self.alias) and composes persona itself.
response = await self._call_llm(
"You rewrite test files to remove mock usage.",
prompt,
temperature=0.0,
graph_context=self._graph_context,
)
except Exception as exc:
print(
f" [U:Qa] G157-1: LLM retry failed: {exc}",
flush=True,
)
return None
# Extract code from response (may have ```python fences)
code = response.strip()
if "```python" in code:
parts = code.split("```python", 1)
if len(parts) > 1:
code = parts[1].split("```")[0]
elif "```" in code:
parts = code.split("```", 1)
if len(parts) > 1:
code = parts[1].split("```")[0]
# Validate no mocks remain
violations = self._validate_no_mocks(code)
if violations:
print(
f" [U:Qa] G157-1: retry still has "
f"{len(violations)} violations — rejecting",
flush=True,
)
return None
print(
f" [U:Qa] G157-1: {filename} rewritten successfully",
flush=True,
)
return code
def _rewrite_relative_imports(self):
"""Rewrite relative imports to absolute imports in all .py files.
G113: Relative imports (from orchestrata_ultimate.agents.volks.qa. import X, from orchestrata_ultimate.agents.volks.qa.module import Y)
only work with installed packages. Rewrite them to absolute imports
using the package directory name so they work with sys.path.
G115: Also scan root-level files — strip the dot prefix so
'from orchestrata_ultimate.agents.volks.qa.module import X' becomes 'from module import X'.
"""
count = self._rewrite_root_relative_imports()
count += self._rewrite_package_relative_imports()
if count:
print(f" [U:Qa] G113/G115: rewrote relative imports in {count} files")
def _rewrite_root_relative_imports(self) -> int:
"""G115: Strip the dot prefix from relative imports in root files."""
count = 0
for fname in os.listdir(self.workspace):
if not fname.endswith(".py"):
continue
fpath = os.path.join(self.workspace, fname)
if not os.path.isfile(fpath):
continue
try:
with open(fpath, encoding="utf-8") as f:
content = f.read()
except Exception:
continue
original = content
# from orchestrata_ultimate.agents.volks.qa.module import X → from module import X
content = re.sub(
r"^from \.(\w+) import",
r"from \1 import",
content,
flags=re.MULTILINE,
)
# from orchestrata_ultimate.agents.volks.qa. import X → import X (less common at root)
content = re.sub(
r"^from \. import (\w+)",
r"import \1",
content,
flags=re.MULTILINE,
)
if content != original:
with open(fpath, "w", encoding="utf-8") as f:
f.write(content)
count += 1
return count
def _rewrite_package_relative_imports(self) -> int:
"""G113: Rewrite relative imports inside package dirs with pkg prefix."""
count = 0
for entry in os.listdir(self.workspace):
subdir = os.path.join(self.workspace, entry)
if not os.path.isdir(subdir):
continue
if entry.startswith((".", "_", "test")):
continue
init = os.path.join(subdir, "__init__.py")
if not os.path.exists(init):
continue
pkg_name = entry
for fname in os.listdir(subdir):
if not fname.endswith(".py"):
continue
fpath = os.path.join(subdir, fname)
try:
with open(fpath, encoding="utf-8") as f:
content = f.read()
except Exception:
continue
original = content
# from orchestrata_ultimate.agents.volks.qa.module import X → from pkg_name.module import X
content = re.sub(
r"^from \.(\w+) import",
f"from {pkg_name}.\\1 import",
content,
flags=re.MULTILINE,
)
# from orchestrata_ultimate.agents.volks.qa. import X → from pkg_name import X
content = re.sub(
r"^from \. import",
f"from {pkg_name} import",
content,
flags=re.MULTILINE,
)
if content != original:
with open(fpath, "w", encoding="utf-8") as f:
f.write(content)
count += 1
return count
def _ensure_init_files(self):
"""Add __init__.py to subdirs with .py files so imports work."""
for root, _dirs, files in os.walk(self.workspace):
if root == self.workspace:
continue
if any(f.endswith(".py") for f in files):
init = os.path.join(root, "__init__.py")
if not os.path.exists(init):
with open(init, "w") as f:
f.write("")
@traces(
spec="031",
story="US-031-005",
reqs=["REQ-031-055"],
)
async def _run_tests(self) -> tuple[bool, str]:
"""Run pytest on the workspace test files.
Deps and import validation are handled in Phase A.5 (_handle_review).
"""
await self._mock_gate()
venv_python = self._get_venv_python()
self._import_gate(venv_python)
try:
pytest_args = self._build_pytest_command(venv_python)
test_env = self._build_test_env()
result = subprocess.run(
pytest_args,
cwd=self.workspace,
capture_output=True,
text=True,
timeout=300,
env=test_env,
)
output = result.stdout + "\n" + result.stderr
passed = result.returncode == 0
return passed, output
except subprocess.TimeoutExpired:
return False, "Tests timed out after 300 seconds"
except Exception as e:
return False, f"Failed to run tests: {e}"
@traces(spec="031", story="US-031-004", reqs=["REQ-031-052", "REQ-031-053"])
async def _mock_gate(self):
"""G-MOCK: rewrite (or delete) workspace test files that use mocks.
G-77-08: skip .venv and site-packages — only scan workspace source
dirs. G-193-03: retry an LLM rewrite before deleting (was
delete-only).
"""
for root, dirs, files in os.walk(self.workspace):
dirs[:] = [
d
for d in dirs
if not d.startswith(".")
and d != "site-packages"
and d != "__pycache__"
and not d.startswith("Module:")
]
for fname in files:
if fname.startswith("test_") and fname.endswith(".py"):
fpath = os.path.join(root, fname)
with open(fpath, encoding="utf-8") as f:
content = f.read()
violations = self._validate_no_mocks(content)
if violations:
details = "; ".join(
f"L{ln}: {desc}" for ln, desc in violations[:3]
)
print(f" [U:Qa] MOCK GATE FAIL {fname}: {details}")
# G-193-03: attempt LLM rewrite before deleting
rewritten = await self._retry_without_mocks(fname, content)
if rewritten is not None:
with open(fpath, "w", encoding="utf-8") as f:
f.write(rewritten)
print(f" [U:Qa] Rewrote {fname} (mocks removed)")
else:
os.remove(fpath)
print(f" [U:Qa] Deleted {fname}")
@traces(spec="031", reqs=["REQ-031-055"])
def _import_gate(self, venv_python: str) -> None:
"""G-77-10: delete workspace test files that fail import validation.
Generated test files may carry wrong import paths from the U:Ar
design spec; validate every test module loads before pytest runs.
"""
monorepo_syspath = (
"import sys, os, glob; "
"sys.path.insert(0, os.getcwd()); "
"_s=os.path.join(os.getcwd(),'src'); "
"(sys.path.insert(0,_s) if os.path.isdir(_s) and _s not in sys.path else None); "
# G-92b-01: nested package dirs (J:Aider may create workspace/pkg-name/src/)
"[sys.path.insert(0,p) for p in glob.glob(os.path.join(os.getcwd(),'*/src')) "
"if os.path.isdir(p) and p not in sys.path]; "
"_m='/opt/orchestrata/packages'; "
"[sys.path.append(p) for p in glob.glob(f'{_m}/*/src') "
"if p not in sys.path] if os.path.isdir(_m) else None; "
)
for root, dirs, files in os.walk(self.workspace):
dirs[:] = [
d
for d in dirs
if not d.startswith(".") and d != "site-packages" and d != "__pycache__"
]
for fname in files:
if not fname.startswith("test_") or not fname.endswith(".py"):
continue
fpath = os.path.join(root, fname)
rel = os.path.relpath(fpath, self.workspace)
check = subprocess.run(
[
venv_python,
"-c",
monorepo_syspath
+ f"import importlib.util, sys; spec=importlib.util.spec_from_file_location('_t', r'{fpath}'); mod=importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)",
],
cwd=self.workspace,
capture_output=True,
text=True,
timeout=15,
env=self._venv_env(),
)
if check.returncode != 0:
stderr = check.stderr.strip().splitlines()
err = stderr[-1] if stderr else "import error"
hint = self._get_import_hint(venv_python, err)
print(f" [U:Qa] IMPORT GATE FAIL {rel}: {err}{hint}")
os.remove(fpath)
print(f" [U:Qa] Deleted {rel} (bad imports)")
@traces(spec="031", reqs=["REQ-031-055"])
def _build_pytest_command(self, venv_python: str) -> list[str]:
"""Build the pytest argv: coverage, ignored dirs, and target files."""
# G98-2: Only run QA-generated test files, not project-internal ones
# This prevents conftest conflicts between workspace root and nested dirs
qa_test_files = [
f for f in self.test_files_created if not f.startswith((".", "/"))
]
# G122-2: add per-test timeout to prevent hanging tests
# G219: Add --cov for coverage measurement toward goal criteria
src_dir = os.path.join(self.workspace, "src")
cov_args = []
if os.path.isdir(src_dir):
# Find the package name under src/
for entry in os.listdir(src_dir):
pkg_path = os.path.join(src_dir, entry)
if os.path.isdir(pkg_path) and not entry.startswith("."):
cov_args = [f"--cov={entry}", "--cov-report=term-missing"]
break
pytest_args = [
venv_python,
"-m",
"pytest",
"-v",
"--tb=short",
"--timeout=30",
] + cov_args
# G102-1: Ignore project-internal test dirs to prevent
# collection conflicts with QA-generated test files
for entry in os.listdir(self.workspace):
test_dir = os.path.join(self.workspace, entry, "tests")
if os.path.isdir(test_dir):
pytest_args.extend(["--ignore", test_dir])
# G126-3: Ignore top-level tests/ dir — Junior may generate
# tests/conftest.py with unavailable deps (testcontainers)
top_tests = os.path.join(self.workspace, "tests")
if os.path.isdir(top_tests):
pytest_args.extend(["--ignore", top_tests])
if qa_test_files:
pytest_args.extend(qa_test_files)
return pytest_args
@traces(spec="031", reqs=["REQ-031-055"])
def _build_test_env(self) -> dict:
"""Build the pytest subprocess env: forced Redis vars + PYTHONPATH."""
# G100-1 + G129-1: Force-set infra env vars so generated
# tests always connect to the correct Redis instance.
# Direct assignment (not setdefault) to override empty
# values that may leak through uv/venv environment layers.
test_env = self._venv_env()
test_env["REDIS_HOST"] = REDIS_HOST
test_env["REDIS_PORT"] = str(REDIS_PORT)
# G104-2: Set PYTHONPATH so subprocess calls like
# `python -m package.cli` can find the project package
py_paths = [self.workspace]
src_dir = os.path.join(self.workspace, "src")
if os.path.isdir(src_dir):
py_paths.append(src_dir)
for entry in os.listdir(self.workspace):
nested_src = os.path.join(self.workspace, entry, "src")
if os.path.isdir(nested_src):
py_paths.append(nested_src)
existing = test_env.get("PYTHONPATH", "")
test_env["PYTHONPATH"] = os.pathsep.join(
py_paths + ([existing] if existing else [])
)
return test_env