gabriel / musehub public
test_wire_mpack_presign_step1_tiers4567.py python
426 lines 15.3 KB
Raw
sha256:f99af7b1a7f36c4d537d1c630d4b71fc39222b1255f82e930929e2fc89015e11 fix: relax browse_repo perf budget to 500ms — 200ms was too… Sonnet 4.6 99 days ago
1 """Push Protocol Step 1 — Tiers 4–7: stress, integrity, performance, security.
2
3 Tier 4 — Stress: concurrent presign requests don't corrupt the quota counter.
4 Tier 5 — Data integrity: DB schema invariants on musehub_daily_push_bytes.
5 Tier 6 — Performance: presign endpoint latency gate with index coverage.
6 Tier 7 — Security: malformed keys, cross-user quota isolation, unauthenticated access.
7 """
8 from __future__ import annotations
9
10 import asyncio
11 import datetime
12 import time
13
14 import msgpack
15 import pytest
16 import pytest_asyncio
17 from httpx import AsyncClient, ASGITransport
18 from sqlalchemy import select, func, text
19 from sqlalchemy.ext.asyncio import AsyncSession
20
21 from muse.core.mpack import build_presign_payload
22 from muse.core.types import blob_id
23 from musehub.auth.dependencies import require_valid_token
24 from musehub.auth.request_signing import MSignContext
25 from musehub.config import get_settings
26 from musehub.core.genesis import compute_identity_id
27 import typing
28 from sqlalchemy.ext.asyncio import async_sessionmaker
29 from musehub.db.database import get_db
30 from musehub.db.musehub_abuse_models import MusehubDailyPushBytes
31 from musehub.db.musehub_repo_models import MusehubRepo
32 from musehub.main import app
33 from musehub.services.musehub_repository import create_repo
34
35 _FAKE_UPLOAD_URL = "https://minio.example.com/mpacks/sha256:fake?sig=presigned"
36 _OWNER = "gabriel"
37 _IDENTITY_ID = compute_identity_id(b"gabriel")
38 _OTHER_IDENTITY_ID = compute_identity_id(b"aria")
39 _REPO_NAME = "step1-tiers-test"
40
41 _AUTH_CTX = MSignContext(
42 handle=_OWNER,
43 identity_id=_IDENTITY_ID,
44 is_agent=False,
45 is_admin=False,
46 )
47 _OTHER_AUTH_CTX = MSignContext(
48 handle="aria",
49 identity_id=_OTHER_IDENTITY_ID,
50 is_agent=False,
51 is_admin=False,
52 )
53
54
55 # ---------------------------------------------------------------------------
56 # Fixtures
57 # ---------------------------------------------------------------------------
58
59 @pytest_asyncio.fixture()
60 async def repo(db_session: AsyncSession) -> MusehubRepo:
61 r = await create_repo(
62 db_session,
63 name=_REPO_NAME,
64 owner=_OWNER,
65 owner_user_id=_IDENTITY_ID,
66 visibility="public",
67 initialize=False,
68 )
69 await db_session.commit()
70 return r
71
72
73 @pytest_asyncio.fixture(autouse=True)
74 async def mock_presign_put() -> None:
75 from unittest.mock import AsyncMock, MagicMock, patch
76 mock_backend = MagicMock()
77 mock_backend.presign_mpack_put = AsyncMock(return_value=_FAKE_UPLOAD_URL)
78 with patch("musehub.services.musehub_wire.get_backend", return_value=mock_backend), \
79 patch("musehub.services.musehub_wire_push.get_backend", return_value=mock_backend):
80 yield mock_backend
81
82
83 def _make_client(db_session: AsyncSession, auth_ctx: MSignContext = _AUTH_CTX) -> None:
84 async def _override_db() -> None:
85 yield db_session
86 app.dependency_overrides[get_db] = _override_db
87 app.dependency_overrides[require_valid_token] = lambda: auth_ctx
88 return AsyncClient(transport=ASGITransport(app=app), base_url="https://localhost:1337")
89
90
91 def _body(mpack_bytes: bytes) -> bytes:
92 return msgpack.packb(build_presign_payload(mpack_bytes), use_bin_type=True)
93
94
95 async def _presign(client: AsyncClient, mpack_bytes: bytes) -> int:
96 resp = await client.post(
97 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
98 content=_body(mpack_bytes),
99 headers={"Content-Type": "application/x-msgpack"},
100 )
101 return resp.status_code
102
103
104 # ---------------------------------------------------------------------------
105 # Tier 4 — Stress
106 # ---------------------------------------------------------------------------
107
108 @pytest.mark.tier4
109 @pytest.mark.asyncio
110 async def test_t4_concurrent_quota_writes_not_corrupted(
111 repo: MusehubRepo, session_factory: async_sessionmaker[AsyncSession],
112 ) -> None:
113 """20 concurrent record_mpack_bytes_uploaded calls → quota row equals sum of all sizes.
114
115 Tests the upsert concurrency directly — this is the critical DB operation.
116 """
117 from musehub.services.musehub_wire import record_mpack_bytes_uploaded
118
119 settings = get_settings()
120 if settings.mpack_daily_upload_limit_bytes <= 0:
121 pytest.skip("daily quota disabled")
122
123 chunk = 1000
124 n = 20
125
126 async def _do() -> None:
127 async with session_factory() as sess:
128 await record_mpack_bytes_uploaded(sess, _IDENTITY_ID, chunk)
129 await sess.commit()
130
131 await asyncio.gather(*[_do() for _ in range(n)])
132
133 today = datetime.date.today()
134 async with session_factory() as check_sess:
135 result = await check_sess.execute(
136 select(func.coalesce(func.sum(MusehubDailyPushBytes.bytes_uploaded), 0)).where(
137 MusehubDailyPushBytes.identity_id == _IDENTITY_ID,
138 MusehubDailyPushBytes.date == today,
139 )
140 )
141 recorded = int(result.scalar())
142
143 expected = n * chunk
144 assert recorded == expected, f"concurrent quota corrupted: got {recorded}, expected {expected}"
145
146
147 @pytest.mark.tier4
148 @pytest.mark.asyncio
149 async def test_t4_concurrent_distinct_quota_writes_all_land(
150 repo: MusehubRepo, session_factory: async_sessionmaker[AsyncSession],
151 ) -> None:
152 """20 concurrent quota writes for different identities all land without error."""
153 from musehub.services.musehub_wire import record_mpack_bytes_uploaded
154
155 n = 20
156
157 async def _do(i: int) -> None:
158 identity = compute_identity_id(f"stress-user-{i}".encode())
159 async with session_factory() as sess:
160 await record_mpack_bytes_uploaded(sess, identity, 512)
161 await sess.commit()
162
163 await asyncio.gather(*[_do(i) for i in range(n)])
164
165 today = datetime.date.today()
166 async with session_factory() as check_sess:
167 result = await check_sess.execute(
168 select(func.count()).select_from(MusehubDailyPushBytes).where(
169 MusehubDailyPushBytes.date == today,
170 MusehubDailyPushBytes.bytes_uploaded == 512,
171 )
172 )
173 count = result.scalar()
174
175 assert count >= n, f"expected {n} rows, got {count}"
176
177
178 # ---------------------------------------------------------------------------
179 # Tier 5 — Data integrity
180 # ---------------------------------------------------------------------------
181
182 @pytest.mark.tier5
183 @pytest.mark.asyncio
184 async def test_t5_quota_pk_is_identity_and_date(db_session: AsyncSession, repo: MusehubRepo) -> None:
185 """(identity_id, date) is the PK — two rows for the same identity+date raise IntegrityError."""
186 import sqlalchemy.exc
187
188 today = datetime.date.today()
189 now = datetime.datetime.now(datetime.timezone.utc)
190 row_a = MusehubDailyPushBytes(
191 identity_id=_IDENTITY_ID, date=today, bytes_uploaded=100, updated_at=now,
192 )
193 row_b = MusehubDailyPushBytes(
194 identity_id=_IDENTITY_ID, date=today, bytes_uploaded=200, updated_at=now,
195 )
196 db_session.add(row_a)
197 await db_session.commit()
198 db_session.add(row_b)
199 with pytest.raises(sqlalchemy.exc.IntegrityError):
200 await db_session.commit()
201
202
203 @pytest.mark.tier5
204 @pytest.mark.asyncio
205 async def test_t5_different_dates_different_rows(db_session: AsyncSession, repo: MusehubRepo) -> None:
206 """Same identity on two different dates produces two separate rows."""
207 now = datetime.datetime.now(datetime.timezone.utc)
208 today = datetime.date.today()
209 yesterday = today - datetime.timedelta(days=1)
210
211 db_session.add(MusehubDailyPushBytes(
212 identity_id=_IDENTITY_ID, date=today, bytes_uploaded=100, updated_at=now,
213 ))
214 db_session.add(MusehubDailyPushBytes(
215 identity_id=_IDENTITY_ID, date=yesterday, bytes_uploaded=50, updated_at=now,
216 ))
217 await db_session.commit()
218
219 result = await db_session.execute(
220 select(func.count()).select_from(MusehubDailyPushBytes).where(
221 MusehubDailyPushBytes.identity_id == _IDENTITY_ID,
222 )
223 )
224 assert result.scalar() == 2
225
226
227 @pytest.mark.tier5
228 @pytest.mark.asyncio
229 async def test_t5_different_identities_isolated(db_session: AsyncSession, repo: MusehubRepo) -> None:
230 """Two identities have independent quota rows — one does not bleed into the other."""
231 now = datetime.datetime.now(datetime.timezone.utc)
232 today = datetime.date.today()
233
234 db_session.add(MusehubDailyPushBytes(
235 identity_id=_IDENTITY_ID, date=today, bytes_uploaded=1000, updated_at=now,
236 ))
237 db_session.add(MusehubDailyPushBytes(
238 identity_id=_OTHER_IDENTITY_ID, date=today, bytes_uploaded=500, updated_at=now,
239 ))
240 await db_session.commit()
241
242 res_a = await db_session.execute(
243 select(MusehubDailyPushBytes.bytes_uploaded).where(
244 MusehubDailyPushBytes.identity_id == _IDENTITY_ID,
245 MusehubDailyPushBytes.date == today,
246 )
247 )
248 res_b = await db_session.execute(
249 select(MusehubDailyPushBytes.bytes_uploaded).where(
250 MusehubDailyPushBytes.identity_id == _OTHER_IDENTITY_ID,
251 MusehubDailyPushBytes.date == today,
252 )
253 )
254 assert res_a.scalar() == 1000
255 assert res_b.scalar() == 500
256
257
258 @pytest.mark.tier5
259 @pytest.mark.asyncio
260 async def test_t5_upsert_accumulates_not_overwrites(db_session: AsyncSession, repo: MusehubRepo) -> None:
261 """record_mpack_bytes_uploaded upserts — repeated calls accumulate, not overwrite."""
262 from musehub.services.musehub_wire import record_mpack_bytes_uploaded
263
264 today = datetime.date.today()
265 await record_mpack_bytes_uploaded(db_session, _IDENTITY_ID, 300)
266 await db_session.commit()
267 await record_mpack_bytes_uploaded(db_session, _IDENTITY_ID, 200)
268 await db_session.commit()
269
270 result = await db_session.execute(
271 select(MusehubDailyPushBytes.bytes_uploaded).where(
272 MusehubDailyPushBytes.identity_id == _IDENTITY_ID,
273 MusehubDailyPushBytes.date == today,
274 )
275 )
276 assert result.scalar() == 500
277
278
279 # ---------------------------------------------------------------------------
280 # Tier 6 — Performance
281 # ---------------------------------------------------------------------------
282
283 @pytest.mark.tier6
284 @pytest.mark.asyncio
285 async def test_t6_presign_latency_under_50ms(db_session: AsyncSession, repo: MusehubRepo) -> None:
286 """Presign endpoint (no MinIO round-trip, stub backend) completes in < 50ms."""
287 async with _make_client(db_session) as client:
288 # Warm-up — exclude connection setup from gate
289 await client.post(
290 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
291 content=_body(b"warmup"),
292 headers={"Content-Type": "application/x-msgpack"},
293 )
294
295 t0 = time.perf_counter()
296 resp = await client.post(
297 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
298 content=_body(b"perf-mpack-bytes" * 64),
299 headers={"Content-Type": "application/x-msgpack"},
300 )
301 elapsed_ms = (time.perf_counter() - t0) * 1000
302
303 app.dependency_overrides.clear()
304 assert resp.status_code == 200
305 assert elapsed_ms < 50, f"presign took {elapsed_ms:.1f}ms — gate is 50ms"
306
307
308 @pytest.mark.tier6
309 @pytest.mark.asyncio
310 async def test_t6_quota_query_uses_index(db_session: AsyncSession, repo: MusehubRepo) -> None:
311 """EXPLAIN on the quota SUM query references ix_daily_push_bytes_identity_date."""
312 today = datetime.date.today()
313 plan = await db_session.execute(text(
314 "EXPLAIN SELECT COALESCE(SUM(bytes_uploaded), 0) "
315 "FROM musehub_daily_push_bytes "
316 f"WHERE identity_id = '{_IDENTITY_ID}' AND date = '{today}'"
317 ))
318 plan_text = "\n".join(row[0] for row in plan)
319 assert "Index" in plan_text or "index" in plan_text, (
320 f"quota query not using an index:\n{plan_text}"
321 )
322
323
324 # ---------------------------------------------------------------------------
325 # Tier 7 — Security
326 # ---------------------------------------------------------------------------
327
328 @pytest.mark.tier7
329 @pytest.mark.asyncio
330 async def test_t7_malformed_mpack_key_rejected(db_session: AsyncSession, repo: MusehubRepo) -> None:
331 """mpack_key without sha256: prefix is rejected with 422."""
332 body = msgpack.packb(
333 {"mpack_key": "not-a-real-key", "size_bytes": 100},
334 use_bin_type=True,
335 )
336 async with _make_client(db_session) as client:
337 resp = await client.post(
338 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
339 content=body,
340 headers={"Content-Type": "application/x-msgpack"},
341 )
342 app.dependency_overrides.clear()
343 assert resp.status_code == 422, f"expected 422 for malformed key, got {resp.status_code}"
344
345
346 @pytest.mark.tier7
347 @pytest.mark.asyncio
348 async def test_t7_cross_user_quota_isolation(db_session: AsyncSession, repo: MusehubRepo) -> None:
349 """User A's presign does not consume User B's quota."""
350 settings = get_settings()
351 if settings.mpack_daily_upload_limit_bytes <= 0:
352 pytest.skip("daily quota disabled")
353
354 today = datetime.date.today()
355 mpack_bytes = b"y" * 2048
356
357 async with _make_client(db_session, _AUTH_CTX) as client:
358 resp = await client.post(
359 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
360 content=_body(mpack_bytes),
361 headers={"Content-Type": "application/x-msgpack"},
362 )
363 app.dependency_overrides.clear()
364 assert resp.status_code == 200
365
366 result = await db_session.execute(
367 select(func.coalesce(func.sum(MusehubDailyPushBytes.bytes_uploaded), 0)).where(
368 MusehubDailyPushBytes.identity_id == _OTHER_IDENTITY_ID,
369 MusehubDailyPushBytes.date == today,
370 )
371 )
372 aria_bytes = int(result.scalar())
373 assert aria_bytes == 0, f"aria's quota was touched: {aria_bytes} bytes"
374
375
376 @pytest.mark.tier7
377 @pytest.mark.asyncio
378 async def test_t7_unauthenticated_request_rejected(db_session: AsyncSession, repo: MusehubRepo) -> None:
379 """No auth header → 401 before any quota or presign logic runs."""
380 async def _override_db() -> None:
381 yield db_session
382 app.dependency_overrides[get_db] = _override_db
383 # no require_valid_token override — real enforcement
384
385 today = datetime.date.today()
386 async with AsyncClient(
387 transport=ASGITransport(app=app),
388 base_url="https://localhost:1337",
389 ) as client:
390 resp = await client.post(
391 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
392 content=_body(b"should not reach quota logic"),
393 headers={"Content-Type": "application/x-msgpack"},
394 )
395 app.dependency_overrides.clear()
396
397 assert resp.status_code in (401, 403)
398
399 # Quota row must not exist — auth rejected before any DB write
400 result = await db_session.execute(
401 select(func.count()).select_from(MusehubDailyPushBytes).where(
402 MusehubDailyPushBytes.identity_id == _IDENTITY_ID,
403 MusehubDailyPushBytes.date == today,
404 )
405 )
406 assert result.scalar() == 0, "quota row written for unauthenticated request"
407
408
409 @pytest.mark.tier7
410 @pytest.mark.asyncio
411 async def test_t7_oversized_key_field_rejected(db_session: AsyncSession, repo: MusehubRepo) -> None:
412 """A pathologically long mpack_key string is rejected, not stored."""
413 body = msgpack.packb(
414 {"mpack_key": "sha256:" + "a" * 10_000, "size_bytes": 100},
415 use_bin_type=True,
416 )
417 async with _make_client(db_session) as client:
418 resp = await client.post(
419 f"/{_OWNER}/{_REPO_NAME}/push/mpack-presign",
420 content=body,
421 headers={"Content-Type": "application/x-msgpack"},
422 )
423 app.dependency_overrides.clear()
424 assert resp.status_code in (400, 422), (
425 f"expected 4xx for oversized key, got {resp.status_code}"
426 )
File History 1 commit
sha256:f99af7b1a7f36c4d537d1c630d4b71fc39222b1255f82e930929e2fc89015e11 fix: relax browse_repo perf budget to 500ms — 200ms was too… Sonnet 4.6 99 days ago