Skip to content
Merged
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: 5 additions & 1 deletion tests/control_plane/canonical_authority_fixture.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import os
from pathlib import Path
import subprocess
import tempfile


def initialize_canonical_authority(runtime_root: Path, goal_id: str, projection: dict, *, state_path: Path, provider: str = "file") -> dict:
Expand Down Expand Up @@ -45,9 +46,12 @@ def initialize_canonical_authority(runtime_root: Path, goal_id: str, projection:
def isolate_sqlite_runtime(tmp_path, monkeypatch):
monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite")
# Do not reuse an Effect runtime started by the minimum-Node CI step.
# Each CLI subprocess resolves its own tempfile root from this environment.
# CLI subprocesses resolve their root from the environment; the current
# process may already have cached a different root in tempfile.tempdir.
# Patch both for the same lifetime, restoring the cache at fixture teardown.
for variable in ("TMPDIR", "TEMP", "TMP"):
monkeypatch.setenv(variable, str(tmp_path))
monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))


def single_snapshot_page(result: dict, goal_id: str = "goal-a") -> dict:
Expand Down
39 changes: 39 additions & 0 deletions tests/control_plane/test_canonical_authority_fixture.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
"""Canonical CLI tests must not reuse a host's warmed Effect runtime."""
from __future__ import annotations

import subprocess
import sys
import tempfile

import pytest

from canonical_authority_fixture import isolate_sqlite_runtime
from loopx.control_plane.effect_runtime import _runtime_dir


@pytest.mark.parametrize("cached", [False, True])
def test_runtime_isolation_covers_process_and_cli_temp_directories(tmp_path, monkeypatch, cached):
shared = tmp_path / "shared"
shared.mkdir()
monkeypatch.setattr(tempfile, "tempdir", str(shared) if cached else None)

isolate_sqlite_runtime(tmp_path, monkeypatch)

assert tempfile.gettempdir() == str(tmp_path)
assert _runtime_dir().parent == tmp_path
child_root = subprocess.check_output(
[sys.executable, "-c", "import tempfile; print(tempfile.gettempdir())"], text=True,
).strip()
assert child_root == str(tmp_path)


def test_runtime_isolation_restores_the_process_cache(tmp_path, monkeypatch):
shared = tmp_path / "shared"
monkeypatch.setattr(tempfile, "tempdir", str(shared))

with monkeypatch.context() as isolated:
isolate_sqlite_runtime(tmp_path, isolated)
assert _runtime_dir().parent == tmp_path

assert tempfile.tempdir == str(shared)
assert _runtime_dir().parent == shared
Loading