gabriel / musehub public
test_musehub_symlog_service.py python
144 lines 5.5 KB
Raw
sha256:4fba330283cca0a6bdd067f2c0ad014062fcfda7e98b19ce8cd1e20290252b08 feat: muse#53 item 4 server side — musehub_symlog_entries Sonnet 5 minor ⚠ breaking 2 days ago
1 """Unit tests for musehub/services/musehub_symlog.py (muse#53 item 4).
2
3 Tests the service layer directly (no HTTP): batched multi-symbol
4 store/list round-trip, per-symbol isolation, dedup via content-hash
5 entry_id, and 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_symlog as symlog_svc
14 from tests.factories import create_repo
15
16
17 def _entry_dict(old_cid: str, new_cid: str, commit_id: str, **overrides: str) -> dict:
18 base = {
19 "old_content_id": old_cid,
20 "new_content_id": new_cid,
21 "commit_id": commit_id,
22 "author": "gabriel",
23 "operation": "symbol-created: compute_total",
24 "timestamp": datetime.now(timezone.utc).isoformat(),
25 }
26 base.update(overrides)
27 return base
28
29
30 async def test_store_and_list_round_trip(db_session: AsyncSession) -> None:
31 repo = await create_repo(db_session, visibility="public")
32 stored, skipped = await symlog_svc.store_symlog_entries(
33 db_session, str(repo.repo_id),
34 {"billing.py::compute_total": [_entry_dict("0" * 64, "a" * 64, "c" * 64)]},
35 )
36 await db_session.commit()
37 assert stored == 1
38 assert skipped == 0
39
40 symbols = await symlog_svc.list_symlog_entries(db_session, str(repo.repo_id))
41 assert "billing.py::compute_total" in symbols
42 entries = symbols["billing.py::compute_total"]
43 assert len(entries) == 1
44 assert entries[0]["old_content_id"] == "0" * 64
45 assert entries[0]["new_content_id"] == "a" * 64
46 assert entries[0]["commit_id"] == "c" * 64
47
48
49 async def test_repush_same_entry_is_skipped(db_session: AsyncSession) -> None:
50 """Re-pushing an identical entry (same content hash) is a true
51 no-op -- it must not duplicate the row."""
52 repo = await create_repo(db_session, visibility="public")
53 entry = _entry_dict("0" * 64, "a" * 64, "c" * 64, timestamp="2026-01-01T00:00:00+00:00")
54
55 await symlog_svc.store_symlog_entries(
56 db_session, str(repo.repo_id), {"billing.py::compute_total": [entry]},
57 )
58 await db_session.commit()
59
60 stored, skipped = await symlog_svc.store_symlog_entries(
61 db_session, str(repo.repo_id), {"billing.py::compute_total": [entry]},
62 )
63 await db_session.commit()
64 assert stored == 0
65 assert skipped == 1
66
67 symbols = await symlog_svc.list_symlog_entries(db_session, str(repo.repo_id))
68 assert len(symbols["billing.py::compute_total"]) == 1
69
70
71 async def test_batched_multi_symbol_push(db_session: AsyncSession) -> None:
72 repo = await create_repo(db_session, visibility="public")
73 stored, skipped = await symlog_svc.store_symlog_entries(
74 db_session, str(repo.repo_id),
75 {
76 "billing.py::compute_total": [_entry_dict("0" * 64, "a" * 64, "c" * 64)],
77 "billing.py::apply_discount": [_entry_dict("0" * 64, "b" * 64, "c" * 64)],
78 },
79 )
80 await db_session.commit()
81 assert stored == 2
82 assert skipped == 0
83
84 symbols = await symlog_svc.list_symlog_entries(db_session, str(repo.repo_id))
85 assert set(symbols.keys()) == {"billing.py::compute_total", "billing.py::apply_discount"}
86
87
88 async def test_entries_isolated_per_symbol(db_session: AsyncSession) -> None:
89 repo = await create_repo(db_session, visibility="public")
90 await symlog_svc.store_symlog_entries(
91 db_session, str(repo.repo_id),
92 {"billing.py::a": [_entry_dict("0" * 64, "a" * 64, "c" * 64)]},
93 )
94 await symlog_svc.store_symlog_entries(
95 db_session, str(repo.repo_id),
96 {"billing.py::b": [_entry_dict("0" * 64, "b" * 64, "c" * 64)]},
97 )
98 await db_session.commit()
99
100 symbols = await symlog_svc.list_symlog_entries(db_session, str(repo.repo_id))
101 assert len(symbols["billing.py::a"]) == 1
102 assert len(symbols["billing.py::b"]) == 1
103 assert symbols["billing.py::a"][0]["new_content_id"] == "a" * 64
104 assert symbols["billing.py::b"][0]["new_content_id"] == "b" * 64
105
106
107 async def test_list_ordered_oldest_first_within_symbol(db_session: AsyncSession) -> None:
108 repo = await create_repo(db_session, visibility="public")
109 await symlog_svc.store_symlog_entries(
110 db_session, str(repo.repo_id),
111 {"billing.py::compute_total": [
112 _entry_dict("0" * 64, "a" * 64, "c1" + "0" * 62, timestamp="2026-01-01T00:00:00+00:00"),
113 _entry_dict("a" * 64, "b" * 64, "c2" + "0" * 62, timestamp="2026-01-02T00:00:00+00:00"),
114 ]},
115 )
116 await db_session.commit()
117
118 symbols = await symlog_svc.list_symlog_entries(db_session, str(repo.repo_id))
119 entries = symbols["billing.py::compute_total"]
120 assert len(entries) == 2
121 assert entries[0]["new_content_id"] == "a" * 64
122 assert entries[1]["new_content_id"] == "b" * 64
123
124
125 async def test_empty_batch_is_noop(db_session: AsyncSession) -> None:
126 repo = await create_repo(db_session, visibility="public")
127 stored, skipped = await symlog_svc.store_symlog_entries(
128 db_session, str(repo.repo_id), {},
129 )
130 assert (stored, skipped) == (0, 0)
131
132
133 async def test_entry_missing_commit_id_is_skipped_silently(db_session: AsyncSession) -> None:
134 repo = await create_repo(db_session, visibility="public")
135 stored, skipped = await symlog_svc.store_symlog_entries(
136 db_session, str(repo.repo_id),
137 {"billing.py::compute_total": [_entry_dict("0" * 64, "a" * 64, "")]},
138 )
139 await db_session.commit()
140 assert stored == 0
141 assert skipped == 0
142
143 symbols = await symlog_svc.list_symlog_entries(db_session, str(repo.repo_id))
144 assert symbols == {}
File History 1 commit
sha256:4fba330283cca0a6bdd067f2c0ad014062fcfda7e98b19ce8cd1e20290252b08 feat: muse#53 item 4 server side — musehub_symlog_entries Sonnet 5 minor ⚠ 2 days ago