"""Unit and integration tests for ``muse.core.doc_history``. Coverage: - :func:`get_symbol_version_events` with empty, single, and multi-entry index. - :func:`infer_since_version` with tagged and untagged events. - :func:`infer_last_changed_version` with various event sequences. - :func:`detect_stale_docstring` with insufficient history, stable, and stale symbols. - :func:`generate_changelog` with added/removed/changed/breaking classifications. - :func:`_build_commit_to_version_map` determinism with multiple tags. """ from __future__ import annotations import datetime import hashlib import pathlib import pytest from muse.core.doc_history import ( ChangelogReport, StaleInfo, SymbolVersionEvent, _build_commit_to_version_map, detect_stale_docstring, generate_changelog, get_symbol_version_events, infer_last_changed_version, infer_since_version, ) from muse.domain import SemVerBump from muse.core.indices import ( SymbolHistoryEntry, SymbolHistoryIndex, save_symbol_history, ) from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.types import Manifest, fake_id, long_id _REPO_ID = fake_id("test-repo-123") from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.tags import ( TagRecord, write_tag, ) from muse.core.paths import heads_dir, muse_dir # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Create a minimal .muse/ repository skeleton.""" dot_muse = muse_dir(tmp_path) dot_muse.mkdir() import json as _json (dot_muse / "repo.json").write_text(_json.dumps({"repo_id": _REPO_ID, "name": "test"})) refs = dot_muse / "refs" / "heads" refs.mkdir(parents=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") return tmp_path def _write_commit( root: pathlib.Path, label: str, parent_id: str | None = None, sem_ver_bump: SemVerBump = "none", breaking_changes: list[str] | None = None, ) -> CommitRecord: manifest: Manifest = {} snapshot_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snapshot_id, manifest=manifest)) committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) message = f"test commit {label}" parent_ids = [parent_id] if parent_id else [] commit_id = compute_commit_id( parent_ids=parent_ids, snapshot_id=snapshot_id, message=message, committed_at_iso=committed_at.isoformat(), author="test", ) commit = CommitRecord( commit_id=commit_id, branch="main", snapshot_id=snapshot_id, message=message, committed_at=committed_at, author="test", parent_commit_id=parent_id, sem_ver_bump=sem_ver_bump, breaking_changes=breaking_changes or [], ) write_commit(root, commit) (heads_dir(root) / "main").write_text(commit_id) return commit def _write_tag(root: pathlib.Path, tag_name: str, commit_id: str) -> None: tag = TagRecord( repo_id=_REPO_ID, tag_id=fake_id(tag_name + "-tag"), commit_id=commit_id, tag=tag_name, ) write_tag(root, tag) def _make_entry( commit_id: str, op: str = "insert", content_id: str = "c1", body_hash: str = "b1", signature_id: str = "s1", ) -> SymbolHistoryEntry: return SymbolHistoryEntry( commit_id=commit_id, committed_at="2026-01-01T00:00:00+00:00", op=op, content_id=content_id, body_hash=body_hash, signature_id=signature_id, ) # --------------------------------------------------------------------------- # Tests: get_symbol_version_events # --------------------------------------------------------------------------- class TestGetSymbolVersionEvents: def test_empty_index(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) events = get_symbol_version_events(root, _REPO_ID, "foo.py::bar") assert events == [] def test_address_not_in_index(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) index: SymbolHistoryIndex = { "other.py::baz": [_make_entry("abc123")] } save_symbol_history(root, index) events = get_symbol_version_events(root, _REPO_ID, "foo.py::bar") assert events == [] def test_single_entry_no_commit(self, tmp_path: pathlib.Path) -> None: """When the commit is not in the store, events still have sem_ver_bump=None.""" root = _make_repo(tmp_path) index: SymbolHistoryIndex = { "foo.py::bar": [_make_entry(fake_id("deadbeef01"))] } save_symbol_history(root, index) events = get_symbol_version_events(root, _REPO_ID, "foo.py::bar") assert len(events) == 1 assert events[0]["op"] == "insert" assert events[0]["sem_ver_bump"] is None assert events[0]["version"] is None assert events[0]["breaking"] is False def test_event_with_commit_and_tag(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) rec = _write_commit(root, "a", sem_ver_bump="minor") cid = rec.commit_id _write_tag(root, "v1.0.0", cid) index: SymbolHistoryIndex = {"foo.py::bar": [_make_entry(cid)]} save_symbol_history(root, index) events = get_symbol_version_events(root, _REPO_ID, "foo.py::bar") assert len(events) == 1 assert events[0]["version"] == "v1.0.0" assert events[0]["sem_ver_bump"] == "minor" assert events[0]["breaking"] is False def test_event_with_breaking_commit(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) rec = _write_commit(root, "b", sem_ver_bump="major", breaking_changes=["Removed foo()"]) cid = rec.commit_id index: SymbolHistoryIndex = { "foo.py::bar": [_make_entry(cid, op="replace")] } save_symbol_history(root, index) events = get_symbol_version_events(root, _REPO_ID, "foo.py::bar") assert events[0]["breaking"] is True def test_multiple_events_ordered(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) rec1 = _write_commit(root, "1") rec2 = _write_commit(root, "2") entries = [ _make_entry(rec1.commit_id, op="insert"), _make_entry(rec2.commit_id, op="replace", content_id="c2"), ] index: SymbolHistoryIndex = {"foo.py::bar": entries} save_symbol_history(root, index) events = get_symbol_version_events(root, _REPO_ID, "foo.py::bar") assert len(events) == 2 assert events[0]["op"] == "insert" assert events[1]["op"] == "replace" # --------------------------------------------------------------------------- # Tests: infer_since_version # --------------------------------------------------------------------------- class TestInferSinceVersion: def test_empty_events(self) -> None: assert infer_since_version([]) is None def test_single_untagged(self) -> None: ev = SymbolVersionEvent( commit_id="abc", committed_at="2026-01-01T00:00:00+00:00", op="insert", version=None, sem_ver_bump=None, breaking=False, ) assert infer_since_version([ev]) is None def test_insert_with_version(self) -> None: ev = SymbolVersionEvent( commit_id="abc", committed_at="2026-01-01T00:00:00+00:00", op="insert", version="v1.0.0", sem_ver_bump="minor", breaking=False, ) assert infer_since_version([ev]) == "v1.0.0" def test_prefers_insert_over_replace(self) -> None: ev1 = SymbolVersionEvent( commit_id="a", committed_at="2026-01-01T00:00:00+00:00", op="insert", version="v1.0.0", sem_ver_bump=None, breaking=False, ) ev2 = SymbolVersionEvent( commit_id="b", committed_at="2026-02-01T00:00:00+00:00", op="replace", version="v2.0.0", sem_ver_bump=None, breaking=False, ) assert infer_since_version([ev1, ev2]) == "v1.0.0" def test_fallback_to_first_event_with_version(self) -> None: ev1 = SymbolVersionEvent( commit_id="a", committed_at="2026-01-01T00:00:00+00:00", op="replace", # not "insert" version="v0.9.0", sem_ver_bump=None, breaking=False, ) assert infer_since_version([ev1]) == "v0.9.0" # --------------------------------------------------------------------------- # Tests: infer_last_changed_version # --------------------------------------------------------------------------- class TestInferLastChangedVersion: def test_empty(self) -> None: assert infer_last_changed_version([]) is None def test_only_insert(self) -> None: ev = SymbolVersionEvent( commit_id="a", committed_at="2026-01-01T00:00:00+00:00", op="insert", version="v1.0.0", sem_ver_bump=None, breaking=False, ) # "insert" is not "replace"/"delete" so returns None. assert infer_last_changed_version([ev]) is None def test_replace_returns_version(self) -> None: ev1 = SymbolVersionEvent( commit_id="a", committed_at="2026-01-01T00:00:00+00:00", op="insert", version="v1.0.0", sem_ver_bump=None, breaking=False, ) ev2 = SymbolVersionEvent( commit_id="b", committed_at="2026-02-01T00:00:00+00:00", op="replace", version="v1.1.0", sem_ver_bump=None, breaking=False, ) assert infer_last_changed_version([ev1, ev2]) == "v1.1.0" def test_newest_first_scan(self) -> None: """infer_last_changed_version scans newest-first.""" events = [ SymbolVersionEvent( commit_id="a", committed_at="2026-01-01T00:00:00+00:00", op="replace", version="v1.0.0", sem_ver_bump=None, breaking=False, ), SymbolVersionEvent( commit_id="b", committed_at="2026-02-01T00:00:00+00:00", op="replace", version="v2.0.0", sem_ver_bump=None, breaking=False, ), ] assert infer_last_changed_version(events) == "v2.0.0" # --------------------------------------------------------------------------- # Tests: detect_stale_docstring # --------------------------------------------------------------------------- class TestDetectStaleDocstring: def test_empty_index(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) info = detect_stale_docstring(root, "foo.py::bar") assert info["is_stale"] is False assert info["last_doc_commit"] is None def test_single_entry_not_stale(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) index: SymbolHistoryIndex = { "foo.py::bar": [_make_entry("abc")] } save_symbol_history(root, index) info = detect_stale_docstring(root, "foo.py::bar") assert info["is_stale"] is False def test_stable_body_and_sig(self, tmp_path: pathlib.Path) -> None: """Two events with same hashes — nothing changed.""" root = _make_repo(tmp_path) entries = [ _make_entry("a1", body_hash="bh1", signature_id="sg1"), _make_entry("a2", op="replace", body_hash="bh1", signature_id="sg1"), ] index: SymbolHistoryIndex = {"foo.py::bar": entries} save_symbol_history(root, index) info = detect_stale_docstring(root, "foo.py::bar") assert info["is_stale"] is False def test_sig_changed_after_body(self, tmp_path: pathlib.Path) -> None: """Signature changed after body → stale.""" root = _make_repo(tmp_path) entries = [ _make_entry("a1", body_hash="bh1", signature_id="sg1"), _make_entry("a2", op="replace", body_hash="bh2", signature_id="sg1"), # body changed _make_entry("a3", op="replace", body_hash="bh2", signature_id="sg2"), # sig changed ] index: SymbolHistoryIndex = {"foo.py::bar": entries} save_symbol_history(root, index) info = detect_stale_docstring(root, "foo.py::bar") assert info["is_stale"] is True assert info["signature_changed"] is True def test_not_stale_when_body_last(self, tmp_path: pathlib.Path) -> None: """Body changed last — no staleness.""" root = _make_repo(tmp_path) entries = [ _make_entry("a1", body_hash="bh1", signature_id="sg1"), _make_entry("a2", op="replace", body_hash="bh1", signature_id="sg2"), # sig changed _make_entry("a3", op="replace", body_hash="bh2", signature_id="sg2"), # body changed ] index: SymbolHistoryIndex = {"foo.py::bar": entries} save_symbol_history(root, index) info = detect_stale_docstring(root, "foo.py::bar") # body changed after sig → body_changed = True → is_stale = True assert info["is_stale"] is True assert info["body_changed"] is True # --------------------------------------------------------------------------- # Tests: generate_changelog # --------------------------------------------------------------------------- class TestGenerateChangelog: def test_unresolvable_to_ref(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = generate_changelog(root, _REPO_ID, "v0.9", "v999") assert result["from_ref"] == "v0.9" assert result["to_ref"] == "v999" assert result["added"] == [] assert result["removed"] == [] assert result["changed"] == [] assert result["breaking"] == [] def test_empty_range(self, tmp_path: pathlib.Path) -> None: """With no commits in range, all sections are empty.""" root = _make_repo(tmp_path) rec = _write_commit(root, "f") _write_tag(root, "v1.0", rec.commit_id) result = generate_changelog(root, _REPO_ID, "v1.0", "v1.0") assert result["added"] == [] def test_added_symbol(self, tmp_path: pathlib.Path) -> None: """A symbol with only 'insert' events in range appears in 'added'.""" root = _make_repo(tmp_path) rec = _write_commit(root, "c") cid = rec.commit_id _write_tag(root, "v1.1", cid) entries = [_make_entry(cid, op="insert")] index: SymbolHistoryIndex = {"foo.py::new_fn": entries} save_symbol_history(root, index) result = generate_changelog(root, _REPO_ID, "v1.0", "v1.1") added_addrs = [e["address"] for e in result["added"]] assert "foo.py::new_fn" in added_addrs def test_breaking_symbol(self, tmp_path: pathlib.Path) -> None: """A symbol in a commit with breaking_changes appears in 'breaking'.""" root = _make_repo(tmp_path) rec = _write_commit(root, "d", breaking_changes=["Removed API"]) cid = rec.commit_id _write_tag(root, "v2.0", cid) entries = [_make_entry(cid, op="replace")] index: SymbolHistoryIndex = {"foo.py::changed_fn": entries} save_symbol_history(root, index) result = generate_changelog(root, _REPO_ID, "v1.0", "v2.0") breaking_addrs = [e["address"] for e in result["breaking"]] assert "foo.py::changed_fn" in breaking_addrs def test_sorted_output(self, tmp_path: pathlib.Path) -> None: """Output entries are sorted by address.""" root = _make_repo(tmp_path) rec = _write_commit(root, "e") cid = rec.commit_id _write_tag(root, "v1.2", cid) index: SymbolHistoryIndex = { "z.py::b": [_make_entry(cid, op="insert")], "a.py::a": [_make_entry(cid, op="insert")], } save_symbol_history(root, index) result = generate_changelog(root, _REPO_ID, "v0.9", "v1.2") addrs = [e["address"] for e in result["added"]] assert addrs == sorted(addrs) # --------------------------------------------------------------------------- # Tests: _build_commit_to_version_map # --------------------------------------------------------------------------- class TestBuildCommitToVersionMap: def test_empty(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) result = _build_commit_to_version_map(root, _REPO_ID) assert result == {} def test_single_tag(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) rec = _write_commit(root, "a") _write_tag(root, "v1.0", rec.commit_id) result = _build_commit_to_version_map(root, _REPO_ID) assert result[rec.commit_id] == "v1.0" def test_deterministic_with_multiple_tags(self, tmp_path: pathlib.Path) -> None: """When a commit has multiple tags, the last-sorted tag wins.""" root = _make_repo(tmp_path) rec = _write_commit(root, "b") cid = rec.commit_id _write_tag(root, "v1.0.0", cid) _write_tag(root, "v1.0.1", cid) result = _build_commit_to_version_map(root, _REPO_ID) # Sorted: "v1.0.0" < "v1.0.1" — last sorted wins assert result[cid] == "v1.0.1"