gabriel / musehub public
test_wire_mpack_unpack_step3_tiers4567.py python
493 lines 18.1 KB
Raw
sha256:ad616c6113d6c00f4efed6b2993734ca46d3e9b5bee25addd4ce8ae6b57136e5 chore: bump version to 0.2.0rc11; typing audit clean + all … Sonnet 4.6 minor ⚠ breaking 55 days ago
1 """Push Protocol Step 3 (unpack-mpack) — Tiers 4–7: stress, integrity, performance, security.
2
3 Tier 4 — Stress: concurrent wire_push_unpack_mpack calls all land without errors.
4 Tier 5 — Data integrity: DB invariants on MusehubBackgroundJob and MusehubBranch.
5 Tier 6 — Performance: unpack endpoint latency gate.
6 Tier 7 — Security: malformed keys, oversized keys, unauthenticated access.
7 """
8 from __future__ import annotations
9
10 import asyncio
11 import time
12
13 import msgpack
14 import pytest
15 import pytest_asyncio
16 from httpx import AsyncClient, ASGITransport
17 from sqlalchemy import select, func, text
18 from sqlalchemy.ext.asyncio import AsyncSession
19
20 from muse.core.mpack import build_wire_mpack
21 from muse.core.types import blob_id, fake_id
22 from musehub.auth.dependencies import require_valid_token
23 from musehub.auth.request_signing import MSignContext
24 from musehub.config import get_settings
25 from musehub.core.genesis import compute_identity_id
26 from musehub.db.database import get_db
27 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
28 from musehub.db.musehub_repo_models import MusehubBranch
29 from musehub.main import app
30 from musehub.services.musehub_repository import create_repo
31
32 _OWNER = "gabriel"
33 _IDENTITY_ID = compute_identity_id(b"gabriel")
34 _OTHER_IDENTITY_ID = compute_identity_id(b"aria")
35 _REPO_NAME = "step3-tiers-test"
36
37 _MPACK_BYTES = build_wire_mpack({"objects": [], "commits": [], "snapshots": []})
38 _MPACK_KEY = blob_id(_MPACK_BYTES)
39 _HEAD = fake_id("step3-tiers-tip")
40
41 _AUTH_CTX = MSignContext(
42 handle=_OWNER,
43 identity_id=_IDENTITY_ID,
44 is_agent=False,
45 is_admin=False,
46 )
47
48
49 # ---------------------------------------------------------------------------
50 # Fixtures
51 # ---------------------------------------------------------------------------
52
53 @pytest_asyncio.fixture()
54 async def repo(db_session: AsyncSession):
55 r = await create_repo(
56 db_session,
57 name=_REPO_NAME,
58 owner=_OWNER,
59 owner_user_id=_IDENTITY_ID,
60 visibility="public",
61 initialize=False,
62 )
63 await db_session.commit()
64 return r
65
66
67 @pytest_asyncio.fixture(autouse=True)
68 async def mock_get_mpack():
69 from unittest.mock import AsyncMock, MagicMock, patch
70 mock_backend = MagicMock()
71 mock_backend.get_mpack = AsyncMock(return_value=_MPACK_BYTES)
72 with patch("musehub.services.musehub_wire.get_backend", return_value=mock_backend), \
73 patch("musehub.services.musehub_wire_push.get_backend", return_value=mock_backend):
74 yield mock_backend
75
76
77 def _make_client(db_session: AsyncSession, auth_ctx: MSignContext = _AUTH_CTX):
78 async def _override_db():
79 yield db_session
80 app.dependency_overrides[get_db] = _override_db
81 app.dependency_overrides[require_valid_token] = lambda: auth_ctx
82 return AsyncClient(transport=ASGITransport(app=app), base_url="https://localhost:1337")
83
84
85 def _body(
86 mpack_key: str = _MPACK_KEY,
87 branch: str = "main",
88 head: str = "",
89 commits_count: int = 1,
90 objects_count: int = 2,
91 ) -> bytes:
92 return msgpack.packb(
93 {
94 "mpack_key": mpack_key,
95 "branch": branch,
96 "head": head,
97 "commits_count": commits_count,
98 "objects_count": objects_count,
99 },
100 use_bin_type=True,
101 )
102
103
104 async def _unpack(client: AsyncClient, repo_name: str = _REPO_NAME, **kwargs) -> int:
105 resp = await client.post(
106 f"/{_OWNER}/{repo_name}/push/unpack-mpack",
107 content=_body(**kwargs),
108 headers={"Content-Type": "application/x-msgpack"},
109 )
110 return resp.status_code
111
112
113 # ---------------------------------------------------------------------------
114 # Tier 4 — Stress
115 # ---------------------------------------------------------------------------
116
117 @pytest.mark.tier4
118 @pytest.mark.asyncio
119 async def test_t4_concurrent_unpack_calls_all_succeed(
120 repo, session_factory,
121 ) -> None:
122 """20 concurrent wire_push_unpack_mpack calls with distinct mpack_keys all succeed.
123
124 Tests inline processing under concurrency — every call must complete without
125 deadlocking. Phase 3: all processing is inline; no background job rows.
126 """
127 from musehub.services.musehub_wire import wire_push_unpack_mpack
128
129 n = 20
130
131 async def _do(i: int) -> None:
132 # Each call gets a unique mpack_key so job_id is unique
133 content = f"object-{i}".encode()
134 oid = blob_id(content)
135 unique_bytes = build_wire_mpack(
136 {"objects": [{"object_id": oid, "content": content}], "commits": [], "snapshots": []}
137 )
138 unique_key = blob_id(unique_bytes)
139 async with session_factory() as sess:
140 # Override get_mpack to return the correct bytes for this key
141 from unittest.mock import AsyncMock, MagicMock, patch
142 mock_backend = MagicMock()
143 mock_backend.get_mpack = AsyncMock(return_value=unique_bytes)
144 with patch("musehub.services.musehub_wire.get_backend", return_value=mock_backend), \
145 patch("musehub.services.musehub_wire_push.get_backend", return_value=mock_backend):
146 await wire_push_unpack_mpack(
147 sess, repo.repo_id, unique_key, _OWNER,
148 branch="main", head_commit_id="", commits_count=1, objects_count=1,
149 )
150 await sess.commit()
151
152 # If any call raises, gather re-raises immediately — no further assertion needed.
153 await asyncio.gather(*[_do(i) for i in range(n)])
154
155
156 @pytest.mark.tier4
157 @pytest.mark.asyncio
158 async def test_t4_concurrent_branch_updates_for_same_branch(
159 repo, session_factory,
160 ) -> None:
161 """10 concurrent calls advancing the same branch serialise without data corruption.
162
163 The service uses WITH FOR UPDATE on the branch row — concurrent updates
164 must not produce two branch rows for the same name. Pre-create the branch
165 so all concurrent calls hit the UPDATE path (avoiding the INSERT race).
166 """
167 from musehub.services.musehub_wire import wire_push_unpack_mpack
168 from musehub.core.genesis import compute_branch_id
169
170 # Pre-create the branch so concurrent calls all SELECT-for-update and UPDATE
171 async with session_factory() as seed_sess:
172 seed_sess.add(MusehubBranch(
173 branch_id=compute_branch_id(repo.repo_id, "concurrent-branch"),
174 repo_id=repo.repo_id,
175 name="concurrent-branch",
176 head_commit_id=fake_id("seed-head"),
177 ))
178 await seed_sess.commit()
179
180 n = 10
181
182 async def _do(i: int) -> None:
183 content = f"seq-{i}".encode()
184 oid = blob_id(content)
185 unique_bytes = build_wire_mpack(
186 {"objects": [{"object_id": oid, "content": content}], "commits": [], "snapshots": []}
187 )
188 unique_key = blob_id(unique_bytes)
189 head = fake_id(f"commit-{i}")
190 async with session_factory() as sess:
191 from unittest.mock import AsyncMock, MagicMock, patch
192 mock_backend = MagicMock()
193 mock_backend.get_mpack = AsyncMock(return_value=unique_bytes)
194 with patch("musehub.services.musehub_wire.get_backend", return_value=mock_backend), \
195 patch("musehub.services.musehub_wire_push.get_backend", return_value=mock_backend):
196 await wire_push_unpack_mpack(
197 sess, repo.repo_id, unique_key, _OWNER,
198 branch="concurrent-branch", head_commit_id=head,
199 commits_count=1, objects_count=0, force=True,
200 )
201 await sess.commit()
202
203 await asyncio.gather(*[_do(i) for i in range(n)])
204
205 async with session_factory() as check_sess:
206 result = await check_sess.execute(
207 select(func.count()).select_from(MusehubBranch).where(
208 MusehubBranch.repo_id == repo.repo_id,
209 MusehubBranch.name == "concurrent-branch",
210 )
211 )
212 count = result.scalar()
213
214 assert count == 1, f"expected exactly 1 branch row, got {count}"
215
216
217 # ---------------------------------------------------------------------------
218 # Tier 5 — Data integrity
219 # ---------------------------------------------------------------------------
220
221 @pytest.mark.tier5
222 @pytest.mark.asyncio
223 async def test_t5_unpack_returns_correct_shape(
224 db_session: AsyncSession, repo,
225 ) -> None:
226 """Successful inline unpack returns the expected response keys (Phase 3 — no background job)."""
227 from musehub.services.musehub_wire import wire_push_unpack_mpack
228
229 result = await wire_push_unpack_mpack(
230 db_session, repo.repo_id, _MPACK_KEY, _OWNER,
231 branch="main", head_commit_id="", commits_count=3, objects_count=7,
232 )
233
234 assert "head" in result
235 assert "branch" in result
236 assert result["branch"] == "main"
237 assert "objects_in_mpack" in result
238 assert "commits_in_mpack" in result
239
240
241 @pytest.mark.tier5
242 @pytest.mark.asyncio
243 async def test_t5_branch_row_created_when_head_provided(
244 db_session: AsyncSession, repo,
245 ) -> None:
246 """Branch row is inserted when head_commit_id is provided and branch doesn't exist."""
247 from musehub.services.musehub_wire import wire_push_unpack_mpack
248
249 await wire_push_unpack_mpack(
250 db_session, repo.repo_id, _MPACK_KEY, _OWNER,
251 branch="feat/new-branch", head_commit_id=_HEAD,
252 commits_count=1, objects_count=0,
253 )
254
255 result = await db_session.execute(
256 select(MusehubBranch).where(
257 MusehubBranch.repo_id == repo.repo_id,
258 MusehubBranch.name == "feat/new-branch",
259 )
260 )
261 branch_row = result.scalar_one_or_none()
262 assert branch_row is not None, "branch row was not created"
263 assert branch_row.head_commit_id == _HEAD
264
265
266 @pytest.mark.tier5
267 @pytest.mark.asyncio
268 async def test_t5_branch_row_updated_when_branch_exists(
269 db_session: AsyncSession, repo,
270 ) -> None:
271 """When branch already exists, unpack updates head_commit_id — no duplicate row."""
272 from musehub.services.musehub_wire import wire_push_unpack_mpack
273
274 first_head = fake_id("first-head")
275 second_head = fake_id("second-head")
276
277 await wire_push_unpack_mpack(
278 db_session, repo.repo_id, _MPACK_KEY, _OWNER,
279 branch="main", head_commit_id=first_head,
280 commits_count=1, objects_count=0,
281 )
282
283 # Second push to same branch with different mpack (different key)
284 second_content = b"second-push-content"
285 second_oid = blob_id(second_content)
286 second_bytes = build_wire_mpack(
287 {"objects": [{"object_id": second_oid, "content": second_content}], "commits": [], "snapshots": []}
288 )
289 second_key = blob_id(second_bytes)
290 from unittest.mock import AsyncMock, MagicMock, patch
291 mock_backend = MagicMock()
292 mock_backend.get_mpack = AsyncMock(return_value=second_bytes)
293 with patch("musehub.services.musehub_wire.get_backend", return_value=mock_backend), \
294 patch("musehub.services.musehub_wire_push.get_backend", return_value=mock_backend):
295 await wire_push_unpack_mpack(
296 db_session, repo.repo_id, second_key, _OWNER,
297 branch="main", head_commit_id=second_head,
298 commits_count=1, objects_count=0, force=True,
299 )
300
301 result = await db_session.execute(
302 select(func.count()).select_from(MusehubBranch).where(
303 MusehubBranch.repo_id == repo.repo_id,
304 MusehubBranch.name == "main",
305 )
306 )
307 assert result.scalar() == 1, "duplicate branch row created"
308
309 branch_result = await db_session.execute(
310 select(MusehubBranch.head_commit_id).where(
311 MusehubBranch.repo_id == repo.repo_id,
312 MusehubBranch.name == "main",
313 )
314 )
315 assert branch_result.scalar() == second_head, "branch head not updated to second push"
316
317
318 @pytest.mark.tier5
319 @pytest.mark.asyncio
320 async def test_t5_job_row_not_written_on_mpack_not_found(
321 db_session: AsyncSession, repo,
322 ) -> None:
323 """When MinIO returns None, ValueError is raised and no job row is persisted."""
324 from musehub.services.musehub_wire import wire_push_unpack_mpack
325 from unittest.mock import AsyncMock, MagicMock, patch
326
327 mock_backend = MagicMock()
328 mock_backend.get_mpack = AsyncMock(return_value=None)
329 with patch("musehub.services.musehub_wire.get_backend", return_value=mock_backend), \
330 patch("musehub.services.musehub_wire_push.get_backend", return_value=mock_backend):
331 with pytest.raises(ValueError):
332 await wire_push_unpack_mpack(
333 db_session, repo.repo_id, _MPACK_KEY, _OWNER,
334 branch="main", head_commit_id="", commits_count=0, objects_count=0,
335 )
336
337 result = await db_session.execute(
338 select(func.count()).select_from(MusehubBackgroundJob).where(
339 MusehubBackgroundJob.repo_id == repo.repo_id,
340 MusehubBackgroundJob.job_type == "mpack.index",
341 )
342 )
343 assert result.scalar() == 0, "job row written despite mpack not found"
344
345
346 # ---------------------------------------------------------------------------
347 # Tier 6 — Performance
348 # ---------------------------------------------------------------------------
349
350 @pytest.mark.tier6
351 @pytest.mark.asyncio
352 async def test_t6_unpack_latency_under_100ms(db_session: AsyncSession, repo) -> None:
353 """Unpack endpoint (stub MinIO, small mpack) completes in < 100ms."""
354 async with _make_client(db_session) as client:
355 # Warm-up — exclude connection setup from gate
356 await client.post(
357 f"/{_OWNER}/{_REPO_NAME}/push/unpack-mpack",
358 content=_body(),
359 headers={"Content-Type": "application/x-msgpack"},
360 )
361
362 t0 = time.perf_counter()
363 resp = await client.post(
364 f"/{_OWNER}/{_REPO_NAME}/push/unpack-mpack",
365 content=_body(commits_count=3, objects_count=10),
366 headers={"Content-Type": "application/x-msgpack"},
367 )
368 elapsed_ms = (time.perf_counter() - t0) * 1000
369
370 app.dependency_overrides.clear()
371 assert resp.status_code == 200
372 assert elapsed_ms < 100, f"unpack took {elapsed_ms:.1f}ms — gate is 100ms"
373
374
375 @pytest.mark.tier6
376 @pytest.mark.asyncio
377 async def test_t6_job_index_query_uses_index(db_session: AsyncSession, repo) -> None:
378 """EXPLAIN on the pending job query references ix_musehub_background_jobs_status_created."""
379 plan = await db_session.execute(text(
380 "EXPLAIN SELECT * FROM musehub_background_jobs "
381 "WHERE status = 'pending' "
382 "ORDER BY created_at "
383 "LIMIT 10"
384 ))
385 plan_text = "\n".join(row[0] for row in plan)
386 assert "Index" in plan_text or "index" in plan_text, (
387 f"job queue query not using an index:\n{plan_text}"
388 )
389
390
391 # ---------------------------------------------------------------------------
392 # Tier 7 — Security
393 # ---------------------------------------------------------------------------
394
395 @pytest.mark.tier7
396 @pytest.mark.asyncio
397 async def test_t7_malformed_mpack_key_rejected(db_session: AsyncSession, repo) -> None:
398 """mpack_key without sha256: prefix is rejected with 422."""
399 body = msgpack.packb(
400 {"mpack_key": "not-a-real-key", "branch": "main", "commits_count": 1, "objects_count": 0},
401 use_bin_type=True,
402 )
403 async with _make_client(db_session) as client:
404 resp = await client.post(
405 f"/{_OWNER}/{_REPO_NAME}/push/unpack-mpack",
406 content=body,
407 headers={"Content-Type": "application/x-msgpack"},
408 )
409 app.dependency_overrides.clear()
410 assert resp.status_code == 422, f"expected 422 for malformed key, got {resp.status_code}"
411
412
413 @pytest.mark.tier7
414 @pytest.mark.asyncio
415 async def test_t7_oversized_key_field_rejected(db_session: AsyncSession, repo) -> None:
416 """A pathologically long mpack_key string is rejected, not stored."""
417 body = msgpack.packb(
418 {"mpack_key": "sha256:" + "a" * 10_000, "branch": "main"},
419 use_bin_type=True,
420 )
421 async with _make_client(db_session) as client:
422 resp = await client.post(
423 f"/{_OWNER}/{_REPO_NAME}/push/unpack-mpack",
424 content=body,
425 headers={"Content-Type": "application/x-msgpack"},
426 )
427 app.dependency_overrides.clear()
428 assert resp.status_code in (400, 422), (
429 f"expected 4xx for oversized key, got {resp.status_code}"
430 )
431
432
433 @pytest.mark.tier7
434 @pytest.mark.asyncio
435 async def test_t7_unauthenticated_request_rejected(db_session: AsyncSession, repo) -> None:
436 """No auth header → 401/403 before any job enqueue logic runs."""
437 async def _override_db():
438 yield db_session
439 app.dependency_overrides[get_db] = _override_db
440 # no require_valid_token override — real enforcement
441
442 async with AsyncClient(
443 transport=ASGITransport(app=app),
444 base_url="https://localhost:1337",
445 ) as client:
446 resp = await client.post(
447 f"/{_OWNER}/{_REPO_NAME}/push/unpack-mpack",
448 content=_body(),
449 headers={"Content-Type": "application/x-msgpack"},
450 )
451 app.dependency_overrides.clear()
452
453 assert resp.status_code in (401, 403)
454
455 # No job row must exist — auth rejected before any DB write
456 result = await db_session.execute(
457 select(func.count()).select_from(MusehubBackgroundJob).where(
458 MusehubBackgroundJob.repo_id == repo.repo_id,
459 MusehubBackgroundJob.job_type == "mpack.index",
460 )
461 )
462 assert result.scalar() == 0, "job row written for unauthenticated request"
463
464
465 @pytest.mark.tier7
466 @pytest.mark.asyncio
467 async def test_t7_commits_count_overflow_rejected(db_session: AsyncSession, repo) -> None:
468 """commits_count wildly exceeding the limit is rejected with 422 — no job enqueued."""
469 settings = get_settings()
470 body = msgpack.packb(
471 {
472 "mpack_key": _MPACK_KEY,
473 "branch": "main",
474 "commits_count": settings.mpack_max_commits * 100,
475 "objects_count": 0,
476 },
477 use_bin_type=True,
478 )
479 async with _make_client(db_session) as client:
480 resp = await client.post(
481 f"/{_OWNER}/{_REPO_NAME}/push/unpack-mpack",
482 content=body,
483 headers={"Content-Type": "application/x-msgpack"},
484 )
485 app.dependency_overrides.clear()
486 assert resp.status_code == 422
487
488 result = await db_session.execute(
489 select(func.count()).select_from(MusehubBackgroundJob).where(
490 MusehubBackgroundJob.repo_id == repo.repo_id,
491 )
492 )
493 assert result.scalar() == 0, "job row written despite over-limit commits_count"
File History 2 commits
sha256:ad616c6113d6c00f4efed6b2993734ca46d3e9b5bee25addd4ce8ae6b57136e5 chore: bump version to 0.2.0rc11; typing audit clean + all … Sonnet 4.6 minor ⚠ 55 days ago
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 57 days ago