# SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc """Shared isolation harness for API baseline tests and baseline regeneration. Used by: - tests/test_api_baselines.py - module-scoped fixtures - tests/verify_api.py - verify/update CLI mode Keeps both paths on identical isolation so generated baselines match the test oracle. """ from __future__ import annotations import atexit import json import os import shutil import subprocess import tempfile import threading from contextlib import contextmanager from pathlib import Path from typing import Iterator from unittest.mock import patch FROZEN_DATE = "2026-04-15" FROZEN_TZ_OFFSET = -7 # The fixture journal is built once per process from an immutable snapshot of # HEAD rather than copied file-by-file from the live working tree. make dev / # make sandbox write runtime artifacts into tests/fixtures/journal (AGENTS.md # §6), so a concurrent writer can delete or replace a tracked file between # `git ls-files` and the copy — raising FileNotFoundError at fixture *setup* # (a spurious pytest ERROR, not a real failure). `git archive HEAD` reads the # committed tree from the object store, never the working tree, so it is immune # to concurrent working-tree mutation. _REPO_ROOT = Path(__file__).resolve().parent.parent _FIXTURE_JOURNAL_REL = "tests/fixtures/journal" _LIVE_FIXTURE_JOURNAL = (_REPO_ROOT / _FIXTURE_JOURNAL_REL).resolve() _snapshot_lock = threading.Lock() _snapshot_root: Path | None = None def _fixture_journal_snapshot() -> Path: """Return an immutable, process-scoped snapshot of the tracked fixture journal. Built once per process via `git archive HEAD` → tar, which reads HEAD from the git object store and is therefore unaffected by concurrent writes to the live tests/fixtures/journal tree. The same set of git-tracked files that `git ls-files` selects (working tree clean ⇒ HEAD ≡ index), with symlinks preserved as symlinks. The temp dir is removed at interpreter exit. """ global _snapshot_root with _snapshot_lock: if _snapshot_root is not None and _snapshot_root.exists(): return _snapshot_root tmp = Path(tempfile.mkdtemp(prefix="solstone-journal-snapshot-")) archive = subprocess.run( ["git", "archive", "HEAD", "--", _FIXTURE_JOURNAL_REL], cwd=str(_REPO_ROOT), capture_output=True, check=True, ) subprocess.run( ["tar", "-x", "-C", str(tmp)], input=archive.stdout, check=True, ) # `git archive` preserves the tests/fixtures/journal/ prefix in the tar. _snapshot_root = (tmp / _FIXTURE_JOURNAL_REL).resolve() atexit.register(shutil.rmtree, tmp, ignore_errors=True) return _snapshot_root def copytree_tracked(src: Path, dst: Path) -> None: """Copy only git-tracked files from src to dst. For paths within tests/fixtures/journal (the journal fixture and any subtree of it), copy from an immutable, process-scoped snapshot of HEAD instead of enumerating and copying from the live working tree — see `_fixture_journal_snapshot` for why this defeats the concurrent-write race. For any other src, fall back to enumerating tracked files from the live tree. """ src = Path(src).resolve() dst = Path(dst) try: rel = src.relative_to(_LIVE_FIXTURE_JOURNAL) except ValueError: rel = None if rel is not None: snap_src = _fixture_journal_snapshot() / rel shutil.copytree(snap_src, dst, symlinks=True, dirs_exist_ok=True) return result = subprocess.run( ["git", "ls-files", "."], cwd=str(src), capture_output=True, text=True, check=True, ) for rel_path in result.stdout.splitlines(): if not rel_path: continue src_file = src / rel_path dst_file = dst / rel_path dst_file.parent.mkdir(parents=True, exist_ok=True) if src_file.is_symlink(): os.symlink(os.readlink(src_file), dst_file) else: shutil.copy2(src_file, dst_file) def prepare_isolated_journal(dst: Path) -> Path: """Copy the git-tracked fixture journal into dst and return the absolute path.""" src = Path("tests/fixtures/journal").resolve() dst = dst.resolve() copytree_tracked(src, dst) return dst @contextmanager def isolated_app_env(journal: Path) -> Iterator[Path]: """Patch env so create_app(journal) is fully isolated.""" journal = Path(journal).resolve() overrides = { "SOLSTONE_JOURNAL": str(journal), # Match tests/conftest.py's deterministic unit-test runtime. Without # these, baseline regeneration probes host GPU/supervisor state while # test_api_baselines compares against the isolated test contract. "SOL_SKIP_SUPERVISOR_CHECK": "1", "SOLSTONE_DISABLE_CONVEY_SIDE_RUNTIMES": "1", } previous = {key: os.environ.get(key) for key in overrides} os.environ.update(overrides) try: from solstone.think.providers import local_cuda, local_install, local_vulkan deterministic_devices = [ local_vulkan.VulkanDevice( 1, "NVIDIA GeForce GTX 1660 Ti", local_vulkan.VK_TYPE_DISCRETE, 6390, ) ] with ( patch.object( local_cuda, "probe_nvidia_gpu", lambda: local_cuda.NvidiaProbe( index=None, compute_cap=None, driver_cuda_version=None, vram_mib=None, tiering_memory_mib=None, memory_source=local_cuda.MEMORY_SOURCE_UNAVAILABLE, detected=False, ), ), patch.object( local_install, "probe_cuda_runtime_artifact_trust", lambda _pin, **_kwargs: local_cuda.ArtifactTrust.ABSENT, ), patch.object( local_install, "has_persisted_installed_cuda_target", lambda **_kwargs: False, ), patch.object( local_vulkan, "detect_gpus", lambda: deterministic_devices, ), ): yield journal finally: for key, value in previous.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value def make_test_client(journal: Path): """Create a Flask test client for an isolated journal.""" from solstone.convey import create_app app = create_app(journal=str(Path(journal).resolve())) app.config["TESTING"] = True client = app.test_client() return client def mark_setup_complete(journal: Path, completed_at: int = 1700000000000) -> None: """Mark a minimal test journal as past first-run setup.""" config_path = Path(journal) / "config" / "journal.json" config_path.parent.mkdir(parents=True, exist_ok=True) if config_path.exists(): config = json.loads(config_path.read_text(encoding="utf-8")) else: config = {} config["setup"] = {"completed_at": completed_at} config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")