gabriel / musehub public
test_musehub_reflog_service.py python
119 lines 4.3 KB
Raw
sha256:80674069cfe8fe062c526a103e1cd0f855009d1cd06662ca3ae8393e1545a400 feat: muse#54 item 4 server side — musehub_reflog_entries Sonnet 5 minor ⚠ breaking 21 hours ago
1 """Unit tests for musehub/services/musehub_reflog.py (muse#54 item 4).
2
3 Tests the service layer directly (no HTTP): append-only store/list
4 round-trip, per-branch isolation, dedup via content-hash entry_id, and
5 default handling for missing/empty fields.
6 """
7 from __future__ import annotations
8
9 from datetime import datetime, timezone
10
11 from sqlalchemy.ext.asyncio import AsyncSession
12
13 from musehub.services import musehub_reflog as reflog_svc
14 from tests.factories import create_repo
15
16
17 def _entry_dict(old_id: str, new_id: str, **overrides: str) -> dict:
18 base = {
19 "old_id": old_id,
20 "new_id": new_id,
21 "author": "gabriel",
22 "operation": "commit: c1",
23 "timestamp": datetime.now(timezone.utc).isoformat(),
24 }
25 base.update(overrides)
26 return base
27
28
29 async def test_store_and_list_round_trip(db_session: AsyncSession) -> None:
30 repo = await create_repo(db_session, visibility="public")
31 stored, skipped = await reflog_svc.store_reflog_entries(
32 db_session, str(repo.repo_id), "main",
33 [_entry_dict("0" * 64, "a" * 64)],
34 )
35 await db_session.commit()
36 assert stored == 1
37 assert skipped == 0
38
39 entries = await reflog_svc.list_reflog_entries(db_session, str(repo.repo_id), "main")
40 assert len(entries) == 1
41 assert entries[0]["old_id"] == "0" * 64
42 assert entries[0]["new_id"] == "a" * 64
43 assert entries[0]["operation"] == "commit: c1"
44
45
46 async def test_repush_same_entry_is_skipped(db_session: AsyncSession) -> None:
47 """Re-pushing an identical entry (same content hash) is a true
48 no-op -- it must not duplicate the row."""
49 repo = await create_repo(db_session, visibility="public")
50 entry = _entry_dict("0" * 64, "a" * 64, timestamp="2026-01-01T00:00:00+00:00")
51
52 await reflog_svc.store_reflog_entries(db_session, str(repo.repo_id), "main", [entry])
53 await db_session.commit()
54
55 stored, skipped = await reflog_svc.store_reflog_entries(
56 db_session, str(repo.repo_id), "main", [entry],
57 )
58 await db_session.commit()
59 assert stored == 0
60 assert skipped == 1
61
62 entries = await reflog_svc.list_reflog_entries(db_session, str(repo.repo_id), "main")
63 assert len(entries) == 1
64
65
66 async def test_entries_isolated_per_branch(db_session: AsyncSession) -> None:
67 repo = await create_repo(db_session, visibility="public")
68 await reflog_svc.store_reflog_entries(
69 db_session, str(repo.repo_id), "main", [_entry_dict("0" * 64, "a" * 64)],
70 )
71 await reflog_svc.store_reflog_entries(
72 db_session, str(repo.repo_id), "dev", [_entry_dict("0" * 64, "b" * 64)],
73 )
74 await db_session.commit()
75
76 main_entries = await reflog_svc.list_reflog_entries(db_session, str(repo.repo_id), "main")
77 dev_entries = await reflog_svc.list_reflog_entries(db_session, str(repo.repo_id), "dev")
78 assert len(main_entries) == 1
79 assert len(dev_entries) == 1
80 assert main_entries[0]["new_id"] == "a" * 64
81 assert dev_entries[0]["new_id"] == "b" * 64
82
83
84 async def test_list_ordered_oldest_first(db_session: AsyncSession) -> None:
85 repo = await create_repo(db_session, visibility="public")
86 await reflog_svc.store_reflog_entries(
87 db_session, str(repo.repo_id), "main",
88 [
89 _entry_dict("0" * 64, "a" * 64, timestamp="2026-01-01T00:00:00+00:00"),
90 _entry_dict("a" * 64, "b" * 64, timestamp="2026-01-02T00:00:00+00:00"),
91 ],
92 )
93 await db_session.commit()
94
95 entries = await reflog_svc.list_reflog_entries(db_session, str(repo.repo_id), "main")
96 assert len(entries) == 2
97 assert entries[0]["new_id"] == "a" * 64
98 assert entries[1]["new_id"] == "b" * 64
99
100
101 async def test_empty_entries_list_is_noop(db_session: AsyncSession) -> None:
102 repo = await create_repo(db_session, visibility="public")
103 stored, skipped = await reflog_svc.store_reflog_entries(
104 db_session, str(repo.repo_id), "main", [],
105 )
106 assert (stored, skipped) == (0, 0)
107
108
109 async def test_entry_missing_new_id_is_skipped_silently(db_session: AsyncSession) -> None:
110 repo = await create_repo(db_session, visibility="public")
111 stored, skipped = await reflog_svc.store_reflog_entries(
112 db_session, str(repo.repo_id), "main", [_entry_dict("0" * 64, "")],
113 )
114 await db_session.commit()
115 assert stored == 0
116 assert skipped == 0
117
118 entries = await reflog_svc.list_reflog_entries(db_session, str(repo.repo_id), "main")
119 assert entries == []
File History 1 commit
sha256:80674069cfe8fe062c526a103e1cd0f855009d1cd06662ca3ae8393e1545a400 feat: muse#54 item 4 server side — musehub_reflog_entries Sonnet 5 minor ⚠ 21 hours ago