618 lines
22 KiB
Python
618 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Repository-executor boundaries with an explicit in-memory process harness.
|
|
|
|
The marked tests retain the concrete adapter command contracts for the whole
|
|
suite. Skinny tests exercise inherited container behavior through a nominal
|
|
subclass whose stationary command result never starts Docker or a host process.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from typing import Iterable, Sequence
|
|
|
|
import pytest
|
|
|
|
from band_aider.judgement.repo_exec import (
|
|
DockerContainerRepoExec,
|
|
HostRepoExec,
|
|
RepoExecError,
|
|
)
|
|
from band_aider.judgement.code_based_reasoning import CodeBasedReasoning
|
|
from tests.band_aider_loop_impl import _file_manager
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _CommandCall:
|
|
"""One command requested from the in-memory container harness."""
|
|
|
|
command: tuple[str, ...]
|
|
cwd: Path | None
|
|
env: dict[str, str] | None
|
|
input: bytes | str | None
|
|
|
|
|
|
class _RecordingDockerContainerRepoExec(DockerContainerRepoExec):
|
|
"""Concrete container adapter whose command boundary returns stationary results."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
returncode: int = 0,
|
|
stdout: bytes = b"",
|
|
stderr: bytes = b"",
|
|
) -> None:
|
|
super().__init__(container="ctr-123", repo_root=Path("/testbed"), python_bin="python")
|
|
self._returncode = returncode
|
|
self._stdout = stdout
|
|
self._stderr = stderr
|
|
self.calls: list[_CommandCall] = []
|
|
|
|
def run_cli(
|
|
self,
|
|
command: Sequence[str],
|
|
*,
|
|
cwd: Path | None = None,
|
|
env: dict[str, str] | None = None,
|
|
capture_output: bool = True,
|
|
text: bool = True,
|
|
check: bool = False,
|
|
timeout: int | None = None,
|
|
input: bytes | str | None = None,
|
|
unset_env: Iterable[str] = (),
|
|
) -> subprocess.CompletedProcess:
|
|
"""Record the requested command and return its fixed local result."""
|
|
|
|
del capture_output, check, timeout, unset_env
|
|
self.calls.append(_CommandCall(tuple(command), cwd, env, input))
|
|
stdout: str | bytes = self._stdout.decode() if text else self._stdout
|
|
stderr: str | bytes = self._stderr.decode() if text else self._stderr
|
|
return subprocess.CompletedProcess(list(command), self._returncode, stdout, stderr)
|
|
|
|
|
|
class _RecordingHostRepoExec(HostRepoExec):
|
|
"""Concrete host adapter type with a stationary, non-host command result."""
|
|
|
|
def __init__(self) -> None:
|
|
self._repo_root = Path("/testbed")
|
|
self.calls: list[_CommandCall] = []
|
|
|
|
def run_cli(
|
|
self,
|
|
command: Sequence[str],
|
|
*,
|
|
cwd: Path | None = None,
|
|
env: dict[str, str] | None = None,
|
|
capture_output: bool = True,
|
|
text: bool = True,
|
|
check: bool = False,
|
|
timeout: int | None = None,
|
|
input: bytes | str | None = None,
|
|
unset_env: Iterable[str] = (),
|
|
) -> subprocess.CompletedProcess:
|
|
"""Record the command instead of invoking the operating system."""
|
|
|
|
del capture_output, text, check, timeout, unset_env
|
|
self.calls.append(_CommandCall(tuple(command), cwd, env, input))
|
|
return subprocess.CompletedProcess(list(command), 0, "", "")
|
|
|
|
|
|
def test_code_reasons_read_current_code_through_container_executor() -> None:
|
|
"""Mechanical citations use ClassMeta's real container read without host file access."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
source = manager.require_src("src/foo.py")
|
|
executor = _RecordingDockerContainerRepoExec(stdout=b"def foo():\n return 42\n")
|
|
manager.repo_exec = executor
|
|
gate = CodeBasedReasoning(file_manager=manager, files=[source])
|
|
|
|
assert gate(["return 42"])
|
|
assert not gate(["return 1"])
|
|
assert len(executor.calls) == 4
|
|
assert all(map(lambda call: call.command[-1] == "/testbed/src/foo.py", executor.calls))
|
|
|
|
|
|
def test_code_reasons_reject_and_log_container_read_failure(caplog: pytest.LogCaptureFixture) -> None:
|
|
"""A failed container operation supplies correction evidence rather than acceptance."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
source = manager.require_src("src/foo.py")
|
|
manager.repo_exec = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"executor unavailable")
|
|
gate = CodeBasedReasoning(file_manager=manager, files=[source])
|
|
|
|
assert not gate(["return 1"])
|
|
assert "src/foo.py" in gate.validation.stringify()
|
|
assert "executor unavailable" in gate.validation.stringify()
|
|
assert "Cannot retrieve live code" in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize("returncode", (0, 1))
|
|
def test_both_reason_buckets_require_readable_container_source(returncode: int) -> None:
|
|
"""Container code supplies presence and absence evidence only after a successful read."""
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
source = manager.require_src("src/foo.py")
|
|
manager.repo_exec = _RecordingDockerContainerRepoExec(
|
|
returncode=returncode, stdout=b"def foo():\n return 42\n", stderr=b"unavailable",
|
|
)
|
|
gate = CodeBasedReasoning(
|
|
file_manager=manager, files=[source], evidence_lines=["return 1"],
|
|
)
|
|
|
|
assert gate(["return 42"], negative_reason=["return 1"]) is (returncode == 0)
|
|
# Failed execution cannot turn an unknown file into evidence that the old line disappeared.
|
|
if returncode:
|
|
assert "negative_reason" in gate.validation.stringify()
|
|
assert "Cannot retrieve live code" in gate.validation.stringify()
|
|
|
|
|
|
def test_docker_container_repo_exec_decorates_with_container_env() -> None:
|
|
"""Verify the concrete Docker adapter owns container-side environment decoration."""
|
|
|
|
executor = DockerContainerRepoExec(container="ctr-123", repo_root=Path("/testbed"), python_bin="python")
|
|
|
|
command = executor._decorate_with_env_and_input(
|
|
["env"],
|
|
cwd=Path("/testbed"),
|
|
env={"KEY": "value"},
|
|
needs_stdin=False,
|
|
)
|
|
|
|
assert command[:2] == ["docker", "exec"]
|
|
assert "-e" in command
|
|
assert "KEY=value" in command
|
|
assert command[-1] == "env"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_docker_container_repo_exec_whole_suite_uses_owned_stationary_docker_shim(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
) -> None:
|
|
"""Exercise the concrete subprocess boundary without starting a Docker daemon."""
|
|
|
|
shim_root = tmp_path_factory.mktemp("docker-repo-exec")
|
|
docker_shim = shim_root / "docker"
|
|
docker_shim.write_text(
|
|
f"#!{sys.executable}\n"
|
|
"import json\n"
|
|
"import sys\n"
|
|
"print(json.dumps({'arguments': sys.argv[1:], 'input': sys.stdin.read()}, sort_keys=True))\n",
|
|
encoding="utf-8",
|
|
)
|
|
docker_shim.chmod(0o755)
|
|
child_program = (
|
|
"import json\n"
|
|
"from pathlib import Path\n"
|
|
"from band_aider.judgement.repo_exec import DockerContainerRepoExec\n"
|
|
"executor = DockerContainerRepoExec(\n"
|
|
" container='ctr-stationary',\n"
|
|
" repo_root=Path('/container-repo'),\n"
|
|
" python_bin='python',\n"
|
|
")\n"
|
|
"result = executor.run_cli(\n"
|
|
" ('python', '-c', \"print('inside')\"),\n"
|
|
" cwd=Path('/container-repo'),\n"
|
|
" env={'FIXED': 'stationary'},\n"
|
|
" input='stationary input',\n"
|
|
" unset_env=('UNSET',),\n"
|
|
")\n"
|
|
"print(json.dumps({'returncode': result.returncode, 'stdout': result.stdout, 'stderr': result.stderr}, sort_keys=True))\n"
|
|
)
|
|
|
|
completed = subprocess.run(
|
|
(sys.executable, "-c", child_program),
|
|
env={"PATH": str(shim_root)},
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=4,
|
|
)
|
|
|
|
assert (completed.returncode, completed.stderr) == (0, "")
|
|
adapter_result = json.loads(completed.stdout)
|
|
assert adapter_result["returncode"] == 0
|
|
assert adapter_result["stderr"] == ""
|
|
assert json.loads(adapter_result["stdout"]) == {
|
|
"arguments": [
|
|
"exec",
|
|
"-i",
|
|
"--workdir",
|
|
"/container-repo",
|
|
"-e",
|
|
"FIXED=stationary",
|
|
"ctr-stationary",
|
|
"env",
|
|
"-u",
|
|
"UNSET",
|
|
"python",
|
|
"-c",
|
|
"print('inside')",
|
|
],
|
|
"input": "stationary input",
|
|
}
|
|
|
|
|
|
def test_docker_container_repo_exec_decorates_stdin_and_unset_environment_explicitly() -> None:
|
|
"""Container command metadata carries only stationary caller-provided environment data."""
|
|
|
|
executor = DockerContainerRepoExec(
|
|
container="ctr-123",
|
|
repo_root=Path("/testbed"),
|
|
python_bin="python",
|
|
)
|
|
|
|
command = executor._decorate_with_env_and_input(
|
|
("python", "-c", "print('stationary')"),
|
|
cwd=None,
|
|
env={"VISIBLE": "fixed"},
|
|
needs_stdin=True,
|
|
unset_env=("UNWANTED",),
|
|
)
|
|
|
|
assert command == [
|
|
"docker",
|
|
"exec",
|
|
"-i",
|
|
"--workdir",
|
|
"/testbed",
|
|
"-e",
|
|
"VISIBLE=fixed",
|
|
"ctr-123",
|
|
"env",
|
|
"-u",
|
|
"UNWANTED",
|
|
"python",
|
|
"-c",
|
|
"print('stationary')",
|
|
]
|
|
|
|
|
|
def test_container_write_forwards_bytes_through_stationary_stdin_harness() -> None:
|
|
"""Verify inherited write behavior supplies encoded content through its boundary."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec()
|
|
executor.write_text(Path("pkg/module.py"), "print('x')\n")
|
|
|
|
call = executor.calls[-1]
|
|
assert call.command[:2] == ("python", "-c")
|
|
assert call.command[-1] == "/testbed/pkg/module.py"
|
|
assert call.input == b"print('x')\n"
|
|
|
|
|
|
def test_container_python_argument_passing_has_no_stray_separator() -> None:
|
|
"""Verify inherited container Python calls pass the path as argv[1]."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec()
|
|
executor.exists(Path("pkg/a.py"))
|
|
|
|
command = executor.calls[-1].command
|
|
assert command[0:2] == ("python", "-c")
|
|
assert command[-1] == "/testbed/pkg/a.py"
|
|
assert "--" not in command
|
|
|
|
|
|
def test_container_python_bin_can_be_configured_without_environment_access() -> None:
|
|
"""Verify an explicit interpreter token remains local to the adapter configuration."""
|
|
|
|
executor = DockerContainerRepoExec(
|
|
container="ctr-python",
|
|
repo_root=Path("/testbed"),
|
|
python_bin="/opt/venv/bin/python",
|
|
)
|
|
|
|
assert executor.python_bin() == "/opt/venv/bin/python"
|
|
|
|
|
|
def test_container_predicates_raise_on_stationary_infrastructure_failure() -> None:
|
|
"""Verify an infrastructure result is distinct from an absent project path."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec(returncode=125, stderr=b"container is stopped")
|
|
|
|
with pytest.raises(RepoExecError, match="ctr-123.*exists.*125.*container is stopped"):
|
|
executor.exists(Path("/testbed/project.py"))
|
|
|
|
|
|
def test_container_is_file_raises_on_stationary_missing_interpreter() -> None:
|
|
"""Verify a command failure is never misreported as a false predicate."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec(returncode=127, stderr=b"python: not found")
|
|
|
|
with pytest.raises(RepoExecError, match="ctr-123.*is_file.*127.*python: not found"):
|
|
executor.is_file(Path("/testbed/project.py"))
|
|
|
|
|
|
def test_container_list_files_rejects_stationary_invalid_json() -> None:
|
|
"""Verify malformed command output remains an explicit container failure."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec(stdout=b"not-json", stderr=b"oops")
|
|
|
|
with pytest.raises(RuntimeError, match="JSON parse failed"):
|
|
executor.list_files(Path("/testbed"), patterns=["*.py"], recursive=True)
|
|
|
|
|
|
def test_container_list_files_returns_sorted_stationary_paths() -> None:
|
|
"""Verify inherited list parsing returns deterministic lexical container paths."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec(stdout=b'["/testbed/b.py", "/testbed/a.py"]')
|
|
|
|
assert executor.list_files(Path("/testbed"), patterns=["*.py"], recursive=True) == [
|
|
Path("/testbed/a.py"),
|
|
Path("/testbed/b.py"),
|
|
]
|
|
|
|
|
|
def test_container_read_operations_decode_stationary_memory_responses() -> None:
|
|
"""Read contracts decode bytes and preserve each requested container path."""
|
|
|
|
one = _RecordingDockerContainerRepoExec(stdout=b"module = 1\n")
|
|
many_payload = json.dumps(
|
|
{
|
|
"/testbed/a.py": base64.b64encode(b"a = 1\n").decode("ascii"),
|
|
"/testbed/b.py": base64.b64encode(b"b = 2\n").decode("ascii"),
|
|
},
|
|
).encode("utf-8")
|
|
many = _RecordingDockerContainerRepoExec(stdout=many_payload)
|
|
binary = _RecordingDockerContainerRepoExec(stdout=b"\x00\xff")
|
|
|
|
assert one.read_text(Path("module.py")) == "module = 1\n"
|
|
assert many.read_text_many((Path("a.py"), Path("b.py"))) == {
|
|
Path("/testbed/a.py"): "a = 1\n",
|
|
Path("/testbed/b.py"): "b = 2\n",
|
|
}
|
|
assert binary.read_bytes(Path("payload.bin")) == b"\x00\xff"
|
|
|
|
|
|
def test_container_predicates_distinguish_present_and_absent_stationary_paths() -> None:
|
|
"""Predicate absence remains a normal false result instead of an adapter fault."""
|
|
|
|
present = _RecordingDockerContainerRepoExec()
|
|
absent = _RecordingDockerContainerRepoExec(returncode=44)
|
|
|
|
assert present.exists(Path("present.py")) is True
|
|
assert present.is_file(Path("present.py")) is True
|
|
assert present.is_dir(Path("folder")) is True
|
|
assert absent.exists(Path("missing.py")) is False
|
|
assert absent.is_file(Path("missing.py")) is False
|
|
assert absent.is_dir(Path("missing")) is False
|
|
|
|
|
|
def test_container_mutations_accept_stationary_successful_boundary_results() -> None:
|
|
"""Every mutation uses the adapter command boundary without a Docker process."""
|
|
|
|
executor = _RecordingDockerContainerRepoExec(
|
|
stdout=json.dumps(["/testbed/cache", "/testbed/module.py"]).encode("utf-8"),
|
|
)
|
|
|
|
executor.write_bytes(Path("payload.bin"), b"payload")
|
|
executor.append_text(Path("notes.txt"), "note\n")
|
|
executor.mkdir(Path("cache"), parents=False, exist_ok=False)
|
|
assert executor.list_directory(Path(".")) == [
|
|
Path("/testbed/cache"),
|
|
Path("/testbed/module.py"),
|
|
]
|
|
executor.replace(Path("draft.py"), Path("module.py"))
|
|
executor.move(Path("module.py"), Path("archive/module.py"))
|
|
executor.move_many(((Path("one.py"), Path("archive/one.py")),))
|
|
executor.move_many(())
|
|
executor.delete(Path("absent.py"), missing_ok=True)
|
|
|
|
assert len(executor.calls) == 8
|
|
|
|
|
|
def test_container_missing_reads_raise_file_not_found_without_a_docker_process() -> None:
|
|
"""The adapter maps its stationary absent-path result to normal read absence."""
|
|
|
|
text = _RecordingDockerContainerRepoExec(returncode=44, stderr=b"missing")
|
|
binary = _RecordingDockerContainerRepoExec(returncode=44, stderr=b"missing")
|
|
|
|
with pytest.raises(FileNotFoundError, match="read_text"):
|
|
text.read_text(Path("missing.txt"))
|
|
with pytest.raises(FileNotFoundError, match="read_bytes"):
|
|
binary.read_bytes(Path("missing.bin"))
|
|
assert _RecordingDockerContainerRepoExec().read_text_many(()) == {}
|
|
|
|
|
|
def test_container_write_failures_remain_owned_adapter_failures() -> None:
|
|
"""Stationary nonzero writes cannot silently report a successful mutation."""
|
|
|
|
text = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"read-only")
|
|
binary = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"read-only")
|
|
append = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"read-only")
|
|
|
|
with pytest.raises(RepoExecError, match="write_text"):
|
|
text.write_text(Path("read-only.txt"), "fixed")
|
|
with pytest.raises(RepoExecError, match="write_bytes"):
|
|
binary.write_bytes(Path("read-only.bin"), b"fixed")
|
|
with pytest.raises(OSError, match="append"):
|
|
append.append_text(Path("read-only.txt"), "fixed")
|
|
|
|
|
|
def test_container_directory_failures_remain_explicit() -> None:
|
|
"""Stationary directory mutation and discovery failures remain observable errors."""
|
|
|
|
mkdir = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"denied")
|
|
directory = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"denied")
|
|
listing = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"denied")
|
|
|
|
with pytest.raises(OSError, match="mkdir"):
|
|
mkdir.mkdir(Path("blocked"))
|
|
with pytest.raises(OSError, match="list directory"):
|
|
directory.list_directory(Path("blocked"))
|
|
with pytest.raises(RuntimeError, match="list_files failed"):
|
|
listing.list_files(Path("blocked"), ("*.py",), recursive=False)
|
|
|
|
|
|
def test_container_relocation_failures_remain_explicit() -> None:
|
|
"""Stationary replace and move failures never fabricate a completed handoff."""
|
|
|
|
replace = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"denied")
|
|
move = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"denied")
|
|
many = _RecordingDockerContainerRepoExec(returncode=1, stderr=b"denied")
|
|
|
|
with pytest.raises(OSError, match="replace"):
|
|
replace.replace(Path("draft.py"), Path("final.py"))
|
|
with pytest.raises(RepoExecError, match="move"):
|
|
move.move(Path("draft.py"), Path("archive/final.py"))
|
|
with pytest.raises(RepoExecError, match="move_many"):
|
|
many.move_many(((Path("draft.py"), Path("archive/final.py")),))
|
|
|
|
|
|
def test_container_exec_path_mapping_and_rejection() -> None:
|
|
"""Verify concrete container paths stay lexical and reject traversal outside the repo."""
|
|
|
|
executor = DockerContainerRepoExec(container="ctr-123", repo_root=Path("/testbed"), python_bin="python")
|
|
|
|
assert executor.path_in_exec(Path("/testbed")) == "/testbed"
|
|
assert executor.path_in_exec(Path("pkg/file.py")) == "/testbed/pkg/file.py"
|
|
with pytest.raises(ValueError):
|
|
executor.path_in_exec(Path("/etc/passwd"))
|
|
with pytest.raises(ValueError):
|
|
executor.path_in_exec(Path("/testbed/../etc/passwd"))
|
|
|
|
|
|
def test_host_repo_exec_command_contract_has_a_stationary_integration_harness() -> None:
|
|
"""Keep the concrete HostRepoExec command contract isolated from the skinny suite."""
|
|
|
|
executor = _RecordingHostRepoExec()
|
|
completed = executor.run_cli(["python", "-c", "print('ok')"], cwd=Path("/testbed"), env={"X": "1"})
|
|
|
|
assert completed.returncode == 0
|
|
assert executor.calls == [
|
|
_CommandCall(("python", "-c", "print('ok')"), Path("/testbed"), {"X": "1"}, None)
|
|
]
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_host_repo_exec_whole_suite_round_trips_a_real_local_fixture(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
) -> None:
|
|
"""Verify the explicit whole-suite adapter tier owns host-file behavior."""
|
|
|
|
root = tmp_path_factory.mktemp("host-repo-exec")
|
|
executor = HostRepoExec(root)
|
|
source = root / "pkg" / "module.py"
|
|
binary = root / "pkg" / "payload.bin"
|
|
staged = root / "pkg" / "staged.py"
|
|
moved = root / "archive" / "module.py"
|
|
|
|
executor.write_text(source, "before\n")
|
|
executor.append_text(source, "after\n")
|
|
executor.write_bytes(binary, b"\x00\x01")
|
|
executor.write_text(staged, "staged\n")
|
|
|
|
assert executor.read_text(source) == "before\nafter\n"
|
|
assert executor.read_text_many((source, staged)) == {
|
|
source: "before\nafter\n",
|
|
staged: "staged\n",
|
|
}
|
|
assert executor.read_bytes(binary) == b"\x00\x01"
|
|
assert executor.exists(source)
|
|
assert executor.is_file(source)
|
|
assert executor.is_dir(root / "pkg")
|
|
assert executor.path_in_exec(source) == str(source.resolve())
|
|
assert executor.decorate_cli(("python", "-V")) == ["python", "-V"]
|
|
|
|
executor.replace(staged, root / "pkg" / "replaced.py")
|
|
executor.move(source, moved)
|
|
executor.move_many(((binary, root / "archive" / "payload.bin"),))
|
|
|
|
assert executor.list_directory(root / "archive") == [
|
|
root / "archive" / "module.py",
|
|
root / "archive" / "payload.bin",
|
|
]
|
|
assert executor.list_files(root, ("*.py",), recursive=True) == [
|
|
root / "archive" / "module.py",
|
|
root / "pkg" / "replaced.py",
|
|
]
|
|
|
|
executor.delete(root / "archive")
|
|
|
|
assert not executor.exists(root / "archive")
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_host_repo_exec_whole_suite_runs_a_stationary_local_command(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
) -> None:
|
|
"""Verify the labeled integration adapter can execute a fixed local command."""
|
|
|
|
root = tmp_path_factory.mktemp("host-command")
|
|
executor = HostRepoExec(root)
|
|
|
|
completed = executor.run_cli(
|
|
(sys.executable, "-c", "print('STATIONARY_HOST_RESULT')"),
|
|
cwd=root,
|
|
env={},
|
|
)
|
|
|
|
assert (completed.returncode, completed.stdout, completed.stderr) == (
|
|
0,
|
|
"STATIONARY_HOST_RESULT\n",
|
|
"",
|
|
)
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_host_repo_exec_whole_suite_passes_stationary_input_and_unsets_only_requested_keys(
|
|
tmp_path_factory: pytest.TempPathFactory,
|
|
) -> None:
|
|
"""The labeled host tier proves command input and explicit environment removal."""
|
|
|
|
root = tmp_path_factory.mktemp("host-command-input")
|
|
executor = HostRepoExec(root)
|
|
completed = executor.run_cli(
|
|
(
|
|
sys.executable,
|
|
"-c",
|
|
"import os, sys; print(os.getenv('VISIBLE')); print(os.getenv('HIDDEN')); print(sys.stdin.read())",
|
|
),
|
|
cwd=root,
|
|
env={"VISIBLE": "stationary", "HIDDEN": "remove"},
|
|
unset_env=("HIDDEN",),
|
|
input="stationary input",
|
|
)
|
|
|
|
assert completed.stdout == "stationary\nNone\nstationary input\n"
|
|
|
|
|
|
def test_memory_repo_exec_runs_stationary_commands_in_the_skinny_suite() -> None:
|
|
"""Verify the default executor returns a local success without creating a process."""
|
|
|
|
executor = MemoryRepoExec()
|
|
|
|
assert executor.run_cli(["python", "-c", "print('ok')"]).returncode == 0
|
|
|
|
|
|
def test_memory_repo_exec_persists_text_without_host_disk_access() -> None:
|
|
"""Verify the skinny executor stores and replaces lexical project files in memory."""
|
|
|
|
executor = MemoryRepoExec()
|
|
staged = executor.root / "pkg" / "module.tmp"
|
|
final = executor.root / "pkg" / "module.py"
|
|
executor.write_text(staged, "before")
|
|
executor.replace(staged, final)
|
|
|
|
assert executor.read_text(final) == "before"
|
|
|
|
|
|
def test_memory_repo_exec_lists_lexical_files_deterministically() -> None:
|
|
"""Verify skinny discovery observes only its explicit in-memory file inventory."""
|
|
|
|
executor = MemoryRepoExec()
|
|
executor.write_text(executor.root / "pkg" / "b.py", "")
|
|
executor.write_text(executor.root / "pkg" / "a.py", "")
|
|
|
|
assert executor.list_files(executor.root, ["*.py"], recursive=True) == [
|
|
executor.root / "pkg" / "a.py",
|
|
executor.root / "pkg" / "b.py",
|
|
]
|