test_snapshot.py
python
| 1 | """Unit tests for maestro.muse_cli.snapshot. |
| 2 | |
| 3 | All tests are pure — no database, no network, no Typer runner. |
| 4 | They verify the deterministic hash derivation contract documented in |
| 5 | snapshot.py's module docstring. |
| 6 | """ |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import hashlib |
| 10 | import pathlib |
| 11 | |
| 12 | import pytest |
| 13 | |
| 14 | from maestro.muse_cli.snapshot import ( |
| 15 | build_snapshot_manifest, |
| 16 | compute_commit_id, |
| 17 | compute_snapshot_id, |
| 18 | hash_file, |
| 19 | walk_workdir, |
| 20 | ) |
| 21 | |
| 22 | |
| 23 | # --------------------------------------------------------------------------- |
| 24 | # hash_file |
| 25 | # --------------------------------------------------------------------------- |
| 26 | |
| 27 | |
| 28 | def test_hash_file_known_digest(tmp_path: pathlib.Path) -> None: |
| 29 | f = tmp_path / "hello.txt" |
| 30 | f.write_bytes(b"hello") |
| 31 | expected = hashlib.sha256(b"hello").hexdigest() |
| 32 | assert hash_file(f) == expected |
| 33 | |
| 34 | |
| 35 | def test_hash_file_empty_file(tmp_path: pathlib.Path) -> None: |
| 36 | f = tmp_path / "empty.mid" |
| 37 | f.write_bytes(b"") |
| 38 | assert hash_file(f) == hashlib.sha256(b"").hexdigest() |
| 39 | |
| 40 | |
| 41 | def test_hash_file_different_content_different_digest(tmp_path: pathlib.Path) -> None: |
| 42 | a = tmp_path / "a.mid" |
| 43 | b = tmp_path / "b.mid" |
| 44 | a.write_bytes(b"MIDI-A") |
| 45 | b.write_bytes(b"MIDI-B") |
| 46 | assert hash_file(a) != hash_file(b) |
| 47 | |
| 48 | |
| 49 | # --------------------------------------------------------------------------- |
| 50 | # walk_workdir / build_snapshot_manifest |
| 51 | # --------------------------------------------------------------------------- |
| 52 | |
| 53 | |
| 54 | def test_walk_workdir_returns_relative_posix_paths(tmp_path: pathlib.Path) -> None: |
| 55 | (tmp_path / "bass.mid").write_bytes(b"bass") |
| 56 | (tmp_path / "drums.mp3").write_bytes(b"drums") |
| 57 | result = walk_workdir(tmp_path) |
| 58 | assert "bass.mid" in result |
| 59 | assert "drums.mp3" in result |
| 60 | |
| 61 | |
| 62 | def test_walk_workdir_excludes_hidden_files(tmp_path: pathlib.Path) -> None: |
| 63 | (tmp_path / "track.mid").write_bytes(b"data") |
| 64 | (tmp_path / ".DS_Store").write_bytes(b"mac junk") |
| 65 | result = walk_workdir(tmp_path) |
| 66 | assert ".DS_Store" not in result |
| 67 | assert "track.mid" in result |
| 68 | |
| 69 | |
| 70 | def test_walk_workdir_recurses_into_subdirectories(tmp_path: pathlib.Path) -> None: |
| 71 | sub = tmp_path / "loops" |
| 72 | sub.mkdir() |
| 73 | (sub / "beat.mid").write_bytes(b"beat") |
| 74 | result = walk_workdir(tmp_path) |
| 75 | assert "loops/beat.mid" in result |
| 76 | |
| 77 | |
| 78 | def test_walk_workdir_empty_directory(tmp_path: pathlib.Path) -> None: |
| 79 | result = walk_workdir(tmp_path) |
| 80 | assert result == {} |
| 81 | |
| 82 | |
| 83 | def test_build_snapshot_manifest_same_as_walk_workdir(tmp_path: pathlib.Path) -> None: |
| 84 | (tmp_path / "x.mid").write_bytes(b"x") |
| 85 | assert build_snapshot_manifest(tmp_path) == walk_workdir(tmp_path) |
| 86 | |
| 87 | |
| 88 | # --------------------------------------------------------------------------- |
| 89 | # compute_snapshot_id |
| 90 | # --------------------------------------------------------------------------- |
| 91 | |
| 92 | |
| 93 | def test_snapshot_id_is_deterministic(tmp_path: pathlib.Path) -> None: |
| 94 | (tmp_path / "a.mid").write_bytes(b"A") |
| 95 | m1 = walk_workdir(tmp_path) |
| 96 | m2 = walk_workdir(tmp_path) |
| 97 | assert compute_snapshot_id(m1) == compute_snapshot_id(m2) |
| 98 | |
| 99 | |
| 100 | def test_snapshot_id_changes_when_content_changes(tmp_path: pathlib.Path) -> None: |
| 101 | f = tmp_path / "a.mid" |
| 102 | f.write_bytes(b"original") |
| 103 | snap1 = compute_snapshot_id(walk_workdir(tmp_path)) |
| 104 | f.write_bytes(b"modified") |
| 105 | snap2 = compute_snapshot_id(walk_workdir(tmp_path)) |
| 106 | assert snap1 != snap2 |
| 107 | |
| 108 | |
| 109 | def test_snapshot_id_is_order_independent() -> None: |
| 110 | """snapshot_id must not depend on dict insertion order.""" |
| 111 | m1 = {"b.mid": "bbb", "a.mid": "aaa"} |
| 112 | m2 = {"a.mid": "aaa", "b.mid": "bbb"} |
| 113 | assert compute_snapshot_id(m1) == compute_snapshot_id(m2) |
| 114 | |
| 115 | |
| 116 | def test_snapshot_id_is_sha256_hex(tmp_path: pathlib.Path) -> None: |
| 117 | (tmp_path / "f.mid").write_bytes(b"data") |
| 118 | sid = compute_snapshot_id(walk_workdir(tmp_path)) |
| 119 | assert len(sid) == 64 |
| 120 | assert all(c in "0123456789abcdef" for c in sid) |
| 121 | |
| 122 | |
| 123 | # --------------------------------------------------------------------------- |
| 124 | # compute_commit_id |
| 125 | # --------------------------------------------------------------------------- |
| 126 | |
| 127 | |
| 128 | def test_commit_id_is_deterministic() -> None: |
| 129 | cid1 = compute_commit_id([], "snap1", "first commit", "2026-01-01T00:00:00+00:00") |
| 130 | cid2 = compute_commit_id([], "snap1", "first commit", "2026-01-01T00:00:00+00:00") |
| 131 | assert cid1 == cid2 |
| 132 | |
| 133 | |
| 134 | def test_commit_id_changes_with_different_message() -> None: |
| 135 | cid1 = compute_commit_id([], "snap1", "take 1", "2026-01-01T00:00:00+00:00") |
| 136 | cid2 = compute_commit_id([], "snap1", "take 2", "2026-01-01T00:00:00+00:00") |
| 137 | assert cid1 != cid2 |
| 138 | |
| 139 | |
| 140 | def test_commit_id_changes_with_different_snapshot() -> None: |
| 141 | cid1 = compute_commit_id([], "snap-A", "msg", "2026-01-01T00:00:00+00:00") |
| 142 | cid2 = compute_commit_id([], "snap-B", "msg", "2026-01-01T00:00:00+00:00") |
| 143 | assert cid1 != cid2 |
| 144 | |
| 145 | |
| 146 | def test_commit_id_parent_order_does_not_matter() -> None: |
| 147 | """commit_id must be stable regardless of parent_ids list order.""" |
| 148 | cid1 = compute_commit_id(["p1", "p2"], "snap", "msg", "2026-01-01T00:00:00+00:00") |
| 149 | cid2 = compute_commit_id(["p2", "p1"], "snap", "msg", "2026-01-01T00:00:00+00:00") |
| 150 | assert cid1 == cid2 |
| 151 | |
| 152 | |
| 153 | def test_commit_id_is_sha256_hex() -> None: |
| 154 | cid = compute_commit_id([], "snap", "msg", "2026-01-01T00:00:00+00:00") |
| 155 | assert len(cid) == 64 |
| 156 | assert all(c in "0123456789abcdef" for c in cid) |
| 157 | |
| 158 | |
| 159 | @pytest.mark.parametrize( |
| 160 | "parent_ids,snapshot_id,message,ts", |
| 161 | [ |
| 162 | ([], "abc", "boom bap demo take 1", "2026-02-01T12:00:00+00:00"), |
| 163 | (["p1"], "def", "ambient take 3", "2026-02-02T08:00:00+00:00"), |
| 164 | (["p1", "p2"], "ghi", "merge feature/drums", "2026-02-27T00:00:00+00:00"), |
| 165 | ], |
| 166 | ) |
| 167 | def test_commit_id_parametrized_deterministic( |
| 168 | parent_ids: list[str], snapshot_id: str, message: str, ts: str |
| 169 | ) -> None: |
| 170 | assert compute_commit_id(parent_ids, snapshot_id, message, ts) == compute_commit_id( |
| 171 | parent_ids, snapshot_id, message, ts |
| 172 | ) |