from __future__ import annotations import json import re import sqlite3 from pathlib import Path from typing import Any import pytest from opensquilla.recovery import inspect_profile, reconcile_profile FIXTURE_ROOT = Path(__file__).with_name("fixtures") DESKTOP_MANIFEST = FIXTURE_ROOT / "desktop" / "released-profiles.json" DESKTOP_SNAPSHOTS = FIXTURE_ROOT / "desktop" / "frozen-profile-snapshots.json" def _load_manifest(path: Path) -> dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) assert payload["schema_version"] == 1 return payload DESKTOP_CASES = _load_manifest(DESKTOP_MANIFEST)["cases"] DESKTOP_SNAPSHOT_CASES = { snapshot["id"]: snapshot for snapshot in _load_manifest(DESKTOP_SNAPSHOTS)["snapshots"] } def _toml_string(value: Path) -> str: # JSON basic strings use the same escaping required by TOML basic strings. return json.dumps(str(value), ensure_ascii=False) def _write_config(home: Path, template_name: str) -> None: template = (FIXTURE_ROOT / "templates" / template_name).read_text(encoding="utf-8") rendered = template.replace("{{STATE_DIR_TOML}}", _toml_string(home / "state")) rendered = rendered.replace( "{{LEGACY_WORKSPACE_TOML}}", _toml_string(home / "state" / "workspace") ) home.mkdir(parents=True, exist_ok=True) (home / "config.toml").write_text(rendered, encoding="utf-8") def _write_workspace(path: Path, *, identity: str) -> None: path.mkdir(parents=True, exist_ok=False) for name in ("USER.md", "SOUL.md", "IDENTITY.md", "MEMORY.md"): (path / name).write_text(f"synthetic {identity} {name}\n", encoding="utf-8") def _write_session_database(home: Path, *, release_tag: str) -> None: state = home / "state" state.mkdir(parents=True, exist_ok=True) with sqlite3.connect(state / "sessions.db") as connection: connection.execute( "CREATE TABLE synthetic_sessions (session_id TEXT PRIMARY KEY, transcript TEXT)" ) connection.execute( "INSERT INTO synthetic_sessions VALUES (?, ?)", (f"session-{release_tag}", "synthetic transcript; contains no user data"), ) def _write_released_config(home: Path, template_name: str) -> None: template_path = DESKTOP_SNAPSHOTS.parent / template_name template = template_path.read_text(encoding="utf-8") rendered = template.replace("{{STATE_DIR_TOML}}", _toml_string(home / "state")) rendered = rendered.replace( "{{LEGACY_WORKSPACE_TOML}}", _toml_string(home / "state" / "workspace") ) assert "{{" not in rendered, f"unresolved fixture token in {template_path}" home.mkdir(parents=True, exist_ok=True) (home / "config.toml").write_text(rendered, encoding="utf-8") def _materialize_snapshot_entry( home: Path, case: dict[str, Any], entry: dict[str, Any] ) -> None: relative = Path(entry["path"]) assert not relative.is_absolute() and ".." not in relative.parts target = home / relative kind = entry["kind"] if kind == "config": assert relative == Path("config.toml") _write_released_config(home, entry["template"]) elif kind == "sqlite_sessions": assert relative == Path("state/sessions.db") _write_session_database(home, release_tag=case["release_tag"]) elif kind == "identity_markdown": target.parent.mkdir(parents=True, exist_ok=True) identity = entry.get("identity", "release-layout") target.write_text( f"synthetic {case['id']} {identity} {target.name}\n", encoding="utf-8" ) elif kind == "text": target.parent.mkdir(parents=True, exist_ok=True) target.write_text(entry["content"], encoding="utf-8") else: # pragma: no cover - malformed checked-in fixture raise AssertionError(f"unknown frozen snapshot entry kind: {kind}") def _build_desktop_fixture(tmp_path: Path, case: dict[str, Any]) -> Path: home = tmp_path / "ApplicationData" / "OpenSquilla" / "opensquilla" snapshot = DESKTOP_SNAPSHOT_CASES[case["id"]] assert snapshot["release_tag"] == case["release_tag"] seen: set[str] = set() for entry in snapshot["tree"]: assert entry["path"] not in seen, f"duplicate frozen path: {entry['path']}" seen.add(entry["path"]) _materialize_snapshot_entry(home, case, entry) return home def _tree_snapshot(root: Path) -> dict[str, tuple[str, bytes | None]]: snapshot: dict[str, tuple[str, bytes | None]] = {} for path in sorted(root.rglob("*")): relative = path.relative_to(root).as_posix() if path.is_dir(): snapshot[relative] = ("directory", None) elif path.is_file(): snapshot[relative] = ("file", path.read_bytes()) else: snapshot[relative] = ("special", None) return snapshot def _read_session(home: Path) -> tuple[str, str]: with sqlite3.connect(f"file:{home / 'state' / 'sessions.db'}?mode=ro", uri=True) as connection: row = connection.execute("SELECT session_id, transcript FROM synthetic_sessions").fetchone() assert row is not None return str(row[0]), str(row[1]) @pytest.mark.parametrize("case", DESKTOP_CASES, ids=lambda case: case["id"]) def test_released_desktop_profile_inspect_and_reconcile( tmp_path: Path, case: dict[str, Any] ) -> None: home = _build_desktop_fixture(tmp_path, case) database = home / "state" / "sessions.db" database_identity = (database.stat().st_dev, database.stat().st_ino) session_before = _read_session(home) tree_before_inspect = _tree_snapshot(home) inspected = inspect_profile(home, profile_kind="desktop-primary") assert (inspected.outcome, inspected.stable_code) == tuple(case["expected_inspect"]) assert inspected.effective_workspace == home / case["expected_effective_workspace"] assert _tree_snapshot(home) == tree_before_inspect, "inspect must be completely read-only" canonical_before = _tree_snapshot(home / "workspace") if (home / "workspace").is_dir() else None legacy_before = ( _tree_snapshot(home / "state" / "workspace") if (home / "state" / "workspace").is_dir() else None ) reconciled = reconcile_profile(home, profile_kind="desktop-primary") assert (reconciled.outcome, reconciled.stable_code) == tuple(case["expected_reconcile"]) assert reconciled.effective_workspace == home / case["expected_effective_workspace"] assert database.is_file() assert (database.stat().st_dev, database.stat().st_ino) == database_identity assert _read_session(home) == session_before assert not (home / "state" / "state" / "workspace").exists() layout = case["layout_template"] if layout == "pre-rc3-nested" and case["config_template"] == "desktop-unpinned.toml": assert (home / "workspace" / "SOUL.md").is_file() assert not (home / "state" / "workspace").exists() if case["include_nested_roles"]: assert (home / "skills" / "synthetic-skill" / "SKILL.md").is_file() assert (home / "session-archive" / "archive.json").is_file() assert (home / "router" / "calibration.json").is_file() assert (home / ".env").is_file() assert (home / "state" / "approvals.json").is_file() elif layout == "pre-rc3-nested": assert not (home / "workspace").exists() assert _tree_snapshot(home / "state" / "workspace") == legacy_before elif layout == "rc3-relocated-clean": assert _tree_snapshot(home / "workspace") == canonical_before assert not (home / "state" / "workspace").exists() else: # The RC3 stale pin conflict is intentionally preserved byte-for-byte. assert _tree_snapshot(home / "workspace") == canonical_before assert _tree_snapshot(home / "state" / "workspace") == legacy_before assert (home / "media" / "synthetic.txt").is_file() assert not (home / "state" / "state" / "sessions.db").exists() def test_released_desktop_manifest_freezes_verified_path_contract() -> None: manifest = _load_manifest(DESKTOP_MANIFEST) cases = manifest["cases"] tags = {case["release_tag"] for case in cases} assert tags == {"v0.4.0", "v0.4.1", "v0.5.0rc1", "v0.5.0rc2", "v0.5.0rc3"} assert all( case["gateway_env_home"] == "H/state" for case in cases if case["release_tag"] != "v0.5.0rc3" ) assert all( case["gateway_env_home"] == "H" for case in cases if case["release_tag"] == "v0.5.0rc3" ) assert manifest["provenance"]["rc3_relocation_allowlist"] == [ "skills", "skills-taps.json", "skills-lock.json", "workspace", "session-archive", "router", ".env", "state/*", ] def test_released_desktop_cases_use_tag_proven_frozen_tree_snapshots() -> None: """Every upgrade case must be materialized from an audited release snapshot.""" manifest = _load_manifest(DESKTOP_MANIFEST) assert DESKTOP_SNAPSHOTS.is_file() snapshots = _load_manifest(DESKTOP_SNAPSHOTS) by_id = {snapshot["id"]: snapshot for snapshot in snapshots["snapshots"]} assert set(by_id) == {case["id"] for case in manifest["cases"]} for case in manifest["cases"]: snapshot = by_id[case["id"]] source = snapshot["source"] assert snapshot["release_tag"] == case["release_tag"] assert re.fullmatch(r"[0-9a-f]{40}", source["release_commit"]) assert re.fullmatch(r"[0-9a-f]{40}", source["desktop_main_blob"]) assert re.fullmatch(r"[0-9a-f]{40}", source["python_paths_blob"]) assert source["desktop_main_path"] == "desktop/electron/src/main.ts" assert source["python_paths_path"] == "src/opensquilla/paths.py" assert source["gateway_env_home"] == case["gateway_env_home"] entries = {entry["path"]: entry for entry in snapshot["tree"]} assert entries["config.toml"]["kind"] == "config" assert entries["state/sessions.db"]["kind"] == "sqlite_sessions" assert entries["media/synthetic.txt"]["kind"] == "text" assert any( path.endswith("/USER.md") and entry["kind"] == "identity_markdown" for path, entry in entries.items() )