""" Rate limiting test stubs — D-11, D-12, Assumption A1. Wave 0 xfail scaffold for Phase 6 plan 06-04. D-11: trusted-proxy CIDR logic in get_client_ip (deps/utils.py replacement). D-12: per-account rate limiter keyed by user_id (100 req/min). A1: slowapi key_func ordering assumption — request.state.current_user must be set as the FIRST line of the handler body before slowapi reads the key. All tests use strict=False so an unexpected pass during development never breaks CI. Implementation lands in plan 06-04. """ from __future__ import annotations import pytest # ── D-11: trusted-proxy CIDR logic in get_client_ip ────────────────────────── @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_get_client_ip_untrusted_returns_direct_peer(): """When request.client.host is 8.8.8.8 (not in trusted CIDRs), ignore X-Forwarded-For and return the direct peer IP "8.8.8.8".""" pytest.xfail("not implemented yet") @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_get_client_ip_trusted_proxy_reads_xff_leftmost(): """When request.client.host is 127.0.0.1 (trusted), return the leftmost address from X-Forwarded-For: "1.2.3.4, 5.6.7.8" → "1.2.3.4".""" pytest.xfail("not implemented yet") @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_get_client_ip_trusted_proxy_no_xff_falls_back(): """When the direct peer is trusted but no X-Forwarded-For header is present, return the direct peer IP as fallback.""" pytest.xfail("not implemented yet") @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_get_client_ip_invalid_peer_returns_none_or_string(): """When request.client is None, get_client_ip returns None without raising.""" pytest.xfail("not implemented yet") # ── D-12: per-account rate limiter keyed by user_id ────────────────────────── @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_account_limiter_key_uses_user_id(): """_account_key(request) where request.state.current_user has id=UUID(...) returns str(user.id), not request.client.host.""" pytest.xfail("not implemented yet") @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_account_limiter_key_falls_back_to_ip_when_no_user(): """When request.state.current_user is missing, the key function returns the direct peer IP — must not crash (Pitfall 3 guard).""" pytest.xfail("not implemented yet") # ── A1: key_func ordering assumption — unit verification ───────────────────── @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_account_limiter_key_ordering_assumption(async_client, auth_user): """A1 verification: construct a FastAPI test endpoint that sets request.state.current_user as its first line, decorated with @account_limiter.limit("100/minute"); call it 101 times with the same user; assert the 101st response is 429 AND the limiter recorded the key as str(user.id), not the IP.""" pytest.xfail("not implemented yet") # ── D-12: full integration — 429 after 100 requests/minute ─────────────────── @pytest.mark.xfail(strict=False, reason="implementation in 06-04") async def test_authenticated_endpoint_429_after_100_per_minute( async_client, auth_user ): """Full integration: GET /api/documents/ called 101 times with the same auth_user returns 429 on the 101st request.""" pytest.xfail("not implemented yet")