Skip to content

test(cli): isolate machine transport and local telemetry fixtures #8195

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangruiteng merged 1 commit into from
Nov 7, 2026
Merged

test(cli): isolate machine transport and local telemetry fixtures #8195

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
12 changes: 10 additions & 2 deletions tests/control_plane/test_source_cli_entrypoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,10 @@ def run_entry(
return subprocess.run(
[sys.executable, "-c", script],
cwd=REPO_ROOT,
env={**os.environ, "PYTHONPATH": str(REPO_ROOT), "LOOPX_USAGE_PING": "0", **(env or {})},
# This caller consumes machine JSON; keep the parent's model-facing
# context transport intact, with explicit native overrides still allowed.
env={**os.environ, "PYTHONPATH": str(REPO_ROOT), "LOOPX_USAGE_PING": "0",
"LOOPX_CONTEXT_DELIVERY": "off", **(env or {})},
text=True,
capture_output=True,
check=False,
Expand Down Expand Up @@ -192,8 +195,9 @@ def replace_process(*args):


@pytest.mark.parametrize("entry", ["module", "console"])
@pytest.mark.parametrize("parent_context", ["on", "off"])
def test_outer_controller_scope_preserves_reads_other_goals_and_recovery(
tmp_path: Path, entry: str,
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, entry: str, parent_context: str,
) -> None:
from examples.control_plane.quota_plan_fixtures import (
SCOPED_AGENT_ID, write_cli_fixture, write_codex_heartbeat_fixture,
Expand All @@ -203,6 +207,7 @@ def test_outer_controller_scope_preserves_reads_other_goals_and_recovery(
load_scheduler_state,
)

monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", parent_context)
registry, runtime, _project = write_cli_fixture(tmp_path / "fixture", scoped_agents=True)
prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"]
env = {"CODEX_HOME": str(tmp_path / "codex-home"), "CODEX_THREAD_ID": "",
Expand Down Expand Up @@ -280,6 +285,7 @@ def selected_ack(goal: str) -> list[str]:
assert json.loads(recovered.stdout)["scheduler_state_mutated"] is True
assert load_scheduler_state(runtime, goal_id="needs-operator", agent_id=SCOPED_AGENT_ID,
state_key=state_key) is not None
assert os.environ["LOOPX_CONTEXT_DELIVERY"] == parent_context


def test_source_module_observes_usage_once_and_preserves_failure_result() -> None:
Expand Down Expand Up @@ -323,6 +329,8 @@ def test_source_first_usage_disclosure_keeps_json_pure_and_does_not_send(tmp_pat
from loopx import usage_ping
for key in ("CI", "DO_NOT_TRACK", "LOOPX_USAGE_PING", "LOOPX_USAGE_POLICY"):
os.environ.pop(key, None)
# Disposable telemetry state requires explicit qualification to numeric loopback.
os.environ["LOOPX_USAGE_PING"] = "1"
os.environ["LOOPX_USAGE_PING_ENDPOINT"] = "http://127.0.0.1:1/v1/ping"
usage_ping.select_default_runtime_root = lambda: Path({str(state)!r})
sent = []
Expand Down
Loading