test_hash_object.py
python
| 1 | """Tests for ``muse hash-object``. |
| 2 | |
| 3 | All async tests inject an in-memory SQLite session and a ``tmp_path`` repo |
| 4 | root — no real Postgres or running process required. |
| 5 | |
| 6 | Coverage: |
| 7 | - Pure hash computation (no write) |
| 8 | - Write mode: DB insertion + on-disk store |
| 9 | - Idempotency: writing the same object twice is a no-op |
| 10 | - Stdin mode |
| 11 | - CLI flag validation (file + --stdin, neither flag) |
| 12 | - File-not-found and non-file path errors |
| 13 | - Outside-repo exit |
| 14 | """ |
| 15 | from __future__ import annotations |
| 16 | |
| 17 | import hashlib |
| 18 | import json |
| 19 | import pathlib |
| 20 | import uuid |
| 21 | |
| 22 | import pytest |
| 23 | import typer |
| 24 | from sqlalchemy.ext.asyncio import AsyncSession |
| 25 | |
| 26 | from maestro.muse_cli.commands.hash_object import ( |
| 27 | HashObjectResult, |
| 28 | _hash_object_async, |
| 29 | hash_bytes, |
| 30 | ) |
| 31 | from maestro.muse_cli.errors import ExitCode |
| 32 | from maestro.muse_cli.models import MuseCliObject |
| 33 | from maestro.muse_cli.object_store import object_path, objects_dir |
| 34 | |
| 35 | |
| 36 | # --------------------------------------------------------------------------- |
| 37 | # Helpers |
| 38 | # --------------------------------------------------------------------------- |
| 39 | |
| 40 | |
| 41 | def _init_muse_repo(root: pathlib.Path) -> str: |
| 42 | """Create a minimal ``.muse/`` directory so ``require_repo()`` succeeds.""" |
| 43 | rid = str(uuid.uuid4()) |
| 44 | muse = root / ".muse" |
| 45 | (muse / "refs" / "heads").mkdir(parents=True) |
| 46 | (muse / "repo.json").write_text(json.dumps({"repo_id": rid, "schema_version": "1"})) |
| 47 | (muse / "HEAD").write_text("refs/heads/main") |
| 48 | (muse / "refs" / "heads" / "main").write_text("") |
| 49 | return rid |
| 50 | |
| 51 | |
| 52 | # --------------------------------------------------------------------------- |
| 53 | # hash_bytes — pure unit tests |
| 54 | # --------------------------------------------------------------------------- |
| 55 | |
| 56 | |
| 57 | def test_hash_bytes_returns_sha256_hex() -> None: |
| 58 | """hash_bytes returns the SHA-256 hex digest of the input.""" |
| 59 | content = b"hello muse" |
| 60 | expected = hashlib.sha256(content).hexdigest() |
| 61 | assert hash_bytes(content) == expected |
| 62 | |
| 63 | |
| 64 | def test_hash_bytes_is_64_chars() -> None: |
| 65 | """hash_bytes output is always exactly 64 lowercase hex characters.""" |
| 66 | result = hash_bytes(b"") |
| 67 | assert len(result) == 64 |
| 68 | assert result == result.lower() |
| 69 | assert all(c in "0123456789abcdef" for c in result) |
| 70 | |
| 71 | |
| 72 | def test_hash_bytes_empty_input() -> None: |
| 73 | """hash_bytes of empty bytes is the well-known SHA-256 of empty string.""" |
| 74 | expected = hashlib.sha256(b"").hexdigest() |
| 75 | assert hash_bytes(b"") == expected |
| 76 | |
| 77 | |
| 78 | def test_hash_bytes_deterministic() -> None: |
| 79 | """hash_bytes produces the same output for identical inputs.""" |
| 80 | content = b"drum-pattern-001" |
| 81 | assert hash_bytes(content) == hash_bytes(content) |
| 82 | |
| 83 | |
| 84 | # --------------------------------------------------------------------------- |
| 85 | # _hash_object_async — compute-only (write=False) |
| 86 | # --------------------------------------------------------------------------- |
| 87 | |
| 88 | |
| 89 | @pytest.mark.anyio |
| 90 | async def test_hash_object_compute_only_returns_correct_id( |
| 91 | muse_cli_db_session: AsyncSession, |
| 92 | ) -> None: |
| 93 | """_hash_object_async returns the SHA-256 digest without touching the DB.""" |
| 94 | content = b"kick.mid raw bytes" |
| 95 | expected = hash_bytes(content) |
| 96 | |
| 97 | result = await _hash_object_async( |
| 98 | session=muse_cli_db_session, |
| 99 | content=content, |
| 100 | write=False, |
| 101 | ) |
| 102 | |
| 103 | assert result.object_id == expected |
| 104 | assert result.stored is False |
| 105 | assert result.already_existed is False |
| 106 | |
| 107 | |
| 108 | @pytest.mark.anyio |
| 109 | async def test_hash_object_compute_only_does_not_insert_db_row( |
| 110 | muse_cli_db_session: AsyncSession, |
| 111 | ) -> None: |
| 112 | """write=False must not insert a MuseCliObject row.""" |
| 113 | content = b"snare.mid" |
| 114 | result = await _hash_object_async( |
| 115 | session=muse_cli_db_session, |
| 116 | content=content, |
| 117 | write=False, |
| 118 | ) |
| 119 | |
| 120 | row = await muse_cli_db_session.get(MuseCliObject, result.object_id) |
| 121 | assert row is None |
| 122 | |
| 123 | |
| 124 | # --------------------------------------------------------------------------- |
| 125 | # _hash_object_async — write mode |
| 126 | # --------------------------------------------------------------------------- |
| 127 | |
| 128 | |
| 129 | @pytest.mark.anyio |
| 130 | async def test_hash_object_write_inserts_db_row( |
| 131 | muse_cli_db_session: AsyncSession, |
| 132 | tmp_path: pathlib.Path, |
| 133 | ) -> None: |
| 134 | """write=True inserts a MuseCliObject row with correct size_bytes.""" |
| 135 | _init_muse_repo(tmp_path) |
| 136 | content = b"bass.mid content" |
| 137 | result = await _hash_object_async( |
| 138 | session=muse_cli_db_session, |
| 139 | content=content, |
| 140 | write=True, |
| 141 | repo_root=tmp_path, |
| 142 | ) |
| 143 | await muse_cli_db_session.commit() |
| 144 | |
| 145 | row = await muse_cli_db_session.get(MuseCliObject, result.object_id) |
| 146 | assert row is not None |
| 147 | assert row.size_bytes == len(content) |
| 148 | assert result.stored is True |
| 149 | assert result.already_existed is False |
| 150 | |
| 151 | |
| 152 | @pytest.mark.anyio |
| 153 | async def test_hash_object_write_creates_on_disk_file( |
| 154 | muse_cli_db_session: AsyncSession, |
| 155 | tmp_path: pathlib.Path, |
| 156 | ) -> None: |
| 157 | """write=True writes the object bytes to .muse/objects/<object_id>.""" |
| 158 | _init_muse_repo(tmp_path) |
| 159 | content = b"keys.mid data" |
| 160 | |
| 161 | result = await _hash_object_async( |
| 162 | session=muse_cli_db_session, |
| 163 | content=content, |
| 164 | write=True, |
| 165 | repo_root=tmp_path, |
| 166 | ) |
| 167 | |
| 168 | stored_path = object_path(tmp_path, result.object_id) |
| 169 | assert stored_path.exists() |
| 170 | assert stored_path.read_bytes() == content |
| 171 | |
| 172 | |
| 173 | @pytest.mark.anyio |
| 174 | async def test_hash_object_write_is_idempotent_db( |
| 175 | muse_cli_db_session: AsyncSession, |
| 176 | tmp_path: pathlib.Path, |
| 177 | ) -> None: |
| 178 | """Writing the same object twice leaves exactly one DB row (idempotent).""" |
| 179 | _init_muse_repo(tmp_path) |
| 180 | content = b"repeat-object" |
| 181 | |
| 182 | result1 = await _hash_object_async( |
| 183 | session=muse_cli_db_session, |
| 184 | content=content, |
| 185 | write=True, |
| 186 | repo_root=tmp_path, |
| 187 | ) |
| 188 | await muse_cli_db_session.commit() |
| 189 | |
| 190 | result2 = await _hash_object_async( |
| 191 | session=muse_cli_db_session, |
| 192 | content=content, |
| 193 | write=True, |
| 194 | repo_root=tmp_path, |
| 195 | ) |
| 196 | await muse_cli_db_session.commit() |
| 197 | |
| 198 | assert result1.object_id == result2.object_id |
| 199 | assert result2.already_existed is True |
| 200 | |
| 201 | |
| 202 | @pytest.mark.anyio |
| 203 | async def test_hash_object_write_matches_commit_hash( |
| 204 | muse_cli_db_session: AsyncSession, |
| 205 | tmp_path: pathlib.Path, |
| 206 | ) -> None: |
| 207 | """hash-object -w produces the same ID that muse commit would assign.""" |
| 208 | from maestro.muse_cli.snapshot import hash_file |
| 209 | |
| 210 | _init_muse_repo(tmp_path) |
| 211 | content = b"midi-track-data" |
| 212 | src_file = tmp_path / "track.mid" |
| 213 | src_file.write_bytes(content) |
| 214 | |
| 215 | commit_hash = hash_file(src_file) |
| 216 | result = await _hash_object_async( |
| 217 | session=muse_cli_db_session, |
| 218 | content=content, |
| 219 | write=True, |
| 220 | repo_root=tmp_path, |
| 221 | ) |
| 222 | |
| 223 | assert result.object_id == commit_hash |
| 224 | |
| 225 | |
| 226 | # --------------------------------------------------------------------------- |
| 227 | # HashObjectResult — unit tests |
| 228 | # --------------------------------------------------------------------------- |
| 229 | |
| 230 | |
| 231 | def test_hash_object_result_fields() -> None: |
| 232 | """HashObjectResult stores object_id, stored, and already_existed correctly.""" |
| 233 | oid = "a" * 64 |
| 234 | r = HashObjectResult(object_id=oid, stored=True, already_existed=False) |
| 235 | assert r.object_id == oid |
| 236 | assert r.stored is True |
| 237 | assert r.already_existed is False |
| 238 | |
| 239 | |
| 240 | def test_hash_object_result_defaults_already_existed_false() -> None: |
| 241 | """already_existed defaults to False when not provided.""" |
| 242 | r = HashObjectResult(object_id="b" * 64, stored=False) |
| 243 | assert r.already_existed is False |
| 244 | |
| 245 | |
| 246 | # --------------------------------------------------------------------------- |
| 247 | # CLI integration tests |
| 248 | # --------------------------------------------------------------------------- |
| 249 | |
| 250 | |
| 251 | def test_hash_object_prints_sha256_for_file(tmp_path: pathlib.Path) -> None: |
| 252 | """CLI prints the correct SHA-256 hash for a given file.""" |
| 253 | import os |
| 254 | from typer.testing import CliRunner |
| 255 | from maestro.muse_cli.app import cli |
| 256 | |
| 257 | _init_muse_repo(tmp_path) |
| 258 | content = b"groove-data" |
| 259 | src = tmp_path / "groove.mid" |
| 260 | src.write_bytes(content) |
| 261 | expected = hash_bytes(content) |
| 262 | |
| 263 | runner = CliRunner() |
| 264 | prev = os.getcwd() |
| 265 | try: |
| 266 | os.chdir(tmp_path) |
| 267 | result = runner.invoke(cli, ["hash-object", str(src)], catch_exceptions=False) |
| 268 | finally: |
| 269 | os.chdir(prev) |
| 270 | |
| 271 | assert result.exit_code == 0 |
| 272 | assert expected in result.output |
| 273 | |
| 274 | |
| 275 | def test_hash_object_file_and_stdin_mutually_exclusive(tmp_path: pathlib.Path) -> None: |
| 276 | """Providing both a file and --stdin exits with USER_ERROR.""" |
| 277 | import os |
| 278 | from typer.testing import CliRunner |
| 279 | from maestro.muse_cli.app import cli |
| 280 | |
| 281 | _init_muse_repo(tmp_path) |
| 282 | src = tmp_path / "f.mid" |
| 283 | src.write_bytes(b"x") |
| 284 | |
| 285 | runner = CliRunner() |
| 286 | prev = os.getcwd() |
| 287 | try: |
| 288 | os.chdir(tmp_path) |
| 289 | result = runner.invoke( |
| 290 | cli, ["hash-object", "--stdin", str(src)], catch_exceptions=False |
| 291 | ) |
| 292 | finally: |
| 293 | os.chdir(prev) |
| 294 | |
| 295 | assert result.exit_code == ExitCode.USER_ERROR |
| 296 | |
| 297 | |
| 298 | def test_hash_object_no_args_exits_user_error(tmp_path: pathlib.Path) -> None: |
| 299 | """Calling hash-object with neither a file nor --stdin exits USER_ERROR.""" |
| 300 | import os |
| 301 | from typer.testing import CliRunner |
| 302 | from maestro.muse_cli.app import cli |
| 303 | |
| 304 | _init_muse_repo(tmp_path) |
| 305 | |
| 306 | runner = CliRunner() |
| 307 | prev = os.getcwd() |
| 308 | try: |
| 309 | os.chdir(tmp_path) |
| 310 | result = runner.invoke(cli, ["hash-object", ""], catch_exceptions=False) |
| 311 | finally: |
| 312 | os.chdir(prev) |
| 313 | |
| 314 | assert result.exit_code == ExitCode.USER_ERROR |
| 315 | |
| 316 | |
| 317 | def test_hash_object_missing_file_exits_user_error(tmp_path: pathlib.Path) -> None: |
| 318 | """A non-existent file path exits with USER_ERROR and a clear message.""" |
| 319 | import os |
| 320 | from typer.testing import CliRunner |
| 321 | from maestro.muse_cli.app import cli |
| 322 | |
| 323 | _init_muse_repo(tmp_path) |
| 324 | |
| 325 | runner = CliRunner() |
| 326 | prev = os.getcwd() |
| 327 | try: |
| 328 | os.chdir(tmp_path) |
| 329 | result = runner.invoke( |
| 330 | cli, ["hash-object", "nonexistent.mid"], catch_exceptions=False |
| 331 | ) |
| 332 | finally: |
| 333 | os.chdir(prev) |
| 334 | |
| 335 | assert result.exit_code == ExitCode.USER_ERROR |
| 336 | assert "not found" in result.output.lower() or "File not found" in result.output |
| 337 | |
| 338 | |
| 339 | def test_hash_object_outside_repo_exits_repo_not_found(tmp_path: pathlib.Path) -> None: |
| 340 | """``muse hash-object`` outside a .muse/ directory exits REPO_NOT_FOUND.""" |
| 341 | import os |
| 342 | from typer.testing import CliRunner |
| 343 | from maestro.muse_cli.app import cli |
| 344 | |
| 345 | src = tmp_path / "f.mid" |
| 346 | src.write_bytes(b"data") |
| 347 | |
| 348 | runner = CliRunner() |
| 349 | prev = os.getcwd() |
| 350 | try: |
| 351 | os.chdir(tmp_path) |
| 352 | result = runner.invoke(cli, ["hash-object", str(src)], catch_exceptions=False) |
| 353 | finally: |
| 354 | os.chdir(prev) |
| 355 | |
| 356 | assert result.exit_code == ExitCode.REPO_NOT_FOUND |