test(07.2-01): mount FakeRedis on test_auth_deps app + add NBF-check xfail stubs
- Import FakeRedis from tests.test_auth_api (no redefinition) - Set test_app.state.redis = FakeRedis() in make_test_app() (Pitfall 4 guard) - Add xfail(strict=False) stubs: reject iat<nbf, allow iat>nbf, fail-open on Redis error - All 7 existing auth-dep tests unaffected (zero regressions)
This commit is contained in:
@@ -7,6 +7,11 @@ Tests verify:
|
|||||||
- get_current_user raises HTTP 401 when user.is_active is False
|
- get_current_user raises HTTP 401 when user.is_active is False
|
||||||
- get_current_admin raises HTTP 403 when user.role == "user"
|
- get_current_admin raises HTTP 403 when user.role == "user"
|
||||||
- get_current_admin returns user when user.role == "admin"
|
- get_current_admin returns user when user.role == "admin"
|
||||||
|
|
||||||
|
Phase 7.2 additions (Wave 0 scaffolding):
|
||||||
|
- FakeRedis attached to app.state.redis so Wave 1 NBF check in get_current_user
|
||||||
|
does not raise AttributeError: 'State' object has no attribute 'redis'
|
||||||
|
- Three xfail stubs for NBF check behaviours (reject, allow, fail-open)
|
||||||
"""
|
"""
|
||||||
import uuid
|
import uuid
|
||||||
|
|
||||||
@@ -16,11 +21,18 @@ from httpx import ASGITransport, AsyncClient
|
|||||||
from fastapi import FastAPI, Depends
|
from fastapi import FastAPI, Depends
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
|
from tests.test_auth_api import FakeRedis
|
||||||
|
|
||||||
|
|
||||||
# ── Minimal test app with /test/me and /test/admin routes ─────────────────────
|
# ── Minimal test app with /test/me and /test/admin routes ─────────────────────
|
||||||
|
|
||||||
def make_test_app():
|
def make_test_app():
|
||||||
"""Create a minimal FastAPI app that exercises the auth deps."""
|
"""Create a minimal FastAPI app that exercises the auth deps.
|
||||||
|
|
||||||
|
Phase 7.2: app.state.redis is set to a FakeRedis() instance so that the
|
||||||
|
NBF check in get_current_user (Wave 1) can call request.app.state.redis
|
||||||
|
without raising AttributeError (Pitfall 4 from Phase 7.2 RESEARCH.md).
|
||||||
|
"""
|
||||||
from deps.auth import get_current_user, get_current_admin
|
from deps.auth import get_current_user, get_current_admin
|
||||||
from db.models import User
|
from db.models import User
|
||||||
|
|
||||||
@@ -34,12 +46,20 @@ def make_test_app():
|
|||||||
async def admin_only(_admin: User = Depends(get_current_admin)):
|
async def admin_only(_admin: User = Depends(get_current_admin)):
|
||||||
return {"role": _admin.role}
|
return {"role": _admin.role}
|
||||||
|
|
||||||
|
# Phase 7.2: mount FakeRedis so NBF check in get_current_user works
|
||||||
|
test_app.state.redis = FakeRedis()
|
||||||
|
|
||||||
return test_app
|
return test_app
|
||||||
|
|
||||||
|
|
||||||
@pytest_asyncio.fixture
|
@pytest_asyncio.fixture
|
||||||
async def auth_client(db_session: AsyncSession):
|
async def auth_client(db_session: AsyncSession):
|
||||||
"""Async HTTP test client for the auth-dep test app."""
|
"""Async HTTP test client for the auth-dep test app.
|
||||||
|
|
||||||
|
Phase 7.2: each invocation of make_test_app() creates a fresh FakeRedis
|
||||||
|
instance (instantiated inside make_test_app). The NBF check in
|
||||||
|
get_current_user reads request.app.state.redis — this fixture provides it.
|
||||||
|
"""
|
||||||
from deps.db import get_db
|
from deps.db import get_db
|
||||||
|
|
||||||
app = make_test_app()
|
app = make_test_app()
|
||||||
@@ -158,3 +178,76 @@ def test_deps_auth_has_http_403():
|
|||||||
source = f.read()
|
source = f.read()
|
||||||
assert re.search(r"HTTP_403_FORBIDDEN", source), \
|
assert re.search(r"HTTP_403_FORBIDDEN", source), \
|
||||||
"deps/auth.py must raise HTTP 403 for non-admin access"
|
"deps/auth.py must raise HTTP 403 for non-admin access"
|
||||||
|
|
||||||
|
|
||||||
|
# ── Phase 7.2 Wave 0 stubs: NBF check behaviours ─────────────────────────────
|
||||||
|
# These three tests are xfailed with strict=False. When Wave 1 (Plan 02) adds
|
||||||
|
# the NBF check to get_current_user, the stubs are promoted to real assertions
|
||||||
|
# by replacing `assert False, "stub"` with the actual assertion logic.
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 1 — NBF check not yet implemented in get_current_user")
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_current_user_rejects_token_when_iat_before_user_nbf(auth_client, db_session):
|
||||||
|
"""Token with iat < user_nbf in Redis should return 401 'Session invalidated'."""
|
||||||
|
from services.auth import create_access_token
|
||||||
|
import time
|
||||||
|
|
||||||
|
user = await _create_user(db_session, role="user")
|
||||||
|
token = create_access_token(str(user.id), "user")
|
||||||
|
|
||||||
|
# Pre-populate Redis with a future nbf (token's iat will be < this value)
|
||||||
|
future_ts = int(time.time()) + 3600
|
||||||
|
fake_redis = auth_client._transport.app.state.redis
|
||||||
|
await fake_redis.set(f"user_nbf:{user.id}", str(future_ts).encode())
|
||||||
|
|
||||||
|
resp = await auth_client.get(
|
||||||
|
"/test/me", headers={"Authorization": f"Bearer {token}"}
|
||||||
|
)
|
||||||
|
# Wave 1 will implement the NBF check; this assertion is the target behaviour
|
||||||
|
assert False, "stub — Wave 1 will fill in: assert resp.status_code == 401 and 'Session invalidated' in resp.json()['detail']"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 1 — NBF check not yet implemented in get_current_user")
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_current_user_allows_token_when_iat_after_user_nbf(auth_client, db_session):
|
||||||
|
"""Token with iat > user_nbf in Redis should return 200 (token is valid)."""
|
||||||
|
from services.auth import create_access_token
|
||||||
|
import time
|
||||||
|
|
||||||
|
user = await _create_user(db_session, role="user")
|
||||||
|
token = create_access_token(str(user.id), "user")
|
||||||
|
|
||||||
|
# Pre-populate Redis with a past nbf (token's iat will be > this value)
|
||||||
|
past_ts = int(time.time()) - 3600
|
||||||
|
fake_redis = auth_client._transport.app.state.redis
|
||||||
|
await fake_redis.set(f"user_nbf:{user.id}", str(past_ts).encode())
|
||||||
|
|
||||||
|
resp = await auth_client.get(
|
||||||
|
"/test/me", headers={"Authorization": f"Bearer {token}"}
|
||||||
|
)
|
||||||
|
# Wave 1 will implement the NBF check; this assertion is the target behaviour
|
||||||
|
assert False, "stub — Wave 1 will fill in: assert resp.status_code == 200"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.xfail(strict=False, reason="Phase 7.2 Wave 1 — NBF check not yet implemented in get_current_user")
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_get_current_user_failopen_on_redis_error(auth_client, db_session):
|
||||||
|
"""Redis error during NBF check must fail-open (return 200, not crash)."""
|
||||||
|
from services.auth import create_access_token
|
||||||
|
|
||||||
|
class _BrokenRedis:
|
||||||
|
async def get(self, key):
|
||||||
|
raise RuntimeError("simulated redis down")
|
||||||
|
|
||||||
|
user = await _create_user(db_session, role="user")
|
||||||
|
token = create_access_token(str(user.id), "user")
|
||||||
|
|
||||||
|
# Override app.state.redis with a broken redis for this test
|
||||||
|
auth_client._transport.app.state.redis = _BrokenRedis()
|
||||||
|
|
||||||
|
resp = await auth_client.get(
|
||||||
|
"/test/me", headers={"Authorization": f"Bearer {token}"}
|
||||||
|
)
|
||||||
|
# Wave 1 will implement fail-open (D-04): assert resp.status_code == 200
|
||||||
|
assert False, "stub — Wave 1 will fill in: assert resp.status_code == 200"
|
||||||
|
|||||||
Reference in New Issue
Block a user