test_core_object_availability.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Tests for muse/core/object_availability.py — ObjectState model. |
| 2 | |
| 3 | Every object in a Muse repo is in one of three states: |
| 4 | PRESENT — bytes exist in the local .muse/objects/ store |
| 5 | PROMISED — not local but a promisor remote is configured; can be fetched |
| 6 | MISSING — not local AND no promisor remote; genuine data loss risk |
| 7 | |
| 8 | Coverage: |
| 9 | Unit: object_state() for all three states |
| 10 | load_promisor_remotes() reads config correctly |
| 11 | Edge cases: no remotes configured, promisor=false explicit opt-out, |
| 12 | mixed promisor and non-promisor remotes |
| 13 | Invariants: PRESENT when file exists regardless of remotes |
| 14 | MISSING only when no promisor remote AND file absent |
| 15 | """ |
| 16 | |
| 17 | from __future__ import annotations |
| 18 | from collections.abc import Mapping |
| 19 | |
| 20 | import json |
| 21 | import pathlib |
| 22 | |
| 23 | import pytest |
| 24 | |
| 25 | from muse.core.types import blob_id |
| 26 | from muse.core.object_availability import ObjectState, load_promisor_remotes, object_state |
| 27 | from muse.core.object_store import write_object |
| 28 | from muse.core.paths import config_toml_path, muse_dir |
| 29 | |
| 30 | |
| 31 | # --------------------------------------------------------------------------- |
| 32 | # Helpers |
| 33 | # --------------------------------------------------------------------------- |
| 34 | |
| 35 | |
| 36 | |
| 37 | def _init_repo(path: pathlib.Path, remotes: Mapping[str, object] | None = None) -> pathlib.Path: |
| 38 | dot_muse = muse_dir(path) |
| 39 | for d in ("commits", "snapshots", "objects", "refs/heads"): |
| 40 | (dot_muse / d).mkdir(parents=True, exist_ok=True) |
| 41 | (dot_muse / "HEAD").write_text("ref: refs/heads/main") |
| 42 | (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "avail-test", "domain": "code"})) |
| 43 | if remotes: |
| 44 | _write_remotes(path, remotes) |
| 45 | return path |
| 46 | |
| 47 | |
| 48 | def _write_remotes(repo: pathlib.Path, remotes: Mapping[str, object]) -> None: |
| 49 | """Write a minimal config.toml with the given remotes. |
| 50 | |
| 51 | remotes format: {"name": {"url": "...", "promisor": True/False}} |
| 52 | """ |
| 53 | lines = [] |
| 54 | for name, cfg in remotes.items(): |
| 55 | lines.append(f'[remotes.{name}]') |
| 56 | lines.append(f'url = "{cfg["url"]}"') |
| 57 | if "promisor" in cfg: |
| 58 | val = "true" if cfg["promisor"] else "false" |
| 59 | lines.append(f'promisor = {val}') |
| 60 | (config_toml_path(repo)).write_text("\n".join(lines) + "\n") |
| 61 | |
| 62 | |
| 63 | _OBJ_CONTENT = b"test object content for availability" |
| 64 | _OBJ_ID = blob_id(_OBJ_CONTENT) |
| 65 | |
| 66 | |
| 67 | # --------------------------------------------------------------------------- |
| 68 | # ObjectState enum |
| 69 | # --------------------------------------------------------------------------- |
| 70 | |
| 71 | class TestObjectStateEnum: |
| 72 | def test_has_present(self) -> None: |
| 73 | assert ObjectState.PRESENT == "present" |
| 74 | |
| 75 | def test_has_promised(self) -> None: |
| 76 | assert ObjectState.PROMISED == "promised" |
| 77 | |
| 78 | def test_has_missing(self) -> None: |
| 79 | assert ObjectState.MISSING == "missing" |
| 80 | |
| 81 | def test_is_string_enum(self) -> None: |
| 82 | assert isinstance(ObjectState.PRESENT, str) |
| 83 | |
| 84 | |
| 85 | # --------------------------------------------------------------------------- |
| 86 | # object_state() — PRESENT |
| 87 | # --------------------------------------------------------------------------- |
| 88 | |
| 89 | class TestObjectStatePresent: |
| 90 | def test_present_when_file_exists_no_remotes(self, tmp_path: pathlib.Path) -> None: |
| 91 | repo = _init_repo(tmp_path) |
| 92 | content = b"content-no-remotes" |
| 93 | oid = blob_id(content) |
| 94 | write_object(repo, oid, content) |
| 95 | assert object_state(repo, oid, []) == ObjectState.PRESENT |
| 96 | |
| 97 | def test_present_when_file_exists_with_promisor(self, tmp_path: pathlib.Path) -> None: |
| 98 | repo = _init_repo(tmp_path) |
| 99 | content = b"content-with-promisor" |
| 100 | oid = blob_id(content) |
| 101 | write_object(repo, oid, content) |
| 102 | # Even with promisors, PRESENT wins — file is local |
| 103 | assert object_state(repo, oid, ["local"]) == ObjectState.PRESENT |
| 104 | |
| 105 | def test_present_takes_priority_over_promisor(self, tmp_path: pathlib.Path) -> None: |
| 106 | repo = _init_repo(tmp_path) |
| 107 | content = b"present content bytes" |
| 108 | obj_id = blob_id(content) |
| 109 | write_object(repo, obj_id, content) |
| 110 | state = object_state(repo, obj_id, ["remote-a", "remote-b"]) |
| 111 | assert state == ObjectState.PRESENT |
| 112 | |
| 113 | |
| 114 | # --------------------------------------------------------------------------- |
| 115 | # object_state() — PROMISED |
| 116 | # --------------------------------------------------------------------------- |
| 117 | |
| 118 | class TestObjectStatePromised: |
| 119 | def test_promised_when_absent_with_one_promisor(self, tmp_path: pathlib.Path) -> None: |
| 120 | repo = _init_repo(tmp_path) |
| 121 | state = object_state(repo, _OBJ_ID, ["local"]) |
| 122 | assert state == ObjectState.PROMISED |
| 123 | |
| 124 | def test_promised_when_absent_with_multiple_promisors(self, tmp_path: pathlib.Path) -> None: |
| 125 | repo = _init_repo(tmp_path) |
| 126 | state = object_state(repo, _OBJ_ID, ["origin", "backup"]) |
| 127 | assert state == ObjectState.PROMISED |
| 128 | |
| 129 | def test_promised_not_present(self, tmp_path: pathlib.Path) -> None: |
| 130 | repo = _init_repo(tmp_path) |
| 131 | state = object_state(repo, _OBJ_ID, ["local"]) |
| 132 | assert state != ObjectState.PRESENT |
| 133 | |
| 134 | |
| 135 | # --------------------------------------------------------------------------- |
| 136 | # object_state() — MISSING |
| 137 | # --------------------------------------------------------------------------- |
| 138 | |
| 139 | class TestObjectStateMissing: |
| 140 | def test_missing_when_absent_no_remotes(self, tmp_path: pathlib.Path) -> None: |
| 141 | repo = _init_repo(tmp_path) |
| 142 | state = object_state(repo, _OBJ_ID, []) |
| 143 | assert state == ObjectState.MISSING |
| 144 | |
| 145 | def test_missing_not_promised(self, tmp_path: pathlib.Path) -> None: |
| 146 | repo = _init_repo(tmp_path) |
| 147 | state = object_state(repo, _OBJ_ID, []) |
| 148 | assert state != ObjectState.PROMISED |
| 149 | |
| 150 | |
| 151 | # --------------------------------------------------------------------------- |
| 152 | # load_promisor_remotes() — reads config |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | class TestLoadPromisorRemotes: |
| 156 | def test_empty_when_no_config(self, tmp_path: pathlib.Path) -> None: |
| 157 | repo = _init_repo(tmp_path) |
| 158 | result = load_promisor_remotes(repo) |
| 159 | assert result == [] |
| 160 | |
| 161 | def test_all_remotes_are_promisors_by_default(self, tmp_path: pathlib.Path) -> None: |
| 162 | repo = _init_repo(tmp_path, remotes={ |
| 163 | "local": {"url": "https://localhost:1337/gabriel/muse"}, |
| 164 | "staging": {"url": "https://staging.musehub.ai/gabriel/muse"}, |
| 165 | }) |
| 166 | result = load_promisor_remotes(repo) |
| 167 | assert "local" in result |
| 168 | assert "staging" in result |
| 169 | |
| 170 | def test_explicit_promisor_true_included(self, tmp_path: pathlib.Path) -> None: |
| 171 | repo = _init_repo(tmp_path, remotes={ |
| 172 | "origin": {"url": "https://localhost:1337/gabriel/muse", "promisor": True}, |
| 173 | }) |
| 174 | assert "origin" in load_promisor_remotes(repo) |
| 175 | |
| 176 | def test_explicit_promisor_false_excluded(self, tmp_path: pathlib.Path) -> None: |
| 177 | repo = _init_repo(tmp_path, remotes={ |
| 178 | "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, |
| 179 | }) |
| 180 | assert "mirror" not in load_promisor_remotes(repo) |
| 181 | |
| 182 | def test_mixed_promisors(self, tmp_path: pathlib.Path) -> None: |
| 183 | repo = _init_repo(tmp_path, remotes={ |
| 184 | "local": {"url": "https://localhost:1337/g/muse"}, # default → promisor |
| 185 | "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, |
| 186 | }) |
| 187 | result = load_promisor_remotes(repo) |
| 188 | assert "local" in result |
| 189 | assert "mirror" not in result |
| 190 | |
| 191 | def test_returns_list_of_strings(self, tmp_path: pathlib.Path) -> None: |
| 192 | repo = _init_repo(tmp_path, remotes={ |
| 193 | "local": {"url": "https://localhost:1337/g/muse"}, |
| 194 | }) |
| 195 | result = load_promisor_remotes(repo) |
| 196 | assert isinstance(result, list) |
| 197 | assert all(isinstance(r, str) for r in result) |
| 198 | |
| 199 | def test_no_promisor_when_all_opted_out(self, tmp_path: pathlib.Path) -> None: |
| 200 | repo = _init_repo(tmp_path, remotes={ |
| 201 | "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, |
| 202 | }) |
| 203 | assert load_promisor_remotes(repo) == [] |