fix(extensions): W3 add tests for auth-jwt and sqlalchemy, prep incompatibleWith

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:**
 

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

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

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()