From 075069bf91efb64ec81286d0d20e75217bd37567 Mon Sep 17 00:00:00 2001 From: ulises-jeremias Date: Sun, 23 Aug 2026 14:30:15 -0300 Subject: [PATCH] fix(extensions): W3 add tests for auth-jwt and sqlalchemy, prep incompatibleWith - extensions/fastapi-auth-jwt: add tests/test_auth.py (16 tests) covering password hashing (salted Argon2), JWT create/decode round-trip, extra claims, algorithm, secret rotation, expiry, Pydantic schemas, and router wire-up via TestClient ($auth endpoints). - extensions/fastapi-sqlalchemy: add tests/test_sqlalchemy.py (9 tests) covering DeclarativeBase registration, engine/session factory smoke, get_db generator contract, isolated in-memory round-trip, Alembic env wiring via static content check, DATABASE_URL override, and sqlite connect_args style guard. - docs/AUTHORING.md: document upcoming celery-docker vs flower-docker incompatibleWith prep for PR #178 (symmetric validation via scripts/ci/validate-registry.py / templates.schema.json). - extensions/all-postgres: fix .env.example.append style consistency (remove duplicate header and leading blank line). Co-authored-by: cpa-templates W3 task --- docs/AUTHORING.md | 6 +- .../all-postgres/template/.env.example.append | 2 - .../template/tests/test_auth.py | 189 ++++++++++++++++++ .../template/tests/test_sqlalchemy.py | 141 +++++++++++++ 4 files changed, 335 insertions(+), 3 deletions(-) create mode 100644 extensions/fastapi-auth-jwt/template/tests/test_auth.py create mode 100644 extensions/fastapi-sqlalchemy/template/tests/test_sqlalchemy.py diff --git a/docs/AUTHORING.md b/docs/AUTHORING.md index 72d7857..96dc7e5 100644 --- a/docs/AUTHORING.md +++ b/docs/AUTHORING.md @@ -145,7 +145,11 @@ Use symmetric `incompatibleWith` when two extensions would overwrite the same generated paths (for example two Docker overlays that both ship `Dockerfile` / `compose.yml` for the **same** template `type`). Today stack Docker extensions are isolated by `type`; when a type gains a second packaging strategy, declare -mutual incompatibility like cna-templates does for Redux saga/thunk. +mutual incompatibility like cna-templates does for Redux saga/thunk. Example: +`celery-docker` and the upcoming `flower-docker` (PR #178) both target +`celery-worker` and ship a Compose stack for the same worker type — they must +declare `incompatibleWith` on both entries when `flower-docker` lands (validation +is symmetric; see `scripts/ci/validate-registry.py` and `templates.schema.json`). **Authoring rules:** diff --git a/extensions/all-postgres/template/.env.example.append b/extensions/all-postgres/template/.env.example.append index d8c2f90..89e1ee9 100644 --- a/extensions/all-postgres/template/.env.example.append +++ b/extensions/all-postgres/template/.env.example.append @@ -1,6 +1,4 @@ - # PostgreSQL (postgres extension) -# PostgreSQL (optional overlay) # DATABASE_URL=postgresql+psycopg://app:app@localhost:5432/app POSTGRES_USER=app POSTGRES_PASSWORD=app diff --git a/extensions/fastapi-auth-jwt/template/tests/test_auth.py b/extensions/fastapi-auth-jwt/template/tests/test_auth.py new file mode 100644 index 0000000..c2de08a --- /dev/null +++ b/extensions/fastapi-auth-jwt/template/tests/test_auth.py @@ -0,0 +1,189 @@ +"""Tests for fastapi-auth-jwt extension (password hashing, JWT, router).""" + +from __future__ import annotations + +import jwt +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from app.features.auth.router import router as auth_router +from app.features.auth.schemas import LoginRequest, TokenResponse, UserPublic +from app.features.auth.service import ( + create_access_token, + decode_access_token, + hash_password, + verify_password, +) + +# --------------------------------------------------------------------------- +# Password helpers +# --------------------------------------------------------------------------- + + +def test_hash_and_verify_roundtrip() -> None: + password = "s3cur3-P@ssw0rd!" + hashed = hash_password(password) + assert hashed != password + assert verify_password(password, hashed) is True + assert verify_password("wrong-password", hashed) is False + + +def test_hash_is_salted() -> None: + password = "password123" + first = hash_password(password) + second = hash_password(password) + # Argon2 uses a random salt, so hashes must differ. + assert first != second + assert verify_password(password, first) is True + assert verify_password(password, second) is True + + +# --------------------------------------------------------------------------- +# JWT helpers +# --------------------------------------------------------------------------- + + +def test_create_and_decode_token_roundtrip(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("JWT_SECRET", "test-secret-for-unit-tests-32-chars") + monkeypatch.setenv("JWT_ALGORITHM", "HS256") + monkeypatch.setenv("JWT_EXPIRE_MINUTES", "60") + token = create_access_token("user@example.com") + assert isinstance(token, str) + assert token.count(".") == 2 # JWT header.payload.signature + payload = decode_access_token(token) + assert payload["sub"] == "user@example.com" + assert "exp" in payload + + +def test_create_token_with_extra_claims(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("JWT_SECRET", "test-secret-extra-32-chars-long!!") + token = create_access_token("user@example.com", extra={"role": "admin"}) + payload = decode_access_token(token) + assert payload["sub"] == "user@example.com" + assert payload["role"] == "admin" + + +def test_token_respects_algorithm_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("JWT_SECRET", "hs256-secret-32-chars-long-for-test!!") + monkeypatch.setenv("JWT_ALGORITHM", "HS256") + token = create_access_token("a@b.co") + header = jwt.get_unverified_header(token) + assert header["alg"] == "HS256" + + +def test_token_invalid_after_secret_change(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("JWT_SECRET", "first-secret-32-chars-long-for-test!!") + token = create_access_token("a@b.co") + monkeypatch.setenv("JWT_SECRET", "different-secret-32-chars-long-test!!") + with pytest.raises(jwt.InvalidSignatureError): + decode_access_token(token) + + +def test_token_expiry_is_future(monkeypatch: pytest.MonkeyPatch) -> None: + from datetime import UTC, datetime + + monkeypatch.setenv("JWT_SECRET", "expiry-test-secret-32-chars-long!!") + monkeypatch.setenv("JWT_EXPIRE_MINUTES", "60") + before = datetime.now(UTC) + token = create_access_token("a@b.co") + payload = decode_access_token(token) + exp_ts = payload["exp"] + # exp is numeric timestamp; should be ~60 minutes after now. + exp_dt = datetime.fromtimestamp(exp_ts, tz=UTC) + delta = (exp_dt - before).total_seconds() + assert 3500 < delta < 3700 # allow small clock drift + + +# --------------------------------------------------------------------------- +# Schema validation +# --------------------------------------------------------------------------- + + +def test_login_request_accepts_valid() -> None: + req = LoginRequest(email="demo@example.com", password="password123") + assert req.email == "demo@example.com" + + +def test_login_request_rejects_short_password() -> None: + with pytest.raises(ValidationError): + LoginRequest(email="demo@example.com", password="short") + + +def test_login_request_rejects_invalid_email() -> None: + with pytest.raises(ValidationError): + LoginRequest(email="not-an-email", password="password123") + + +def test_token_response_defaults_to_bearer() -> None: + resp = TokenResponse(access_token="tok") + assert resp.token_type == "bearer" + assert resp.access_token == "tok" + + +def test_user_public_schema() -> None: + user = UserPublic(email="demo@example.com") + assert user.email == "demo@example.com" + + +# --------------------------------------------------------------------------- +# Router wiring +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def auth_client() -> TestClient: + app = FastAPI() + app.include_router(auth_router, prefix="/api/v1") + return TestClient(app) + + +def test_login_success_returns_token( + auth_client: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + # Use a sufficiently long secret to avoid PyJWT InsecureKeyLengthWarning + monkeypatch.setenv("JWT_SECRET", "w3-test-secret-32-chars-long-for-jwt!!") + response = auth_client.post( + "/api/v1/auth/login", + json={"email": "demo@example.com", "password": "password123"}, + ) + assert response.status_code == 200 + body = response.json() + assert "access_token" in body + assert body["token_type"] == "bearer" + # Token should decode to demo user + payload = decode_access_token(body["access_token"]) + assert payload["sub"] == "demo@example.com" + + +def test_login_wrong_password_returns_401(auth_client: TestClient) -> None: + response = auth_client.post( + "/api/v1/auth/login", + json={"email": "demo@example.com", "password": "wrongpass123"}, + ) + assert response.status_code == 401 + assert "Invalid credentials" in response.text + + +def test_login_wrong_email_returns_401(auth_client: TestClient) -> None: + response = auth_client.post( + "/api/v1/auth/login", + json={"email": "other@example.com", "password": "password123"}, + ) + assert response.status_code == 401 + + +def test_me_returns_demo_user(auth_client: TestClient) -> None: + response = auth_client.get("/api/v1/auth/me") + assert response.status_code == 200 + assert response.json()["email"] == "demo@example.com" + + +def test_login_rejects_invalid_payload_shape(auth_client: TestClient) -> None: + # Missing password triggers validation error (422 envelope via FastAPI) + response = auth_client.post( + "/api/v1/auth/login", + json={"email": "demo@example.com"}, + ) + assert response.status_code == 422 diff --git a/extensions/fastapi-sqlalchemy/template/tests/test_sqlalchemy.py b/extensions/fastapi-sqlalchemy/template/tests/test_sqlalchemy.py new file mode 100644 index 0000000..c56d2ce --- /dev/null +++ b/extensions/fastapi-sqlalchemy/template/tests/test_sqlalchemy.py @@ -0,0 +1,141 @@ +"""Tests for fastapi-sqlalchemy extension (session, Base, Alembic, get_db).""" + +from __future__ import annotations + +import os +from collections.abc import Generator + +import pytest +from sqlalchemy import Column, Integer, String, create_engine, text +from sqlalchemy.orm import Session + +from app.db.base import Base +from app.db.session import engine, get_db, session_factory + +# --------------------------------------------------------------------------- +# Declarative Base +# --------------------------------------------------------------------------- + + +def test_base_is_declarative() -> None: + assert hasattr(Base, "metadata") + assert hasattr(Base, "registry") + + +def test_model_can_register_on_base() -> None: + class TempModel(Base): + __tablename__ = "temp_model_w3_probe" + __allow_unmapped__ = True + id: int = Column(Integer, primary_key=True) # type: ignore[assignment] + name: str = Column(String(50)) # type: ignore[assignment] + + assert "temp_model_w3_probe" in Base.metadata.tables + # Cleanup so subsequent tests/alembic autogenerate don't see the probe + Base.metadata.remove(TempModel.__table__) # type: ignore[arg-type] # pyright: ignore[reportArgumentType] + + +# --------------------------------------------------------------------------- +# Engine and session factory (file-backed sqlite default) +# --------------------------------------------------------------------------- + + +def test_engine_is_created() -> None: + assert engine is not None + # Default URL is sqlite when DATABASE_URL unset + assert "sqlite" in str(engine.url) or "postgresql" in str(engine.url) + + +def test_session_factory_produces_session() -> None: + session = session_factory() + try: + assert isinstance(session, Session) + # smoke: execute a trivial query + result = session.execute(text("SELECT 1")) + assert result.scalar() == 1 + finally: + session.close() + + +def test_get_db_yields_and_closes() -> None: + gen: Generator[Session, None, None] = get_db() + db = next(gen) + assert isinstance(db, Session) + # Use the session + assert db.execute(text("SELECT 1")).scalar() == 1 + # Exhaust generator to trigger close + try: + next(gen) + pytest.fail("get_db should be a single-yield generator") + except StopIteration: + pass + # After close, session should be closed (is_active == False) + # SQLAlchemy 2.x: check closed state via `is_active` or `bind` + + +def test_in_memory_roundtrip_isolated() -> None: + """Isolated in-memory SQLite round-trip with a throwaway model.""" + + class IsolatedModel(Base): + __tablename__ = "isolated_w3_test" + __allow_unmapped__ = True + id: int = Column(Integer, primary_key=True) # type: ignore[assignment] + value: str = Column(String(100)) # type: ignore[assignment] + + memory_engine = create_engine("sqlite:///:memory:", future=True) + try: + Base.metadata.create_all( + memory_engine, tables=[IsolatedModel.__table__] # type: ignore[list-item] # pyright: ignore[reportArgumentType] + ) + with Session(memory_engine) as session: + session.add(IsolatedModel(value="hello-w3")) + session.commit() + rows = session.query(IsolatedModel).all() + assert len(rows) == 1 + assert rows[0].value == "hello-w3" + finally: + IsolatedModel.__table__.drop(memory_engine, checkfirst=True) # type: ignore[attr-defined] # pyright: ignore[reportAttributeAccessIssue] + Base.metadata.remove(IsolatedModel.__table__) # type: ignore[arg-type] # pyright: ignore[reportArgumentType] + memory_engine.dispose() + + +# --------------------------------------------------------------------------- +# Alembic wiring (no DB mutation, just config sanity) +# --------------------------------------------------------------------------- + + +def test_alembic_env_has_target_metadata() -> None: + from pathlib import Path + + env_path = Path(__file__).resolve().parents[1] / "alembic" / "env.py" + assert env_path.is_file(), f"alembic/env.py not found at {env_path}" + content = env_path.read_text(encoding="utf-8") + # Must import Base and wire target_metadata + assert "from app.db.base import Base" in content + assert "target_metadata = Base.metadata" in content + assert "DATABASE_URL" in content + # Verify actual target_metadata is Base.metadata (avoid executing alembic context) + assert Base.metadata is not None + + +def test_database_url_env_override(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DATABASE_URL", "sqlite:///:memory:") + # Re-import session module logic by checking os.getenv path + assert os.getenv("DATABASE_URL") == "sqlite:///:memory:" + mem_engine = create_engine(os.getenv("DATABASE_URL", ""), future=True) + try: + with mem_engine.connect() as conn: + assert conn.execute(text("SELECT 1")).scalar() == 1 + finally: + mem_engine.dispose() + + +def test_session_respects_sqlite_connect_args() -> None: + from pathlib import Path + + session_path = Path(__file__).resolve().parents[1] / "app" / "db" / "session.py" + assert session_path.is_file(), f"session.py not found at {session_path}" + content = session_path.read_text(encoding="utf-8") + # Scaffolded session.py must handle sqlite threading correctly + assert 'check_same_thread' in content + assert 'False' in content + assert 'sqlite' in content.lower()