test_core_object_availability.py
python
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
131 days ago
| 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 | |
| 29 | |
| 30 | # --------------------------------------------------------------------------- |
| 31 | # Helpers |
| 32 | # --------------------------------------------------------------------------- |
| 33 | |
| 34 | def _sha(data: bytes) -> str: |
| 35 | return blob_id(data) |
| 36 | |
| 37 | |
| 38 | def _init_repo(path: pathlib.Path, remotes: Mapping[str, object] | None = None) -> pathlib.Path: |
| 39 | muse = path / ".muse" |
| 40 | for d in ("commits", "snapshots", "objects", "refs/heads"): |
| 41 | (muse / d).mkdir(parents=True, exist_ok=True) |
| 42 | (muse / "HEAD").write_text("ref: refs/heads/main") |
| 43 | (muse / "repo.json").write_text(json.dumps({"repo_id": "avail-test", "domain": "code"})) |
| 44 | if remotes: |
| 45 | _write_remotes(path, remotes) |
| 46 | return path |
| 47 | |
| 48 | |
| 49 | def _write_remotes(repo: pathlib.Path, remotes: Mapping[str, object]) -> None: |
| 50 | """Write a minimal config.toml with the given remotes. |
| 51 | |
| 52 | remotes format: {"name": {"url": "...", "promisor": True/False}} |
| 53 | """ |
| 54 | lines = [] |
| 55 | for name, cfg in remotes.items(): |
| 56 | lines.append(f'[remotes.{name}]') |
| 57 | lines.append(f'url = "{cfg["url"]}"') |
| 58 | if "promisor" in cfg: |
| 59 | val = "true" if cfg["promisor"] else "false" |
| 60 | lines.append(f'promisor = {val}') |
| 61 | (repo / ".muse" / "config.toml").write_text("\n".join(lines) + "\n") |
| 62 | |
| 63 | |
| 64 | _OBJ_CONTENT = b"test object content for availability" |
| 65 | _OBJ_ID = _sha(_OBJ_CONTENT) |
| 66 | |
| 67 | |
| 68 | # --------------------------------------------------------------------------- |
| 69 | # ObjectState enum |
| 70 | # --------------------------------------------------------------------------- |
| 71 | |
| 72 | class TestObjectStateEnum: |
| 73 | def test_has_present(self) -> None: |
| 74 | assert ObjectState.PRESENT == "present" |
| 75 | |
| 76 | def test_has_promised(self) -> None: |
| 77 | assert ObjectState.PROMISED == "promised" |
| 78 | |
| 79 | def test_has_missing(self) -> None: |
| 80 | assert ObjectState.MISSING == "missing" |
| 81 | |
| 82 | def test_is_string_enum(self) -> None: |
| 83 | assert isinstance(ObjectState.PRESENT, str) |
| 84 | |
| 85 | |
| 86 | # --------------------------------------------------------------------------- |
| 87 | # object_state() — PRESENT |
| 88 | # --------------------------------------------------------------------------- |
| 89 | |
| 90 | class TestObjectStatePresent: |
| 91 | def test_present_when_file_exists_no_remotes(self, tmp_path: pathlib.Path) -> None: |
| 92 | repo = _init_repo(tmp_path) |
| 93 | content = b"content-no-remotes" |
| 94 | oid = _sha(content) |
| 95 | write_object(repo, oid, content) |
| 96 | assert object_state(repo, oid, []) == ObjectState.PRESENT |
| 97 | |
| 98 | def test_present_when_file_exists_with_promisor(self, tmp_path: pathlib.Path) -> None: |
| 99 | repo = _init_repo(tmp_path) |
| 100 | content = b"content-with-promisor" |
| 101 | oid = _sha(content) |
| 102 | write_object(repo, oid, content) |
| 103 | # Even with promisors, PRESENT wins — file is local |
| 104 | assert object_state(repo, oid, ["local"]) == ObjectState.PRESENT |
| 105 | |
| 106 | def test_present_takes_priority_over_promisor(self, tmp_path: pathlib.Path) -> None: |
| 107 | repo = _init_repo(tmp_path) |
| 108 | content = b"present content bytes" |
| 109 | obj_id = _sha(content) |
| 110 | write_object(repo, obj_id, content) |
| 111 | state = object_state(repo, obj_id, ["remote-a", "remote-b"]) |
| 112 | assert state == ObjectState.PRESENT |
| 113 | |
| 114 | |
| 115 | # --------------------------------------------------------------------------- |
| 116 | # object_state() — PROMISED |
| 117 | # --------------------------------------------------------------------------- |
| 118 | |
| 119 | class TestObjectStatePromised: |
| 120 | def test_promised_when_absent_with_one_promisor(self, tmp_path: pathlib.Path) -> None: |
| 121 | repo = _init_repo(tmp_path) |
| 122 | state = object_state(repo, _OBJ_ID, ["local"]) |
| 123 | assert state == ObjectState.PROMISED |
| 124 | |
| 125 | def test_promised_when_absent_with_multiple_promisors(self, tmp_path: pathlib.Path) -> None: |
| 126 | repo = _init_repo(tmp_path) |
| 127 | state = object_state(repo, _OBJ_ID, ["origin", "backup"]) |
| 128 | assert state == ObjectState.PROMISED |
| 129 | |
| 130 | def test_promised_not_present(self, tmp_path: pathlib.Path) -> None: |
| 131 | repo = _init_repo(tmp_path) |
| 132 | state = object_state(repo, _OBJ_ID, ["local"]) |
| 133 | assert state != ObjectState.PRESENT |
| 134 | |
| 135 | |
| 136 | # --------------------------------------------------------------------------- |
| 137 | # object_state() — MISSING |
| 138 | # --------------------------------------------------------------------------- |
| 139 | |
| 140 | class TestObjectStateMissing: |
| 141 | def test_missing_when_absent_no_remotes(self, tmp_path: pathlib.Path) -> None: |
| 142 | repo = _init_repo(tmp_path) |
| 143 | state = object_state(repo, _OBJ_ID, []) |
| 144 | assert state == ObjectState.MISSING |
| 145 | |
| 146 | def test_missing_not_promised(self, tmp_path: pathlib.Path) -> None: |
| 147 | repo = _init_repo(tmp_path) |
| 148 | state = object_state(repo, _OBJ_ID, []) |
| 149 | assert state != ObjectState.PROMISED |
| 150 | |
| 151 | |
| 152 | # --------------------------------------------------------------------------- |
| 153 | # load_promisor_remotes() — reads config |
| 154 | # --------------------------------------------------------------------------- |
| 155 | |
| 156 | class TestLoadPromisorRemotes: |
| 157 | def test_empty_when_no_config(self, tmp_path: pathlib.Path) -> None: |
| 158 | repo = _init_repo(tmp_path) |
| 159 | result = load_promisor_remotes(repo) |
| 160 | assert result == [] |
| 161 | |
| 162 | def test_all_remotes_are_promisors_by_default(self, tmp_path: pathlib.Path) -> None: |
| 163 | repo = _init_repo(tmp_path, remotes={ |
| 164 | "local": {"url": "https://localhost:1337/gabriel/muse"}, |
| 165 | "staging": {"url": "https://staging.musehub.ai/gabriel/muse"}, |
| 166 | }) |
| 167 | result = load_promisor_remotes(repo) |
| 168 | assert "local" in result |
| 169 | assert "staging" in result |
| 170 | |
| 171 | def test_explicit_promisor_true_included(self, tmp_path: pathlib.Path) -> None: |
| 172 | repo = _init_repo(tmp_path, remotes={ |
| 173 | "origin": {"url": "https://localhost:1337/gabriel/muse", "promisor": True}, |
| 174 | }) |
| 175 | assert "origin" in load_promisor_remotes(repo) |
| 176 | |
| 177 | def test_explicit_promisor_false_excluded(self, tmp_path: pathlib.Path) -> None: |
| 178 | repo = _init_repo(tmp_path, remotes={ |
| 179 | "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, |
| 180 | }) |
| 181 | assert "mirror" not in load_promisor_remotes(repo) |
| 182 | |
| 183 | def test_mixed_promisors(self, tmp_path: pathlib.Path) -> None: |
| 184 | repo = _init_repo(tmp_path, remotes={ |
| 185 | "local": {"url": "https://localhost:1337/g/muse"}, # default → promisor |
| 186 | "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, |
| 187 | }) |
| 188 | result = load_promisor_remotes(repo) |
| 189 | assert "local" in result |
| 190 | assert "mirror" not in result |
| 191 | |
| 192 | def test_returns_list_of_strings(self, tmp_path: pathlib.Path) -> None: |
| 193 | repo = _init_repo(tmp_path, remotes={ |
| 194 | "local": {"url": "https://localhost:1337/g/muse"}, |
| 195 | }) |
| 196 | result = load_promisor_remotes(repo) |
| 197 | assert isinstance(result, list) |
| 198 | assert all(isinstance(r, str) for r in result) |
| 199 | |
| 200 | def test_no_promisor_when_all_opted_out(self, tmp_path: pathlib.Path) -> None: |
| 201 | repo = _init_repo(tmp_path, remotes={ |
| 202 | "mirror": {"url": "http://mirror.example.com/muse", "promisor": False}, |
| 203 | }) |
| 204 | assert load_promisor_remotes(repo) == [] |
File History
2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9
docs: expand cache plan with all seven testing tiers and do…
Sonnet 4.6
131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c
docs: docstring sprint for-each-ref→hotspots — idiomatic ru…
Sonnet 4.6
patch
137 days ago