3280 lines
124 KiB
Python
3280 lines
124 KiB
Python
"""Focused Python compiler state and Memento-scoped policy coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
from band_aider.context.failing_test import FailingTest
|
|
from dataclasses import replace
|
|
import logging
|
|
from pathlib import Path
|
|
from subprocess import CompletedProcess
|
|
from typing import cast, Sequence, TextIO
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.edited_test_inventory import EditedTestInventory
|
|
from band_aider.context.memento import Memento
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.repo_exec import RepoExec, RepoExecError
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from band_aider.loop.beta_loop import BetaLoop
|
|
from band_aider.loop.bootstrap_loop import BootstrapLoop
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
from band_aider.strategies.bom import ExternalReference, ReferenceOperation
|
|
from band_aider.strategies.compiler import (
|
|
BlackboardCompiler,
|
|
CompileAttempt,
|
|
CompileError,
|
|
ReferenceError,
|
|
RemediationCompiler,
|
|
RepoExecCompileError,
|
|
StaticCompileError,
|
|
TestRunCompileError,
|
|
TypeCheckError,
|
|
)
|
|
from band_aider.strategies.python.python_compiler import (
|
|
AlphaPythonCompiler,
|
|
BetaPythonCompiler,
|
|
PythonCompiler,
|
|
PythonCompilerFactory,
|
|
)
|
|
from band_aider.strategies.python.python_bom_editor import PythonBomEditor
|
|
from band_aider.strategies.python.python_source_discovery import PythonSourceDiscovery
|
|
from tests.support.completion_clients import StaticOpenAICompatibleClient
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _Bom:
|
|
def __init__(self, references: list[ExternalReference] = []) -> None:
|
|
self.references = list(references)
|
|
|
|
def inventory(self, *, include_development: bool = False) -> list[ExternalReference]:
|
|
del include_development
|
|
return list(self.references)
|
|
|
|
def apply(self, operation: ReferenceOperation) -> str:
|
|
"""Accept a directive that is outside this compiler-test fixture's concern."""
|
|
|
|
return f"Applied {operation.operation} for {operation.name}."
|
|
|
|
|
|
class _TargetedRunner(TestRunnerStrategy):
|
|
"""Record Beta's inner callable execution without invoking pytest."""
|
|
|
|
def failure_reports(self, test_files, result):
|
|
"""Retain the scripted execution's origin for workflow tests using this fake."""
|
|
return [FailingTest.from_result(test_files[0], result)]
|
|
|
|
def matching_tests(self, test_files: list[ClassMeta], result: TestRunResult) -> dict[str, tuple[str, ...]]:
|
|
"""Reject unexpected final-exam discovery in this isolated runner scenario."""
|
|
raise AssertionError("This scenario does not provide failed-suite method discovery.")
|
|
|
|
def __init__(
|
|
self,
|
|
results: list[TestRunResult | Exception] | None = None,
|
|
) -> None:
|
|
"""Store deterministic inner-test outcomes."""
|
|
|
|
self.results = list(results or [])
|
|
self.calls: list[tuple[list[ClassMeta], int | None, tuple[str, ...]]] = []
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Record one targeted inner run and return its configured outcome."""
|
|
|
|
self.calls.append((test_files, max_failures, tuple(test_names)))
|
|
# Consume explicitly supplied outcomes before falling back to a clean targeted run.
|
|
if self.results:
|
|
result = self.results.pop(0)
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return result
|
|
return TestRunResult("pytest", TestRunOutcome.Success, 0, "")
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Provide the unrelated protocol operation for factory fixtures."""
|
|
|
|
del test_root
|
|
return TestRunResult("pytest", TestRunOutcome.Success, 0, "")
|
|
|
|
|
|
class _CompilerParser(CodeParser):
|
|
"""Create deterministic metadata without reading a host path or model endpoint."""
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return one minimal record for the supplied in-memory path."""
|
|
|
|
del prompt
|
|
return ClassMeta(str(file_name), "", [], {}, content_type=content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Consume no external state while preserving the parser protocol."""
|
|
|
|
del stream
|
|
return self.parse_file(file_name, prompt, content_type)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Return metadata for each supplied path in input order."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _FileManager(FileManager):
|
|
"""Use the production manager API over seed metadata held entirely in memory."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
repo_exec: MemoryRepoExec,
|
|
src_classes: list[ClassMeta],
|
|
test_classes: list[ClassMeta],
|
|
client: OpenAICompatibleClient | None = None,
|
|
) -> None:
|
|
"""Seed current records before production indexing captures their baseline."""
|
|
|
|
self._seed_src_classes = list(src_classes)
|
|
self._seed_test_classes = list(test_classes)
|
|
self.refresh_test_calls: list[tuple[str, str, str]] = []
|
|
self.upsert_test_calls: list[tuple[ClassMeta, str, str]] = []
|
|
super().__init__(
|
|
parser=_CompilerParser(),
|
|
client=client or StaticOpenAICompatibleClient('{"cdc": ""}'),
|
|
repo_root=repo_root,
|
|
src_root=src_root,
|
|
test_root=test_root,
|
|
actor="test",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
repo_exec=repo_exec,
|
|
prompt="Original task",
|
|
)
|
|
|
|
def resync(self) -> None:
|
|
"""Rebuild indexes from this scenario's explicit in-memory metadata only."""
|
|
|
|
self.src_classes = list(self._seed_src_classes)
|
|
self.test_classes = list(self._seed_test_classes)
|
|
self.src_index = {item.location: index for index, item in enumerate(self.src_classes)}
|
|
self.test_index = {item.location: index for index, item in enumerate(self.test_classes)}
|
|
self.src_dirty.clear()
|
|
self.test_dirty.clear()
|
|
|
|
def git_diff_for(self, location: str | Path) -> str:
|
|
"""Keep compiler scenarios independent from a host Git checkout."""
|
|
|
|
del location
|
|
return ""
|
|
|
|
def refresh_test(
|
|
self,
|
|
location: str,
|
|
prompt: str,
|
|
tailing_logs: str = "",
|
|
) -> ClassMeta:
|
|
"""Record test-metadata refreshes before retaining normal manager semantics."""
|
|
|
|
self.refresh_test_calls.append((location, prompt, tailing_logs))
|
|
return super().refresh_test(location, prompt, tailing_logs)
|
|
|
|
def upsert_test(
|
|
self,
|
|
latest: ClassMeta,
|
|
prompt: str,
|
|
tailing_logs: str = "",
|
|
) -> None:
|
|
"""Record metadata writes before retaining normal manager semantics."""
|
|
|
|
self.upsert_test_calls.append((latest, prompt, tailing_logs))
|
|
super().upsert_test(latest, prompt, tailing_logs)
|
|
|
|
def source_path(self, location: str | Path) -> Path:
|
|
"""Resolve source metadata when a scenario deliberately uses relative roots."""
|
|
|
|
path = self.repo_path(location)
|
|
root = self.src_root if self.src_root.is_absolute() else self.repo_root / self.src_root
|
|
path.relative_to(root)
|
|
return path
|
|
|
|
def test_path(self, location: str | Path) -> Path:
|
|
"""Resolve test metadata when a scenario deliberately uses relative roots."""
|
|
|
|
path = self.repo_path(location)
|
|
root = self.test_root if self.test_root.is_absolute() else self.repo_root / self.test_root
|
|
path.relative_to(root)
|
|
return path
|
|
|
|
|
|
def _meta(location: str, content_type: ContentType) -> ClassMeta:
|
|
return ClassMeta(
|
|
location=location,
|
|
summary="",
|
|
relevance=[],
|
|
relevant_symbols={},
|
|
content_type=content_type,
|
|
)
|
|
|
|
|
|
def _write(repo_exec: RepoExec, path: Path, source: str) -> None:
|
|
"""Store a scenario source file through its explicit repository boundary."""
|
|
|
|
repo_exec.write_text(path, source, encoding="utf-8")
|
|
|
|
|
|
def _compiler(
|
|
memory_root: Path,
|
|
*,
|
|
sources: dict[str, str],
|
|
tests: dict[str, str] | None = None,
|
|
edit_location: str,
|
|
replacement: str,
|
|
context_name: str,
|
|
frontier_locations: tuple[str, ...] = (),
|
|
readonly_locations: tuple[str, ...] = (),
|
|
upstream_locations: tuple[str, ...] = (),
|
|
references: list[ExternalReference] | None = None,
|
|
client: OpenAICompatibleClient | None = None,
|
|
runner: _TargetedRunner | None = None,
|
|
candidate: bool = True,
|
|
previous_errors: Sequence[CompileError] = (),
|
|
blackboard: EditedTestInventory | None = None,
|
|
is_remediation: bool = False,
|
|
remediation_originator_location: str | None = None,
|
|
failure_error_message: str = "",
|
|
failure_stack_trace: Sequence[str] = (),
|
|
failure_raw_output: str = "",
|
|
failure_test_name: str = "",
|
|
failure_outcome: TestRunOutcome | None = None,
|
|
) -> tuple[PythonCompiler, Memento, _FileManager]:
|
|
"""Build one compiler with explicit accepted and candidate test state."""
|
|
|
|
src_root = memory_root / "src"
|
|
test_root = memory_root / "tests"
|
|
repo_exec = MemoryRepoExec(memory_root)
|
|
for location, source in sources.items():
|
|
_write(repo_exec, src_root / location, source)
|
|
for location, source in (tests or {}).items():
|
|
_write(repo_exec, test_root / location, source)
|
|
source_meta = [
|
|
_meta(f"src/{location}", ContentType.Source)
|
|
for location in sources
|
|
]
|
|
test_meta = [
|
|
_meta(f"tests/{location}", ContentType.Test)
|
|
for location in (tests or {})
|
|
]
|
|
manager = _FileManager(
|
|
repo_root=memory_root,
|
|
src_root=src_root,
|
|
test_root=test_root,
|
|
repo_exec=repo_exec,
|
|
src_classes=source_meta,
|
|
test_classes=test_meta,
|
|
client=client,
|
|
)
|
|
target = memory_root / edit_location
|
|
memento = Memento(target, repo_exec).capture()
|
|
_write(repo_exec, target, replacement)
|
|
context = {
|
|
"AlphaLoop": AlphaLoop,
|
|
"BetaLoop": BetaLoop,
|
|
"BootstrapLoop": BootstrapLoop,
|
|
}[context_name]
|
|
locations = set(frontier_locations)
|
|
readonly = set(readonly_locations)
|
|
upstream = set(upstream_locations)
|
|
frontier = lambda: [
|
|
[meta for meta in manager.src_classes if meta.location in locations]
|
|
]
|
|
dependency_strategy = Mock(spec=DependencyDiscoveryStrategy)
|
|
dependency_strategy.upstream_sources_for.return_value = list(filter(
|
|
lambda meta: meta.location in upstream,
|
|
manager.src_classes,
|
|
))
|
|
readonly_files: list[Path] = []
|
|
# Resolve the scenario's supplied readonly paths from current managed metadata.
|
|
for meta in (*manager.src_classes, *manager.test_classes):
|
|
# Select only locations declared readonly by this compiler scenario.
|
|
if meta.location in readonly:
|
|
readonly_files.append(meta.resolve_path(manager))
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Implement the requested change.",
|
|
editable_files=[target],
|
|
readonly_files=readonly_files,
|
|
pinned_context=[],
|
|
backlog_guidance=["Keep references legal."],
|
|
context=context,
|
|
is_remediation=is_remediation,
|
|
remediation_originator_test=manager.current_test(remediation_originator_location) if remediation_originator_location is not None else None,
|
|
failure_error_message=failure_error_message,
|
|
failure_stack_trace=list(failure_stack_trace),
|
|
failure_raw_output=failure_raw_output,
|
|
failure_test_name=failure_test_name,
|
|
failure_outcome=failure_outcome,
|
|
)
|
|
attempt = CompileAttempt(
|
|
request=request,
|
|
mementos=(memento,),
|
|
frontier_provider=None if context is BootstrapLoop else frontier,
|
|
candidate_mementos=(memento,) if candidate else (),
|
|
previous_errors=previous_errors,
|
|
blackboard=blackboard,
|
|
)
|
|
compiler_factory = PythonCompilerFactory(
|
|
file_manager=manager,
|
|
repo_exec=repo_exec,
|
|
bom_editor=_Bom(references or []),
|
|
dependency_strategy=dependency_strategy,
|
|
repo_root=memory_root,
|
|
source_discovery=PythonSourceDiscovery(
|
|
src_root=src_root,
|
|
test_root=test_root,
|
|
repo_exec=repo_exec,
|
|
repo_root=memory_root,
|
|
),
|
|
test_runner=runner or _TargetedRunner(),
|
|
)
|
|
compiler = compiler_factory.build(attempt)
|
|
assert isinstance(compiler, PythonCompiler)
|
|
return compiler, memento, manager
|
|
|
|
|
|
def test_reference_delta_uses_memento_pre_and_post_for_relative_imports(memory_root: Path) -> None:
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/base.py": "class Old: pass\nclass New: pass\n",
|
|
"pkg/consumer.py": "from .base import Old\n",
|
|
},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="from .base import New\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/base.py", "src/pkg/consumer.py"),
|
|
)
|
|
|
|
delta = compiler._changed_references()
|
|
|
|
assert [item.declaration for item in delta.added] == ["from .base import New"]
|
|
assert [item.declaration for item in delta.removed] == ["from .base import Old"]
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_removed_only_references_remain_legal(memory_root: Path) -> None:
|
|
"""Removing an old import never creates a new reference-validation blocker."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": "import missing_legacy\n"},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_compile_validates_an_unchanged_supplied_python_file(memory_root: Path) -> None:
|
|
"""Syntax validation describes supplied state even without a Memento delta."""
|
|
|
|
compiler, memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": "def broken(:\n pass\n"},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="def broken(:\n pass\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
candidate=False,
|
|
)
|
|
|
|
assert memento.diff == ""
|
|
[error] = compiler.compile([memory_root / "src/pkg/consumer.py"])
|
|
|
|
assert "SyntaxError" in error.message
|
|
|
|
|
|
def test_compiler_factory_binds_python_collaborators_before_building_attempt(
|
|
memory_root: Path,
|
|
) -> None:
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
factory = PythonCompilerFactory(
|
|
file_manager=manager,
|
|
repo_exec=manager.repo_exec,
|
|
bom_editor=compiler.bom_editor,
|
|
dependency_strategy=compiler.dependency_strategy,
|
|
repo_root=memory_root,
|
|
source_discovery=PythonSourceDiscovery(
|
|
src_root=manager.src_root,
|
|
test_root=manager.test_root,
|
|
repo_exec=manager.repo_exec,
|
|
repo_root=memory_root,
|
|
),
|
|
test_runner=_TargetedRunner(),
|
|
)
|
|
|
|
rebuilt = factory.build(compiler.attempt)
|
|
|
|
assert isinstance(rebuilt, PythonCompiler)
|
|
assert rebuilt.file_manager is manager
|
|
assert rebuilt.bom_editor is compiler.bom_editor
|
|
assert rebuilt.dependency_strategy is compiler.dependency_strategy
|
|
assert rebuilt.mementos == compiler.mementos
|
|
|
|
|
|
def test_compiler_inventory_exit_returns_cleanup_to_its_owner(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Borrowing compilers share inventory while exit makes its owner forget it."""
|
|
|
|
owner, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
factory = PythonCompilerFactory(
|
|
file_manager=manager,
|
|
repo_exec=manager.repo_exec,
|
|
bom_editor=owner.bom_editor,
|
|
dependency_strategy=owner.dependency_strategy,
|
|
repo_root=memory_root,
|
|
source_discovery=PythonSourceDiscovery(
|
|
src_root=manager.src_root,
|
|
test_root=manager.test_root,
|
|
repo_exec=manager.repo_exec,
|
|
repo_root=memory_root,
|
|
),
|
|
test_runner=_TargetedRunner(),
|
|
)
|
|
|
|
with owner.get_blackboard() as inventory:
|
|
borrower = cast(
|
|
BlackboardCompiler,
|
|
factory.build(replace(owner.attempt, blackboard=inventory)),
|
|
)
|
|
assert borrower.get_blackboard() is inventory
|
|
|
|
assert not hasattr(owner, "blackboard")
|
|
|
|
|
|
def test_compiler_factory_fails_closed_for_an_unsupported_workflow_context(
|
|
memory_root: Path,
|
|
) -> None:
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
unsupported_request = replace(
|
|
compiler.attempt.request,
|
|
context=type("UnsupportedWorkflow", (), {}),
|
|
)
|
|
factory = PythonCompilerFactory(
|
|
file_manager=manager,
|
|
repo_exec=manager.repo_exec,
|
|
bom_editor=compiler.bom_editor,
|
|
dependency_strategy=compiler.dependency_strategy,
|
|
repo_root=memory_root,
|
|
source_discovery=PythonSourceDiscovery(
|
|
src_root=manager.src_root,
|
|
test_root=manager.test_root,
|
|
repo_exec=manager.repo_exec,
|
|
repo_root=memory_root,
|
|
),
|
|
test_runner=_TargetedRunner(),
|
|
)
|
|
rebuilt = factory.build(
|
|
CompileAttempt(
|
|
request=unsupported_request,
|
|
mementos=compiler.mementos,
|
|
frontier_provider=compiler.frontier_provider,
|
|
)
|
|
)
|
|
|
|
[error] = rebuilt.compile([memory_root / "src/pkg/consumer.py"])
|
|
|
|
assert error.message == "Compilation requires a supported workflow context."
|
|
|
|
|
|
def test_unchanged_beta_file_reports_an_inherited_pyright_error_initially(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Beta cannot enter pytest while its supplied baseline has a type error."""
|
|
|
|
runner = _TargetedRunner()
|
|
source = (
|
|
"def test_widget() -> None:\n"
|
|
" count: int = 'wrong'\n"
|
|
" assert count\n"
|
|
)
|
|
compiler, memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": source},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=source,
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
candidate=False,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
checked_paths: list[tuple[Path, ...]] = []
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: checked_paths.append(tuple(paths)) or [
|
|
TypeCheckError("Type str is not assignable to type int")
|
|
]
|
|
|
|
assert memento.diff == ""
|
|
[error] = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert isinstance(error, TypeCheckError)
|
|
assert "not assignable to type" in error.message
|
|
assert checked_paths == [(memory_root / "tests/test_widget.py",)]
|
|
assert runner.calls == []
|
|
|
|
|
|
def test_unchanged_beta_file_reports_an_inherited_unresolved_import_initially(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Current import resolution runs even when Beta did not introduce the import."""
|
|
|
|
runner = _TargetedRunner()
|
|
source = "from pkg.missing import Widget\n\ndef test_widget() -> None:\n assert Widget\n"
|
|
compiler, memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": source},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=source,
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
candidate=False,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
checked_paths: list[tuple[Path, ...]] = []
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: checked_paths.append(tuple(paths)) or [
|
|
TypeCheckError('Import "pkg.missing" could not be resolved')
|
|
]
|
|
|
|
assert memento.diff == ""
|
|
[error] = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert isinstance(error, TypeCheckError)
|
|
assert error.message == 'Import "pkg.missing" could not be resolved'
|
|
assert checked_paths == [(memory_root / "tests/test_widget.py",)]
|
|
assert runner.calls == []
|
|
|
|
|
|
def test_beta_compiler_targets_an_existing_changed_top_level_test(memory_root: Path) -> None:
|
|
"""A changed existing pytest function becomes one inner targeted selector."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 2 == 2\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
assert runner.calls == [
|
|
([manager.test_classes[0]], 1, ("test_widget",))
|
|
]
|
|
|
|
|
|
def test_beta_compiler_targets_an_existing_changed_test_class_method(memory_root: Path) -> None:
|
|
"""A changed existing Test* method becomes one inner targeted selector."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"class TestWidget:\n"
|
|
" def test_render(self):\n"
|
|
" assert 1 == 1\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"class TestWidget:\n"
|
|
" def test_render(self):\n"
|
|
" assert 2 == 2\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
assert runner.calls == [
|
|
([manager.test_classes[0]], 1, ("TestWidget.test_render",))
|
|
]
|
|
|
|
|
|
def test_beta_compiler_excludes_an_identical_callable_moved_by_an_unrelated_edit(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Line movement cannot turn an unchanged callable definition into a test target."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="\ndef test_widget():\n assert 1 == 1\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
|
|
assert compiler._retested_test_names() == ()
|
|
|
|
|
|
def test_beta_compiler_excludes_a_changed_helper_callable(memory_root: Path) -> None:
|
|
"""Only ordinary pytest naming conventions can admit an updated callable."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def helper():\n return 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def helper():\n return 2\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
|
|
assert compiler._retested_test_names() == ()
|
|
|
|
|
|
def test_beta_compiler_excludes_a_deleted_test_from_candidate_runs(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A deleted PRE identity has no POST callable available for an exact run."""
|
|
|
|
deleted, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="",
|
|
context_name="BetaLoop",
|
|
)
|
|
assert isinstance(deleted, BetaPythonCompiler)
|
|
|
|
assert deleted._retested_test_names() == ()
|
|
|
|
|
|
def test_beta_compiler_targets_a_renamed_post_identity_as_created(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A renamed POST pytest identity enters the created-test candidate set."""
|
|
|
|
renamed, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget_renamed():\n assert 2 == 2\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
assert isinstance(renamed, BetaPythonCompiler)
|
|
|
|
assert renamed._retested_test_names() == ("test_widget_renamed",)
|
|
|
|
|
|
def test_beta_compiler_targets_new_post_only_tests_but_excludes_import_only_edits(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A created pytest test runs while an indirect import edit does not."""
|
|
|
|
added, _memento, _manager = _compiler(
|
|
memory_root / "added",
|
|
sources={},
|
|
tests={"test_widget.py": ""},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 1 == 1\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
imported, _memento, _manager = _compiler(
|
|
memory_root / "imported",
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="import pytest\n\ndef test_widget():\n assert 1 == 1\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
assert isinstance(added, BetaPythonCompiler)
|
|
assert isinstance(imported, BetaPythonCompiler)
|
|
|
|
assert added._retested_test_names() == ("test_widget",)
|
|
assert imported._retested_test_names() == ()
|
|
|
|
|
|
def test_beta_compiler_uses_only_the_candidate_memento_for_test_selection(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""An earlier accepted update is not rerun for a later import-only candidate."""
|
|
|
|
compiler, outer, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 2 == 2\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
target = memory_root / "tests/test_widget.py"
|
|
inner = Memento.first_available(target, outer.repo_exec).capture()
|
|
_write(outer.repo_exec, target, "import pytest\n\ndef test_widget():\n assert 2 == 2\n")
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.attempt = replace(compiler.attempt, candidate_mementos=(inner,))
|
|
|
|
assert compiler._retested_test_names() == ()
|
|
inner.rollback()
|
|
outer.rollback()
|
|
|
|
|
|
def test_beta_compiler_excludes_rolled_back_candidate_edits(memory_root: Path) -> None:
|
|
"""A reverted inner save point cannot enter the outer episode's target set."""
|
|
|
|
compiler, outer, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 1 == 1\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
target = memory_root / "tests/test_widget.py"
|
|
inner = Memento.first_available(target, outer.repo_exec).capture()
|
|
_write(outer.repo_exec, target, "def test_widget():\n assert 2 == 2\n")
|
|
inner.rollback()
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
|
|
assert compiler._retested_test_names() == ()
|
|
outer.rollback()
|
|
|
|
|
|
def test_beta_compiler_runs_pytest_after_a_code_failure(memory_root: Path) -> None:
|
|
"""Code diagnostics and current semantic evidence share one candidate result."""
|
|
|
|
runner = _TargetedRunner([
|
|
TestRunResult(
|
|
"pytest",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
"current assertion failed",
|
|
["assert 2 == 1"],
|
|
)
|
|
])
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 2 == 2\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: [StaticCompileError("mechanically invalid")]
|
|
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert errors == [
|
|
StaticCompileError("mechanically invalid"),
|
|
TestRunCompileError(
|
|
message=(
|
|
"Test command failed with return code 1.\nassert 2 == 1\ncurrent assertion failed"
|
|
),
|
|
test_name="test_widget",
|
|
error_message="Test command failed with return code 1.",
|
|
stack_trace=("assert 2 == 1",),
|
|
raw_output="current assertion failed",
|
|
),
|
|
]
|
|
assert runner.calls[0][2] == ("test_widget",)
|
|
assert compiler.blackboard.monitored == ("test_widget",)
|
|
assert compiler.blackboard.failures == ()
|
|
|
|
|
|
def test_compiler_runs_every_code_validation_stage_after_diagnostics(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Ordinary code failures cannot hide evidence from later validation stages."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": "value = 1\n"},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
calls: list[str] = []
|
|
syntax_error = StaticCompileError("syntax diagnostic")
|
|
type_error = TypeCheckError("type diagnostic")
|
|
reference_error = ReferenceError(
|
|
"reference diagnostic",
|
|
bad_reference=["import missing"],
|
|
available_references=[],
|
|
closest_fuzzy_match=[],
|
|
)
|
|
compiler.shallow_compile = lambda paths: calls.append("syntax") or [syntax_error]
|
|
compiler.type_check = lambda paths: calls.append("pyright") or [type_error]
|
|
compiler.validate_references = lambda paths: calls.append("references") or [
|
|
reference_error
|
|
]
|
|
|
|
assert compiler.compile([memory_root / "src/pkg/consumer.py"]) == [
|
|
syntax_error,
|
|
type_error,
|
|
reference_error,
|
|
]
|
|
assert calls == ["syntax", "pyright", "references"]
|
|
|
|
|
|
def test_beta_compiler_stops_after_a_static_environment_failure(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""An unavailable Pyright prevents downstream checks without changing inventory."""
|
|
|
|
failure = TestRunCompileError("old failure", test_name="test_widget")
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((failure,))
|
|
runner = _TargetedRunner()
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert False\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert True\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
calls: list[str] = []
|
|
unavailable = RepoExecCompileError("Unable to run Pyright")
|
|
compiler.shallow_compile = lambda paths: calls.append("syntax") or []
|
|
compiler.type_check = lambda paths: calls.append("pyright") or [unavailable]
|
|
compiler.validate_references = lambda paths: calls.append("references") or []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == [unavailable]
|
|
assert calls == ["syntax", "pyright"]
|
|
assert runner.calls == []
|
|
assert inventory.monitored == ("test_widget",)
|
|
assert inventory.failures == (failure,)
|
|
|
|
|
|
def test_beta_compiler_contains_a_missing_editor_snapshot(memory_root: Path) -> None:
|
|
"""Missing editor source remains a repository setup error in Beta."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert True\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert True\n",
|
|
context_name="BetaLoop",
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.mementos = ()
|
|
compiler.attempt = replace(
|
|
compiler.attempt,
|
|
mementos=(),
|
|
candidate_mementos=(),
|
|
)
|
|
|
|
[error] = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert isinstance(error, RepoExecCompileError)
|
|
assert error.message == "Compilation requires editor-owned ClassMeta Mementos."
|
|
assert compiler.blackboard.monitored == ()
|
|
|
|
|
|
def test_beta_compiler_stops_at_a_test_environment_failure(memory_root: Path) -> None:
|
|
"""An unavailable pytest environment stops validation without committing state."""
|
|
|
|
failure = TestRunCompileError("old failure", test_name="test_widget")
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((failure,))
|
|
runner = _TargetedRunner([RepoExecError("test environment unavailable")])
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert False\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert True\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
[error] = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert isinstance(error, RepoExecCompileError)
|
|
assert "test environment unavailable" in error.message
|
|
assert inventory.monitored == ("test_widget",)
|
|
assert inventory.failures == (failure,)
|
|
|
|
|
|
def test_beta_compiler_returns_targeted_test_failures_without_metadata_mutation(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Failed inner tests become repair evidence without refreshing approved metadata."""
|
|
|
|
runner = _TargetedRunner(
|
|
[
|
|
TestRunResult(
|
|
"pytest",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
"tests/test_widget.py::test_widget FAILED\nassertion failed",
|
|
)
|
|
]
|
|
)
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert 1 == 1\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 2 == 2\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
|
|
[error] = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert isinstance(error, TestRunCompileError)
|
|
assert error.test_name == "test_widget"
|
|
assert "assertion failed" in error.message
|
|
assert manager.refresh_test_calls == []
|
|
assert manager.upsert_test_calls == []
|
|
|
|
|
|
def test_beta_compiler_seeds_every_named_failure_on_an_unchanged_baseline(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Static cleanliness cannot clear aggregate failures that name exact tests."""
|
|
|
|
compiler, memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def test_alpha():\n"
|
|
" assert True\n\n"
|
|
"def test_beta():\n"
|
|
" assert True\n\n"
|
|
"def test_clear():\n"
|
|
" assert True\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def test_alpha():\n"
|
|
" assert True\n\n"
|
|
"def test_beta():\n"
|
|
" assert True\n\n"
|
|
"def test_clear():\n"
|
|
" assert True\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
candidate=False,
|
|
failure_error_message="test_alpha failed before test_beta failed",
|
|
failure_stack_trace=("tests/test_widget.py::test_beta",),
|
|
failure_raw_output="original aggregate output",
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler_calls: list[tuple[str, tuple[Path, ...]]] = []
|
|
compiler.shallow_compile = lambda paths: compiler_calls.append(
|
|
("syntax", tuple(paths))
|
|
) or []
|
|
compiler.type_check = lambda paths: compiler_calls.append(
|
|
("pyright", tuple(paths))
|
|
) or []
|
|
compiler.validate_references = lambda paths: compiler_calls.append(
|
|
("reference-policy", tuple(paths))
|
|
) or []
|
|
|
|
assert memento.diff == ""
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
expected_paths = (memory_root / "tests/test_widget.py",)
|
|
assert compiler_calls == [
|
|
("syntax", expected_paths),
|
|
("pyright", expected_paths),
|
|
("reference-policy", expected_paths),
|
|
]
|
|
test_errors = [error for error in errors if isinstance(error, TestRunCompileError)]
|
|
assert len(test_errors) == len(errors)
|
|
assert [error.test_name for error in test_errors] == ["test_alpha", "test_beta"]
|
|
assert [error.raw_output for error in test_errors] == [
|
|
"original aggregate output",
|
|
"original aggregate output",
|
|
]
|
|
|
|
|
|
def test_beta_compiler_runs_identified_overlapping_names(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Initial containment is broad, while later exact execution is key-local."""
|
|
|
|
inventory = EditedTestInventory(lambda: None)
|
|
baseline, _memento, _manager = _compiler(
|
|
memory_root / "baseline",
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def test_item():\n"
|
|
" assert False\n\n"
|
|
"def test_item_long():\n"
|
|
" assert False\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def test_item():\n"
|
|
" assert False\n\n"
|
|
"def test_item_long():\n"
|
|
" assert False\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
candidate=False,
|
|
blackboard=inventory,
|
|
failure_error_message="test_item_long failed",
|
|
)
|
|
assert isinstance(baseline, BetaPythonCompiler)
|
|
baseline.shallow_compile = lambda paths: []
|
|
baseline.type_check = lambda paths: []
|
|
baseline.validate_references = lambda paths: []
|
|
baseline_errors = baseline.compile(
|
|
[memory_root / "baseline/tests/test_widget.py"]
|
|
)
|
|
runner = _TargetedRunner(
|
|
[
|
|
TestRunResult(
|
|
"pytest item",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
"tests/test_widget.py::test_item FAILED with latest error",
|
|
)
|
|
]
|
|
)
|
|
candidate, _memento, manager = _compiler(
|
|
memory_root / "candidate",
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def test_item():\n"
|
|
" assert False\n\n"
|
|
"def test_item_long():\n"
|
|
" assert False\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def test_item():\n"
|
|
" assert 1 == 2\n\n"
|
|
"def test_item_long():\n"
|
|
" assert False\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(candidate, BetaPythonCompiler)
|
|
candidate.shallow_compile = lambda paths: []
|
|
candidate.type_check = lambda paths: []
|
|
candidate.validate_references = lambda paths: []
|
|
|
|
candidate_errors = candidate.compile(
|
|
[memory_root / "candidate/tests/test_widget.py"]
|
|
)
|
|
inventory.commit_candidate_results()
|
|
|
|
baseline_test_errors = [
|
|
error for error in baseline_errors if isinstance(error, TestRunCompileError)
|
|
]
|
|
assert len(baseline_test_errors) == len(baseline_errors)
|
|
assert [error.test_name for error in baseline_test_errors] == [
|
|
"test_item",
|
|
"test_item_long",
|
|
]
|
|
assert candidate_errors[0] is not baseline_errors[0]
|
|
assert len(candidate_errors) == 1
|
|
assert inventory.monitored == ("test_item",)
|
|
assert runner.calls == [
|
|
([manager.test_classes[0]], 1, ("test_item",)),
|
|
([manager.test_classes[0]], 1, ("test_item_long",)),
|
|
]
|
|
|
|
|
|
def test_beta_compiler_clears_only_the_exact_retested_passing_method(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Only the passing singleton clears; the other monitored test gets fresh evidence."""
|
|
|
|
alpha_error = TestRunCompileError(
|
|
"old alpha failure",
|
|
test_name="test_alpha",
|
|
error_message="alpha error",
|
|
stack_trace=("alpha trace",),
|
|
raw_output="alpha output",
|
|
)
|
|
beta_error = TestRunCompileError(
|
|
"old beta failure",
|
|
test_name="test_beta",
|
|
error_message="beta error",
|
|
stack_trace=("beta trace",),
|
|
raw_output="beta output",
|
|
)
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((alpha_error, beta_error))
|
|
runner = _TargetedRunner([
|
|
TestRunResult("pytest alpha", TestRunOutcome.Failure, 1, "tests/test_widget.py::test_alpha FAILED again"),
|
|
TestRunResult("pytest beta", TestRunOutcome.Success, 0, ""),
|
|
])
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def test_alpha():\n"
|
|
" assert True\n\n"
|
|
"def test_beta():\n"
|
|
" assert False\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def test_alpha():\n"
|
|
" assert True\n\n"
|
|
"def test_beta():\n"
|
|
" assert True\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
inventory.commit_candidate_results()
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].test_name == "test_alpha"
|
|
assert errors[0].raw_output.endswith("FAILED again")
|
|
assert inventory.monitored == ("test_alpha",)
|
|
assert runner.calls == [
|
|
([manager.test_classes[0]], 1, ("test_alpha",)),
|
|
([manager.test_classes[0]], 1, ("test_beta",)),
|
|
]
|
|
|
|
|
|
def test_beta_compiler_replaces_only_the_exact_retested_failing_method(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A singleton failure replaces its key without parsing rendered output."""
|
|
|
|
alpha_error = TestRunCompileError(
|
|
"old alpha failure",
|
|
test_name="test_alpha",
|
|
error_message="alpha error",
|
|
stack_trace=("alpha trace",),
|
|
raw_output="alpha output",
|
|
)
|
|
beta_error = TestRunCompileError(
|
|
"old beta failure",
|
|
test_name="test_beta",
|
|
error_message="old beta error",
|
|
stack_trace=("old beta trace",),
|
|
raw_output="old beta output",
|
|
)
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((alpha_error, beta_error))
|
|
runner = _TargetedRunner(
|
|
[
|
|
TestRunResult("pytest alpha", TestRunOutcome.Success, 0, ""),
|
|
TestRunResult(
|
|
"pytest beta",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
"assertion failed with new beta error",
|
|
["tests/test_widget.py:105: in a_rendered_frame"],
|
|
)
|
|
]
|
|
)
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def test_alpha():\n"
|
|
" assert True\n\n"
|
|
"def test_beta():\n"
|
|
" assert False\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def test_alpha():\n"
|
|
" assert True\n\n"
|
|
"def test_beta():\n"
|
|
" assert 1 == 2\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
inventory.commit_candidate_results()
|
|
|
|
assert len(errors) == 1
|
|
assert inventory.monitored == ("test_beta",)
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].test_name == "test_beta"
|
|
assert errors[0].error_message == "Test command failed with return code 1."
|
|
assert errors[0].stack_trace == (
|
|
"tests/test_widget.py:105: in a_rendered_frame",
|
|
)
|
|
assert errors[0].raw_output.endswith("new beta error")
|
|
|
|
|
|
def test_beta_monitoring_clears_after_helper_repair(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A helper repair reruns unchanged monitored tests and clears old failures."""
|
|
|
|
alpha_error = TestRunCompileError("alpha failure", test_name="test_alpha")
|
|
beta_error = TestRunCompileError("beta failure", test_name="test_beta")
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((alpha_error, beta_error))
|
|
runner = _TargetedRunner()
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def helper():\n"
|
|
" return 1\n\n"
|
|
"def test_alpha():\n"
|
|
" assert helper() == 2\n\n"
|
|
"def test_beta():\n"
|
|
" assert helper() == 2\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def helper():\n"
|
|
" return 2\n\n"
|
|
"def test_alpha():\n"
|
|
" assert helper() == 2\n\n"
|
|
"def test_beta():\n"
|
|
" assert helper() == 2\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
inventory.commit_candidate_results()
|
|
assert tuple(call[2] for call in runner.calls) == (("test_alpha",), ("test_beta",))
|
|
assert inventory.monitored == ()
|
|
assert inventory.failures == ()
|
|
|
|
|
|
def test_beta_compiler_omits_a_deleted_unresolved_test_without_running_it(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A prior failure is stale once its exact identity leaves the accepted file."""
|
|
|
|
old_error = TestRunCompileError(
|
|
"old deleted-test failure",
|
|
test_name="test_deleted",
|
|
)
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((old_error,))
|
|
runner = _TargetedRunner()
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_deleted():\n assert False\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
inventory.commit_candidate_results()
|
|
assert runner.calls == []
|
|
|
|
|
|
def test_beta_compiler_does_not_treat_an_invalid_ast_as_test_removal(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Failed AST discovery retains and executes the existing monitored identity."""
|
|
|
|
old_error = TestRunCompileError(
|
|
"old widget failure",
|
|
test_name="test_widget",
|
|
)
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((old_error,))
|
|
runner = _TargetedRunner([
|
|
TestRunResult(
|
|
"pytest widget",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
"collection failed for invalid source",
|
|
)
|
|
])
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert False\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget(:\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: [StaticCompileError("invalid syntax")]
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
inventory.discard_candidate_results()
|
|
|
|
assert [error.message for error in errors] == [
|
|
"invalid syntax",
|
|
(
|
|
"Test command failed with return code 1.\ncollection failed for invalid source"
|
|
),
|
|
]
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_widget",))]
|
|
assert inventory.monitored == ("test_widget",)
|
|
assert inventory.failures == (old_error,)
|
|
|
|
|
|
def test_beta_compiler_treats_a_renamed_failing_test_as_delete_plus_create(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A rename drops the old failure and executes the new accepted identity."""
|
|
|
|
old_error = TestRunCompileError(
|
|
"old renamed-test failure",
|
|
test_name="test_old",
|
|
)
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((old_error,))
|
|
runner = _TargetedRunner([TestRunResult("pytest new", TestRunOutcome.Success, 0, "")])
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_old():\n assert False\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_new():\n assert True\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
inventory.commit_candidate_results()
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_new",))]
|
|
|
|
|
|
def test_beta_compiler_runs_a_created_test_and_records_its_failure(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A new pytest method expands the catalog before its singleton execution."""
|
|
|
|
runner = _TargetedRunner(
|
|
[
|
|
TestRunResult(
|
|
"pytest new",
|
|
TestRunOutcome.Failure,
|
|
1,
|
|
"tests/test_widget.py::test_new FAILED",
|
|
["tests/test_widget.py::test_new: AssertionError"],
|
|
)
|
|
]
|
|
)
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": ""},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_new():\n assert False\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
[error] = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert isinstance(error, TestRunCompileError)
|
|
assert error.test_name == "test_new"
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_new",))]
|
|
|
|
|
|
def test_beta_compiler_runs_a_created_passing_test_without_catalog_failure(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A successful new exact test leaves no unresolved compiler diagnostic."""
|
|
|
|
runner = _TargetedRunner([TestRunResult("pytest new", TestRunOutcome.Success, 0, "")])
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": ""},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_new():\n assert True\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_new",))]
|
|
|
|
|
|
def test_beta_compiler_replaces_old_failure_with_singleton_timeout(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A timeout is the current semantic failure for its requested test."""
|
|
|
|
old_error = TestRunCompileError(
|
|
"old widget failure",
|
|
test_name="test_widget",
|
|
error_message="old error",
|
|
stack_trace=("old trace",),
|
|
raw_output="old output",
|
|
)
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((old_error,))
|
|
runner = _TargetedRunner([
|
|
TestRunResult(
|
|
"pytest widget",
|
|
TestRunOutcome.Failure,
|
|
124,
|
|
"TIMEOUT: pytest exceeded 1800 seconds",
|
|
["partial stdout", "partial stderr"],
|
|
)
|
|
])
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": "def test_widget():\n assert False\n"},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_widget():\n assert 1 == 2\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
inventory.commit_candidate_results()
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].test_name == "test_widget"
|
|
assert errors[0].error_message == "Test command failed with return code 124."
|
|
assert errors[0].stack_trace == ("partial stdout", "partial stderr")
|
|
assert errors[0] is not old_error
|
|
|
|
|
|
def test_beta_compiler_runs_changed_tests_separately_in_sorted_order(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Each changed catalog key receives its own deterministic pytest process."""
|
|
|
|
runner = _TargetedRunner(
|
|
[
|
|
TestRunResult("pytest alpha", TestRunOutcome.Success, 0, ""),
|
|
TestRunResult("pytest beta", TestRunOutcome.Success, 0, ""),
|
|
]
|
|
)
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={
|
|
"test_widget.py": (
|
|
"def test_beta():\n"
|
|
" assert False\n\n"
|
|
"def test_alpha():\n"
|
|
" assert False\n"
|
|
)
|
|
},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=(
|
|
"def test_beta():\n"
|
|
" assert True\n\n"
|
|
"def test_alpha():\n"
|
|
" assert True\n"
|
|
),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
)
|
|
assert isinstance(compiler, BetaPythonCompiler)
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
compiler.validate_references = lambda paths: []
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
assert runner.calls == [
|
|
([manager.test_classes[0]], 1, ("test_alpha",)),
|
|
([manager.test_classes[0]], 1, ("test_beta",)),
|
|
]
|
|
|
|
|
|
def test_top_level_relative_import_is_not_mistaken_for_a_local_module(memory_root: Path) -> None:
|
|
"""A namespace-root module cannot legally use ``from .x`` imports."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"base.py": "class Base: pass\n", "consumer.py": ""},
|
|
edit_location="src/consumer.py",
|
|
replacement="from .base import Base\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/base.py", "src/consumer.py"),
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference == ["from .base import Base"]
|
|
assert "escapes the known package root" in error.message
|
|
|
|
|
|
def test_alpha_rejects_untraversed_production_reference_and_suggests_legal_symbol(memory_root: Path) -> None:
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/current.py": "",
|
|
"pkg/legal.py": "class Widget: pass\n",
|
|
"pkg/later.py": "class Later: pass\n",
|
|
},
|
|
edit_location="src/pkg/current.py",
|
|
replacement="from pkg.later import Later\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/current.py", "src/pkg/legal.py"),
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference == ["from pkg.later import Later"]
|
|
assert "outside the current/traversed COT" in error.message
|
|
assert "pkg.legal.Widget" in error.available_references
|
|
assert "pkg.later.Later" not in error.available_references
|
|
|
|
|
|
def test_beta_accepts_readonly_production_reference_outside_traversed_cot(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A supplied production file is part of Beta's legal import surface."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/current.py": "class Current: pass\n",
|
|
"pkg/readonly.py": "class LifecycleError(RuntimeError): pass\n",
|
|
},
|
|
tests={"test_current.py": ""},
|
|
edit_location="tests/test_current.py",
|
|
replacement="from pkg.readonly import LifecycleError\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
readonly_locations=("src/pkg/readonly.py",),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_beta_accepts_direct_upstream_of_supplied_files(memory_root: Path) -> None:
|
|
"""One graph edge from editable or readonly evidence is legal to reference."""
|
|
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/current.py": "class Current: pass\n",
|
|
"pkg/readonly.py": "from pkg.upstream import Phase\n",
|
|
"pkg/upstream.py": "class Phase: pass\n",
|
|
},
|
|
tests={"test_current.py": ""},
|
|
edit_location="tests/test_current.py",
|
|
replacement="from pkg.upstream import Phase\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
readonly_locations=("src/pkg/readonly.py",),
|
|
upstream_locations=("src/pkg/upstream.py",),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
upstream_sources_for = cast(
|
|
Mock,
|
|
compiler.dependency_strategy.upstream_sources_for,
|
|
)
|
|
[call] = upstream_sources_for.call_args_list
|
|
supplied = call.args[0]
|
|
assert set(map(lambda meta: meta.location, supplied)) == {
|
|
"src/pkg/readonly.py",
|
|
"tests/test_current.py",
|
|
}
|
|
assert call.kwargs["recursive"] is False
|
|
assert call.args[1] == manager.src_classes
|
|
|
|
|
|
def test_beta_rejects_transitive_source_beyond_direct_upstream_scope(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A second graph hop remains illegal beside an admitted direct dependency."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/current.py": "class Current: pass\n",
|
|
"pkg/readonly.py": "from pkg.upstream import Phase\n",
|
|
"pkg/upstream.py": "from pkg.transitive import Resolution\nclass Phase: pass\n",
|
|
"pkg/transitive.py": "class Resolution: pass\n",
|
|
},
|
|
tests={"test_current.py": ""},
|
|
edit_location="tests/test_current.py",
|
|
replacement="from pkg.transitive import Resolution\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
readonly_locations=("src/pkg/readonly.py",),
|
|
upstream_locations=("src/pkg/upstream.py",),
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert "supplied files, and their direct upstream dependencies" in error.message
|
|
assert "pkg.upstream.Phase" in error.available_references
|
|
assert "pkg.transitive.Resolution" not in error.available_references
|
|
|
|
|
|
def test_alpha_rejects_source_to_test_reference(memory_root: Path) -> None:
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/current.py": ""},
|
|
tests={"test_support.py": "class Support: pass\n"},
|
|
edit_location="src/pkg/current.py",
|
|
replacement="from test_support import Support\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
readonly_locations=("tests/test_support.py",),
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference == ["from test_support import Support"]
|
|
assert "source-to-test" in error.message
|
|
assert all("test_support" not in item for item in error.available_references)
|
|
|
|
|
|
def test_reference_resolver_accepts_exported_constants_and_reexports(memory_root: Path) -> None:
|
|
"""Local AST export facts supplement source-discovery class/function metadata."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "from .constants import VALUE\n",
|
|
"pkg/constants.py": "VALUE = 42\n",
|
|
"pkg/consumer.py": "",
|
|
},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="from pkg import VALUE\nfrom pkg.constants import VALUE as OTHER\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=(
|
|
"src/pkg/__init__.py",
|
|
"src/pkg/constants.py",
|
|
"src/pkg/consumer.py",
|
|
),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_reference_resolver_honors_explicit_all_over_discovered_symbols(memory_root: Path) -> None:
|
|
"""An explicit module export list remains authoritative for local symbols."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/api.py": "__all__ = ['Public']\nclass Public: pass\nclass Private: pass\n",
|
|
"pkg/consumer.py": "",
|
|
},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="from pkg.api import Private\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/api.py", "src/pkg/consumer.py"),
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference == ["from pkg.api import Private"]
|
|
|
|
|
|
def test_beta_rejects_test_to_test_reference(memory_root: Path) -> None:
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/current.py": "class Current: pass\n"},
|
|
tests={"test_current.py": "", "test_helpers.py": "class Helper: pass\n"},
|
|
edit_location="tests/test_current.py",
|
|
replacement="from test_helpers import Helper\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
readonly_locations=("tests/test_helpers.py",),
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference == ["from test_helpers import Helper"]
|
|
assert "test-to-test" in error.message
|
|
|
|
|
|
def test_beta_prefers_legal_production_module_over_same_named_test_module(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""A test-root name collision cannot turn a production dependency into test-to-test."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/foo.py": "class Foo: pass\n",
|
|
},
|
|
tests={
|
|
"pkg/foo.py": "class TestFoo: pass\n",
|
|
"test_consumer.py": "",
|
|
},
|
|
edit_location="tests/test_consumer.py",
|
|
replacement="from pkg.foo import Foo\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/foo.py",),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_from_package_prefers_a_production_reexport_over_a_test_child_module(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Package exports remain legal even if a test module shares their name."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "from .api import Widget\n",
|
|
"pkg/api.py": "class Widget: pass\n",
|
|
},
|
|
tests={"pkg/Widget.py": "class TestOnly: pass\n", "test_consumer.py": ""},
|
|
edit_location="tests/test_consumer.py",
|
|
replacement="from pkg import Widget\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/__init__.py", "src/pkg/api.py"),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_reference_resolver_handles_relative_file_manager_roots(memory_root: Path) -> None:
|
|
"""FileManager roots may be repository-relative without misclassifying source as test."""
|
|
|
|
compiler, _memento, manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/api.py": "class Api: pass\n",
|
|
"pkg/consumer.py": "",
|
|
},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="from pkg.api import Api\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/api.py", "src/pkg/consumer.py"),
|
|
)
|
|
manager.src_root = Path("src")
|
|
manager.test_root = Path("tests")
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_bootstrap_allows_all_production_and_distribution_import_mapping(memory_root: Path) -> None:
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/consumer.py": "",
|
|
"pkg/other.py": "class Other: pass\n",
|
|
},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="from pkg.other import Other\nfrom fancy_pkg.client import Client\n",
|
|
context_name="BootstrapLoop",
|
|
references=[ExternalReference("fancy-dist", ">=1", Path("requirements.txt"), ("fancy_pkg",))],
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_beta_can_use_dev_requirements_without_exposing_them_to_alpha(memory_root: Path) -> None:
|
|
"""requirements.txt stays test-only when a higher-priority BOM exists."""
|
|
|
|
beta, _memento, _manager = _compiler(
|
|
memory_root / "beta",
|
|
sources={"pkg/__init__.py": "", "pkg/current.py": "class Current: pass\n"},
|
|
tests={"test_current.py": ""},
|
|
edit_location="tests/test_current.py",
|
|
replacement="import pytest\n",
|
|
context_name="BetaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
)
|
|
beta.repo_exec.write_text(
|
|
beta.repo_root / "setup.py",
|
|
"from setuptools import setup\nsetup(install_requires=['requests>=2'])\n",
|
|
)
|
|
beta.repo_exec.write_text(beta.repo_root / "requirements.txt", "pytest>=8\n")
|
|
beta.bom_editor = PythonBomEditor(beta.repo_root, beta.repo_exec)
|
|
|
|
assert beta.validate_references([]) == []
|
|
|
|
alpha, _memento, _manager = _compiler(
|
|
memory_root / "alpha",
|
|
sources={"pkg/__init__.py": "", "pkg/current.py": ""},
|
|
edit_location="src/pkg/current.py",
|
|
replacement="import pytest\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/current.py",),
|
|
)
|
|
alpha.repo_exec.write_text(
|
|
alpha.repo_root / "setup.py",
|
|
"from setuptools import setup\nsetup(install_requires=['requests>=2'])\n",
|
|
)
|
|
alpha.repo_exec.write_text(alpha.repo_root / "requirements.txt", "pytest>=8\n")
|
|
alpha.bom_editor = PythonBomEditor(alpha.repo_root, alpha.repo_exec)
|
|
|
|
[error] = alpha.validate_references([])
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference == ["import pytest"]
|
|
|
|
|
|
def test_external_resolution_supplements_distribution_aliases_from_repo_exec(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Target-environment metadata fills aliases that are not in the static table."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="",
|
|
context_name="BootstrapLoop",
|
|
)
|
|
metadata_exec = compiler.repo_exec
|
|
assert isinstance(metadata_exec, MemoryRepoExec)
|
|
metadata_exec.queue_result(
|
|
CompletedProcess(
|
|
[],
|
|
0,
|
|
stdout='{"custom_alias": ["custom-dist"]}',
|
|
stderr="",
|
|
)
|
|
)
|
|
|
|
candidates = compiler._external_candidates_from(
|
|
[ExternalReference("custom-dist", ">=1", Path("requirements.txt"))]
|
|
)
|
|
|
|
assert candidates[0].import_names[0] == "custom_dist"
|
|
assert "custom_alias" in candidates[0].import_names
|
|
assert metadata_exec.commands[0][1] == "-c"
|
|
assert "packages_distributions" in metadata_exec.commands[0][2]
|
|
|
|
|
|
def test_reference_resolver_accepts_standard_library_on_all_supported_interpreters(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Standard-library imports do not require a declared external distribution."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="from pathlib import Path\nimport json\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
|
|
assert compiler.validate_references([]) == []
|
|
|
|
|
|
def test_unchanged_imports_do_not_read_or_revalidate_a_legacy_bom(memory_root: Path) -> None:
|
|
"""A malformed historical manifest cannot block an edit with no import delta."""
|
|
|
|
class _UnreadableBom(_Bom):
|
|
def inventory(self, *, include_development: bool = False) -> list[ExternalReference]:
|
|
del include_development
|
|
raise AssertionError("unchanged imports must not consult the BOM")
|
|
|
|
compiler, memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": "value = 1\n"},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
compiler.bom_editor = _UnreadableBom()
|
|
compiler.shallow_compile = lambda paths: []
|
|
compiler.type_check = lambda paths: []
|
|
reference_calls: list[tuple[Path, ...]] = []
|
|
validate_references = compiler.validate_references
|
|
|
|
def record_reference_policy(paths: Sequence[Path]) -> list[CompileError]:
|
|
reference_calls.append(tuple(paths))
|
|
return validate_references(paths)
|
|
|
|
compiler.validate_references = record_reference_policy
|
|
|
|
assert memento.diff == ""
|
|
assert compiler.compile([memory_root / "src/pkg/consumer.py"]) == []
|
|
assert reference_calls == [(memory_root / "src/pkg/consumer.py",)]
|
|
|
|
|
|
def test_shallow_compile_uses_native_repo_python_without_writing_bytecode(memory_root: Path) -> None:
|
|
"""The syntax pass is target-environment compilation, not host file parsing."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="def malformed(:\n pass\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
|
|
errors = compiler.shallow_compile([memory_root / "src/pkg/consumer.py"])
|
|
|
|
assert len(errors) == 1
|
|
assert "SyntaxError" in errors[0].message
|
|
assert isinstance(compiler.repo_exec, MemoryRepoExec)
|
|
assert compiler.repo_exec.list_files(memory_root, ("__pycache__/*",), recursive=True) == []
|
|
|
|
|
|
def test_type_check_uses_selected_repo_python_module_entrypoint(memory_root: Path) -> None:
|
|
"""Pyright must run in the selected RepoExec runtime rather than host PATH."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
repo_exec = compiler.repo_exec
|
|
assert isinstance(repo_exec, MemoryRepoExec)
|
|
|
|
assert compiler.type_check([memory_root / "src/pkg/consumer.py"]) == []
|
|
assert repo_exec.commands == [
|
|
[
|
|
repo_exec.python_bin(),
|
|
"-m",
|
|
"pyright",
|
|
"--outputjson",
|
|
str(memory_root / "src/pkg/consumer.py"),
|
|
]
|
|
]
|
|
|
|
|
|
def test_type_check_returns_every_pyright_error_as_a_structured_diagnostic(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Pyright error diagnostics remain ordered TypeCheckError values."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
assert isinstance(compiler.repo_exec, MemoryRepoExec)
|
|
compiler.repo_exec.queue_result(
|
|
CompletedProcess(
|
|
[],
|
|
0,
|
|
stdout=(
|
|
'{"generalDiagnostics": ['
|
|
'{"severity": "error", "message": "first", '
|
|
'"file": "src/pkg/consumer.py", '
|
|
'"range": {"start": {"line": 2, "character": 4}}}, '
|
|
'{"severity": "warning", "message": "ignored"}, '
|
|
'{"severity": "error", "message": "second", '
|
|
'"range": {"start": {"line": 5, "character": 1}}}'
|
|
']}'
|
|
),
|
|
stderr="",
|
|
)
|
|
)
|
|
|
|
errors = compiler.type_check([memory_root / "src/pkg/consumer.py"])
|
|
|
|
assert errors == [
|
|
TypeCheckError(
|
|
"first",
|
|
path=Path("src/pkg/consumer.py"),
|
|
line=2,
|
|
column=4,
|
|
),
|
|
TypeCheckError("second", line=5, column=1),
|
|
]
|
|
assert not hasattr(errors[0], "reference_cards")
|
|
|
|
|
|
def test_repo_exec_failure_is_returned_as_compile_error(memory_root: Path) -> None:
|
|
"""Adapter failures never leak beyond the language-neutral compiler boundary."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
assert isinstance(compiler.repo_exec, MemoryRepoExec)
|
|
compiler.repo_exec.queue_result(RepoExecError("selected environment unavailable"))
|
|
|
|
[error] = compiler.shallow_compile([memory_root / "src/pkg/consumer.py"])
|
|
|
|
assert isinstance(error, RepoExecCompileError)
|
|
assert "selected environment unavailable" in error.message
|
|
|
|
|
|
def test_missing_pyright_module_is_a_repo_exec_compile_error(memory_root: Path) -> None:
|
|
"""A target runtime without Pyright must not silently accept the candidate."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
assert isinstance(compiler.repo_exec, MemoryRepoExec)
|
|
compiler.repo_exec.queue_result(
|
|
CompletedProcess([], 1, stdout="", stderr="No module named pyright")
|
|
)
|
|
|
|
[error] = compiler.type_check([memory_root / "src/pkg/consumer.py"])
|
|
|
|
assert isinstance(error, RepoExecCompileError)
|
|
assert "No module named pyright" in error.message
|
|
|
|
|
|
def test_type_check_rejects_malformed_pyright_diagnostic_shapes(memory_root: Path) -> None:
|
|
"""Malformed Pyright JSON remains a repairable RepoExec compiler failure."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
path = memory_root / "src/pkg/consumer.py"
|
|
assert isinstance(compiler.repo_exec, MemoryRepoExec)
|
|
for payload in (
|
|
'{"generalDiagnostics": null}',
|
|
'{"generalDiagnostics": [false]}',
|
|
'{"generalDiagnostics": [{"severity": "error", "range": null}]}',
|
|
):
|
|
compiler.repo_exec.queue_result(CompletedProcess([], 0, stdout=payload, stderr=""))
|
|
|
|
[error] = compiler.type_check([path])
|
|
|
|
assert isinstance(error, RepoExecCompileError)
|
|
|
|
|
|
def test_standard_library_inventory_uses_selected_repo_exec_once_with_fallback(
|
|
memory_root: Path,
|
|
) -> None:
|
|
"""Stdlib resolution is target-runtime scoped and cached per compiler."""
|
|
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={"pkg/__init__.py": "", "pkg/consumer.py": ""},
|
|
edit_location="src/pkg/consumer.py",
|
|
replacement="value = 1\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/consumer.py",),
|
|
)
|
|
repo_exec = compiler.repo_exec
|
|
assert isinstance(repo_exec, MemoryRepoExec)
|
|
repo_exec.queue_result(CompletedProcess([], 0, stdout='["target_only"]', stderr=""))
|
|
|
|
assert compiler._is_stdlib_root("target_only")
|
|
assert compiler._is_stdlib_root("pathlib")
|
|
assert not compiler._is_stdlib_root("not_a_target_stdlib_module")
|
|
assert len(repo_exec.commands) == 1
|
|
command = repo_exec.commands[0]
|
|
assert command[:2] == [repo_exec.python_bin(), "-c"]
|
|
assert "stdlib_module_names" in command[2]
|
|
|
|
|
|
def test_reference_error_ranks_internal_suffix_and_external_prefix_matches(memory_root: Path) -> None:
|
|
class _RaisingClient(StaticOpenAICompatibleClient):
|
|
"""Fail visibly if reference validation incorrectly invokes a model client."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__('{"cdc": ""}')
|
|
self.calls = 0
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
del prompt, logger
|
|
self.calls += 1
|
|
raise AssertionError("Reference validation must not call a completion client.")
|
|
|
|
client = _RaisingClient()
|
|
compiler, _memento, _manager = _compiler(
|
|
memory_root,
|
|
sources={
|
|
"pkg/__init__.py": "",
|
|
"pkg/current.py": "",
|
|
"pkg/good.py": "class Widget: pass\n",
|
|
},
|
|
edit_location="src/pkg/current.py",
|
|
replacement="from pkg.badd import Widget\nimport requestz\n",
|
|
context_name="AlphaLoop",
|
|
frontier_locations=("src/pkg/current.py", "src/pkg/good.py"),
|
|
references=[ExternalReference("requests", ">=2", Path("requirements.txt"), ("requests",))],
|
|
client=client,
|
|
)
|
|
|
|
[error] = compiler.validate_references([])
|
|
|
|
assert isinstance(error, ReferenceError)
|
|
assert error.bad_reference
|
|
assert "pkg.good.Widget" in error.available_references
|
|
assert "pkg.good.Widget" in error.closest_fuzzy_match
|
|
assert "requests>=2" in error.closest_fuzzy_match
|
|
assert client.calls == 0
|
|
assert not hasattr(error, "llm_judgement")
|
|
assert "SEMANTIC JUDGEMENT" not in error.message
|
|
|
|
|
|
def test_stabilization_frontier_provider_yields_current_classmeta_not_index(memory_root: Path) -> None:
|
|
"""A live manager record replaces a stale topology record in the owned loop query."""
|
|
|
|
current = _meta("src/pkg/current.py", ContentType.Source)
|
|
repo_exec = MemoryRepoExec(memory_root)
|
|
manager = _FileManager(
|
|
repo_root=memory_root,
|
|
src_root=memory_root / "src",
|
|
test_root=memory_root / "tests",
|
|
repo_exec=repo_exec,
|
|
src_classes=[current],
|
|
test_classes=[],
|
|
)
|
|
stale = _meta("src/pkg/current.py", ContentType.Source)
|
|
context = StabilizationContext(
|
|
goal="stabilize current metadata",
|
|
classes=[stale],
|
|
upstream_file_manager=manager,
|
|
root_change=stale,
|
|
file_manager=manager,
|
|
)
|
|
|
|
class _ReferenceFrontierLoop(StabilizationLoop):
|
|
"""Expose the inherited frontier query without starting a workflow."""
|
|
|
|
def __init__(self, current_context: StabilizationContext) -> None:
|
|
self.context = current_context
|
|
|
|
loop = _ReferenceFrontierLoop(context)
|
|
|
|
[frontier] = list(loop.reference_frontiers())
|
|
|
|
assert frontier == [current]
|
|
|
|
|
|
def test_beta_monitoring_does_not_readd_identified_passes(memory_root: Path) -> None:
|
|
"""A cleared unchanged test stays cleared across helper edits and compiler reconstruction."""
|
|
|
|
inventory = EditedTestInventory(lambda: None)
|
|
runner = _TargetedRunner([TestRunResult("pytest value", TestRunOutcome.Success, 0, "")])
|
|
test = "\ndef test_value():\n assert helper() == 2\n"
|
|
sources = tuple(f"def helper():\n return {value}\n" + test for value in (1, 2, 3))
|
|
# The second helper edit may regress this cleared test; outer validation owns that discovery.
|
|
for before, after in zip(sources, sources[1:]):
|
|
compiler, _, _ = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": before},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=after,
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
failure_raw_output="FAILED tests/test_widget.py::test_value",
|
|
previous_errors=(TestRunCompileError("stale failure", test_name="test_value"),),
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
inventory.commit_candidate_results()
|
|
assert inventory.monitored == ()
|
|
assert inventory.failures == ()
|
|
|
|
assert tuple(call[2] for call in runner.calls) == (("test_value",),)
|
|
|
|
|
|
def test_beta_monitoring_readmits_touched_tests(memory_root: Path) -> None:
|
|
"""A previously passing test reenters monitoring when its definition changes again."""
|
|
|
|
inventory = EditedTestInventory(lambda: None)
|
|
runner = _TargetedRunner([
|
|
TestRunResult("pytest value", TestRunOutcome.Success, 0, ""),
|
|
TestRunResult("pytest value", TestRunOutcome.Failure, 1, "tests/test_widget.py::test_value FAILED anew"),
|
|
])
|
|
original = "def test_value():\n assert False\n"
|
|
passing = "def test_value():\n assert True\n"
|
|
compiler, _, _ = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": original},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=passing,
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
inventory.commit_candidate_results()
|
|
assert inventory.monitored == ()
|
|
|
|
compiler, _, _ = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": passing},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=original,
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
inventory.commit_candidate_results()
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].raw_output.endswith("FAILED anew")
|
|
assert inventory.monitored == ("test_value",)
|
|
assert tuple(call[2] for call in runner.calls) == (("test_value",), ("test_value",))
|
|
|
|
|
|
def test_beta_monitoring_survives_static_rejection(memory_root: Path) -> None:
|
|
"""A static-invalid deletion neither clears monitored evidence nor adds a new test."""
|
|
|
|
failure = TestRunCompileError("old failure", test_name="test_value")
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures((failure,))
|
|
runner = _TargetedRunner()
|
|
original = "def helper():\n return 1\n\ndef test_value():\n assert helper() == 2\n"
|
|
compiler, memento, manager = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": original},
|
|
edit_location="tests/test_widget.py",
|
|
replacement="def test_new():\n assert True\n\ninvalid =\n",
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
errors = compiler.compile([memory_root / "tests/test_widget.py"])
|
|
|
|
assert errors
|
|
assert all(isinstance(error, StaticCompileError) for error in errors)
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_value",))]
|
|
assert inventory.monitored == ("test_value",)
|
|
assert inventory.failures == (failure,)
|
|
inventory.discard_candidate_results()
|
|
memento.rollback()
|
|
assert manager.repo_exec.read_text(memento.file_path) == original
|
|
|
|
repaired, _, _ = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": original},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=original.replace("return 1", "return 2"),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
assert repaired.compile([memory_root / "tests/test_widget.py"]) == []
|
|
inventory.commit_candidate_results()
|
|
assert inventory.monitored == ()
|
|
assert tuple(call[2] for call in runner.calls) == (
|
|
("test_value",),
|
|
("test_value",),
|
|
)
|
|
|
|
|
|
def test_beta_monitoring_excludes_unadmitted_tests(memory_root: Path) -> None:
|
|
"""Current definitions and stale diagnostic records alone do not admit untouched tests."""
|
|
|
|
inventory = EditedTestInventory(lambda: None)
|
|
inventory.add_identified_failures(())
|
|
runner = _TargetedRunner()
|
|
original = "def helper():\n return 1\n\ndef test_unrelated():\n assert False\n"
|
|
compiler, _, _ = _compiler(
|
|
memory_root,
|
|
sources={},
|
|
tests={"test_widget.py": original},
|
|
edit_location="tests/test_widget.py",
|
|
replacement=original.replace("return 1", "return 2"),
|
|
context_name="BetaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
previous_errors=(TestRunCompileError("not admitted", test_name="test_unrelated"),),
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "tests/test_widget.py"]) == []
|
|
assert inventory.monitored == ()
|
|
assert runner.calls == []
|
|
|
|
|
|
def test_alpha_without_failure_remains_static_only(memory_root: Path) -> None:
|
|
"""Ordinary Alpha work neither executes tests nor consumes Beta's inventory."""
|
|
|
|
inventory = EditedTestInventory(lambda: None)
|
|
failure = TestRunCompileError("incoming failure", test_name="test_value")
|
|
inventory.add_identified_failures((failure,))
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root,
|
|
sources={"value.py": "value = 1\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
runner=runner,
|
|
blackboard=inventory,
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
assert runner.calls == []
|
|
assert inventory.monitored == ("test_value",)
|
|
assert inventory.failures == (failure,)
|
|
|
|
|
|
@pytest.mark.parametrize("failure_detail", (
|
|
"FAILED tests/test_widget.py::test_value - AssertionError",
|
|
"tests/test_widget.py:2: in test_value\n assert value() == 2",
|
|
'File "tests/test_widget.py", line 2, in test_value\nAssertionError',
|
|
))
|
|
def test_alpha_runs_only_the_trace_selected_test(memory_root: Path, failure_detail: str) -> None:
|
|
"""A source-only edit reruns one readonly test without admitting its neighbors."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_widget.py": "def test_value():\n assert False\n\ndef test_unrelated():\n assert False\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_widget.py",
|
|
readonly_locations=("tests/test_widget.py",),
|
|
failure_stack_trace=(failure_detail,),
|
|
runner=runner,
|
|
)
|
|
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_value",))]
|
|
|
|
|
|
@pytest.mark.parametrize("failure_detail", (
|
|
"FAILED tests/test_widget.py::TestSecond::test_value[param.with.dots]",
|
|
"tests/test_widget.py:7: in TestSecond.test_value",
|
|
"___ TestSecond.test_value ___\ntests/test_widget.py:7: AssertionError",
|
|
))
|
|
def test_alpha_resolves_the_qualified_test_method(memory_root: Path, failure_detail: str) -> None:
|
|
"""Same-named methods retain their class owner and use method-level execution."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_widget.py": (
|
|
"class TestFirst:\n def test_value(self):\n assert False\n\n"
|
|
"class TestSecond:\n def test_value(self):\n assert False\n"
|
|
)},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_widget.py",
|
|
readonly_locations=("tests/test_widget.py",),
|
|
failure_stack_trace=(failure_detail,),
|
|
runner=runner,
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
expected = "TestSecond.test_value[param.with.dots]" if "[param.with.dots]" in failure_detail else "TestSecond.test_value"
|
|
assert tuple(call[2] for call in runner.calls) == ((expected,),)
|
|
|
|
|
|
@pytest.mark.parametrize("failure_detail,expected", (
|
|
("test_value failed", None),
|
|
("FAILED tests/test_second.py::test_value", "tests/test_second.py"),
|
|
("tests/test_second.py:2: in test_value", "tests/test_second.py"),
|
|
("___ test_value ___\ntests/test_second.py:2: AssertionError", "tests/test_second.py"),
|
|
("test_value_extra failed", None),
|
|
("tests/test_second.py:2: AssertionError", "tests/test_second.py"),
|
|
("FAILED tests/test_unmanaged.py::test_value_extra", None),
|
|
))
|
|
def test_alpha_requires_one_supported_test_owner(
|
|
memory_root: Path, failure_detail: str, expected: str | None,
|
|
) -> None:
|
|
"""Evidence must disambiguate identical test names within supplied readonly scope."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={
|
|
"test_first.py": "def test_value():\n assert False\n",
|
|
"test_second.py": "def test_value():\n assert False\n",
|
|
"test_unmanaged.py": "def test_value_extra():\n assert False\n",
|
|
},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_second.py",
|
|
readonly_locations=("tests/test_first.py", "tests/test_second.py"),
|
|
failure_stack_trace=(failure_detail,),
|
|
runner=runner,
|
|
)
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
# Only evidence resolving one managed readonly test permits execution.
|
|
if expected is None:
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert "Cannot identify the Alpha remediation test" in errors[0].message
|
|
assert runner.calls == []
|
|
else:
|
|
assert errors == []
|
|
assert len(runner.calls) == 1
|
|
assert runner.calls[0][0][0].location == expected
|
|
|
|
|
|
@pytest.mark.parametrize("result", (
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 1, "fresh assertion failure", ["fresh traceback"]),
|
|
TestRunResult("pytest", TestRunOutcome.Success, 5, "no tests collected"),
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 2, "collection failed"),
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 124, "TIMEOUT"),
|
|
RuntimeError("runner unavailable"),
|
|
))
|
|
def test_alpha_requires_successful_test_execution(
|
|
memory_root: Path, result: TestRunResult | Exception,
|
|
) -> None:
|
|
"""An unsuccessful exact-test run blocks the edit and retains current evidence."""
|
|
|
|
runner = _TargetedRunner([result])
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value():\n assert False\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_raw_output="FAILED tests/test_value.py::test_value",
|
|
failure_stack_trace=("tests/test_value.py:2: in test_value",),
|
|
runner=runner,
|
|
)
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
assert len(errors) == 1
|
|
assert len(runner.calls) == 1
|
|
# Execution failures retain their compiler-owned failure family.
|
|
if isinstance(result, Exception):
|
|
assert isinstance(errors[0], RepoExecCompileError)
|
|
assert str(result) in errors[0].message
|
|
else:
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].raw_output == result.output
|
|
assert errors[0].stack_trace == tuple(result.stack_trace)
|
|
|
|
|
|
def test_alpha_retests_after_an_earlier_pass(memory_root: Path) -> None:
|
|
"""A later source candidate cannot reuse a passing result for unchanged test code."""
|
|
|
|
runner = _TargetedRunner([
|
|
TestRunResult("pytest", TestRunOutcome.Success, 0, "passed"),
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 1, "source regression"),
|
|
])
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value():\n assert False\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_raw_output="FAILED tests/test_value.py::test_value",
|
|
failure_stack_trace=("tests/test_value.py:2: in test_value",),
|
|
runner=runner,
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
compiler.blackboard.commit_candidate_results()
|
|
compiler.repo_exec.write_text(memory_root / "src/value.py", "value = 3\n")
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].raw_output == "source regression"
|
|
assert tuple(call[2] for call in runner.calls) == (("test_value",), ("test_value",))
|
|
|
|
|
|
@pytest.mark.parametrize("trace,expected", (
|
|
(
|
|
("tests/test_widget.py:900: in test_outer", "tests/test_widget.py:1: in test_inner"),
|
|
"test_outer",
|
|
),
|
|
(
|
|
("tests/test_widget.py:900: in test_inner", "tests/test_widget.py:1: in test_outer"),
|
|
"test_inner",
|
|
),
|
|
(
|
|
("tests/test_widget.py:1: in test_outer", "tests/test_widget.py:900: in test_inner"),
|
|
"test_outer",
|
|
),
|
|
))
|
|
def test_alpha_selects_the_first_named_test(
|
|
memory_root: Path, trace: tuple[str, ...], expected: str,
|
|
) -> None:
|
|
"""Traceback order selects the outer test independently of source order and line numbers."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_widget.py": (
|
|
"def test_inner():\n assert False\n\n"
|
|
"def test_outer():\n test_inner()\n"
|
|
)},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_widget.py",
|
|
readonly_locations=("tests/test_widget.py",),
|
|
failure_stack_trace=trace,
|
|
failure_error_message="test_inner is the deeper call",
|
|
failure_raw_output="FAILED tests/test_widget.py::test_inner",
|
|
runner=runner,
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
assert runner.calls == [([manager.test_classes[0]], 1, (expected,))]
|
|
|
|
|
|
def test_alpha_skips_unmatched_readonly_files(memory_root: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""An unrelated unreadable test and readonly source cannot interfere with the selected test."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n", "helper.py": "helper = 1\n"},
|
|
tests={
|
|
"test_widget.py": "def test_value():\n assert False\n",
|
|
"test_unrelated.py": "this is not valid Python!\n",
|
|
},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_widget.py",
|
|
readonly_locations=("src/helper.py", "tests/test_unrelated.py", "tests/test_widget.py"),
|
|
failure_stack_trace=("tests/test_widget.py:2: in test_value", "src/helper.py:1: in helper"),
|
|
runner=runner,
|
|
)
|
|
read_text = Mock(wraps=compiler.repo_exec.read_text)
|
|
monkeypatch.setattr(compiler.repo_exec, "read_text", read_text)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ("test_value",))]
|
|
read_paths = tuple(call.args[0] for call in read_text.call_args_list)
|
|
assert memory_root / "tests/test_unrelated.py" not in read_paths
|
|
assert memory_root / "src/helper.py" not in read_paths
|
|
|
|
|
|
@pytest.mark.parametrize("trace", (
|
|
("tests/test_widget.py:2: in test_value",),
|
|
("tests/test_widget.py:2: in test_value", "FAILED tests/test_widget.py::TestSecond::test_value"),
|
|
))
|
|
def test_alpha_rejects_ambiguous_method_ownership(memory_root: Path, trace: tuple[str, ...]) -> None:
|
|
"""A later class-qualified name cannot resolve an ambiguous earlier test frame."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_widget.py": (
|
|
"class TestFirst:\n def test_value(self):\n assert False\n\n"
|
|
"class TestSecond:\n def test_value(self):\n assert False\n"
|
|
)},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_widget.py",
|
|
readonly_locations=("tests/test_widget.py",),
|
|
failure_stack_trace=trace,
|
|
runner=runner,
|
|
)
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert "Ambiguous test" in errors[0].message
|
|
assert runner.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("first_passes", (True, False))
|
|
def test_alpha_runs_only_origin_with_supporting_test_files(
|
|
memory_root: Path, first_passes: bool,
|
|
) -> None:
|
|
"""Supporting test files never become additional Alpha execution assignments."""
|
|
|
|
first_result = TestRunResult("pytest", TestRunOutcome.Success if first_passes else TestRunOutcome.Failure, 0 if first_passes else 1, "first result")
|
|
runner = _TargetedRunner([first_result, TestRunResult("pytest", TestRunOutcome.Success, 0, "second passed")])
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={
|
|
"test_first.py": "def test_first():\n assert False\n",
|
|
"test_second.py": "def test_second():\n assert False\n",
|
|
},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_first.py",
|
|
readonly_locations=("tests/test_first.py", "tests/test_second.py"),
|
|
failure_stack_trace=("tests/test_first.py:2: in test_first", "tests/test_second.py:2: in test_second"),
|
|
runner=runner,
|
|
)
|
|
|
|
compiler.edit_request.readonly_files.reverse()
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert len(runner.calls) == 1
|
|
assert runner.calls[0] == ([manager.test_classes[0]], 1, ("test_first",))
|
|
# Only the originating test result controls acceptance.
|
|
if first_passes:
|
|
assert errors == []
|
|
else:
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].raw_output == "first result"
|
|
|
|
|
|
def test_alpha_requires_trace_identity(memory_root: Path) -> None:
|
|
"""Description-only evidence remains visible but cannot substitute for a test in the trace."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value():\n assert False\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_error_message="test_value failed",
|
|
failure_raw_output="FAILED tests/test_value.py::test_value",
|
|
runner=runner,
|
|
)
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert "test_value failed\nFAILED tests/test_value.py::test_value" in errors[0].message
|
|
assert runner.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("candidate", (False, True))
|
|
@pytest.mark.parametrize("refresh_metadata", (False, True))
|
|
@pytest.mark.parametrize("relative_root", (False, True))
|
|
def test_alpha_deleted_test_clears_its_ledger_failure(
|
|
memory_root: Path, candidate: bool, refresh_metadata: bool, relative_root: bool,
|
|
) -> None:
|
|
"""Confirmed file deletion clears its failure even after discovery forgets the file."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True, failure_test_name="test_value",
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value():\n assert False\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_raw_output="FAILED tests/test_value.py::test_value",
|
|
failure_stack_trace=("tests/test_value.py:2: in test_value",),
|
|
runner=runner,
|
|
candidate=candidate,
|
|
)
|
|
compiler.repo_exec.delete(memory_root / "tests/test_value.py")
|
|
# Readonly scope remains authoritative across supported root configurations.
|
|
if relative_root:
|
|
manager.test_root = Path("tests")
|
|
# Refreshed discovery must not erase evidence of the historical test's deletion.
|
|
if refresh_metadata:
|
|
manager.test_classes.clear()
|
|
manager.test_index.clear()
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert errors == []
|
|
assert runner.calls == []
|
|
compiler.blackboard.commit_candidate_results()
|
|
assert compiler.blackboard.failures == ()
|
|
assert compiler.blackboard.monitored == ()
|
|
|
|
|
|
@pytest.mark.parametrize("commit", (False, True))
|
|
@pytest.mark.parametrize("outcome", ("failure", "pass", "removal", "deletion"))
|
|
def test_alpha_ledger_tracks_accepted_results_across_compilers(
|
|
memory_root: Path, commit: bool, outcome: str,
|
|
) -> None:
|
|
"""New compiler instances retain accepted evidence without reseeding historical failures."""
|
|
|
|
runner = _TargetedRunner([TestRunResult("pytest", TestRunOutcome.Success if outcome != "failure" else TestRunOutcome.Failure, 1 if outcome == "failure" else 0, "fresh result")])
|
|
test_source = "def test_value():\n assert False\n"
|
|
compiler, memento, manager = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": test_source},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_stack_trace=("tests/test_value.py:2: in test_value",),
|
|
failure_raw_output="historical failure",
|
|
runner=runner,
|
|
candidate=False,
|
|
)
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
inventory = compiler.get_blackboard()
|
|
paths = [memory_root / "src/value.py"]
|
|
initial = compiler.compile(paths)
|
|
assert len(initial) == 1
|
|
assert inventory.failures == tuple(initial)
|
|
assert runner.calls == []
|
|
factory = PythonCompilerFactory(
|
|
file_manager=manager,
|
|
repo_exec=compiler.repo_exec,
|
|
bom_editor=compiler.bom_editor,
|
|
dependency_strategy=compiler.dependency_strategy,
|
|
repo_root=memory_root,
|
|
source_discovery=compiler.source_discovery,
|
|
test_runner=runner,
|
|
)
|
|
test_path = memory_root / "tests/test_value.py"
|
|
# Simulate candidate-time relevance changes independently from source execution.
|
|
if outcome == "removal":
|
|
compiler.repo_exec.write_text(test_path, "def test_replacement():\n assert True\n")
|
|
elif outcome == "deletion":
|
|
compiler.repo_exec.delete(test_path)
|
|
candidate = factory.build(replace(compiler.attempt, blackboard=inventory, candidate_mementos=(memento,)))
|
|
candidate_errors = candidate.compile(paths)
|
|
assert inventory.failures == tuple(initial)
|
|
# Candidate results remain provisional until the existing transaction owner decides.
|
|
if outcome == "failure":
|
|
assert len(candidate_errors) == 1
|
|
assert isinstance(candidate_errors[0], TestRunCompileError)
|
|
assert candidate_errors[0].raw_output == "fresh result"
|
|
else:
|
|
assert candidate_errors == []
|
|
# Commit and discard must follow the same source/result ownership as Beta.
|
|
if commit:
|
|
inventory.commit_candidate_results()
|
|
else:
|
|
inventory.discard_candidate_results()
|
|
compiler.repo_exec.write_text(test_path, test_source)
|
|
baseline = factory.build(replace(compiler.attempt, blackboard=inventory))
|
|
assert isinstance(baseline, PythonCompiler)
|
|
baseline.type_check = Mock(return_value=[TypeCheckError("new static error")])
|
|
errors = baseline.compile(paths)
|
|
assert errors[0] == TypeCheckError("new static error")
|
|
# Accepted fresh evidence survives static failures; rejected evidence never replaces it.
|
|
if not commit:
|
|
assert tuple(errors[1:]) == tuple(initial) == inventory.failures
|
|
elif outcome == "failure":
|
|
assert tuple(errors[1:]) == tuple(candidate_errors) == inventory.failures
|
|
else:
|
|
assert errors == [TypeCheckError("new static error")]
|
|
assert inventory.failures == ()
|
|
assert inventory.monitored == ()
|
|
assert len(runner.calls) == (1 if outcome in {"failure", "pass"} else 0)
|
|
receipts = inventory.receipts
|
|
assert len(receipts) == int(commit and outcome in {"failure", "pass"})
|
|
# Removal is absence of an execution, never a fabricated passing receipt.
|
|
if receipts:
|
|
assert receipts[0].test_name == "test_value"
|
|
assert receipts[0].test.location == "tests/test_value.py"
|
|
assert receipts[0].test_definition == test_source.rstrip("\n")
|
|
assert receipts[0].passed is (outcome == "pass")
|
|
assert receipts[0].result.output == "fresh result"
|
|
|
|
|
|
@pytest.mark.parametrize("candidate", (False, True))
|
|
@pytest.mark.parametrize("current_source", (
|
|
"",
|
|
"def test_new():\n assert False\n",
|
|
"def test_inner():\n assert False\n",
|
|
"def test_old_extra():\n assert False\n",
|
|
))
|
|
def test_alpha_removed_first_identity_clears_only_its_failure(
|
|
memory_root: Path, candidate: bool, current_source: str,
|
|
) -> None:
|
|
"""Absent historical identity clears without selecting a replacement or deeper test."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True, failure_test_name="test_old",
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": current_source},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_stack_trace=("tests/test_value.py:2: in test_old", "tests/test_value.py:5: in test_inner"),
|
|
runner=runner,
|
|
candidate=candidate,
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
compiler.blackboard.commit_candidate_results()
|
|
assert compiler.blackboard.failures == ()
|
|
assert compiler.blackboard.monitored == ()
|
|
assert runner.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("deleted_file", (False, True))
|
|
def test_alpha_removed_origin_never_substitutes_supporting_test(
|
|
memory_root: Path, deleted_file: bool,
|
|
) -> None:
|
|
"""Confirmed origin removal retains its policy without executing a same-named support test."""
|
|
|
|
runner = _TargetedRunner([TestRunResult("pytest", TestRunOutcome.Failure, 1, "live failure")])
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True, failure_test_name="test_old",
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={
|
|
"test_first.py": "def test_new():\n assert True\n",
|
|
"test_second.py": "def test_old():\n assert False\n",
|
|
},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_first.py",
|
|
readonly_locations=("tests/test_first.py", "tests/test_second.py"),
|
|
failure_stack_trace=("tests/test_first.py:2: in test_old", "tests/test_second.py:2: in test_old"),
|
|
runner=runner,
|
|
)
|
|
# Both callable and file removal are independent of another file's failure.
|
|
if deleted_file:
|
|
compiler.repo_exec.delete(memory_root / "tests/test_first.py")
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
compiler.blackboard.commit_candidate_results()
|
|
|
|
assert errors == []
|
|
assert compiler.blackboard.failures == ()
|
|
assert compiler.blackboard.monitored == ()
|
|
assert runner.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("unreadable", (False, True))
|
|
def test_alpha_failed_discovery_preserves_accepted_ledger_failure(
|
|
memory_root: Path, unreadable: bool,
|
|
) -> None:
|
|
"""Unreadable or invalid current definitions cannot clear an accepted failure."""
|
|
|
|
runner = _TargetedRunner([TestRunResult("pytest", TestRunOutcome.Failure, 1, "fresh accepted failure")])
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_old():\n assert False\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_stack_trace=("tests/test_value.py:2: in test_old",),
|
|
runner=runner,
|
|
)
|
|
accepted = compiler.compile([memory_root / "src/value.py"])
|
|
compiler.blackboard.commit_candidate_results()
|
|
compiler.repo_exec.write_text(memory_root / "tests/test_value.py", "def test_old(:\n")
|
|
# A repository read failure occurs after independent static validation.
|
|
if unreadable:
|
|
compiler.shallow_compile = Mock(return_value=[])
|
|
compiler.type_check = Mock(return_value=[])
|
|
compiler.validate_references = Mock(return_value=[])
|
|
compiler.repo_exec.read_text = Mock(side_effect=OSError("readonly test unreadable"))
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert "Cannot identify the Alpha remediation test" in errors[0].message
|
|
assert compiler.blackboard.failures == tuple(accepted)
|
|
assert compiler.blackboard.monitored == ("tests/test_value.py::test_old",)
|
|
assert len(runner.calls) == 1
|
|
|
|
|
|
def test_alpha_static_candidate_keeps_cleared_failure_out_of_ledger(memory_root: Path) -> None:
|
|
"""An accepted pass stays cleared when a later candidate cannot reach runtime checks."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value():\n assert True\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_stack_trace=("tests/test_value.py:2: in test_value",),
|
|
runner=runner,
|
|
)
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
compiler.blackboard.commit_candidate_results()
|
|
compiler.type_check = Mock(return_value=[TypeCheckError("new static error")])
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
compiler.blackboard.discard_candidate_results()
|
|
|
|
assert errors == [TypeCheckError("new static error")]
|
|
assert compiler.blackboard.failures == ()
|
|
assert len(runner.calls) == 1
|
|
|
|
|
|
def test_alpha_restored_test_records_a_new_failure(memory_root: Path) -> None:
|
|
"""Restoring a removed test creates fresh execution evidence without reviving stale evidence."""
|
|
|
|
runner = _TargetedRunner([TestRunResult("pytest", TestRunOutcome.Failure, 1, "restored failure")])
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True, failure_test_name="test_old",
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_new():\n assert True\n"},
|
|
edit_location="src/value.py",
|
|
replacement="value = 2\n",
|
|
context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_stack_trace=("tests/test_value.py:2: in test_old",),
|
|
runner=runner,
|
|
)
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
compiler.blackboard.commit_candidate_results()
|
|
assert compiler.blackboard.failures == ()
|
|
assert runner.calls == []
|
|
compiler.repo_exec.write_text(memory_root / "tests/test_value.py", "def test_old():\n assert False\n")
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
compiler.blackboard.commit_candidate_results()
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].raw_output == "restored failure"
|
|
assert compiler.blackboard.failures == tuple(errors)
|
|
assert compiler.blackboard.monitored == ("tests/test_value.py::test_old",)
|
|
assert tuple(call[2] for call in runner.calls) == (("test_old",),)
|
|
|
|
|
|
@pytest.mark.parametrize("focus", (None, "source", "outside_readonly"))
|
|
def test_alpha_invalid_origin_cannot_fall_back_to_readonly_support(memory_root: Path, focus: str | None) -> None:
|
|
"""Incomplete execution authority is a compiler error even when another test could pass."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True, sources={"value.py": "value = 1\n"},
|
|
tests={"test_origin.py": "def test_origin(): pass\n", "test_support.py": "def test_support(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 2\n", context_name="AlphaLoop",
|
|
readonly_locations=("tests/test_support.py",),
|
|
failure_stack_trace=("tests/test_origin.py:1: in test_origin", "tests/test_support.py:1: in test_support"),
|
|
runner=runner,
|
|
)
|
|
compiler.edit_request.remediation_originator_test = {
|
|
None: None, "source": manager.src_classes[0], "outside_readonly": manager.test_classes[0],
|
|
}[focus]
|
|
|
|
errors = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert "Cannot identify the Alpha remediation test" in errors[0].message
|
|
assert runner.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("with_originator", (False, True))
|
|
def test_ordinary_alpha_does_not_infer_remediation_from_evidence(
|
|
memory_root: Path, with_originator: bool,
|
|
) -> None:
|
|
"""Ordinary work never acquires a remediation test obligation from incidental evidence."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 2\n", context_name="AlphaLoop",
|
|
is_remediation=False,
|
|
remediation_originator_location="tests/test_value.py" if with_originator else None,
|
|
readonly_locations=("tests/test_value.py",),
|
|
failure_stack_trace=("tests/test_value.py:1: in test_value",), runner=runner,
|
|
)
|
|
|
|
assert compiler.compile([memory_root / "src/value.py"]) == []
|
|
assert runner.calls == []
|
|
assert compiler.blackboard.failures == ()
|
|
|
|
|
|
@pytest.mark.parametrize("with_originator", (False, True))
|
|
def test_alpha_remediation_cannot_become_ordinary_work_when_evidence_is_missing(
|
|
memory_root: Path, with_originator: bool,
|
|
) -> None:
|
|
"""Remediation status survives missing evidence and produces a retained diagnostic."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 2\n", context_name="AlphaLoop",
|
|
is_remediation=True,
|
|
remediation_originator_location="tests/test_value.py" if with_originator else None,
|
|
readonly_locations=("tests/test_value.py",), runner=runner,
|
|
)
|
|
|
|
[error] = compiler.compile([memory_root / "src/value.py"])
|
|
|
|
assert isinstance(error, TestRunCompileError)
|
|
assert "Cannot identify the Alpha remediation test" in error.message
|
|
assert compiler.edit_request.is_remediation
|
|
assert runner.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize("result", (
|
|
TestRunResult("pytest", TestRunOutcome.Success, 0, "passed"),
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 1, "fresh failure", ["frame"]),
|
|
TestRunResult("pytest", TestRunOutcome.Timeout, 124, "TIMEOUT", run_info="PROCESS DEADLINE SECONDS: 60"),
|
|
TestRunResult("pytest", TestRunOutcome.Error, 2, "ERROR collecting tests/test_value.py"),
|
|
))
|
|
def test_alpha_recheck_executes_without_a_candidate_or_static_admission(memory_root: Path, result: TestRunResult) -> None:
|
|
"""Remediation gates rerun the origin even when no source edit was proposed."""
|
|
|
|
runner = _TargetedRunner([result])
|
|
compiler, memento, manager = _compiler(
|
|
memory_root, is_remediation=True, candidate=False,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "def test_value():\n assert False\n", "test_support.py": "def test_support(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 1\n", context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py", "tests/test_support.py"),
|
|
failure_error_message="old error", failure_stack_trace=("tests/test_value.py:2: in test_value",), runner=runner,
|
|
)
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
assert RemediationCompiler in type(compiler).__mro__
|
|
compiler.shallow_compile = Mock(side_effect=AssertionError("test execution must not require static admission"))
|
|
compiler.type_check = Mock(side_effect=AssertionError("test execution must not run Pyright"))
|
|
|
|
assert compiler.recheck() is result
|
|
assert runner.calls == [([manager.current_test("tests/test_value.py")], 1, ("test_value",))]
|
|
assert compiler.blackboard.monitored == ()
|
|
assert memento.pre == "value = 1\n"
|
|
assert memento.diff == ""
|
|
|
|
|
|
@pytest.mark.parametrize("result", (
|
|
TestRunResult("pytest", TestRunOutcome.Success, 5, "no tests collected"),
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 3, "internal runner error"),
|
|
TestRunResult("pytest", TestRunOutcome.Failure, 2, "interrupted before test completion"),
|
|
RuntimeError("environment unavailable"),
|
|
))
|
|
def test_alpha_recheck_requires_a_completed_test_run(memory_root: Path, result: TestRunResult | Exception) -> None:
|
|
"""Empty collection and execution faults cannot become pass or stale evidence."""
|
|
|
|
runner = _TargetedRunner([result])
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True, candidate=False,
|
|
sources={"value.py": "value = 1\n"}, tests={"test_value.py": "def test_value(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 1\n", context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py", readonly_locations=("tests/test_value.py",),
|
|
failure_error_message="old error", failure_stack_trace=("tests/test_value.py:2: in test_value",), runner=runner,
|
|
)
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
assert isinstance(compiler.recheck(), CompileError)
|
|
assert len(runner.calls) == 1
|
|
|
|
|
|
def test_alpha_recheck_removed_origin_is_not_a_passing_execution(memory_root: Path) -> None:
|
|
"""The existing removal policy cannot fabricate a fresh successful test run."""
|
|
|
|
runner = _TargetedRunner()
|
|
compiler, _, _ = _compiler(
|
|
memory_root, is_remediation=True, candidate=False,
|
|
sources={"value.py": "value = 1\n"}, tests={"test_value.py": "def test_replacement(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 1\n", context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py", readonly_locations=("tests/test_value.py",),
|
|
failure_error_message="old error", failure_stack_trace=("tests/test_value.py:2: in test_value",), runner=runner,
|
|
)
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
assert isinstance(compiler.recheck(), CompileError)
|
|
assert runner.calls == []
|
|
|
|
|
|
def test_alpha_executes_and_receipts_the_exact_parameterized_origin(memory_root: Path) -> None:
|
|
"""AST ownership lookup cannot erase a surviving parameterized execution obligation."""
|
|
result = TestRunResult("pytest exact", TestRunOutcome.Success, 0, "passed")
|
|
runner = _TargetedRunner([result])
|
|
compiler, _, manager = _compiler(
|
|
memory_root, is_remediation=True, candidate=False,
|
|
sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "class TestValue:\n def test_value(self, value):\n assert value\n"},
|
|
edit_location="src/value.py", replacement="value = 1\n", context_name="AlphaLoop",
|
|
remediation_originator_location="tests/test_value.py", readonly_locations=("tests/test_value.py",),
|
|
failure_error_message="old error", failure_stack_trace=("tests/test_value.py:3: in test_value",), runner=runner,
|
|
)
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
compiler.edit_request.failure_test_location = "tests/test_value.py"
|
|
compiler.edit_request.failure_test_name = "TestValue.test_value[param.with.dots]"
|
|
|
|
observed = compiler.recheck()
|
|
|
|
assert observed is result
|
|
assert runner.calls == [([manager.current_test("tests/test_value.py")], 1, ("TestValue.test_value[param.with.dots]",))]
|
|
assert compiler.blackboard.receipts[0].test_name == "TestValue.test_value[param.with.dots]"
|
|
assert "def test_value(self, value):" in compiler.blackboard.receipts[0].test_definition
|