"""Tests for muse/core/object_availability.py — ObjectState model. Every object in a Muse repo is in one of three states: PRESENT — bytes exist in the local .muse/objects/ store PROMISED — not local but a promisor remote is configured; can be fetched MISSING — not local AND no promisor remote; genuine data loss risk Coverage: Unit: object_state() for all three states load_promisor_remotes() reads config correctly Edge cases: no remotes configured, promisor=false explicit opt-out, mixed promisor and non-promisor remotes Invariants: PRESENT when file exists regardless of remotes MISSING only when no promisor remote AND file absent """ from __future__ import annotations from collections.abc import Mapping import json import pathlib import pytest from muse.core.types import blob_id from muse.core.object_availability import ObjectState, load_promisor_remotes, object_state from muse.core.object_store import write_object from muse.core.paths import config_toml_path, muse_dir # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(path: pathlib.Path, remotes: Mapping[str, object] | None = None) -> pathlib.Path: dot_muse = muse_dir(path) for d in ("commits", "snapshots", "objects", "refs/heads"): (dot_muse / d).mkdir(parents=True, exist_ok=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main") (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "avail-test", "domain": "code"})) if remotes: _write_remotes(path, remotes) return path def _write_remotes(repo: pathlib.Path, remotes: Mapping[str, object]) -> None: """Write a minimal config.toml with the given remotes. remotes format: {"name": {"url": "...", "promisor": True/False}} """ lines = [] for name, cfg in remotes.items(): lines.append(f'[remotes.{name}]') lines.append(f'url = "{cfg["url"]}"') if "promisor" in cfg: val = "true" if cfg["promisor"] else "false" lines.append(f'promisor = {val}') (config_toml_path(repo)).write_text("\n".join(lines) + "\n") _OBJ_CONTENT = b"test object content for availability" _OBJ_ID = blob_id(_OBJ_CONTENT) # --------------------------------------------------------------------------- # ObjectState enum # --------------------------------------------------------------------------- class TestObjectStateEnum: def test_has_present(self) -> None: assert ObjectState.PRESENT == "present" def test_has_promised(self) -> None: assert ObjectState.PROMISED == "promised" def test_has_missing(self) -> None: assert ObjectState.MISSING == "missing" def test_is_string_enum(self) -> None: assert isinstance(ObjectState.PRESENT, str) # --------------------------------------------------------------------------- # object_state() — PRESENT # --------------------------------------------------------------------------- class TestObjectStatePresent: def test_present_when_file_exists_no_remotes(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) content = b"content-no-remotes" oid = blob_id(content) write_object(repo, oid, content) assert object_state(repo, oid, []) == ObjectState.PRESENT def test_present_when_file_exists_with_promisor(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) content = b"content-with-promisor" oid = blob_id(content) write_object(repo, oid, content) # Even with promisors, PRESENT wins — file is local assert object_state(repo, oid, ["local"]) == ObjectState.PRESENT def test_present_takes_priority_over_promisor(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) content = b"present content bytes" obj_id = blob_id(content) write_object(repo, obj_id, content) state = object_state(repo, obj_id, ["remote-a", "remote-b"]) assert state == ObjectState.PRESENT # --------------------------------------------------------------------------- # object_state() — PROMISED # --------------------------------------------------------------------------- class TestObjectStatePromised: def test_promised_when_absent_with_one_promisor(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) state = object_state(repo, _OBJ_ID, ["local"]) assert state == ObjectState.PROMISED def test_promised_when_absent_with_multiple_promisors(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) state = object_state(repo, _OBJ_ID, ["origin", "backup"]) assert state == ObjectState.PROMISED def test_promised_not_present(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) state = object_state(repo, _OBJ_ID, ["local"]) assert state != ObjectState.PRESENT # --------------------------------------------------------------------------- # object_state() — MISSING # --------------------------------------------------------------------------- class TestObjectStateMissing: def test_missing_when_absent_no_remotes(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) state = object_state(repo, _OBJ_ID, []) assert state == ObjectState.MISSING def test_missing_not_promised(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) state = object_state(repo, _OBJ_ID, []) assert state != ObjectState.PROMISED # --------------------------------------------------------------------------- # load_promisor_remotes() — reads config # --------------------------------------------------------------------------- class TestLoadPromisorRemotes: def test_empty_when_no_config(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path) result = load_promisor_remotes(repo) assert result == [] def test_all_remotes_are_promisors_by_default(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path, remotes={ "local": {"url": "https://localhost:1337/gabriel/muse"}, "staging": {"url": "https://staging.musehub.ai/gabriel/muse"}, }) result = load_promisor_remotes(repo) assert "local" in result assert "staging" in result def test_explicit_promisor_true_included(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path, remotes={ "origin": {"url": "https://localhost:1337/gabriel/muse", "promisor": True}, }) assert "origin" in load_promisor_remotes(repo) def test_explicit_promisor_false_excluded(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path, remotes={ "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, }) assert "mirror" not in load_promisor_remotes(repo) def test_mixed_promisors(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path, remotes={ "local": {"url": "https://localhost:1337/g/muse"}, # default → promisor "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, }) result = load_promisor_remotes(repo) assert "local" in result assert "mirror" not in result def test_returns_list_of_strings(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path, remotes={ "local": {"url": "https://localhost:1337/g/muse"}, }) result = load_promisor_remotes(repo) assert isinstance(result, list) assert all(isinstance(r, str) for r in result) def test_no_promisor_when_all_opted_out(self, tmp_path: pathlib.Path) -> None: repo = _init_repo(tmp_path, remotes={ "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, }) assert load_promisor_remotes(repo) == []