Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions examples/control_plane/cli-output-probe-runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -459,11 +459,13 @@ def main() -> int:
)
args = parser.parse_args()
_install_pytest_import_stub()
probe = _load_module("loopx_cli_output_probe_fixture", args.test_source)
# Bind the frozen workload's helpers before it imports them. LoopX itself
# still comes from the product checkout selected for this subprocess.
fixture_support = _load_module(
"loopx_cli_output_fixture_support",
"tests.control_plane.cli_output_fixture",
Path(__file__).with_name("cli_output_fixture.py"),
)
probe = _load_module("loopx_cli_output_probe_fixture", args.test_source)
# Scenario helpers also call this owner; historical test modules must not
# re-enter their old /tmp implementation from inside the shared outer root.
probe._stable_budget_fixture_root = fixture_support.stable_budget_fixture_root
Expand Down
49 changes: 49 additions & 0 deletions tests/control_plane/test_cli_output_probe_runner.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
from __future__ import annotations

import json
import os
import runpy
import re
import shlex
import shutil
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace
Expand All @@ -21,6 +24,52 @@
)


def test_runner_imports_frozen_support_with_an_older_checkout(tmp_path: Path) -> None:
workload = tmp_path / "workload"
workload.mkdir()
runner = workload / RUNNER.name
shutil.copyfile(RUNNER, runner)
shutil.copyfile(cli_output_fixture.__file__, workload / "cli_output_fixture.py")

historical = tmp_path / "historical"
legacy_tests = historical / "tests" / "control_plane"
legacy_tests.mkdir(parents=True)
for package in (legacy_tests, legacy_tests.parent):
(package / "__init__.py").touch()
(legacy_tests / "cli_output_fixture.py").write_text(
"# Older checkout with no current observation helper.\n", encoding="utf-8"
)
probe_source = workload / "probe.py"
probe_source.write_text(
"from pathlib import Path\n"
"from tests.control_plane.cli_output_fixture import measure_cli_context_for_oracle\n"
"assert Path(measure_cli_context_for_oracle.__code__.co_filename) == "
"Path(__file__).with_name('cli_output_fixture.py')\n"
"print('frozen fixture support loaded')\n"
"raise SystemExit(0)\n",
encoding="utf-8",
)

result = subprocess.run(
[
sys.executable, str(runner),
"--test-source", str(probe_source),
"--semantics-source", str(workload / "unused_semantics.py"),
"--fixture-root", str(workload / "unused_fixture"),
"--receipt", str(workload / "unused_receipt.json"),
],
cwd=workload,
env={
**os.environ,
"PYTHONPATH": os.pathsep.join((str(historical), str(RUNNER.parents[2]))),
"LOOPX_USAGE_PING": "0",
},
capture_output=True, text=True, encoding="utf-8", timeout=30, check=False,
)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "frozen fixture support loaded"


@pytest.fixture
def crowded_turn_probe(monkeypatch: pytest.MonkeyPatch):
commands = probe._surface_commands
Expand Down
Loading