"""Phase 7 TDD: Rate limiting audit for /api/mists/* endpoints. Tests are written RED first. Run before touching mists.py and rate_limits.py to confirm failure, then implement. Gap: - POST /api/mists and POST /api/mists/{id}/fork already have @limiter.limit. - PATCH /api/mists/{id} and DELETE /api/mists/{id} have NO rate limit decorator. - GET read endpoints (explore, get, forks, owner list, embed) have no per-route limit — they fall through to the global 300/min IP bucket, which is too loose. Work: 1. Add MIST_UPDATE_LIMIT and MIST_DELETE_LIMIT to rate_limits.py (keyed on MSign handle like create/fork). 2. Add MIST_READ_LIMIT to rate_limits.py (keyed on IP, read-only endpoints). 3. Decorate update_mist and delete_mist with @limiter.limit(..., key_func=get_msign_handle). 4. Decorate all GET handlers with @limiter.limit(MIST_READ_LIMIT). """ from __future__ import annotations import pytest from httpx import AsyncClient # --------------------------------------------------------------------------- # 1. Rate limit constants defined in rate_limits.py # --------------------------------------------------------------------------- class TestRateLimitConstants: def test_mist_update_limit_defined(self) -> None: """MIST_UPDATE_LIMIT must be defined in rate_limits.""" from musehub.rate_limits import MIST_UPDATE_LIMIT assert isinstance(MIST_UPDATE_LIMIT, str) assert "/" in MIST_UPDATE_LIMIT, ( f"MIST_UPDATE_LIMIT must be a rate string like '20/minute'; got {MIST_UPDATE_LIMIT!r}" ) def test_mist_delete_limit_defined(self) -> None: """MIST_DELETE_LIMIT must be defined in rate_limits.""" from musehub.rate_limits import MIST_DELETE_LIMIT assert isinstance(MIST_DELETE_LIMIT, str) assert "/" in MIST_DELETE_LIMIT def test_mist_read_limit_defined(self) -> None: """MIST_READ_LIMIT must be defined in rate_limits.""" from musehub.rate_limits import MIST_READ_LIMIT assert isinstance(MIST_READ_LIMIT, str) assert "/" in MIST_READ_LIMIT # --------------------------------------------------------------------------- # 2. Mutating endpoints have @limiter.limit decorators (source inspection) # --------------------------------------------------------------------------- class TestMutatingEndpointsHaveLimiters: def test_update_mist_has_limiter_decorator(self) -> None: """update_mist must have a @limiter.limit(...) decorator in mists.py.""" import inspect from musehub.api.routes.musehub import mists as _mod src = inspect.getsource(_mod.update_mist) # The decorator is applied before the function — check the module source # around the function definition to catch the decorator above it. full_src = inspect.getsource(_mod) # Find the update_mist function and check for limiter.limit above it. update_idx = full_src.find("async def update_mist(") assert update_idx != -1 # Look at the 500 chars before the function definition for the decorator. preamble = full_src[max(0, update_idx - 500): update_idx] assert "limiter.limit" in preamble, ( "update_mist must be decorated with @limiter.limit(...); " "no limiter.limit found in the 500 chars before 'async def update_mist'" ) def test_delete_mist_has_limiter_decorator(self) -> None: """delete_mist must have a @limiter.limit(...) decorator in mists.py.""" import inspect from musehub.api.routes.musehub import mists as _mod full_src = inspect.getsource(_mod) delete_idx = full_src.find("async def delete_mist(") assert delete_idx != -1 preamble = full_src[max(0, delete_idx - 500): delete_idx] assert "limiter.limit" in preamble, ( "delete_mist must be decorated with @limiter.limit(...); " "no limiter.limit found in the 500 chars before 'async def delete_mist'" ) def test_update_mist_uses_msign_key_func(self) -> None: """update_mist rate limiter must key on MSign handle, not IP.""" import inspect from musehub.api.routes.musehub import mists as _mod full_src = inspect.getsource(_mod) update_idx = full_src.find("async def update_mist(") preamble = full_src[max(0, update_idx - 500): update_idx] assert "get_msign_handle" in preamble, ( "update_mist limiter must use key_func=get_msign_handle" ) def test_delete_mist_uses_msign_key_func(self) -> None: """delete_mist rate limiter must key on MSign handle, not IP.""" import inspect from musehub.api.routes.musehub import mists as _mod full_src = inspect.getsource(_mod) delete_idx = full_src.find("async def delete_mist(") preamble = full_src[max(0, delete_idx - 500): delete_idx] assert "get_msign_handle" in preamble, ( "delete_mist limiter must use key_func=get_msign_handle" ) # --------------------------------------------------------------------------- # 3. Read endpoints have @limiter.limit decorators # --------------------------------------------------------------------------- class TestReadEndpointsHaveLimiters: def _check_fn_has_limiter(self, fn_name: str) -> None: import inspect from musehub.api.routes.musehub import mists as _mod full_src = inspect.getsource(_mod) fn_idx = full_src.find(f"async def {fn_name}(") assert fn_idx != -1, f"Could not find 'async def {fn_name}(' in mists.py" preamble = full_src[max(0, fn_idx - 500): fn_idx] assert "limiter.limit" in preamble, ( f"{fn_name} must be decorated with @limiter.limit(...); " f"none found in the 500 chars before 'async def {fn_name}'" ) def test_explore_mists_has_limiter(self) -> None: """explore_mists (GET /api/mists/explore) must have @limiter.limit.""" self._check_fn_has_limiter("explore_mists") def test_get_mist_has_limiter(self) -> None: """get_mist (GET /api/mists/{mist_id}) must have @limiter.limit.""" self._check_fn_has_limiter("get_mist") def test_list_mist_forks_has_limiter(self) -> None: """list_mist_forks (GET /api/mists/{mist_id}/forks) must have @limiter.limit.""" self._check_fn_has_limiter("list_mist_forks") def test_list_owner_mists_has_limiter(self) -> None: """list_owner_mists (GET /api/{owner}/mists) must have @limiter.limit.""" self._check_fn_has_limiter("list_owner_mists") def test_get_mist_embed_has_limiter(self) -> None: """get_mist_embed (GET /api/{owner}/mists/{id}/embed) must have @limiter.limit.""" self._check_fn_has_limiter("get_mist_embed") # --------------------------------------------------------------------------- # 4. Read endpoints still return 200 under threshold (regression) # --------------------------------------------------------------------------- class TestReadEndpointsWorkUnderThreshold: @pytest.mark.asyncio async def test_explore_returns_200_under_limit(self, client: AsyncClient) -> None: r = await client.get("/api/mists/explore") assert r.status_code == 200, ( f"GET /api/mists/explore returned {r.status_code} under the rate limit" ) @pytest.mark.asyncio async def test_get_nonexistent_mist_returns_404_not_429( self, client: AsyncClient ) -> None: """A single GET for an unknown mist_id must return 404, not 429.""" r = await client.get("/api/mists/nonexistentId12") assert r.status_code in (404, 200), ( f"GET /api/mists/nonexistentId12 returned {r.status_code}; " "429 under single request means the limit is misconfigured" ) @pytest.mark.asyncio async def test_list_owner_mists_returns_200_under_limit( self, client: AsyncClient ) -> None: r = await client.get("/api/gabriel/mists") assert r.status_code == 200, ( f"GET /api/gabriel/mists returned {r.status_code} under the rate limit" )