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