From c3a8a36e7540888320a3c0c8291a0a4a2ac22a7f Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Thu, 15 Oct 2026 01:08:33 +0530 Subject: [PATCH 1/3] Apply better-py skill rules from the fastapi-template audit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the HIGH and MEDIUM findings from the audit of this repo against the better-py skill (skills/software/better-py in destack): - ruff.toml: drop the hardcoded target-version, remove TCH, stop ignoring ANN401, and ban typing.Any, typing.TYPE_CHECKING, getattr, setattr and hasattr (with an explicit per-file exception for app/auth/openapi.py, matching the existing architecture-test allowlist). ty.toml: drop the hardcoded python-version so both tools read it from requires-python. - Remove the remaining typing.Any uses: a named OpenAPIDocument alias in app/auth/openapi.py, pydantic models instead of raw dict/.json() handling in the entity API tests, the OpenAPI contract test, the Keycloak setup script and the scenario-test client, and a typed TokenClaimOverrides/TokenFactory pair for the integration token helpers instead of **claims: Any. - Type the behave scenario-test context properly (a runner.pyi stub with every attribute used) instead of Any, and replace the getattr fallback in environment.py with context.add_cleanup. - Remove the one remaining if TYPE_CHECKING: block (app/repository/base.py); UUID is not a cycle. - Flatten the four test classes in tests/unit/settings/test_settings.py into plain functions, per better-py's testing.md. - Smaller idiom fixes: Self instead of a quoted return type on two alternate constructors, str.partition instead of str.split, drop a walrus assignment, make the pagination parameters keyword-only, name the repeated entity field constraints, use TypeAdapter instead of cast in the exception tests, and raise OIDCDiscoveryError instead of a bare RuntimeError from OIDC discovery. - seed/__main__.py reads DatabaseSettings instead of os.environ directly, and main() now returns an int with raise SystemExit(main()), matching app/main.py and the skill's script.md. - The OpenAPI snapshot update flag is now a pytest --update-openapi-snapshot option instead of an env var read inside the test. - Makefile: lint and format now also cover .docker/. Deliberately left unchanged, with the reason: - alembic/env.py's os.environ.get("DATABASE_URL") fallback: switching it to DatabaseSettings would make the existing try/except ModuleNotFoundError fallback (which lets alembic run before "app" is importable) inconsistent, and I can't run `make test-integration` here (no Docker) to verify a change to a migration entry point. - .docker/configure_keycloak_client.py's remaining os.environ reads: it configures Keycloak, not the application, so it doesn't belong behind app.settings. - tests/integration/conftest.py's INTEGRATION_TEST_CONTAINER_PASSWORD read: a test-only container password override, not app config. - Docker Compose services already run python in exec form as PID 1 (no shell wrapper), so SIGTERM is delivered without an init process. - Several JUDGEMENT-level findings from the audit (mixins on the SQLAlchemy models, the FastAPI subclass, package-by-layer under import-linter, session-scoped fixtures, generality the tests alone exercise) are architecture choices this template makes on purpose, not violations. Verified: make check (ruff, ruff format, ty --error-on-warning, import-linter, 167 unit tests), make lint-typos. make test-integration was not run (needs Docker, unavailable in this environment); it is required by CI and by this repo's own PR template for database-layer changes, so CI will need to confirm the seed/__main__.py and tests/integration/conftest.py and tests/integration/test_entities_api.py changes. 🤖 Generated with Claude Code Co-Authored-By: Claude --- .docker/configure_keycloak_client.py | 7 +- Makefile | 10 +- app/auth/__init__.py | 3 +- app/auth/openapi.py | 16 +- app/io/entity.py | 13 +- app/logger/configuration.py | 2 +- app/logger/filters.py | 2 +- app/main.py | 5 +- app/repository/base.py | 7 +- app/service/entity.py | 2 + app/settings/__init__.py | 6 +- app/settings/authentication.py | 14 +- app/telemetry.py | 3 +- ruff.toml | 9 +- scenario-tests/app/client.py | 25 +- scenario-tests/features/environment.py | 18 +- scenario-tests/features/steps/entity_steps.py | 101 ++--- scenario-tests/typings/behave/__init__.pyi | 17 +- scenario-tests/typings/behave/runner.pyi | 21 + seed/__main__.py | 14 +- tests/conftest.py | 15 +- tests/integration/conftest.py | 26 +- tests/integration/test_entities_api.py | 61 ++- tests/unit/exceptions/test_exceptions.py | 8 +- .../service/test_entity_service_behaviour.py | 1 + tests/unit/settings/test_settings.py | 382 +++++++++--------- tests/unit/test_openapi_contract.py | 15 +- ty.toml | 1 - 28 files changed, 452 insertions(+), 352 deletions(-) create mode 100644 scenario-tests/typings/behave/runner.pyi diff --git a/.docker/configure_keycloak_client.py b/.docker/configure_keycloak_client.py index cbab49f..c2c22c3 100644 --- a/.docker/configure_keycloak_client.py +++ b/.docker/configure_keycloak_client.py @@ -4,7 +4,6 @@ import logging import os import time -from typing import Any import httpx @@ -42,7 +41,7 @@ def _get_keycloak_client_representation( client: httpx.Client, realm: str, client_id: str, -) -> dict[str, Any]: +) -> dict[str, object]: response = client.get( f"/admin/realms/{realm}/clients", params={"clientId": client_id}, @@ -68,9 +67,7 @@ def configure_keycloak_client() -> None: realm, client_id, ) - representation["redirectUris"] = [ - f"{app_public_url}/docs/oauth2-redirect" - ] + representation["redirectUris"] = [f"{app_public_url}/docs/oauth2-redirect"] representation["webOrigins"] = [app_public_url] response = client.put( f"/admin/realms/{realm}/clients/{representation['id']}", diff --git a/Makefile b/Makefile index f8477c6..56bd553 100644 --- a/Makefile +++ b/Makefile @@ -35,10 +35,10 @@ check: lint test-unit lint: lint-ruff lint-format lint-ty lint-imports lint-typos lint-ruff: - uv run --group lint --all-packages ruff check app tests alembic seed scenario-tests + uv run --group lint --all-packages ruff check app tests alembic seed scenario-tests .docker lint-format: - uv run --group lint --all-packages ruff format --check app tests alembic seed scenario-tests + uv run --group lint --all-packages ruff format --check app tests alembic seed scenario-tests .docker lint-ty: uv run --group lint --group migration --group test --all-packages ty check --error-on-warning app tests alembic seed scenario-tests @@ -50,8 +50,8 @@ lint-typos: uvx typos format: - uv run --group lint --all-packages ruff check --fix app tests alembic seed scenario-tests - uv run --group lint --all-packages ruff format app tests alembic seed scenario-tests + uv run --group lint --all-packages ruff check --fix app tests alembic seed scenario-tests .docker + uv run --group lint --all-packages ruff format app tests alembic seed scenario-tests .docker run: @if [ ! -f .env ]; then \ @@ -90,4 +90,4 @@ test-cov: uv run --group test pytest --cov --cov-report=term-missing openapi-snapshot: - UPDATE_OPENAPI_SNAPSHOT=1 uv run --group test pytest tests/unit/test_openapi_contract.py -q + uv run --group test pytest tests/unit/test_openapi_contract.py -q --update-openapi-snapshot diff --git a/app/auth/__init__.py b/app/auth/__init__.py index 15879e8..bed9c8e 100644 --- a/app/auth/__init__.py +++ b/app/auth/__init__.py @@ -5,7 +5,7 @@ authenticate_request, get_token_validator, ) -from app.auth.openapi import OIDCOpenAPIFastAPI +from app.auth.openapi import OIDCOpenAPIFastAPI, OpenAPIDocument from app.auth.token_validator import AccessTokenValidator __all__: list[str] = [ @@ -14,6 +14,7 @@ "AccessTokenValidator", "AuthClaims", "OIDCOpenAPIFastAPI", + "OpenAPIDocument", "authenticate_request", "get_token_validator", "require_role", diff --git a/app/auth/openapi.py b/app/auth/openapi.py index 4986bf5..f621bb9 100644 --- a/app/auth/openapi.py +++ b/app/auth/openapi.py @@ -10,6 +10,8 @@ from app.exceptions import PROBLEM_JSON_MEDIA_TYPE, ProblemDetails from app.settings import OIDCMetadata, Settings +type OpenAPIDocument = dict[str, Any] + _JSON_MEDIA_TYPE = "application/json" _ERROR_STATUS_PREFIXES = ("4", "5") _GENERATED_VALIDATION_STATUS = str(HTTPStatus.UNPROCESSABLE_CONTENT.value) @@ -19,7 +21,7 @@ _VALIDATION_DESCRIPTION = "Validation Error" -def _use_problem_json_for_error_responses(schema: dict[str, Any]) -> None: +def _use_problem_json_for_error_responses(schema: OpenAPIDocument) -> None: for operations in schema.get("paths", {}).values(): for operation in operations.values(): for status, response in operation.get("responses", {}).items(): @@ -30,7 +32,7 @@ def _use_problem_json_for_error_responses(schema: dict[str, Any]) -> None: content[PROBLEM_JSON_MEDIA_TYPE] = content.pop(_JSON_MEDIA_TYPE) -def _is_generated_validation_response(response: dict[str, Any]) -> bool: +def _is_generated_validation_response(response: OpenAPIDocument) -> bool: return response == { "description": _VALIDATION_DESCRIPTION, "content": { @@ -39,7 +41,7 @@ def _is_generated_validation_response(response: dict[str, Any]) -> bool: } -def _add_problem_details(response: dict[str, Any]) -> None: +def _add_problem_details(response: OpenAPIDocument) -> None: problem_ref = {"$ref": _PROBLEM_DETAILS_REF} media = response.setdefault("content", {}).setdefault(PROBLEM_JSON_MEDIA_TYPE, {}) declared = media.get("schema") @@ -51,7 +53,7 @@ def _add_problem_details(response: dict[str, Any]) -> None: def _drop_unreferenced_components( - schema: dict[str, Any], names: tuple[str, ...] + schema: OpenAPIDocument, names: tuple[str, ...] ) -> None: components = schema.get("components", {}).get("schemas", {}) for name in names: @@ -61,7 +63,7 @@ def _drop_unreferenced_components( del components[name] -def _declare_validation_responses(schema: dict[str, Any], status: HTTPStatus) -> None: +def _declare_validation_responses(schema: OpenAPIDocument, status: HTTPStatus) -> None: rewritten = False for operations in schema.get("paths", {}).values(): for operation in operations.values(): @@ -94,7 +96,7 @@ def _declare_validation_responses(schema: dict[str, Any], status: HTTPStatus) -> def _use_discovered_oidc_endpoints( - schema: dict[str, Any], + schema: OpenAPIDocument, metadata: OIDCMetadata, ) -> None: try: @@ -129,7 +131,7 @@ def __init__( self.validation_status = validation_status self.oidc_metadata: OIDCMetadata | None = None - def openapi(self) -> dict[str, Any]: + def openapi(self) -> OpenAPIDocument: if self.openapi_schema is not None: return self.openapi_schema schema = super().openapi() diff --git a/app/io/entity.py b/app/io/entity.py index 530c583..78b9482 100644 --- a/app/io/entity.py +++ b/app/io/entity.py @@ -1,4 +1,5 @@ from enum import StrEnum +from typing import Annotated from uuid import UUID from pydantic import AwareDatetime, BaseModel, ConfigDict, Field @@ -16,18 +17,22 @@ class OrderDirection(StrEnum): DESC = "desc" +EntityName = Annotated[str, Field(min_length=1, max_length=255)] +EntityDescription = Annotated[str | None, Field(max_length=5000)] + + class EntityCreate(BaseModel): model_config = ConfigDict(extra="forbid") - name: str = Field(..., min_length=1, max_length=255) - description: str | None = Field(None, max_length=5000) + name: EntityName + description: EntityDescription = None class EntityUpdate(BaseModel): model_config = ConfigDict(extra="forbid") - name: str = Field(..., min_length=1, max_length=255) - description: str | None = Field(None, max_length=5000) + name: EntityName + description: EntityDescription = None class EntityResponse(BaseModel): diff --git a/app/logger/configuration.py b/app/logger/configuration.py index 55a9c8a..92275eb 100644 --- a/app/logger/configuration.py +++ b/app/logger/configuration.py @@ -4,7 +4,7 @@ from app.logger.handlers import LOG_LEVEL, get_logfire_handler -LogHandlerFactory = Callable[[], Handler] +type LogHandlerFactory = Callable[[], Handler] __all__ = ["LogHandlerFactory", "configure_logger", "setup_logging"] diff --git a/app/logger/filters.py b/app/logger/filters.py index fe11e2c..f03481e 100644 --- a/app/logger/filters.py +++ b/app/logger/filters.py @@ -12,7 +12,7 @@ def _record_path(record: logging.LogRecord) -> str | None: if isinstance(args, tuple) and len(args) > _UVICORN_ACCESS_PATH_ARG_INDEX: path = args[_UVICORN_ACCESS_PATH_ARG_INDEX] if isinstance(path, str): - return path.split("?", maxsplit=1)[0] + return path.partition("?")[0] message = record.getMessage() for path in HEALTH_ENDPOINT_PATHS: diff --git a/app/main.py b/app/main.py index c434ddc..042701e 100644 --- a/app/main.py +++ b/app/main.py @@ -94,7 +94,7 @@ def app_from_env() -> OIDCOpenAPIFastAPI: return create_app(Settings.from_env()) -def main() -> None: +def main() -> int: settings = Settings.from_env() telemetry.configure_otel(settings.app.app_name) logger.setup_logging() @@ -108,7 +108,8 @@ def main() -> None: proxy_headers=settings.app.proxy_headers, forwarded_allow_ips=settings.app.forwarded_allow_ips, ) + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/app/repository/base.py b/app/repository/base.py index 1c61774..b65bb6c 100644 --- a/app/repository/base.py +++ b/app/repository/base.py @@ -1,6 +1,7 @@ from collections.abc import Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, overload +from typing import overload +from uuid import UUID from sqlalchemy import ColumnElement, delete, func, select from sqlalchemy.ext.asyncio import AsyncSession @@ -8,9 +9,6 @@ from app.model.base import IntegerPrimaryKey, UUIDPrimaryKey -if TYPE_CHECKING: - from uuid import UUID - @dataclass(frozen=True, slots=True) class Ordering: @@ -54,6 +52,7 @@ async def find_all_by_id(self, ids: Sequence[KeyType]) -> Sequence[ModelType]: async def find_all_paginated( self, + *, offset: int = 0, limit: int = 25, ordering: Ordering | None = None, diff --git a/app/service/entity.py b/app/service/entity.py index ae890e2..cbfc9f6 100644 --- a/app/service/entity.py +++ b/app/service/entity.py @@ -24,6 +24,7 @@ async def find_by_id(self, entity_id: UUID) -> Entity | None: ... async def find_all_paginated( self, + *, offset: int = 0, limit: int = 25, ordering: Ordering | None = None, @@ -50,6 +51,7 @@ async def get_by_id(self, entity_id: UUID) -> Entity: async def get_all( self, + *, offset: int = 0, limit: int = 25, order_by: EntityOrderBy = EntityOrderBy.CREATED_AT, diff --git a/app/settings/__init__.py b/app/settings/__init__.py index fb2631c..f99e13f 100644 --- a/app/settings/__init__.py +++ b/app/settings/__init__.py @@ -1,8 +1,11 @@ +from typing import Self + from pydantic import BaseModel, ConfigDict from app.settings.app import ApplicationSettings from app.settings.authentication import ( AuthNSettings, + OIDCDiscoveryError, OIDCMetadata, resolve_oidc_metadata, ) @@ -17,7 +20,7 @@ class Settings(BaseModel): oidc: AuthNSettings @classmethod - def from_env(cls) -> "Settings": + def from_env(cls) -> Self: return cls( app=ApplicationSettings(), database=DatabaseSettings(), @@ -29,6 +32,7 @@ def from_env(cls) -> "Settings": "ApplicationSettings", "AuthNSettings", "DatabaseSettings", + "OIDCDiscoveryError", "OIDCMetadata", "Settings", "resolve_oidc_metadata", diff --git a/app/settings/authentication.py b/app/settings/authentication.py index ec01abb..821a335 100644 --- a/app/settings/authentication.py +++ b/app/settings/authentication.py @@ -1,4 +1,5 @@ import asyncio +from typing import Self import httpx from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -10,6 +11,10 @@ _DISCOVERY_TIMEOUT = httpx.Timeout(5.0, connect=2.0) +class OIDCDiscoveryError(RuntimeError): + pass + + class OIDCMetadata(BaseModel): model_config = ConfigDict(frozen=True) @@ -40,7 +45,7 @@ class AuthNSettings(BaseSettings): token_endpoint: str | None = None @model_validator(mode="after") - def _validate_endpoint_overrides(self) -> "AuthNSettings": + def _validate_endpoint_overrides(self) -> Self: overrides = ( self.jwks_uri, self.issuer, @@ -77,7 +82,7 @@ def _replace_url_base(url: str, source_base: str, target_base: str) -> str: if url == source: return target if url.startswith(f"{source}/"): - return f"{target}{url[len(source) :]}" + return f"{target}{url.removeprefix(source)}" return url @@ -145,7 +150,7 @@ async def _discover_oidc_metadata( await asyncio.sleep(retry_delay_seconds * attempt) message = f"OIDC discovery failed after {attempts} attempts: {discovery_url}" - raise RuntimeError(message) from last_error + raise OIDCDiscoveryError(message) from last_error async def resolve_oidc_metadata( @@ -155,7 +160,8 @@ async def resolve_oidc_metadata( attempts: int = _DISCOVERY_ATTEMPTS, retry_delay_seconds: float = _DISCOVERY_RETRY_DELAY_SECONDS, ) -> OIDCMetadata: - if override := settings.metadata_override(): + override = settings.metadata_override() + if override is not None: return override if attempts < 1: message = "OIDC discovery attempts must be at least 1" diff --git a/app/telemetry.py b/app/telemetry.py index adf60b7..46f9655 100644 --- a/app/telemetry.py +++ b/app/telemetry.py @@ -67,7 +67,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ): await self.app(scope, receive, send) return - if client := scope.get("client"): + client = scope.get("client") + if client: span.set_attribute("client.address", client[0]) token = context.attach(context.set_value(_SERVER_SPAN_KEY, span)) try: diff --git a/ruff.toml b/ruff.toml index 1f55ae4..daf6d54 100644 --- a/ruff.toml +++ b/ruff.toml @@ -1,4 +1,3 @@ -target-version = "py313" line-length = 88 [lint] @@ -12,7 +11,6 @@ select = [ "B", "A", "SIM", - "TCH", "RUF", "FAST", "ASYNC", @@ -54,18 +52,23 @@ ignore = [ "E501", "TRY003", "COM812", - "ANN401", ] [lint.per-file-ignores] "tests/**" = ["S101", "ARG001", "PLC0415", "PLR2004", "SLF001"] "alembic/**" = ["ERA001", "ANN", "INP"] +"app/auth/openapi.py" = ["ANN401", "TID251"] [lint.flake8-tidy-imports] ban-relative-imports = "all" [lint.flake8-tidy-imports.banned-api] "__future__.annotations".msg = "Python 3.13+ does not need `from __future__ import annotations`." +"builtins.getattr".msg = "Use direct attribute access or a typed mapping." +"builtins.setattr".msg = "Use direct attribute assignment." +"builtins.hasattr".msg = "Use direct attribute access or isinstance narrowing." +"typing.Any".msg = "Use object or a precise type. app/auth/openapi.py is the only allowed exception." +"typing.TYPE_CHECKING".msg = "Import plainly. An import cycle means the shared types need their own module." [lint.flake8-quotes] inline-quotes = "double" diff --git a/scenario-tests/app/client.py b/scenario-tests/app/client.py index a68771c..0cbb30b 100644 --- a/scenario-tests/app/client.py +++ b/scenario-tests/app/client.py @@ -1,11 +1,20 @@ import time -from typing import Any +from http import HTTPStatus import httpx +from pydantic import BaseModel, TypeAdapter from app.config import ScenarioTestSettings -HTTP_OK = 200 + +class TokenResponse(BaseModel): + access_token: str = "" + + +class EntityBody(BaseModel): + id: str + name: str + description: str | None class ScenarioTestClient: @@ -26,7 +35,7 @@ def wait_until_ready(self, attempts: int = 30, delay_seconds: float = 2.0) -> No for _ in range(attempts): try: response = self.client.get(self.settings.health_endpoint) - if response.status_code == HTTP_OK: + if response.status_code == HTTPStatus.OK: return except httpx.HTTPError as exc: last_error = exc @@ -52,7 +61,7 @@ def create_access_token(self) -> str: ) response.raise_for_status() - access_token = response.json().get("access_token") + access_token = TokenResponse.model_validate_json(response.content).access_token if not access_token: message = "Keycloak did not return an access token" raise RuntimeError(message) @@ -87,5 +96,9 @@ def delete_entity(self, entity_id: str) -> httpx.Response: return self.client.delete(f"/entities/{entity_id}") @staticmethod - def response_json(response: httpx.Response) -> Any: - return response.json() + def entity_from(response: httpx.Response) -> EntityBody: + return EntityBody.model_validate_json(response.content) + + @staticmethod + def entities_from(response: httpx.Response) -> list[EntityBody]: + return TypeAdapter(list[EntityBody]).validate_json(response.content) diff --git a/scenario-tests/features/environment.py b/scenario-tests/features/environment.py index 6183cc0..ac82a50 100644 --- a/scenario-tests/features/environment.py +++ b/scenario-tests/features/environment.py @@ -1,17 +1,13 @@ -from typing import Any +from behave.runner import Context from app.client import ScenarioTestClient from app.config import ScenarioTestSettings -def before_all(context: Any) -> None: +def before_all(context: Context) -> None: context.settings = ScenarioTestSettings() - context.scenario_client = ScenarioTestClient(context.settings) - context.scenario_client.wait_until_ready() - context.access_token = context.scenario_client.create_access_token() - - -def after_all(context: Any) -> None: - scenario_client = getattr(context, "scenario_client", None) - if scenario_client is not None: - scenario_client.close() + scenario_client = ScenarioTestClient(context.settings) + context.add_cleanup(scenario_client.close) + context.scenario_client = scenario_client + scenario_client.wait_until_ready() + context.access_token = scenario_client.create_access_token() diff --git a/scenario-tests/features/steps/entity_steps.py b/scenario-tests/features/steps/entity_steps.py index f07c060..b575ad9 100644 --- a/scenario-tests/features/steps/entity_steps.py +++ b/scenario-tests/features/steps/entity_steps.py @@ -1,13 +1,18 @@ -import json from base64 import urlsafe_b64decode -from typing import Any +from http import HTTPStatus from behave import given, then, when +from behave.runner import Context +from pydantic import BaseModel -HTTP_OK = 200 -HTTP_CREATED = 201 -HTTP_NO_CONTENT = 204 -HTTP_NOT_FOUND = 404 + +class RealmAccess(BaseModel): + roles: list[str] + + +class AccessTokenClaims(BaseModel): + realm_access: RealmAccess + resource_access: dict[str, object] | None = None def require(condition: object, message: str = "Scenario assertion failed") -> None: @@ -16,68 +21,73 @@ def require(condition: object, message: str = "Scenario assertion failed") -> No @given("the Scenario Tests application is configured") -def step_app_application_is_configured(context: Any) -> None: +def step_app_application_is_configured(context: Context) -> None: require(context.scenario_client is not None) require(bool(context.access_token)) @then("the access token should grant only the entity write role") -def step_access_token_grants_only_entity_write(context: Any) -> None: +def step_access_token_grants_only_entity_write(context: Context) -> None: payload = context.access_token.split(".")[1] - claims = json.loads(urlsafe_b64decode(payload + "=" * (-len(payload) % 4))) - require(claims["realm_access"]["roles"] == ["entities:write"]) - require(not claims.get("resource_access")) + claims = AccessTokenClaims.model_validate_json( + urlsafe_b64decode(payload + "=" * (-len(payload) % 4)) + ) + require(claims.realm_access.roles == ["entities:write"]) + require(not claims.resource_access) @when('I create an entity named "{name}" with description "{description}"') -def step_create_entity(context: Any, name: str, description: str) -> None: +def step_create_entity(context: Context, name: str, description: str) -> None: context.expected_name = name context.expected_description = description context.create_response = context.scenario_client.create_entity(name, description) @then("the entity creation should succeed") -def step_entity_creation_should_succeed(context: Any) -> None: +def step_entity_creation_should_succeed(context: Context) -> None: require( - context.create_response.status_code == HTTP_CREATED, + context.create_response.status_code == HTTPStatus.CREATED, context.create_response.text, ) - entity = context.scenario_client.response_json(context.create_response) - require(bool(entity["id"])) - require(entity["name"] == context.expected_name) - require(entity["description"] == context.expected_description) - context.entity_id = entity["id"] - context.created_entity = entity + entity = context.scenario_client.entity_from(context.create_response) + require(bool(entity.id)) + require(entity.name == context.expected_name) + require(entity.description == context.expected_description) + context.entity_id = entity.id @when("I fetch the created entity") -def step_fetch_created_entity(context: Any) -> None: +def step_fetch_created_entity(context: Context) -> None: context.get_response = context.scenario_client.get_entity(context.entity_id) @then("the fetched entity should match the created entity") -def step_fetched_entity_should_match(context: Any) -> None: - require(context.get_response.status_code == HTTP_OK, context.get_response.text) - entity = context.scenario_client.response_json(context.get_response) - require(entity["id"] == context.entity_id) - require(entity["name"] == context.expected_name) - require(entity["description"] == context.expected_description) +def step_fetched_entity_should_match(context: Context) -> None: + require( + context.get_response.status_code == HTTPStatus.OK, context.get_response.text + ) + entity = context.scenario_client.entity_from(context.get_response) + require(entity.id == context.entity_id) + require(entity.name == context.expected_name) + require(entity.description == context.expected_description) @when("I list entities") -def step_list_entities(context: Any) -> None: +def step_list_entities(context: Context) -> None: context.list_response = context.scenario_client.list_entities() @then("the created entity should be included in the entity list") -def step_created_entity_should_be_listed(context: Any) -> None: - require(context.list_response.status_code == HTTP_OK, context.list_response.text) - entities = context.scenario_client.response_json(context.list_response) - require(any(entity["id"] == context.entity_id for entity in entities)) +def step_created_entity_should_be_listed(context: Context) -> None: + require( + context.list_response.status_code == HTTPStatus.OK, context.list_response.text + ) + entities = context.scenario_client.entities_from(context.list_response) + require(any(entity.id == context.entity_id for entity in entities)) @when('I update the entity to name "{name}" with no description') -def step_update_entity_with_no_description(context: Any, name: str) -> None: +def step_update_entity_with_no_description(context: Context, name: str) -> None: context.expected_name = name context.expected_description = None context.update_response = context.scenario_client.update_entity( @@ -88,37 +98,38 @@ def step_update_entity_with_no_description(context: Any, name: str) -> None: @then("the entity update should succeed") -def step_entity_update_should_succeed(context: Any) -> None: +def step_entity_update_should_succeed(context: Context) -> None: require( - context.update_response.status_code == HTTP_OK, context.update_response.text + context.update_response.status_code == HTTPStatus.OK, + context.update_response.text, ) - entity = context.scenario_client.response_json(context.update_response) - require(entity["id"] == context.entity_id) - require(entity["name"] == context.expected_name) - require(entity["description"] is None) + entity = context.scenario_client.entity_from(context.update_response) + require(entity.id == context.entity_id) + require(entity.name == context.expected_name) + require(entity.description is None) @when("I delete the created entity") -def step_delete_created_entity(context: Any) -> None: +def step_delete_created_entity(context: Context) -> None: context.delete_response = context.scenario_client.delete_entity(context.entity_id) @then("the entity deletion should succeed") -def step_entity_deletion_should_succeed(context: Any) -> None: +def step_entity_deletion_should_succeed(context: Context) -> None: require( - context.delete_response.status_code == HTTP_NO_CONTENT, + context.delete_response.status_code == HTTPStatus.NO_CONTENT, context.delete_response.text, ) @when("I fetch the deleted entity") -def step_fetch_deleted_entity(context: Any) -> None: +def step_fetch_deleted_entity(context: Context) -> None: context.deleted_get_response = context.scenario_client.get_entity(context.entity_id) @then("the deleted entity should not be found") -def step_deleted_entity_should_not_be_found(context: Any) -> None: +def step_deleted_entity_should_not_be_found(context: Context) -> None: require( - context.deleted_get_response.status_code == HTTP_NOT_FOUND, + context.deleted_get_response.status_code == HTTPStatus.NOT_FOUND, context.deleted_get_response.text, ) diff --git a/scenario-tests/typings/behave/__init__.pyi b/scenario-tests/typings/behave/__init__.pyi index b2ca5ac..e438568 100644 --- a/scenario-tests/typings/behave/__init__.pyi +++ b/scenario-tests/typings/behave/__init__.pyi @@ -1,19 +1,14 @@ from collections.abc import Callable -from typing import ParamSpec, TypeVar -P = ParamSpec("P") -R = TypeVar("R") -StepFunction = Callable[P, R] - -def given( +def given[**P, R]( step_text: str, **kwargs: object, -) -> Callable[[StepFunction[P, R]], StepFunction[P, R]]: ... -def when( +) -> Callable[[Callable[P, R]], Callable[P, R]]: ... +def when[**P, R]( step_text: str, **kwargs: object, -) -> Callable[[StepFunction[P, R]], StepFunction[P, R]]: ... -def then( +) -> Callable[[Callable[P, R]], Callable[P, R]]: ... +def then[**P, R]( step_text: str, **kwargs: object, -) -> Callable[[StepFunction[P, R]], StepFunction[P, R]]: ... +) -> Callable[[Callable[P, R]], Callable[P, R]]: ... diff --git a/scenario-tests/typings/behave/runner.pyi b/scenario-tests/typings/behave/runner.pyi new file mode 100644 index 0000000..d212c1c --- /dev/null +++ b/scenario-tests/typings/behave/runner.pyi @@ -0,0 +1,21 @@ +from collections.abc import Callable + +import httpx + +from app.client import ScenarioTestClient +from app.config import ScenarioTestSettings + +class Context: + settings: ScenarioTestSettings + scenario_client: ScenarioTestClient + access_token: str + expected_name: str + expected_description: str | None + entity_id: str + create_response: httpx.Response + get_response: httpx.Response + list_response: httpx.Response + update_response: httpx.Response + delete_response: httpx.Response + deleted_get_response: httpx.Response + def add_cleanup(self, cleanup_func: Callable[[], object]) -> None: ... diff --git a/seed/__main__.py b/seed/__main__.py index 3a23cec..6211ccd 100644 --- a/seed/__main__.py +++ b/seed/__main__.py @@ -1,21 +1,18 @@ import logging -import os from sqlalchemy import create_engine, pool +from app.settings import DatabaseSettings from seed import seed from seed.fixtures import FIXTURES logger = logging.getLogger("seed") -def main() -> None: +def main() -> int: logging.basicConfig(level=logging.INFO, format="%(message)s") - database_url = os.environ.get("DATABASE_URL") - if not database_url: - msg = "DATABASE_URL must be set to seed the database" - raise RuntimeError(msg) - engine = create_engine(database_url, poolclass=pool.NullPool) + database = DatabaseSettings() + engine = create_engine(database.url, poolclass=pool.NullPool) try: with engine.begin() as connection: inserted = seed(connection, FIXTURES) @@ -23,7 +20,8 @@ def main() -> None: engine.dispose() for table, count in inserted.items(): logger.info("%s: %d new rows", table, count) + return 0 if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/tests/conftest.py b/tests/conftest.py index 8706173..2999da7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,7 @@ import asyncio import selectors import sys -from typing import Any +from collections.abc import Callable import pytest @@ -14,8 +14,19 @@ def selector_event_loop() -> asyncio.AbstractEventLoop: return asyncio.SelectorEventLoop(selectors.SelectSelector()) +def pytest_addoption(parser: pytest.Parser) -> None: + parser.addoption( + "--update-openapi-snapshot", + action="store_true", + default=False, + help="Regenerate tests/contract/openapi.json instead of comparing to it.", + ) + + @pytest.fixture -def anyio_backend() -> str | tuple[str, dict[str, Any]]: +def anyio_backend() -> ( + str | tuple[str, dict[str, Callable[[], asyncio.AbstractEventLoop]]] +): if sys.platform == "win32": return "asyncio", {"loop_factory": selector_event_loop} return "asyncio" diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 89c4351..48bc7c9 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -3,9 +3,9 @@ import hmac import json import time -from collections.abc import AsyncGenerator, Callable, Iterator, Sequence +from collections.abc import AsyncGenerator, Iterator from os import environ -from typing import Any +from typing import Protocol, TypedDict, Unpack import httpx import jwt as pyjwt @@ -38,6 +38,18 @@ WithAuth = pytest.mark.with_auth +class TokenClaimOverrides(TypedDict, total=False): + audience: str + issuer: str + expires_in_seconds: int + not_before_in_seconds: int + roles: tuple[str, ...] + + +class TokenFactory(Protocol): + def __call__(self, **claims: Unpack[TokenClaimOverrides]) -> str: ... + + @pytest.fixture(scope="session") def postgres_container() -> Iterator[PostgresContainer]: with PostgresContainer( @@ -140,7 +152,7 @@ def token_claims( issuer: str = TEST_ISSUER, expires_in_seconds: int = 3600, not_before_in_seconds: int = 0, - roles: Sequence[str] = (ENTITY_WRITE_ROLE,), + roles: tuple[str, ...] = (ENTITY_WRITE_ROLE,), ) -> dict[str, object]: now = int(time.time()) return { @@ -159,7 +171,7 @@ def make_test_token( private_key: RSAPrivateKey, *, kid: str = TEST_KID, - **claims: Any, + **claims: Unpack[TokenClaimOverrides], ) -> str: return pyjwt.encode( token_claims(**claims), @@ -169,7 +181,7 @@ def make_test_token( ) -def make_unsigned_token(**claims: Any) -> str: +def make_unsigned_token(**claims: Unpack[TokenClaimOverrides]) -> str: return pyjwt.encode( token_claims(**claims), key="", @@ -196,8 +208,8 @@ def make_hs256_token_signed_with_public_key(private_key: RSAPrivateKey) -> str: @pytest.fixture -def token_factory(signing_private_key: RSAPrivateKey) -> Callable[..., str]: - def factory(**claims: Any) -> str: +def token_factory(signing_private_key: RSAPrivateKey) -> TokenFactory: + def factory(**claims: Unpack[TokenClaimOverrides]) -> str: return make_test_token(signing_private_key, **claims) return factory diff --git a/tests/integration/test_entities_api.py b/tests/integration/test_entities_api.py index ab8f214..985685b 100644 --- a/tests/integration/test_entities_api.py +++ b/tests/integration/test_entities_api.py @@ -1,12 +1,15 @@ from collections.abc import Callable -from typing import Any +from typing import TypedDict from uuid import uuid4 import httpx import pytest from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey +from pydantic import BaseModel from tests.integration.conftest import ( + TokenClaimOverrides, + TokenFactory, WithAuth, make_hs256_token_signed_with_public_key, make_test_token, @@ -16,19 +19,35 @@ pytestmark = pytest.mark.integration +class RequestOptions(TypedDict, total=False): + json: dict[str, str] + + +class ProblemBody(BaseModel): + type: str + title: str + status: int + instance: str + detail: str | list[dict[str, object]] + + +class ValidationErrorItem(BaseModel): + loc: tuple[str | int, ...] + + def _assert_problem_details( response: httpx.Response, *, status_code: int, title: str, -) -> dict[str, Any]: +) -> ProblemBody: assert response.status_code == status_code assert response.headers["content-type"] == "application/problem+json" - body = response.json() - assert body["title"] == title - assert body["status"] == status_code - assert body["type"].endswith("/openapi.json") - assert body["instance"] == str(response.request.url) + body = ProblemBody.model_validate_json(response.content) + assert body.title == title + assert body.status == status_code + assert body.type.endswith("/openapi.json") + assert body.instance == str(response.request.url) return body @@ -42,9 +61,9 @@ def _assert_validation_problem( status_code=422, title="Unprocessable Content", ) - assert any(tuple(error["loc"]) == expected_location for error in body["detail"]), ( - body["detail"] - ) + assert isinstance(body.detail, list), body.detail + errors = [ValidationErrorItem.model_validate(error) for error in body.detail] + assert any(error.loc == expected_location for error in errors), body.detail def _assert_unauthorized(response: httpx.Response) -> None: @@ -100,8 +119,8 @@ async def test_entity_routes_reject_malformed_authorization( ) async def test_entity_routes_reject_invalid_token_claims( app_client: httpx.AsyncClient, - token_factory: Callable[..., str], - token_claims: dict[str, Any], + token_factory: TokenFactory, + token_claims: TokenClaimOverrides, ) -> None: response = await app_client.post( "/entities/", @@ -162,22 +181,22 @@ async def test_entity_routes_reject_cryptographically_invalid_tokens( ) async def test_write_routes_require_the_entity_write_role( app_client: httpx.AsyncClient, - token_factory: Callable[..., str], + token_factory: TokenFactory, method: str, path: str, - kwargs: dict[str, Any], + kwargs: RequestOptions, ) -> None: headers = {"Authorization": f"Bearer {token_factory(roles=())}"} response = await app_client.request(method, path, headers=headers, **kwargs) body = _assert_problem_details(response, status_code=403, title="Forbidden") - assert body["detail"] == "Role 'entities:write' is required" + assert body.detail == "Role 'entities:write' is required" async def test_read_routes_do_not_require_the_entity_write_role( app_client: httpx.AsyncClient, - token_factory: Callable[..., str], + token_factory: TokenFactory, ) -> None: headers = {"Authorization": f"Bearer {token_factory(roles=())}"} @@ -298,7 +317,7 @@ async def test_delete_entity_removes_the_persisted_record( get_response = await app_client.get(f"/entities/{entity_id}") body = _assert_problem_details(get_response, status_code=404, title="Not Found") - assert body["detail"] == f"Entity '{entity_id}' not found" + assert body.detail == f"Entity '{entity_id}' not found" @WithAuth @@ -314,7 +333,7 @@ async def test_missing_entity_operations_return_not_found( app_client: httpx.AsyncClient, method: str, path_template: str, - kwargs: dict[str, Any], + kwargs: RequestOptions, ) -> None: entity_id = uuid4() @@ -323,7 +342,7 @@ async def test_missing_entity_operations_return_not_found( ) body = _assert_problem_details(response, status_code=404, title="Not Found") - assert body["detail"] == f"Entity '{entity_id}' not found" + assert body.detail == f"Entity '{entity_id}' not found" @WithAuth @@ -340,7 +359,7 @@ async def test_missing_entity_operations_return_not_found( ) async def test_create_entity_rejects_invalid_payloads( app_client: httpx.AsyncClient, - payload: dict[str, Any], + payload: dict[str, object], expected_location: tuple[str, ...], ) -> None: response = await app_client.post("/entities/", json=payload) @@ -400,7 +419,7 @@ async def test_get_entity_rejects_invalid_uuid( ) async def test_update_entity_rejects_invalid_payloads( app_client: httpx.AsyncClient, - payload: dict[str, Any], + payload: dict[str, object], expected_location: tuple[str, ...], ) -> None: response = await app_client.put(f"/entities/{uuid4()}", json=payload) diff --git a/tests/unit/exceptions/test_exceptions.py b/tests/unit/exceptions/test_exceptions.py index 7d4588f..77d57e9 100644 --- a/tests/unit/exceptions/test_exceptions.py +++ b/tests/unit/exceptions/test_exceptions.py @@ -1,11 +1,11 @@ -import json from http import HTTPStatus -from typing import cast, override +from typing import override import pytest from fastapi import Request from fastapi.exceptions import RequestValidationError from fastapi.responses import Response +from pydantic import TypeAdapter from sqlalchemy.exc import OperationalError from sqlalchemy.exc import TimeoutError as PoolTimeoutError from starlette.exceptions import HTTPException @@ -27,7 +27,7 @@ def load_json_body(response: Response) -> dict[str, object]: body = response.body raw_body = body.tobytes() if isinstance(body, memoryview) else body - return cast("dict[str, object]", json.loads(raw_body)) + return TypeAdapter(dict[str, object]).validate_json(raw_body) def make_request(path: str = "/entities") -> Request: @@ -95,7 +95,7 @@ def test_handler_maps_validation_errors_to_unprocessable_content() -> None: assert body["title"] == "Unprocessable Content" assert body["status"] == 422 assert body["instance"] == "https://testserver/entities" - detail = cast("list[dict[str, object]]", body["detail"]) + detail = TypeAdapter(list[dict[str, object]]).validate_python(body["detail"]) assert detail[0]["type"] == "missing" assert detail[0]["loc"] == ["body", "name"] diff --git a/tests/unit/service/test_entity_service_behaviour.py b/tests/unit/service/test_entity_service_behaviour.py index 8364bcf..e95106a 100644 --- a/tests/unit/service/test_entity_service_behaviour.py +++ b/tests/unit/service/test_entity_service_behaviour.py @@ -28,6 +28,7 @@ async def find_by_id(self, entity_id: UUID) -> Entity | None: async def find_all_paginated( self, + *, offset: int = 0, limit: int = 25, ordering: Ordering | None = None, diff --git a/tests/unit/settings/test_settings.py b/tests/unit/settings/test_settings.py index 699220f..92029eb 100644 --- a/tests/unit/settings/test_settings.py +++ b/tests/unit/settings/test_settings.py @@ -43,205 +43,207 @@ def _set_env(monkeypatch: pytest.MonkeyPatch, env: dict[str, str]) -> None: monkeypatch.setenv(key, value) -class TestApplicationSettings: - def test_defaults(self) -> None: - settings = ApplicationSettings(_env_file=None) - - assert settings.app_name == "FastAPI Template" - assert settings.host == "127.0.0.1" - assert settings.port == 80 - assert settings.reload is False - assert settings.proxy_headers is False - assert settings.forwarded_allow_ips == "127.0.0.1" +def test_application_settings_defaults() -> None: + settings = ApplicationSettings(_env_file=None) - def test_host_uses_app_host_env(self, monkeypatch: pytest.MonkeyPatch) -> None: - docker_bind_host = str(IPv4Address(0)) - monkeypatch.setenv("APP_HOST", docker_bind_host) + assert settings.app_name == "FastAPI Template" + assert settings.host == "127.0.0.1" + assert settings.port == 80 + assert settings.reload is False + assert settings.proxy_headers is False + assert settings.forwarded_allow_ips == "127.0.0.1" - settings = ApplicationSettings(_env_file=None) - assert settings.host == docker_bind_host +def test_application_settings_host_uses_app_host_env( + monkeypatch: pytest.MonkeyPatch, +) -> None: + docker_bind_host = str(IPv4Address(0)) + monkeypatch.setenv("APP_HOST", docker_bind_host) - def test_proxy_settings_come_from_env( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - monkeypatch.setenv("PROXY_HEADERS", "true") - monkeypatch.setenv("FORWARDED_ALLOW_IPS", "10.0.0.0/8") - - settings = ApplicationSettings(_env_file=None) - - assert settings.proxy_headers is True - assert settings.forwarded_allow_ips == "10.0.0.0/8" - - -class TestDatabaseSettings: - def test_requires_url(self, monkeypatch: pytest.MonkeyPatch) -> None: - _set_env(monkeypatch, {}) - - with pytest.raises(ValidationError): - DatabaseSettings(_env_file=None) - - def test_defaults(self, monkeypatch: pytest.MonkeyPatch) -> None: - _set_env(monkeypatch, _DB_ENV) - - settings = DatabaseSettings(_env_file=None) - - assert settings.url == _DB_ENV["DATABASE_URL"] - assert settings.pool_size == 5 - assert settings.max_overflow == 10 - assert settings.echo is False - assert settings.pool_pre_ping is True - - -class TestSettings: - def test_from_env_composes_every_settings_group( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - _set_env(monkeypatch, {**_DB_ENV, **_AUTHN_ENV}) - - settings = Settings.from_env() - - assert settings.database.url == _DB_ENV["DATABASE_URL"] - assert settings.oidc.client_id == _AUTHN_ENV["OIDC_CLIENT_ID"] - assert settings.app.app_name == "FastAPI Template" - - -class TestAuthNSettings: - def test_requires_fields(self, monkeypatch: pytest.MonkeyPatch) -> None: - _set_env(monkeypatch, {}) - - with pytest.raises(ValidationError): - AuthNSettings(_env_file=None) - - def test_defaults(self, monkeypatch: pytest.MonkeyPatch) -> None: - _set_env(monkeypatch, _AUTHN_ENV) - - settings = AuthNSettings(_env_file=None) - - assert settings.issuer_url == _AUTHN_ENV["OIDC_ISSUER_URL"] - assert settings.internal_url is None - assert settings.client_id == _AUTHN_ENV["OIDC_CLIENT_ID"] - assert settings.docs_client_id == _AUTHN_ENV["OIDC_DOCS_CLIENT_ID"] - assert settings.jwks_cache_ttl_seconds == 300 - assert settings.jwks_uri == _AUTHN_ENV["OIDC_JWKS_URI"] - assert settings.issuer == _AUTHN_ENV["OIDC_ISSUER"] - assert ( - settings.authorization_endpoint == _AUTHN_ENV["OIDC_AUTHORIZATION_ENDPOINT"] - ) - assert settings.token_endpoint == _AUTHN_ENV["OIDC_TOKEN_ENDPOINT"] - - async def test_resolves_discovery_when_overrides_are_not_set( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - env = {k: v for k, v in _AUTHN_ENV.items() if k not in _OVERRIDE_KEYS} - _set_env(monkeypatch, env) - - discovery = { - "jwks_uri": ( - "http://localhost:8080/realms/fastapi-realm/" - "protocol/openid-connect/certs" - ), - "issuer": _AUTHN_ENV["OIDC_ISSUER_URL"], - "authorization_endpoint": "http://localhost:8080/realms/fastapi-realm/auth", - "token_endpoint": "http://localhost:8080/realms/fastapi-realm/token", - } - settings = AuthNSettings(_env_file=None) - assert settings.metadata_override() is None - - def discovery_response(request: httpx.Request) -> httpx.Response: - assert request.url.path.endswith("/.well-known/openid-configuration") - return httpx.Response(200, json=discovery) + settings = ApplicationSettings(_env_file=None) - transport = httpx.MockTransport(discovery_response) - async with httpx.AsyncClient(transport=transport) as client: - metadata = await resolve_oidc_metadata(settings, client=client) + assert settings.host == docker_bind_host - assert metadata.model_dump() == discovery - async def test_internal_discovery_uses_internal_jwks_url( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - env = { - "OIDC_ISSUER_URL": "http://localhost:8080/realms/fastapi-realm", - "OIDC_INTERNAL_URL": _INTERNAL_OIDC_URL, +def test_application_settings_proxy_settings_come_from_env( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("PROXY_HEADERS", "true") + monkeypatch.setenv("FORWARDED_ALLOW_IPS", "10.0.0.0/8") + + settings = ApplicationSettings(_env_file=None) + + assert settings.proxy_headers is True + assert settings.forwarded_allow_ips == "10.0.0.0/8" + + +def test_database_settings_requires_url(monkeypatch: pytest.MonkeyPatch) -> None: + _set_env(monkeypatch, {}) + + with pytest.raises(ValidationError): + DatabaseSettings(_env_file=None) + + +def test_database_settings_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + _set_env(monkeypatch, _DB_ENV) + + settings = DatabaseSettings(_env_file=None) + + assert settings.url == _DB_ENV["DATABASE_URL"] + assert settings.pool_size == 5 + assert settings.max_overflow == 10 + assert settings.echo is False + assert settings.pool_pre_ping is True + + +def test_settings_from_env_composes_every_settings_group( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _set_env(monkeypatch, {**_DB_ENV, **_AUTHN_ENV}) + + settings = Settings.from_env() + + assert settings.database.url == _DB_ENV["DATABASE_URL"] + assert settings.oidc.client_id == _AUTHN_ENV["OIDC_CLIENT_ID"] + assert settings.app.app_name == "FastAPI Template" + + +def test_authn_settings_requires_fields(monkeypatch: pytest.MonkeyPatch) -> None: + _set_env(monkeypatch, {}) + + with pytest.raises(ValidationError): + AuthNSettings(_env_file=None) + + +def test_authn_settings_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + _set_env(monkeypatch, _AUTHN_ENV) + + settings = AuthNSettings(_env_file=None) + + assert settings.issuer_url == _AUTHN_ENV["OIDC_ISSUER_URL"] + assert settings.internal_url is None + assert settings.client_id == _AUTHN_ENV["OIDC_CLIENT_ID"] + assert settings.docs_client_id == _AUTHN_ENV["OIDC_DOCS_CLIENT_ID"] + assert settings.jwks_cache_ttl_seconds == 300 + assert settings.jwks_uri == _AUTHN_ENV["OIDC_JWKS_URI"] + assert settings.issuer == _AUTHN_ENV["OIDC_ISSUER"] + assert settings.authorization_endpoint == _AUTHN_ENV["OIDC_AUTHORIZATION_ENDPOINT"] + assert settings.token_endpoint == _AUTHN_ENV["OIDC_TOKEN_ENDPOINT"] + + +async def test_authn_settings_resolves_discovery_when_overrides_are_not_set( + monkeypatch: pytest.MonkeyPatch, +) -> None: + env = {k: v for k, v in _AUTHN_ENV.items() if k not in _OVERRIDE_KEYS} + _set_env(monkeypatch, env) + + discovery = { + "jwks_uri": ( + "http://localhost:8080/realms/fastapi-realm/protocol/openid-connect/certs" + ), + "issuer": _AUTHN_ENV["OIDC_ISSUER_URL"], + "authorization_endpoint": "http://localhost:8080/realms/fastapi-realm/auth", + "token_endpoint": "http://localhost:8080/realms/fastapi-realm/token", + } + settings = AuthNSettings(_env_file=None) + assert settings.metadata_override() is None + + def discovery_response(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/.well-known/openid-configuration") + return httpx.Response(200, json=discovery) + + transport = httpx.MockTransport(discovery_response) + async with httpx.AsyncClient(transport=transport) as client: + metadata = await resolve_oidc_metadata(settings, client=client) + + assert metadata.model_dump() == discovery + + +async def test_authn_settings_internal_discovery_uses_internal_jwks_url( + monkeypatch: pytest.MonkeyPatch, +) -> None: + env = { + "OIDC_ISSUER_URL": "http://localhost:8080/realms/fastapi-realm", + "OIDC_INTERNAL_URL": _INTERNAL_OIDC_URL, + "OIDC_CLIENT_ID": "fastapi-client", + "OIDC_DOCS_CLIENT_ID": "fastapi-docs", + } + _set_env(monkeypatch, env) + settings = AuthNSettings(_env_file=None) + discovery = { + "jwks_uri": f"{_INTERNAL_OIDC_URL}/protocol/openid-connect/certs", + "issuer": env["OIDC_ISSUER_URL"], + "authorization_endpoint": ( + "http://localhost:8080/realms/fastapi-realm/protocol/openid-connect/auth" + ), + "token_endpoint": f"{_INTERNAL_OIDC_URL}/protocol/openid-connect/token", + } + + def discovery_response(request: httpx.Request) -> httpx.Response: + assert request.url.host == "keycloak" + return httpx.Response(200, json=discovery) + + async with httpx.AsyncClient( + transport=httpx.MockTransport(discovery_response) + ) as client: + metadata = await resolve_oidc_metadata(settings, client=client) + + assert metadata.jwks_uri == discovery["jwks_uri"] + assert metadata.authorization_endpoint == discovery["authorization_endpoint"] + expected_endpoint = discovery["token_endpoint"].replace( + "keycloak", + "localhost", + ) + assert metadata.token_endpoint == expected_endpoint + + +async def test_authn_settings_blank_internal_url_is_treated_as_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _set_env( + monkeypatch, + { + "OIDC_ISSUER_URL": "https://idp.example/realm", + "OIDC_INTERNAL_URL": "", "OIDC_CLIENT_ID": "fastapi-client", "OIDC_DOCS_CLIENT_ID": "fastapi-docs", - } - _set_env(monkeypatch, env) - settings = AuthNSettings(_env_file=None) - discovery = { - "jwks_uri": f"{_INTERNAL_OIDC_URL}/protocol/openid-connect/certs", - "issuer": env["OIDC_ISSUER_URL"], - "authorization_endpoint": ( - "http://localhost:8080/realms/fastapi-realm/" - "protocol/openid-connect/auth" - ), - "token_endpoint": f"{_INTERNAL_OIDC_URL}/protocol/openid-connect/token", - } - - def discovery_response(request: httpx.Request) -> httpx.Response: - assert request.url.host == "keycloak" - return httpx.Response(200, json=discovery) - - async with httpx.AsyncClient( - transport=httpx.MockTransport(discovery_response) - ) as client: - metadata = await resolve_oidc_metadata(settings, client=client) - - assert metadata.jwks_uri == discovery["jwks_uri"] - assert metadata.authorization_endpoint == discovery["authorization_endpoint"] - expected_endpoint = discovery["token_endpoint"].replace( - "keycloak", - "localhost", - ) - assert metadata.token_endpoint == expected_endpoint - - async def test_blank_internal_url_is_treated_as_unset( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - _set_env( - monkeypatch, - { - "OIDC_ISSUER_URL": "https://idp.example/realm", - "OIDC_INTERNAL_URL": "", - "OIDC_CLIENT_ID": "fastapi-client", - "OIDC_DOCS_CLIENT_ID": "fastapi-docs", - }, - ) - settings = AuthNSettings(_env_file=None) - discovery = { - "jwks_uri": "https://idp.example/realm/jwks", - "issuer": "https://idp.example/realm", - "authorization_endpoint": "https://idp.example/realm/authorize", - "token_endpoint": "https://idp.example/realm/token", - } - - def discovery_response(request: httpx.Request) -> httpx.Response: - assert request.url.host == "idp.example" - return httpx.Response(200, json=discovery) - - async with httpx.AsyncClient( - transport=httpx.MockTransport(discovery_response) - ) as client: - metadata = await resolve_oidc_metadata(settings, client=client) - - assert metadata.model_dump() == discovery - - def test_rejects_partial_endpoint_overrides( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - _set_env( - monkeypatch, - { - "OIDC_ISSUER_URL": _AUTHN_ENV["OIDC_ISSUER_URL"], - "OIDC_CLIENT_ID": _AUTHN_ENV["OIDC_CLIENT_ID"], - "OIDC_DOCS_CLIENT_ID": _AUTHN_ENV["OIDC_DOCS_CLIENT_ID"], - "OIDC_JWKS_URI": _AUTHN_ENV["OIDC_JWKS_URI"], - }, - ) - - with pytest.raises(ValidationError, match="must provide"): - AuthNSettings(_env_file=None) + }, + ) + settings = AuthNSettings(_env_file=None) + discovery = { + "jwks_uri": "https://idp.example/realm/jwks", + "issuer": "https://idp.example/realm", + "authorization_endpoint": "https://idp.example/realm/authorize", + "token_endpoint": "https://idp.example/realm/token", + } + + def discovery_response(request: httpx.Request) -> httpx.Response: + assert request.url.host == "idp.example" + return httpx.Response(200, json=discovery) + + async with httpx.AsyncClient( + transport=httpx.MockTransport(discovery_response) + ) as client: + metadata = await resolve_oidc_metadata(settings, client=client) + + assert metadata.model_dump() == discovery + + +def test_authn_settings_rejects_partial_endpoint_overrides( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _set_env( + monkeypatch, + { + "OIDC_ISSUER_URL": _AUTHN_ENV["OIDC_ISSUER_URL"], + "OIDC_CLIENT_ID": _AUTHN_ENV["OIDC_CLIENT_ID"], + "OIDC_DOCS_CLIENT_ID": _AUTHN_ENV["OIDC_DOCS_CLIENT_ID"], + "OIDC_JWKS_URI": _AUTHN_ENV["OIDC_JWKS_URI"], + }, + ) + + with pytest.raises(ValidationError, match="must provide"): + AuthNSettings(_env_file=None) def test_jwks_cache_ttl_must_be_positive(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/unit/test_openapi_contract.py b/tests/unit/test_openapi_contract.py index bd2f1b9..87efa3f 100644 --- a/tests/unit/test_openapi_contract.py +++ b/tests/unit/test_openapi_contract.py @@ -1,14 +1,14 @@ import json -import os from http import HTTPStatus from pathlib import Path -from typing import Annotated, Any +from typing import Annotated import pytest from asgi_lifespan import LifespanManager from fastapi import Header from pydantic import BaseModel +from app.auth import OpenAPIDocument from app.exceptions import ErrorHandling, problem_responses from app.main import create_app from app.settings import Settings @@ -16,7 +16,6 @@ pytestmark = pytest.mark.unit SNAPSHOT_PATH = Path("tests/contract/openapi.json") -UPDATE_SNAPSHOT_ENV = "UPDATE_OPENAPI_SNAPSHOT" PROBLEM_REF = {"$ref": "#/components/schemas/ProblemDetails"} REASON = {"X-Reason": {"schema": {"type": "string"}}} BAD_REQUEST_REF = {"$ref": "#/components/responses/BadRequest"} @@ -38,8 +37,8 @@ def _render(schema: dict[str, object]) -> str: return json.dumps(schema, indent=2, sort_keys=True) + "\n" -def _assert_matches_snapshot(rendered: str) -> None: - if os.environ.get(UPDATE_SNAPSHOT_ENV) == "1": +def _assert_matches_snapshot(rendered: str, *, update_snapshot: bool) -> None: + if update_snapshot: SNAPSHOT_PATH.write_text(rendered, encoding="utf-8") assert SNAPSHOT_PATH.exists(), ( @@ -52,12 +51,14 @@ def _assert_matches_snapshot(rendered: str) -> None: async def test_openapi_document_matches_the_committed_contract( settings: Settings, + request: pytest.FixtureRequest, ) -> None: app = create_app(settings) async with LifespanManager(app): rendered = _render(app.openapi()) - _assert_matches_snapshot(rendered) + update_snapshot = request.config.getoption("--update-openapi-snapshot") + _assert_matches_snapshot(rendered, update_snapshot=update_snapshot) async def test_openapi_document_declares_problem_details_for_error_responses( @@ -81,7 +82,7 @@ async def test_openapi_document_declares_problem_details_for_error_responses( async def _schema_with_validation_status( settings: Settings, status: HTTPStatus -) -> dict[str, Any]: +) -> OpenAPIDocument: app = create_app(settings, ErrorHandling(validation_status=status)) @app.post( diff --git a/ty.toml b/ty.toml index ff5392d..de4e5c2 100644 --- a/ty.toml +++ b/ty.toml @@ -1,5 +1,4 @@ [environment] -python-version = "3.13" extra-paths = ["./scenario-tests/typings"] [rules] From 90e338795f246d76532543b58510b42861b950d6 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Thu, 15 Oct 2026 01:12:42 +0530 Subject: [PATCH 2/3] Reduce cognitive complexity in _declare_validation_responses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SonarCloud flagged this function's complexity at 19 against the 15 allowed, counted as new code because this PR changed its annotation line. Extract the per-operation and description-merge logic into two small functions so the loop body is a single call. Verified: make check, pytest tests/unit/test_openapi_contract.py. 🤖 Generated with Claude Code Co-Authored-By: Claude --- app/auth/openapi.py | 50 +++++++++++++++++++++++++++------------------ 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/app/auth/openapi.py b/app/auth/openapi.py index f621bb9..35c74ea 100644 --- a/app/auth/openapi.py +++ b/app/auth/openapi.py @@ -63,30 +63,40 @@ def _drop_unreferenced_components( del components[name] +def _merged_validation_description(declared: OpenAPIDocument) -> str: + description = declared.get("description") + if not description: + return _VALIDATION_DESCRIPTION + return f"{description} or {_VALIDATION_DESCRIPTION}" + + +def _declare_operation_validation_response( + operation: OpenAPIDocument, status: HTTPStatus +) -> bool: + responses = operation.setdefault("responses", {}) + generated = responses.get(_GENERATED_VALIDATION_STATUS) + if generated is not None and _is_generated_validation_response(generated): + del responses[_GENERATED_VALIDATION_STATUS] + elif not operation.get("parameters") and "requestBody" not in operation: + return False + + declared = responses.get(str(status.value)) + if declared is None: + declared = responses[str(status.value)] = {"description": status.phrase} + elif "$ref" in declared: + return True + else: + declared["description"] = _merged_validation_description(declared) + _add_problem_details(declared) + return True + + def _declare_validation_responses(schema: OpenAPIDocument, status: HTTPStatus) -> None: rewritten = False for operations in schema.get("paths", {}).values(): for operation in operations.values(): - responses = operation.setdefault("responses", {}) - generated = responses.get(_GENERATED_VALIDATION_STATUS) - if generated is not None and _is_generated_validation_response(generated): - del responses[_GENERATED_VALIDATION_STATUS] - elif not operation.get("parameters") and "requestBody" not in operation: - continue - rewritten = True - declared = responses.get(str(status.value)) - if declared is None: - declared = responses[str(status.value)] = {"description": status.phrase} - elif "$ref" in declared: - continue - else: - description = declared.get("description") - declared["description"] = ( - f"{description} or {_VALIDATION_DESCRIPTION}" - if description - else _VALIDATION_DESCRIPTION - ) - _add_problem_details(declared) + if _declare_operation_validation_response(operation, status): + rewritten = True if not rewritten: return _drop_unreferenced_components(schema, _GENERATED_VALIDATION_SCHEMAS) From f61f1e0bdc2238440026bab900726aed818a3877 Mon Sep 17 00:00:00 2001 From: Subhransu De Date: Thu, 15 Oct 2026 01:19:24 +0530 Subject: [PATCH 3/3] Keep seed independent of app; reuse ProblemDetails in API tests Code review fixes: - seed/__main__.py imported app.settings.DatabaseSettings, but the migration image copies only alembic/ and seed/ and installs only the migration group, so `docker compose --profile seed run seed` would fail with ModuleNotFoundError. CI never runs the seed profile, so it would not have caught this. Read DATABASE_URL directly again; the int return and raise SystemExit(main()) stay. - tests/integration/test_entities_api.py declared a ProblemBody model that duplicated app.exceptions.ProblemDetails field for field. Validate responses with ProblemDetails instead. Verified: make check. --- seed/__main__.py | 9 ++++++--- tests/integration/test_entities_api.py | 13 +++---------- 2 files changed, 9 insertions(+), 13 deletions(-) diff --git a/seed/__main__.py b/seed/__main__.py index 6211ccd..4f1acf0 100644 --- a/seed/__main__.py +++ b/seed/__main__.py @@ -1,8 +1,8 @@ import logging +import os from sqlalchemy import create_engine, pool -from app.settings import DatabaseSettings from seed import seed from seed.fixtures import FIXTURES @@ -11,8 +11,11 @@ def main() -> int: logging.basicConfig(level=logging.INFO, format="%(message)s") - database = DatabaseSettings() - engine = create_engine(database.url, poolclass=pool.NullPool) + database_url = os.environ.get("DATABASE_URL") + if not database_url: + msg = "DATABASE_URL must be set to seed the database" + raise RuntimeError(msg) + engine = create_engine(database_url, poolclass=pool.NullPool) try: with engine.begin() as connection: inserted = seed(connection, FIXTURES) diff --git a/tests/integration/test_entities_api.py b/tests/integration/test_entities_api.py index 985685b..b7b92ae 100644 --- a/tests/integration/test_entities_api.py +++ b/tests/integration/test_entities_api.py @@ -7,6 +7,7 @@ from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey from pydantic import BaseModel +from app.exceptions import ProblemDetails from tests.integration.conftest import ( TokenClaimOverrides, TokenFactory, @@ -23,14 +24,6 @@ class RequestOptions(TypedDict, total=False): json: dict[str, str] -class ProblemBody(BaseModel): - type: str - title: str - status: int - instance: str - detail: str | list[dict[str, object]] - - class ValidationErrorItem(BaseModel): loc: tuple[str | int, ...] @@ -40,10 +33,10 @@ def _assert_problem_details( *, status_code: int, title: str, -) -> ProblemBody: +) -> ProblemDetails: assert response.status_code == status_code assert response.headers["content-type"] == "application/problem+json" - body = ProblemBody.model_validate_json(response.content) + body = ProblemDetails.model_validate_json(response.content) assert body.title == title assert body.status == status_code assert body.type.endswith("/openapi.json")