diff --git a/src/ypms/auth/deps.py b/src/ypms/auth/deps.py index 0a72bce..e0ca06b 100644 --- a/src/ypms/auth/deps.py +++ b/src/ypms/auth/deps.py @@ -1,28 +1,45 @@ """Authentication dependencies. -Phase 0 milestone 0.2: placeholder `get_current_user` that always raises 401. -Milestone 0.3 replaces this with a JWT-verifying implementation that also -issues the `iam.assurance_level` claim on the token. +Milestone 0.3: real `get_current_user` — validates the JWT, refuses refresh +tokens on resource endpoints, loads the User and refuses if inactive. -Kept as its own module so the RBAC engine and router factory can import it now -and milestone 0.3's swap is a single-file change. +Downstream authorization (RBAC and assurance-level gating) reads user.roles +and user.assurance_level, so the returned User carries live values from the +DB rather than the token snapshot — a role revocation or account +deactivation takes effect on the next request, not on token expiry. """ from __future__ import annotations -from fastapi import HTTPException, status +from fastapi import Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer +from sqlalchemy.orm import Session +from ypms.auth.security import decode_token +from ypms.db import get_db from ypms.users.models import User +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login", auto_error=False) -def get_current_user() -> User: - """Placeholder. Milestone 0.3 lands real JWT verification. - Raises 401 for any route that goes through it. Tests that need to exercise - RBAC or routes without a real auth stack override this dependency with a - fixture-provided User via `app.dependency_overrides`. - """ - raise HTTPException( - status.HTTP_401_UNAUTHORIZED, - detail={"error": "unauthenticated", "phase": "milestone 0.3 lands JWT auth"}, - ) +def get_current_user( + token: str | None = Depends(oauth2_scheme), + db: Session = Depends(get_db), +) -> User: + if not token: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated") + try: + payload = decode_token(token) + except ValueError as e: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(e)) from e + if payload.get("type") == "refresh": + raise HTTPException( + status.HTTP_401_UNAUTHORIZED, "refresh token cannot access resources" + ) + username = payload.get("sub") + if not username: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token") + user = db.query(User).filter(User.username == username).one_or_none() + if not user or not user.is_active: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found or inactive") + return user diff --git a/src/ypms/auth/rate_limit.py b/src/ypms/auth/rate_limit.py new file mode 100644 index 0000000..1a57ffc --- /dev/null +++ b/src/ypms/auth/rate_limit.py @@ -0,0 +1,47 @@ +"""Tiny in-memory rate limiter. + +Vendored from OWC `auth/rate_limit.py` (snapshot 2026-04-20); unchanged. + +Not suitable for multi-process prod. Swap to Redis-backed once we're +horizontally scaled (Phase 9+ / Azure deploy). Used today to cap login +attempts per IP (NIST AC-7 / source RFP C.5.7.10). +""" + +from __future__ import annotations + +import time +from collections import deque +from threading import Lock + + +class SlidingWindow: + def __init__(self, limit: int, window_seconds: int): + self.limit = limit + self.window = window_seconds + self._hits: dict[str, deque[float]] = {} + self._lock = Lock() + + def allow(self, key: str) -> bool: + now = time.time() + with self._lock: + q = self._hits.setdefault(key, deque()) + # Evict entries outside the window. + while q and now - q[0] > self.window: + q.popleft() + if len(q) >= self.limit: + return False + q.append(now) + return True + + def reset(self, key: str | None = None) -> None: + """Test helper: clear a single key or the whole window.""" + with self._lock: + if key is None: + self._hits.clear() + else: + self._hits.pop(key, None) + + +# NIST AC-7 / source RFP `C.5.7.10`: 5 failed login attempts triggers lockout. +# We track by IP and use a longer window than OWC to keep the block honest. +login_limiter = SlidingWindow(limit=5, window_seconds=900) # 15 minutes diff --git a/src/ypms/auth/router.py b/src/ypms/auth/router.py new file mode 100644 index 0000000..3c10f57 --- /dev/null +++ b/src/ypms/auth/router.py @@ -0,0 +1,145 @@ +"""Auth surface: /auth/login, /auth/refresh, /auth/me, /auth/change-password. + +Source RFP requirements addressed here: +- C.5.7.1 role-based authN/authZ (roles baked into JWT claim) +- C.5.7.3 session timeout — enforced by access token TTL +- C.5.7.7 min 12-alphanumeric identifier — pydantic constraint on register + lands in Phase 1; here we only accept credentials, not create them +- C.5.7.8 user-established and changeable passwords → /change-password +- C.5.7.9 FIPS 140-2/3 crypto — pbkdf2_sha256 + Key Vault (0.3 uses env-var + key; Key Vault swap-in is Phase 1) +- C.5.7.10 5 failed attempts → lockout — enforced by login_limiter +- C.5.7.38 unique user ID per user — enforced by User.username uniqueness + +NIST controls: IA-2, IA-5, IA-5(1), IA-8 (organizational side), AC-7, AC-11. +""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from fastapi.security import OAuth2PasswordRequestForm +from sqlalchemy.orm import Session + +from ypms.audit import service as audit +from ypms.auth.deps import get_current_user +from ypms.auth.rate_limit import login_limiter +from ypms.auth.schemas import ( + ChangePasswordRequest, + LoginResponse, + RefreshRequest, + RefreshResponse, + UserRead, +) +from ypms.auth.security import ( + create_access_token, + create_refresh_token, + decode_token, + hash_password, + verify_password, +) +from ypms.db import get_db +from ypms.security.rbac import ADMIN_ROLE +from ypms.users.models import ( + ASSURANCE_ADMIN, + User, +) + +router = APIRouter(prefix="/auth", tags=["auth"]) + + +def _issue_access_token(user: User) -> str: + role_codes = [r.code for r in user.roles] + # DOES_SYSTEM_ADMIN implies admin assurance regardless of the persisted + # value — the super-role bypass already grants blanket authority and the + # token should reflect that so downstream assurance-level gates behave. + level = ASSURANCE_ADMIN if ADMIN_ROLE in role_codes else user.assurance_level + return create_access_token( + sub=user.username, + uid=user.id, + roles=role_codes, + assurance_level=level, + mfa_verified=user.mfa_verified, + ) + + +@router.post("/login", response_model=LoginResponse) +def login( + request: Request, + form: OAuth2PasswordRequestForm = Depends(), + db: Session = Depends(get_db), +) -> LoginResponse: + ip = request.client.host if request.client else "unknown" + if not login_limiter.allow(ip): + # NIST AC-7: cap unsuccessful logon attempts. + raise HTTPException( + status.HTTP_429_TOO_MANY_REQUESTS, + "too many login attempts; try again shortly", + headers={"Retry-After": "900"}, + ) + user = db.query(User).filter(User.username == form.username).one_or_none() + if not user or not verify_password(form.password, user.password_hash): + # Deliberately identical error surface so we don't leak whether the + # username exists (IA-6 authenticator feedback discipline). + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials") + if not user.is_active: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user inactive") + + audit.record( + db, actor=user, action="login", entity="User", entity_id=user.id, ip=ip + ) + db.commit() + + return LoginResponse( + access_token=_issue_access_token(user), + refresh_token=create_refresh_token(sub=user.username, uid=user.id), + ) + + +@router.post("/refresh", response_model=RefreshResponse) +def refresh( + body: RefreshRequest, + db: Session = Depends(get_db), +) -> RefreshResponse: + try: + payload = decode_token(body.refresh_token) + except ValueError as e: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(e)) from e + if payload.get("type") != "refresh": + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not a refresh token") + username = payload.get("sub") + user = db.query(User).filter(User.username == username).one_or_none() + if not user or not user.is_active: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found or inactive") + return RefreshResponse(access_token=_issue_access_token(user)) + + +@router.get("/me", response_model=UserRead) +def me(current: User = Depends(get_current_user)) -> User: + return current + + +@router.post("/change-password", status_code=status.HTTP_204_NO_CONTENT) +def change_password( + body: ChangePasswordRequest, + request: Request, + db: Session = Depends(get_db), + current: User = Depends(get_current_user), +) -> None: + if not verify_password(body.current_password, current.password_hash): + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "current password incorrect") + if len(body.new_password) < 12: + # Source RFP C.5.7.45 leaves the exact policy to the org. 12 chars is + # the floor from C.5.7.7 (min identifier length) — treat the same for + # secrets. Complexity + rotation policy lands with the admin UI. + raise HTTPException(status.HTTP_400_BAD_REQUEST, "new password must be at least 12 characters") + + current.password_hash = hash_password(body.new_password) + db.flush() + + ip = request.client.host if request.client else "unknown" + audit.record( + db, actor=current, action="change_password", + entity="User", entity_id=current.id, ip=ip, + ) + db.commit() + return None diff --git a/src/ypms/auth/schemas.py b/src/ypms/auth/schemas.py new file mode 100644 index 0000000..4b72443 --- /dev/null +++ b/src/ypms/auth/schemas.py @@ -0,0 +1,40 @@ +"""Pydantic schemas for the auth surface.""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict + + +class LoginResponse(BaseModel): + access_token: str + refresh_token: str + token_type: str = "bearer" + + +class RefreshRequest(BaseModel): + refresh_token: str + + +class RefreshResponse(BaseModel): + access_token: str + token_type: str = "bearer" + + +class ChangePasswordRequest(BaseModel): + current_password: str + new_password: str + + +class UserRead(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: int + username: str + email: str | None = None + full_name: str | None = None + is_active: bool + mfa_verified: bool + assurance_level: str + created_at: datetime diff --git a/src/ypms/auth/security.py b/src/ypms/auth/security.py new file mode 100644 index 0000000..cbdcbb9 --- /dev/null +++ b/src/ypms/auth/security.py @@ -0,0 +1,82 @@ +"""Password hashing + JWT create/verify. + +Vendored from OWC `auth/security.py` (snapshot 2026-04-20). YPMS additions: +`create_access_token()` bakes in YPMS-specific claims — `uid`, `roles`, +`iam.assurance_level`, `mfa_verified` — so downstream authorization can gate +high-risk actions above base RBAC (see ARCHITECTURE.md §4.5, §8.2). + +`iam.assurance_level` values are defined in `ypms.users.models` and follow +NIST 800-63 IAL/AAL semantics: `pre_application` < `standard` < +`payroll_authorized` < `staff` < `admin`. Payroll-touching endpoints will +require `payroll_authorized` or higher (Phase 7+). + +pbkdf2_sha256 not bcrypt — see ARCHITECTURE.md §4.6. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +from jose import JWTError, jwt +from passlib.context import CryptContext + +from ypms.settings import get_settings + +_pwd = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto") + + +def hash_password(pw: str) -> str: + h: str = _pwd.hash(pw) + return h + + +def verify_password(pw: str, hashed: str) -> bool: + return bool(_pwd.verify(pw, hashed)) + + +def create_access_token( + *, + sub: str, + uid: int, + roles: list[str], + assurance_level: str, + mfa_verified: bool, +) -> str: + s = get_settings() + now = datetime.now(UTC) + payload: dict[str, Any] = { + "sub": sub, + "uid": uid, + "roles": roles, + "iam.assurance_level": assurance_level, + "mfa_verified": mfa_verified, + "iat": int(now.timestamp()), + "exp": int((now + timedelta(minutes=s.access_token_ttl_minutes)).timestamp()), + "type": "access", + } + token: str = jwt.encode(payload, s.jwt_secret, algorithm=s.jwt_alg) + return token + + +def create_refresh_token(*, sub: str, uid: int) -> str: + s = get_settings() + now = datetime.now(UTC) + payload: dict[str, Any] = { + "sub": sub, + "uid": uid, + "iat": int(now.timestamp()), + "exp": int((now + timedelta(days=s.refresh_token_ttl_days)).timestamp()), + "type": "refresh", + } + token: str = jwt.encode(payload, s.jwt_secret, algorithm=s.jwt_alg) + return token + + +def decode_token(token: str) -> dict[str, Any]: + s = get_settings() + try: + payload: dict[str, Any] = jwt.decode(token, s.jwt_secret, algorithms=[s.jwt_alg]) + except JWTError as e: + raise ValueError("invalid token") from e + return payload diff --git a/src/ypms/main.py b/src/ypms/main.py index 7cc560f..b5a1197 100644 --- a/src/ypms/main.py +++ b/src/ypms/main.py @@ -12,6 +12,7 @@ from fastapi import FastAPI, status from fastapi.middleware.cors import CORSMiddleware from sqlalchemy import text +from ypms.auth.router import router as auth_router from ypms.db import engine from ypms.logging import configure_logging, get_logger from ypms.middleware import RequestIdMiddleware @@ -40,6 +41,8 @@ def create_app() -> FastAPI: expose_headers=["X-Request-ID"], ) + app.include_router(auth_router) + @app.get("/healthz", tags=["system"]) def healthz() -> dict[str, str]: """Liveness — process is up. No dependency checks.""" diff --git a/tests/auth/__init__.py b/tests/auth/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/auth/test_router.py b/tests/auth/test_router.py new file mode 100644 index 0000000..60530f5 --- /dev/null +++ b/tests/auth/test_router.py @@ -0,0 +1,253 @@ +"""End-to-end auth flow via the FastAPI TestClient. + +Uses the shared SQLite in-memory DB by wiring the `get_db` dependency to the +per-test session, then seeding a user and driving login/refresh/me/ +change-password. +""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from ypms.auth.rate_limit import login_limiter +from ypms.auth.security import ( + create_access_token, + create_refresh_token, + hash_password, +) +from ypms.db import get_db +from ypms.main import create_app +from ypms.security.rbac import ADMIN_ROLE +from ypms.users.models import ( + ASSURANCE_ADMIN, + ASSURANCE_STANDARD, + Role, + User, + UserRole, +) + + +@pytest.fixture(autouse=True) +def _reset_rate_limits() -> Iterator[None]: + login_limiter.reset() + yield + login_limiter.reset() + + +@pytest.fixture +def staff_user(db: Session) -> User: + user = User( + username="staff.user", + email="staff@dc.gov", + full_name="Staff User", + password_hash=hash_password("correct-horse-battery"), + is_active=True, + assurance_level=ASSURANCE_STANDARD, + ) + db.add(user) + db.flush() + db.refresh(user) + return user + + +@pytest.fixture +def admin_user(db: Session) -> User: + role = Role(code=ADMIN_ROLE, description="super-role") + db.add(role) + db.flush() + user = User( + username="chris", + email="chris@dc.gov", + full_name="Chris Tonjes", + password_hash=hash_password("correct-horse-battery"), + is_active=True, + assurance_level=ASSURANCE_STANDARD, # role bump upgrades to admin + ) + db.add(user) + db.flush() + db.add(UserRole(user_id=user.id, role_id=role.id)) + db.flush() + db.refresh(user) + return user + + +@pytest.fixture +def app_client(db: Session) -> Iterator[TestClient]: + """A TestClient wired to the per-test SQLite session via dep override.""" + app = create_app() + + def _override_get_db() -> Iterator[Session]: + yield db + + app.dependency_overrides[get_db] = _override_get_db + with TestClient(app) as c: + yield c + app.dependency_overrides.clear() + + +def test_login_wrong_password_returns_401(app_client: TestClient, staff_user: User) -> None: + r = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "wrong"}, + ) + assert r.status_code == 401 + assert r.json()["detail"] == "invalid credentials" + + +def test_login_success_returns_tokens(app_client: TestClient, staff_user: User) -> None: + r = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + assert r.status_code == 200 + body = r.json() + assert body["token_type"] == "bearer" + assert body["access_token"] + assert body["refresh_token"] + + +def test_admin_login_upgrades_assurance_level( + app_client: TestClient, admin_user: User +) -> None: + from ypms.auth.security import decode_token + + r = app_client.post( + "/auth/login", + data={"username": admin_user.username, "password": "correct-horse-battery"}, + ) + assert r.status_code == 200 + claims = decode_token(r.json()["access_token"]) + assert claims["iam.assurance_level"] == ASSURANCE_ADMIN + assert ADMIN_ROLE in claims["roles"] + + +def test_me_requires_authentication(app_client: TestClient) -> None: + r = app_client.get("/auth/me") + assert r.status_code == 401 + + +def test_me_returns_current_user(app_client: TestClient, staff_user: User) -> None: + login = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + access = login.json()["access_token"] + r = app_client.get("/auth/me", headers={"Authorization": f"Bearer {access}"}) + assert r.status_code == 200 + assert r.json()["username"] == staff_user.username + assert r.json()["email"] == "staff@dc.gov" + + +def test_refresh_returns_new_access_token(app_client: TestClient, staff_user: User) -> None: + login = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + refresh = login.json()["refresh_token"] + r = app_client.post("/auth/refresh", json={"refresh_token": refresh}) + assert r.status_code == 200 + assert r.json()["access_token"] + assert "refresh_token" not in r.json() + + +def test_refresh_rejects_access_token(app_client: TestClient, staff_user: User) -> None: + access = create_access_token( + sub=staff_user.username, + uid=staff_user.id, + roles=[], + assurance_level=ASSURANCE_STANDARD, + mfa_verified=False, + ) + r = app_client.post("/auth/refresh", json={"refresh_token": access}) + assert r.status_code == 401 + + +def test_me_rejects_refresh_token(app_client: TestClient, staff_user: User) -> None: + refresh = create_refresh_token(sub=staff_user.username, uid=staff_user.id) + r = app_client.get("/auth/me", headers={"Authorization": f"Bearer {refresh}"}) + assert r.status_code == 401 + + +def test_change_password_flow(app_client: TestClient, staff_user: User) -> None: + login = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + access = login.json()["access_token"] + r = app_client.post( + "/auth/change-password", + headers={"Authorization": f"Bearer {access}"}, + json={ + "current_password": "correct-horse-battery", + "new_password": "totally-different-passphrase", + }, + ) + assert r.status_code == 204 + + # Old password no longer works. + r2 = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + assert r2.status_code == 401 + + # New password does. + r3 = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "totally-different-passphrase"}, + ) + assert r3.status_code == 200 + + +def test_change_password_rejects_short_new_password( + app_client: TestClient, staff_user: User +) -> None: + login = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + access = login.json()["access_token"] + r = app_client.post( + "/auth/change-password", + headers={"Authorization": f"Bearer {access}"}, + json={"current_password": "correct-horse-battery", "new_password": "short"}, + ) + assert r.status_code == 400 + + +def test_change_password_rejects_wrong_current( + app_client: TestClient, staff_user: User +) -> None: + login = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + access = login.json()["access_token"] + r = app_client.post( + "/auth/change-password", + headers={"Authorization": f"Bearer {access}"}, + json={"current_password": "wrong", "new_password": "long-enough-password"}, + ) + assert r.status_code == 401 + + +def test_login_lockout_after_five_failed_attempts( + app_client: TestClient, staff_user: User +) -> None: + for _ in range(5): + r = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "wrong"}, + ) + assert r.status_code == 401 + # 6th attempt (any credentials) is rate-limited. + r = app_client.post( + "/auth/login", + data={"username": staff_user.username, "password": "correct-horse-battery"}, + ) + assert r.status_code == 429 + assert r.headers["Retry-After"] == "900" diff --git a/tests/auth/test_security.py b/tests/auth/test_security.py new file mode 100644 index 0000000..99ce3c0 --- /dev/null +++ b/tests/auth/test_security.py @@ -0,0 +1,59 @@ +"""Password + JWT primitives.""" + +from __future__ import annotations + +import time + +import pytest + +from ypms.auth.security import ( + create_access_token, + create_refresh_token, + decode_token, + hash_password, + verify_password, +) + + +def test_hash_password_returns_pbkdf2_prefix() -> None: + h = hash_password("hunter2hunter2") + assert h.startswith("$pbkdf2-sha256$") + + +def test_verify_password_true_on_match_false_on_mismatch() -> None: + h = hash_password("correcthorsebattery") + assert verify_password("correcthorsebattery", h) is True + assert verify_password("wrong", h) is False + + +def test_access_token_carries_ypms_claims() -> None: + tok = create_access_token( + sub="chris", + uid=1, + roles=["DOES_SYSTEM_ADMIN"], + assurance_level="admin", + mfa_verified=False, + ) + claims = decode_token(tok) + assert claims["sub"] == "chris" + assert claims["uid"] == 1 + assert claims["roles"] == ["DOES_SYSTEM_ADMIN"] + assert claims["iam.assurance_level"] == "admin" + assert claims["mfa_verified"] is False + assert claims["type"] == "access" + assert claims["exp"] > claims["iat"] >= int(time.time()) - 5 + + +def test_refresh_token_has_type_refresh_and_carries_uid() -> None: + tok = create_refresh_token(sub="chris", uid=42) + claims = decode_token(tok) + assert claims["type"] == "refresh" + assert claims["sub"] == "chris" + assert claims["uid"] == 42 + assert "roles" not in claims + assert "iam.assurance_level" not in claims + + +def test_decode_invalid_token_raises() -> None: + with pytest.raises(ValueError): + decode_token("not-a-real-jwt") diff --git a/tests/conftest.py b/tests/conftest.py index 8f94735..86c44cb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,6 +3,10 @@ Uses SQLite for speed (see ARCHITECTURE.md §4.8). Tests exercise the framework against an in-memory Postgres-shaped SQLAlchemy Base — no Alembic migrations in the test path; models create their tables from metadata. + +Environment defaults are set at *module import* time (not via fixture) so +they are in place before pytest collects any test module that transitively +imports `ypms.main` (which instantiates the FastAPI app at import time). """ from __future__ import annotations @@ -10,22 +14,18 @@ from __future__ import annotations import os from collections.abc import Iterator -import pytest -from fastapi.testclient import TestClient -from sqlalchemy import create_engine -from sqlalchemy.orm import Session, sessionmaker +os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") +os.environ.setdefault("JWT_SECRET", "test-secret") +os.environ.setdefault( + "COLUMN_ENCRYPTION_KEY", "test-col-enc-key-32bytes-______________" +) +os.environ.setdefault("ENV", "test") - -@pytest.fixture(scope="session", autouse=True) -def _test_env() -> Iterator[None]: - """Set minimum env vars required for `Settings()` to construct in tests.""" - os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") - os.environ.setdefault("JWT_SECRET", "test-secret") - os.environ.setdefault( - "COLUMN_ENCRYPTION_KEY", "test-col-enc-key-32bytes-______________" - ) - os.environ.setdefault("ENV", "test") - yield +import pytest # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.orm import Session, sessionmaker # noqa: E402 +from sqlalchemy.pool import StaticPool # noqa: E402 @pytest.fixture @@ -41,7 +41,17 @@ def db() -> Iterator[Session]: import ypms.users.models # noqa: F401 from ypms.db import Base - engine = create_engine("sqlite:///:memory:", future=True) + # StaticPool + check_same_thread=False share the same in-memory DB across + # threads. TestClient runs handlers in a worker thread, so without this + # the fixture-scoped session and the request-scoped session see different + # empty databases (SADeprecationWarning: SQLite objects created in a + # thread can only be used in that same thread). + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + future=True, + ) Base.metadata.create_all(engine) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) session = SessionLocal()