diff --git a/.env.example b/.env.example index b78baf9..31145a0 100644 --- a/.env.example +++ b/.env.example @@ -5,6 +5,7 @@ # its built-in default; uncomment the lines you want to change. # SQLAlchemy database URL. PostgreSQL is the intended production database. +# Alembic reads this too, so `alembic upgrade head` needs nothing else set. FORMS_DATABASE_URL=postgresql+psycopg://forms:forms@localhost:5432/forms # SQLite is supported for local experimentation and backs the test suite. It is diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 63161a0..08cb65f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -13,7 +13,25 @@ concurrency: cancel-in-progress: true jobs: - test: + lint-and-types: + name: Lint, format and types + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + - run: pip install -e ".[dev]" + - name: Ruff lint + run: ruff check . + - name: Ruff format check + run: ruff format --check . + - name: mypy + run: mypy + + fast-tests: + name: Fast tests (SQLite, Python ${{ matrix.python-version }}) runs-on: ubuntu-latest strategy: fail-fast: false @@ -26,7 +44,45 @@ jobs: python-version: ${{ matrix.python-version }} cache: pip - run: pip install -e ".[dev]" - - run: ruff check . - - run: ruff format --check . - - run: mypy - - run: pytest + # The PostgreSQL suite skips itself here: no HYMICAL_TEST_POSTGRES_URL is + # set, which is the same thing that happens on a developer's machine. + - name: pytest + run: pytest + + postgres-integration: + name: PostgreSQL integration tests + runs-on: ubuntu-latest + # Only on the primary supported version. What these tests cover is + # PostgreSQL's behaviour, not the interpreter's, so running them across the + # whole matrix would multiply the runtime and prove nothing extra. + services: + postgres: + image: postgres:17 + env: + # Disposable credentials for a container that exists for one job and is + # reachable only from it. Nothing here is a secret. + POSTGRES_USER: forms + POSTGRES_PASSWORD: forms + POSTGRES_DB: forms + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U forms -d forms" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + - run: pip install -e ".[dev]" + - name: Migrate an empty database to head + run: alembic upgrade head + env: + FORMS_DATABASE_URL: postgresql+psycopg://forms:forms@localhost:5432/forms + - name: pytest (PostgreSQL) + run: pytest tests/integration -m postgres + env: + HYMICAL_TEST_POSTGRES_URL: postgresql+psycopg://forms:forms@localhost:5432/forms diff --git a/README.md b/README.md index 760ecdb..98f0e90 100644 --- a/README.md +++ b/README.md @@ -40,10 +40,10 @@ and no spam protection, so do not expose this to the public internet. | Signed webhook delivery | Implemented | | Durable delivery queue | Implemented | | Retries with backoff | Implemented | +| Schema migrations | Implemented | | API keys / authentication | **Not implemented** | | Manual delivery replay | **Not implemented** | | Rate limiting, spam handling | **Not implemented** | -| Schema migrations | **Not implemented** | | Export, retention, dashboards | **Not implemented** | ## Requirements @@ -81,7 +81,12 @@ See [`.env.example`](.env.example) for every setting and its default. ## Run -Hymical Forms is two processes sharing one database. +Hymical Forms is two processes sharing one database. Migrate it first, then +start them: + +```bash +alembic upgrade head +``` ```bash uvicorn hymical_forms.main:app --reload @@ -96,22 +101,88 @@ It never makes an outbound request. The **worker** claims owed deliveries, sends them, and retries the ones that fail. Running the API alone is fine: submissions are still accepted and nothing is lost, they simply wait until a worker exists. -Missing tables are created at startup, so an empty database is enough to begin. -Startup fails if the database cannot be reached, rather than serving requests -that would only fail later. There is no migration framework yet, so startup -never alters a table that already exists; see [Limitations](#limitations). - -> **Upgrading from an earlier build:** the schema has changed in every release so -> far, most recently by adding the `webhook_deliveries` table and giving -> `delivery_attempts` a `delivery_id` and `attempt_number`. Startup creates -> missing tables but never alters an existing one, so a database created before -> these changes has to be recreated. For local SQLite, delete the file and -> restart. For PostgreSQL, `DROP TABLE delivery_attempts, webhook_deliveries, -> submissions, endpoints;` and restart. There is no in-place upgrade path. -> -> Three consecutive schema changes with no migration tool is the clearest -> remaining infrastructure gap. Alembic is the next thing this project needs, -> and it should arrive before there is a database worth not dropping. +Neither process creates or alters the schema. Both check on startup that the +database is reachable and at the migration revision the build was written +against, and refuse to start otherwise: + +``` +the database is at migration '0001' but this build expects '0002'. +Run 'alembic upgrade head' before starting. +``` + +Migrating is an operator action, run when the operator chooses. See +[Schema migrations](#schema-migrations). + +## Schema migrations + +Alembic owns the schema. Neither the API nor the worker creates or alters a +table: they check on startup that the database is at the revision they were +built against, and stop if it is not. + +### A fresh database + +```bash +createdb forms +export FORMS_DATABASE_URL=postgresql+psycopg://forms:forms@localhost:5432/forms +alembic upgrade head +``` + +That is the whole setup. Migrations read `FORMS_DATABASE_URL`, the same setting +the application reads, so there is nothing extra to configure and no credentials +in any tracked file. To migrate a different database without changing your +environment: + +```bash +alembic -x database_url=postgresql+psycopg://user:pass@host/other upgrade head +``` + +### Upgrading an existing database + +```bash +alembic upgrade head # apply everything outstanding +``` + +Run it before starting the new build. The usual order for a deploy is: stop the +old processes, migrate, start the new ones. Migrating while an old build is +still running is only safe if the change happens to be backwards compatible, +and this project does not promise that for any particular migration. + +Useful alongside it: + +```bash +alembic current # what revision is this database at +alembic history --verbose # what revisions exist +alembic upgrade head --sql # print the SQL instead of applying it, for review +alembic downgrade -1 # step back one revision +``` + +`--sql` is worth knowing about: it lets whoever owns the production database +read the DDL before anything touches it. + +### SQLite + +Migrations run against SQLite too, so local experimentation works the same way: + +```bash +export FORMS_DATABASE_URL=sqlite:///./forms.db +alembic upgrade head +``` + +Migrations that alter a column are written in batch mode, because SQLite cannot +`ALTER` in place and has to rebuild the table instead. This is configured +already; it is not something a migration author has to remember. + +### Writing a migration + +```bash +alembic revision --autogenerate -m "what changed" +``` + +**Read what it produces before committing it.** Autogenerate is a starting +point, not an answer: it does not always render custom column types in a usable +way, and it cannot see anything the models do not declare. The PostgreSQL suite +asserts that migrations and models describe the same schema, so drift fails the +build rather than surfacing in production. Interactive API documentation is served at `http://127.0.0.1:8000/docs`. @@ -520,8 +591,28 @@ ruff format --check . # formatting check mypy # type check ``` -Tests run against an in-memory SQLite database, one per test, so no database -server is needed and nothing is left behind. +### Two test layers + +Most tests run against an in-memory SQLite database, one per test, so `pytest` +needs no services and leaves nothing behind. Their schema is built from the +models and stamped as migrated, rather than replayed migration by migration, +because doing that a few hundred times would cost far more than it proves. + +A smaller suite under `tests/integration/` runs against a real PostgreSQL +database, for the things SQLite cannot model honestly: `SELECT ... FOR UPDATE +SKIP LOCKED`, real constraint enforcement, and genuinely concurrent worker +sessions. It skips itself unless you point it at a database it may destroy: + +```bash +export HYMICAL_TEST_POSTGRES_URL=postgresql+psycopg://forms:forms@localhost:5432/forms_test +pytest tests/integration -m postgres +``` + +One of those tests asserts that the migrations and the models describe the same +schema, which is what keeps the fast suite's shortcut honest. + +CI runs the lint, format and type checks once, the fast suite across Python +3.11 to 3.13, and the PostgreSQL suite once against a PostgreSQL 17 service. ### Layout @@ -529,7 +620,7 @@ server is needed and nothing is left behind. src/hymical_forms/ app.py application assembly and startup config.py typed settings - db.py engine, session, and schema lifecycle + db.py engine and session lifecycle errors.py the shared JSON error envelope delivery.py the outbound webhook request itself ingestion.py domain rules: endpoint IDs, submission validation @@ -538,8 +629,10 @@ src/hymical_forms/ storage.py queries and writes webhooks.py webhook rules: URL validation, payload, signature, retry policy worker.py the delivery worker process + schema.py the boundary between the application and Alembic main.py ASGI entrypoint api/ HTTP routes and response models + migrations/ Alembic environment and revisions ``` `ingestion.py` and `webhooks.py` hold the domain rules and know nothing about @@ -585,17 +678,19 @@ so the claim also performs a conditional update and treats a row as claimed only if that update matched. That guard is redundant under `SKIP LOCKED` and is what makes the claim safe on SQLite. +This is covered by real integration tests: concurrent PostgreSQL sessions claim +disjoint work, a row another worker holds is skipped rather than waited on, and +an expired lease becomes reclaimable by exactly one worker. + ## Limitations - **Delivery is at-least-once, never exactly-once.** A worker that delivers successfully and dies before recording it will have its lease expire, and the next worker will deliver the same event again. Deduplicate on the submission `id` in the signed payload. -- **PostgreSQL worker concurrency is not exercised by the test suite.** Tests run - on SQLite, which cannot model `SELECT ... FOR UPDATE SKIP LOCKED`. The - generated PostgreSQL SQL is asserted, and the claim is written so that it is - also correct without row locking, but two real workers racing on PostgreSQL has - not been run. A PostgreSQL service in CI is the way to close this. +- **Only one migration exists so far.** The upgrade path is real and tested, but + it has only ever been exercised from an empty database to the baseline. Nothing + has yet had to migrate data it cared about. - **A failed delivery is final and cannot be replayed.** Once a delivery reaches `failed`, nothing retries it and there is no manual replay route. - **The lease must outlast a delivery attempt.** A batch is delivered @@ -618,9 +713,11 @@ makes the claim safe on SQLite. to change a destination or rotate a signing secret. - **No API for delivery attempts.** They are recorded, but reading them means querying the database directly. -- **No migration framework.** Startup creates missing tables and nothing else, - so any future change to an existing column has to be applied by hand. - Alembic will arrive when the schema first needs to change. +- **Migrations are applied by hand, one command at a time.** There is no + zero-downtime story and none is claimed: a migration that rewrites a table + will lock it, and a build whose expected revision does not match the database + refuses to start rather than serving against a schema it does not understand. + Plan a deploy as migrate-then-restart. - **No way to read submissions back over the API.** They are stored, but retrieval, export and retention are not implemented. - **No route to list, update or delete endpoints.** diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..b0dd7e3 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,50 @@ +; Alembic configuration for Hymical Forms. +; +; There is deliberately no sqlalchemy.url here. The database URL comes from +; FORMS_DATABASE_URL through the same Settings object the application uses, so +; credentials live in the environment and never in a tracked file. See env.py. +; +; To migrate a database other than the configured one, pass it per invocation: +; alembic -x database_url=postgresql+psycopg://user:pass@host/db upgrade head + +[alembic] +; Package-relative, so migrations ship with an installed wheel and `alembic` +; works from outside a checkout. +script_location = hymical_forms:migrations + +; Filenames sort in apply order, and carry the date they were written. +file_template = %%(rev)s_%%(year)d%%(month).2d%%(day).2d_%%(slug)s + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/pyproject.toml b/pyproject.toml index 84c4f77..5296be4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,6 +20,7 @@ classifiers = [ "Topic :: Internet :: WWW/HTTP :: HTTP Servers", ] dependencies = [ + "alembic>=1.13", # schema migrations, and the startup check that they were applied "fastapi>=0.115", "httpx2>=2.0", # outbound webhook client, and the transport starlette's TestClient uses "psycopg[binary]>=3.1", # PostgreSQL driver for the intended production database @@ -51,6 +52,9 @@ packages = ["src/hymical_forms"] [tool.pytest.ini_options] testpaths = ["tests"] addopts = "-q --strict-markers --strict-config" +markers = [ + "postgres: needs a live PostgreSQL database, named by HYMICAL_TEST_POSTGRES_URL", +] [tool.ruff] target-version = "py311" diff --git a/src/hymical_forms/app.py b/src/hymical_forms/app.py index ada7def..8d817f6 100644 --- a/src/hymical_forms/app.py +++ b/src/hymical_forms/app.py @@ -12,9 +12,10 @@ from hymical_forms import __version__ from hymical_forms.api import endpoints, health, submissions from hymical_forms.config import Settings -from hymical_forms.db import create_engine_from_url, create_session_factory, init_db +from hymical_forms.db import create_engine_from_url, create_session_factory from hymical_forms.errors import register_exception_handlers from hymical_forms.middleware import BodySizeLimitMiddleware +from hymical_forms.schema import verify_schema DESCRIPTION = """\ Hymical Forms accepts HTML form submissions over HTTP so that developers do not @@ -29,14 +30,16 @@ @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: """ - prepare the schema on startup and release the connection pool on shutdown + check the schema on startup and release the connection pool on shutdown :param app: the application starting up :returns: an async context manager wrapping the application's serving life """ - # There is no migration framework yet, so creating missing tables at startup - # is the whole schema story. It is safe to repeat and never alters a table - # that already exists, which also means a changed column needs manual work. - init_db(app.state.engine) + # The application never migrates. It reaches the database, confirms the + # schema is the revision this build was written against, and refuses to serve + # if it is not. Migrating here instead would apply DDL nobody reviewed, at a + # moment nobody chose, races every other replica starting at the same time, + # and would hide the mismatch this check is meant to surface. + verify_schema(app.state.engine) yield app.state.engine.dispose() diff --git a/src/hymical_forms/db.py b/src/hymical_forms/db.py index 3719387..b631a25 100644 --- a/src/hymical_forms/db.py +++ b/src/hymical_forms/db.py @@ -1,5 +1,8 @@ """ -database engine, session, and schema lifecycle +database engine and session lifecycle + +Creating the schema is not here and is not the application's job: Alembic owns +it, and :mod:`hymical_forms.schema` is where the two meet. """ from __future__ import annotations @@ -14,8 +17,6 @@ from sqlalchemy.pool import StaticPool from starlette.requests import Request -from hymical_forms.models import Base - def create_engine_from_url(url: str) -> Engine: """ @@ -64,16 +65,6 @@ def create_session_factory(engine: Engine) -> sessionmaker[Session]: return sessionmaker(bind=engine, expire_on_commit=False) -def init_db(engine: Engine) -> None: - """ - create any tables that do not exist yet - :param engine: the engine whose database should hold the schema - """ - # There is no migration framework yet, so this is the whole schema story: it - # creates missing tables and never alters existing ones. - Base.metadata.create_all(engine) - - def get_session(request: Request) -> Iterator[Session]: """ provide the session a request should do its database work through diff --git a/src/hymical_forms/migrations/env.py b/src/hymical_forms/migrations/env.py new file mode 100644 index 0000000..6f53f05 --- /dev/null +++ b/src/hymical_forms/migrations/env.py @@ -0,0 +1,94 @@ +""" +the Alembic environment + +The schema is taken from the application's SQLAlchemy metadata rather than +declared a second time here, and the database URL is read through the same +Settings object the application uses. Neither this file nor alembic.ini holds +credentials. +""" + +from __future__ import annotations + +from alembic import context +from sqlalchemy import Connection + +from hymical_forms.config import Settings +from hymical_forms.db import create_engine_from_url +from hymical_forms.models import Base + +target_metadata = Base.metadata + + +def database_url() -> str: + """ + work out which database this invocation should migrate + :returns: the SQLAlchemy database URL to run against + """ + # ``alembic -x database_url=...`` wins, so an operator can migrate a database + # other than the one this shell is configured for without exporting anything. + override = context.get_x_argument(as_dictionary=True).get("database_url") + if override: + return override + + # Then a URL handed over programmatically, which is how the tests point a + # migration run at a throwaway database. + from_caller = context.config.attributes.get("database_url") + if from_caller: + return str(from_caller) + + return Settings().database_url + + +def run_migrations_offline() -> None: + """ + emit the migration SQL without connecting to anything + """ + # Useful for handing a reviewable script to whoever owns the production + # database, rather than letting a deploy apply DDL unseen. + context.configure( + url=database_url(), + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + compare_type=True, + ) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + """ + apply the migrations against a live database + """ + # The application's own engine builder, so SQLite gets the same connection + # arguments and foreign key enforcement it gets at runtime. + engine = create_engine_from_url(database_url()) + try: + with engine.connect() as connection: + _run(connection) + finally: + engine.dispose() + + +def _run(connection: Connection) -> None: + """ + configure Alembic against an open connection and apply the migrations + :param connection: the connection to migrate through + """ + context.configure( + connection=connection, + target_metadata=target_metadata, + compare_type=True, + # SQLite cannot ALTER most things in place, so any future migration that + # changes a column has to be rewritten as a table copy. Batch mode does + # that automatically, and is inert on PostgreSQL. + render_as_batch=connection.dialect.name == "sqlite", + ) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/src/hymical_forms/migrations/script.py.mako b/src/hymical_forms/migrations/script.py.mako new file mode 100644 index 0000000..d14ddb0 --- /dev/null +++ b/src/hymical_forms/migrations/script.py.mako @@ -0,0 +1,33 @@ +""" +${message} + +revision: ${up_revision} +revises: ${down_revision | comma,n} +created: ${create_date} +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = ${repr(up_revision)} +down_revision: str | None = ${repr(down_revision)} +branch_labels: str | Sequence[str] | None = ${repr(branch_labels)} +depends_on: str | Sequence[str] | None = ${repr(depends_on)} + + +def upgrade() -> None: + """ + apply this revision + """ + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + """ + undo this revision + """ + ${downgrades if downgrades else "pass"} diff --git a/src/hymical_forms/migrations/versions/0001_20260824_baseline_schema.py b/src/hymical_forms/migrations/versions/0001_20260824_baseline_schema.py new file mode 100644 index 0000000..2689d01 --- /dev/null +++ b/src/hymical_forms/migrations/versions/0001_20260824_baseline_schema.py @@ -0,0 +1,162 @@ +""" +baseline schema + +The whole schema as it stood at the end of interval 5, when migration history +started. Everything before this point was created with ``create_all`` against +development databases that were never released, so there is nothing earlier to +migrate from and the downgrade simply removes what this created. + +Timestamps are written as ``sa.DateTime(timezone=True)`` rather than the +application's ``UtcDateTime`` decorator. The DDL is identical, because that is +exactly what the decorator wraps, and a migration that imports application code +would break the moment that code is refactored. A migration is a frozen record +of a change, not a view of the current models. + +revision: 0001 +revises: +created: 2026-08-24 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0001" +down_revision: str | None = None +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """ + create the schema + """ + op.create_table( + "endpoints", + sa.Column("id", sa.String(length=64), nullable=False), + sa.Column("name", sa.String(length=200), nullable=False), + sa.Column("is_active", sa.Boolean(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("webhook_url", sa.String(length=2048), nullable=True), + sa.Column("webhook_secret", sa.String(length=70), nullable=True), + sa.PrimaryKeyConstraint("id", name=op.f("pk_endpoints")), + # An endpoint has a whole webhook configuration or none of it, so a URL + # can never exist without the secret its payloads are signed with. + sa.CheckConstraint( + "(webhook_url IS NULL) = (webhook_secret IS NULL)", + name=op.f("ck_endpoints_webhook_configuration"), + ), + ) + + op.create_table( + "submissions", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("endpoint_id", sa.String(length=64), nullable=False), + sa.Column("received_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("fields", sa.JSON(), nullable=False), + sa.Column("idempotency_key", sa.String(length=255), nullable=True), + sa.Column("payload_fingerprint", sa.String(length=64), nullable=True), + sa.PrimaryKeyConstraint("id", name=op.f("pk_submissions")), + sa.ForeignKeyConstraint( + ["endpoint_id"], ["endpoints.id"], name=op.f("fk_submissions_endpoint_id_endpoints") + ), + # What actually enforces idempotency. Both backends treat NULLs in a + # unique constraint as distinct, so submissions sent without a key stay + # unrestricted without needing a partial index. + sa.UniqueConstraint( + "endpoint_id", "idempotency_key", name="uq_submissions_endpoint_idempotency_key" + ), + sa.CheckConstraint( + "(idempotency_key IS NULL) = (payload_fingerprint IS NULL)", + name=op.f("ck_submissions_idempotency_identity"), + ), + ) + op.create_index( + op.f("ix_submissions_endpoint_id"), "submissions", ["endpoint_id"], unique=False + ) + + op.create_table( + "webhook_deliveries", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("submission_id", sa.String(length=36), nullable=False), + sa.Column("destination_url", sa.String(length=2048), nullable=False), + sa.Column("signing_secret", sa.String(length=70), nullable=False), + sa.Column("state", sa.String(length=16), nullable=False), + sa.Column("attempts", sa.Integer(), nullable=False), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("claim_expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id", name=op.f("pk_webhook_deliveries")), + sa.ForeignKeyConstraint( + ["submission_id"], + ["submissions.id"], + name=op.f("fk_webhook_deliveries_submission_id_submissions"), + ), + # One delivery per submission, which is what makes an idempotent replay + # unable to queue a second one. + sa.UniqueConstraint("submission_id", name="uq_webhook_deliveries_submission"), + # A delivery is finished exactly when it says it is finished. + sa.CheckConstraint( + "(state IN ('delivered', 'failed')) = (completed_at IS NOT NULL)", + name=op.f("ck_webhook_deliveries_completion"), + ), + ) + # The column a worker scans on every poll. + op.create_index( + op.f("ix_webhook_deliveries_next_attempt_at"), + "webhook_deliveries", + ["next_attempt_at"], + unique=False, + ) + + op.create_table( + "delivery_attempts", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("delivery_id", sa.String(length=36), nullable=False), + sa.Column("submission_id", sa.String(length=36), nullable=False), + sa.Column("attempt_number", sa.Integer(), nullable=False), + sa.Column("destination_url", sa.String(length=2048), nullable=False), + sa.Column("attempted_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("outcome", sa.String(length=32), nullable=False), + sa.Column("response_status", sa.Integer(), nullable=True), + sa.Column("error", sa.String(length=500), nullable=True), + sa.PrimaryKeyConstraint("id", name=op.f("pk_delivery_attempts")), + sa.ForeignKeyConstraint( + ["delivery_id"], + ["webhook_deliveries.id"], + name=op.f("fk_delivery_attempts_delivery_id_webhook_deliveries"), + ), + sa.ForeignKeyConstraint( + ["submission_id"], + ["submissions.id"], + name=op.f("fk_delivery_attempts_submission_id_submissions"), + ), + ) + op.create_index( + op.f("ix_delivery_attempts_delivery_id"), "delivery_attempts", ["delivery_id"], unique=False + ) + op.create_index( + op.f("ix_delivery_attempts_submission_id"), + "delivery_attempts", + ["submission_id"], + unique=False, + ) + + +def downgrade() -> None: + """ + remove the schema this revision created + """ + # Dropped in reverse dependency order so the foreign keys never block a drop. + op.drop_index(op.f("ix_delivery_attempts_submission_id"), table_name="delivery_attempts") + op.drop_index(op.f("ix_delivery_attempts_delivery_id"), table_name="delivery_attempts") + op.drop_table("delivery_attempts") + op.drop_index(op.f("ix_webhook_deliveries_next_attempt_at"), table_name="webhook_deliveries") + op.drop_table("webhook_deliveries") + op.drop_index(op.f("ix_submissions_endpoint_id"), table_name="submissions") + op.drop_table("submissions") + op.drop_table("endpoints") diff --git a/src/hymical_forms/models.py b/src/hymical_forms/models.py index 96284fb..06f8a3c 100644 --- a/src/hymical_forms/models.py +++ b/src/hymical_forms/models.py @@ -11,6 +11,7 @@ CheckConstraint, DateTime, ForeignKey, + MetaData, String, TypeDecorator, UniqueConstraint, @@ -92,6 +93,20 @@ class Base(DeclarativeBase): declarative base for every persisted table """ + # Every constraint and index gets a predictable name rather than one the + # database invents. Alembic needs that to reference them in a later + # migration: an autogenerated foreign key name cannot be dropped portably. + # Constraints that already carry an explicit name keep it. + metadata = MetaData( + naming_convention={ + "ix": "ix_%(table_name)s_%(column_0_N_name)s", + "uq": "uq_%(table_name)s_%(column_0_N_name)s", + "ck": "ck_%(table_name)s_%(constraint_name)s", + "fk": "fk_%(table_name)s_%(column_0_N_name)s_%(referred_table_name)s", + "pk": "pk_%(table_name)s", + } + ) + class Endpoint(Base): """ @@ -110,7 +125,7 @@ class Endpoint(Base): # could trust. CheckConstraint( "(webhook_url IS NULL) = (webhook_secret IS NULL)", - name="ck_endpoints_webhook_configuration", + name="webhook_configuration", ), ) @@ -151,7 +166,7 @@ class Submission(Base): # A submission either carries a full idempotency identity or none of it. CheckConstraint( "(idempotency_key IS NULL) = (payload_fingerprint IS NULL)", - name="ck_submissions_idempotency_identity", + name="idempotency_identity", ), ) @@ -238,7 +253,7 @@ class WebhookDelivery(Base): # A delivery is finished exactly when it says it is finished. CheckConstraint( "(state IN ('delivered', 'failed')) = (completed_at IS NOT NULL)", - name="ck_webhook_deliveries_completion", + name="completion", ), ) diff --git a/src/hymical_forms/schema.py b/src/hymical_forms/schema.py new file mode 100644 index 0000000..9aa9de7 --- /dev/null +++ b/src/hymical_forms/schema.py @@ -0,0 +1,118 @@ +""" +the boundary between the application and Alembic + +Alembic owns the production schema. Nothing here ever migrates a database: the +application only asks whether the schema it was given is the one it was built +for, and says so plainly when it is not. Applying migrations is an operator +action, run deliberately, at a time the operator chooses. +""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path + +from alembic.config import Config +from alembic.migration import MigrationContext +from alembic.script import ScriptDirectory +from sqlalchemy import Engine + +from hymical_forms.models import Base + +MIGRATIONS_PATH = Path(__file__).resolve().parent / "migrations" + + +class SchemaNotReady(RuntimeError): + """ + raised when the database is not at the revision this build expects + """ + + +def alembic_config(database_url: str | None = None) -> Config: + """ + build an Alembic config pointing at the migrations shipped with this package + :param database_url: database to target, or None to let env.py resolve it + :returns: a config usable with Alembic's programmatic API + """ + # Built in code rather than read from alembic.ini so that this works from an + # installed wheel, where the repository's ini file is not present. + config = Config() + config.set_main_option("script_location", str(MIGRATIONS_PATH)) + if database_url is not None: + # Passed through attributes rather than as ``sqlalchemy.url``, so a + # password containing a percent sign is never mangled by ini + # interpolation on its way to the database. + config.attributes["database_url"] = database_url + return config + + +@lru_cache(maxsize=1) +def head_revision() -> str: + """ + read the newest revision in the migrations shipped with this build + :returns: the head revision identifier + """ + # Cached because the migration scripts cannot change while the process runs, + # and this would otherwise scan a directory on every application startup. + script = ScriptDirectory(str(MIGRATIONS_PATH)) + head = script.get_current_head() + if head is None: + raise SchemaNotReady("this build ships no migrations") + return head + + +def current_revision(engine: Engine) -> str | None: + """ + read the revision a database is currently at + :param engine: the engine to inspect through + :returns: the stored revision, or None if the database has never been migrated + """ + with engine.connect() as connection: + return MigrationContext.configure(connection).get_current_revision() + + +def verify_schema(engine: Engine) -> None: + """ + check that the database is reachable and at the revision this build expects + :param engine: the engine to verify + :raises SchemaNotReady: if the database has no schema or is at another revision + """ + # Deliberately a check and not a migration. An application that quietly + # altered the schema it found would make deploys unreviewable and would race + # every other replica starting at the same moment. + expected = head_revision() + found = current_revision(engine) + + if found is None: + raise SchemaNotReady( + "the database has no schema. Run 'alembic upgrade head' before starting." + ) + if found != expected: + raise SchemaNotReady( + f"the database is at migration {found!r} but this build expects {expected!r}. " + "Run 'alembic upgrade head' before starting." + ) + + +def create_all(engine: Engine) -> None: + """ + build the schema straight from the models and record it as fully migrated + :param engine: the engine whose database should hold the schema + """ + # For tests and throwaway databases only. It is much faster than replaying + # migrations for every test, and the stamp is what lets the application's + # startup check accept the result. A PostgreSQL test asserts that what this + # produces and what the migrations produce are the same schema, so the + # shortcut cannot quietly drift away from the real one. + Base.metadata.create_all(engine) + stamp_head(engine) + + +def stamp_head(engine: Engine) -> None: + """ + record a database as being at the newest revision without running migrations + :param engine: the engine whose database should be stamped + """ + with engine.begin() as connection: + context = MigrationContext.configure(connection) + context.stamp(ScriptDirectory(str(MIGRATIONS_PATH)), head_revision()) diff --git a/tests/conftest.py b/tests/conftest.py index e520d95..c4f9d33 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -25,6 +25,7 @@ from hymical_forms.app import create_app from hymical_forms.config import Settings from hymical_forms.delivery import create_webhook_client +from hymical_forms.schema import create_all from hymical_forms.worker import process_batch from webhook_server import WebhookRecorder @@ -161,7 +162,14 @@ def factory(*, seed_endpoint: bool = True, **overrides: Any) -> TestClient: :param overrides: setting values to replace the built-in defaults :returns: a test client closed when the fixture tears down """ - client = stack.enter_context(TestClient(create_app(build_settings(**overrides)))) + # The schema is built from the models and stamped, rather than + # migrated, because replaying migrations for each of a couple of + # hundred tests would cost far more than it proves. That the two + # produce the same schema is asserted once, against PostgreSQL, in + # the integration suite. + app = create_app(build_settings(**overrides)) + create_all(app.state.engine) + client = stack.enter_context(TestClient(app)) if seed_endpoint: create_endpoint(client) return client diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..fb07f04 --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1,6 @@ +""" +the PostgreSQL integration suite + +A package rather than a bare directory so that its ``conftest`` does not +collide with the fast suite's, which sits one level up. +""" diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..e97682a --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,104 @@ +""" +fixtures for the PostgreSQL integration suite + +These tests exist for behaviour SQLite cannot model honestly: real row locking, +real constraint enforcement, and real concurrent worker sessions. They are +skipped unless ``HYMICAL_TEST_POSTGRES_URL`` names a database they may freely +destroy, so the ordinary ``pytest`` run stays fast and needs no services. +""" + +from __future__ import annotations + +import os +from collections.abc import Iterator + +import pytest +from alembic import command +from fastapi.testclient import TestClient +from sqlalchemy import Engine, text +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms.app import create_app +from hymical_forms.db import create_engine_from_url +from hymical_forms.models import Base +from hymical_forms.schema import alembic_config +from integration.support import POSTGRES_URL_VARIABLE, IsolatedSettings, drop_everything + + +def pytest_collection_modifyitems(items: list[pytest.Item]) -> None: + """ + mark everything in this directory as a PostgreSQL test + :param items: the collected tests, modified in place + """ + # Applied here rather than decorated onto every test, so the marker cannot be + # forgotten on a new one. + for item in items: + item.add_marker(pytest.mark.postgres) + + +@pytest.fixture(scope="session") +def postgres_url() -> str: + """ + read the URL of a PostgreSQL database these tests may destroy + :returns: the database URL to run against + """ + url = os.environ.get(POSTGRES_URL_VARIABLE) + if not url: + pytest.skip(f"set {POSTGRES_URL_VARIABLE} to run the PostgreSQL integration suite") + return url + + +@pytest.fixture(scope="session") +def migrated_engine(postgres_url: str) -> Iterator[Engine]: + """ + provide an engine on a database migrated to head by Alembic + :param postgres_url: the database URL to run against + :returns: an iterator yielding the engine, disposed when the session ends + """ + # The schema is built the way production builds it, by running the + # migrations, so this suite is also a standing test that they work. + engine = create_engine_from_url(postgres_url) + drop_everything(engine) + command.upgrade(alembic_config(postgres_url), "head") + yield engine + engine.dispose() + + +@pytest.fixture(autouse=True) +def clean_database(migrated_engine: Engine) -> None: + """ + empty every table before each test + :param migrated_engine: the engine whose database should be emptied + """ + # Before rather than after, so a test that dies part way through cannot leave + # rows that poison whatever runs next. + tables = ", ".join(f'"{table.name}"' for table in Base.metadata.sorted_tables) + with migrated_engine.begin() as connection: + connection.execute(text(f"TRUNCATE {tables} RESTART IDENTITY CASCADE")) + + +@pytest.fixture +def sessions(migrated_engine: Engine) -> sessionmaker[Session]: + """ + provide a session factory that hands out independent connections + :param migrated_engine: the engine sessions should be bound to + :returns: a session factory + """ + # No StaticPool here: each session takes its own connection, which is what + # makes two sessions genuinely two workers. + return sessionmaker(bind=migrated_engine, expire_on_commit=False) + + +@pytest.fixture +def pg_client(postgres_url: str, migrated_engine: Engine) -> Iterator[TestClient]: + """ + provide an API client backed by the migrated PostgreSQL database + :param postgres_url: the database URL to run against + :param migrated_engine: unused, but forces the schema to exist first + :returns: an iterator yielding a test client + """ + app = create_app( + IsolatedSettings(database_url=postgres_url, allow_private_webhook_targets=True) + ) + with TestClient(app) as client: + yield client diff --git a/tests/integration/support.py b/tests/integration/support.py new file mode 100644 index 0000000..4d6a0aa --- /dev/null +++ b/tests/integration/support.py @@ -0,0 +1,133 @@ +""" +helpers for the PostgreSQL integration suite + +Kept out of ``conftest.py`` deliberately. Both ``tests/`` and +``tests/integration/`` end up on sys.path, so a plain ``import conftest`` from a +test in this directory could resolve to either file depending on collection +order. This module's name is unambiguous. +""" + +from __future__ import annotations + +import uuid +from collections.abc import Iterator +from contextlib import contextmanager +from datetime import datetime, timedelta + +from pydantic_settings import SettingsConfigDict +from sqlalchemy import Engine, text +from sqlalchemy.engine import make_url +from sqlalchemy.orm import Session + +from hymical_forms import models +from hymical_forms.config import Settings +from hymical_forms.db import create_engine_from_url +from hymical_forms.webhooks import DeliveryState + +POSTGRES_URL_VARIABLE = "HYMICAL_TEST_POSTGRES_URL" + + +class IsolatedSettings(Settings): + """ + settings that ignore a local ``.env``, so a developer's file cannot skew a run + """ + + model_config = SettingsConfigDict(env_file=None) + + +@contextmanager +def temporary_database(postgres_url: str) -> Iterator[str]: + """ + create an empty database for one test and drop it afterwards + :param postgres_url: a URL on the server the new database should live on + :returns: a context manager yielding the new database's URL + """ + # Migration tests need to own a whole database, so they get one rather than + # fighting the shared schema the rest of the suite runs against. + url = make_url(postgres_url) + name = f"hymical_test_{uuid.uuid4().hex[:12]}" + # ``str(URL)`` masks the password, so rendering has to be explicit or the + # connection arrives with a literal "***" and is refused. + admin = create_engine_from_url( + url.set(database="postgres").render_as_string(hide_password=False) + ) + try: + with admin.connect().execution_options(isolation_level="AUTOCOMMIT") as connection: + connection.execute(text(f'CREATE DATABASE "{name}"')) + try: + yield url.set(database=name).render_as_string(hide_password=False) + finally: + with admin.connect().execution_options(isolation_level="AUTOCOMMIT") as connection: + connection.execute(text(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')) + finally: + admin.dispose() + + +def drop_everything(engine: Engine) -> None: + """ + return a database to being empty, whatever a previous run left in it + :param engine: the engine whose database should be emptied + """ + with engine.begin() as connection: + connection.execute(text("DROP SCHEMA public CASCADE")) + connection.execute(text("CREATE SCHEMA public")) + + +def seed_endpoint(session: Session, endpoint_id: str = "contact-form") -> models.Endpoint: + """ + insert an endpoint with a webhook configured + :param session: the session to insert through + :param endpoint_id: the identifier to give it + :returns: the committed endpoint + """ + endpoint = models.Endpoint( + id=endpoint_id, + name="Contact form", + is_active=True, + webhook_url="https://example.invalid/hook", + webhook_secret="whsec_" + "a" * 64, + ) + session.add(endpoint) + session.commit() + return endpoint + + +def seed_due_deliveries( + session: Session, count: int, *, now: datetime, endpoint_id: str = "contact-form" +) -> list[str]: + """ + insert submissions each owing a delivery that is already due + :param session: the session to insert through + :param count: how many to create + :param now: the instant the deliveries should have become due + :param endpoint_id: the endpoint they belong to + :returns: the delivery ids, oldest first + """ + ids: list[str] = [] + for index in range(count): + submission_id = f"sub_{uuid.uuid4().hex}" + delivery_id = f"whd_{uuid.uuid4().hex}" + session.add( + models.Submission( + id=submission_id, + endpoint_id=endpoint_id, + received_at=now, + fields={"email": [f"dev{index}@example.com"]}, + ) + ) + session.add( + models.WebhookDelivery( + id=delivery_id, + submission_id=submission_id, + destination_url="https://example.invalid/hook", + signing_secret="whsec_" + "a" * 64, + state=DeliveryState.PENDING, + attempts=0, + # Staggered into the past so ordering is deterministic. + next_attempt_at=now - timedelta(seconds=count - index), + created_at=now, + ) + ) + ids.append(delivery_id) + session.commit() + return ids diff --git a/tests/integration/test_claiming_postgres.py b/tests/integration/test_claiming_postgres.py new file mode 100644 index 0000000..58563d1 --- /dev/null +++ b/tests/integration/test_claiming_postgres.py @@ -0,0 +1,251 @@ +""" +worker claiming against real PostgreSQL + +This is the behaviour SQLite cannot model. SQLite silently ignores ``FOR UPDATE`` +and serialises writers anyway, so the fast suite can only ever show that the +conditional update guard works. What matters in production is that two workers +holding two connections are handed different rows and neither waits on the +other, and that can only be shown here. +""" + +from __future__ import annotations + +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms import models, storage +from hymical_forms.webhooks import DeliveryState +from integration.support import seed_due_deliveries, seed_endpoint + +NOW = datetime(2026, 8, 24, 12, 0, tzinfo=UTC) + + +def locking_select(now: datetime, limit: int) -> Any: + """ + build the same locking read the claim uses + :param now: the instant to judge dueness against + :param limit: the most rows to lock + :returns: a select statement that locks and skips locked rows + """ + return ( + select(models.WebhookDelivery) + .where(storage.due_condition(now)) + .order_by(models.WebhookDelivery.next_attempt_at) + .limit(limit) + .with_for_update(skip_locked=True) + ) + + +def test_a_locked_row_is_skipped_rather_than_waited_on( + sessions: sessionmaker[Session], +) -> None: + """ + the defining property of SKIP LOCKED, shown with two real open transactions + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + seed_due_deliveries(setup, 2, now=NOW) + + with sessions() as first, sessions() as second: + # First worker locks a row and deliberately does not commit, standing in + # for a worker that is still mid-claim. + locked_by_first = first.scalars(locking_select(NOW, 1)).all() + assert len(locked_by_first) == 1 + + started = time.monotonic() + locked_by_second = second.scalars(locking_select(NOW, 1)).all() + elapsed = time.monotonic() - started + + assert len(locked_by_second) == 1 + assert {row.id for row in locked_by_first}.isdisjoint({row.id for row in locked_by_second}) + assert elapsed < 5, "the second worker blocked on the first worker's lock" + + first.rollback() + second.rollback() + + +def test_the_only_due_row_is_skipped_when_another_worker_holds_it( + sessions: sessionmaker[Session], +) -> None: + """ + with one row and two workers, the second must come away empty, not blocked + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + seed_due_deliveries(setup, 1, now=NOW) + + with sessions() as first, sessions() as second: + assert len(first.scalars(locking_select(NOW, 10)).all()) == 1 + + started = time.monotonic() + locked_by_second = second.scalars(locking_select(NOW, 10)).all() + elapsed = time.monotonic() - started + + assert locked_by_second == [] + assert elapsed < 5, "the second worker blocked instead of skipping" + + first.rollback() + second.rollback() + + +def test_concurrent_workers_never_claim_the_same_delivery( + sessions: sessionmaker[Session], +) -> None: + """ + six real sessions claiming at once must partition the work, never share it + :param sessions: factory handing out independent connections + """ + workers = 6 + total = 24 + with sessions() as setup: + seed_endpoint(setup) + expected = set(seed_due_deliveries(setup, total, now=NOW)) + + barrier = threading.Barrier(workers) + + def claim() -> list[str]: + barrier.wait() + with sessions() as session: + claimed = storage.claim_due_deliveries(session, now=NOW, lease_seconds=60, limit=total) + return [job.id for job in claimed] + + with ThreadPoolExecutor(max_workers=workers) as pool: + batches = [future.result() for future in [pool.submit(claim) for _ in range(workers)]] + + claimed = [delivery_id for batch in batches for delivery_id in batch] + assert len(claimed) == len(set(claimed)), "a delivery was claimed by more than one worker" + assert set(claimed) == expected, "some due deliveries were never claimed" + + # And the database agrees: everything is processing, held by somebody. + with sessions() as session: + rows = list(session.scalars(select(models.WebhookDelivery))) + assert {row.state for row in rows} == {DeliveryState.PROCESSING} + assert all(row.claim_expires_at == NOW + timedelta(seconds=60) for row in rows) + + +def test_concurrent_workers_share_the_work_out(sessions: sessionmaker[Session]) -> None: + """ + SKIP LOCKED should let workers proceed in parallel, not funnel into one + :param sessions: factory handing out independent connections + """ + workers = 4 + with sessions() as setup: + seed_endpoint(setup) + seed_due_deliveries(setup, 40, now=NOW) + + barrier = threading.Barrier(workers) + + def claim() -> int: + barrier.wait() + with sessions() as session: + return len(storage.claim_due_deliveries(session, now=NOW, lease_seconds=60, limit=5)) + + with ThreadPoolExecutor(max_workers=workers) as pool: + counts = [future.result() for future in [pool.submit(claim) for _ in range(workers)]] + + assert sum(counts) == workers * 5, "workers did not each get a full batch" + assert all(count == 5 for count in counts) + + +def test_a_claim_survives_in_the_database(sessions: sessionmaker[Session]) -> None: + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_due_deliveries(setup, 1, now=NOW)[0] + + with sessions() as worker: + storage.claim_due_deliveries(worker, now=NOW, lease_seconds=60, limit=10) + + # A different connection entirely, so this is the committed state. + with sessions() as observer: + row = observer.get(models.WebhookDelivery, delivery_id) + assert row is not None + assert row.state == DeliveryState.PROCESSING + assert row.claim_expires_at == NOW + timedelta(seconds=60) + + +def test_an_active_lease_is_not_reclaimable(sessions: sessionmaker[Session]) -> None: + with sessions() as setup: + seed_endpoint(setup) + seed_due_deliveries(setup, 1, now=NOW) + with sessions() as first: + assert len(storage.claim_due_deliveries(first, now=NOW, lease_seconds=60, limit=10)) == 1 + + with sessions() as second: + again = storage.claim_due_deliveries( + second, now=NOW + timedelta(seconds=30), lease_seconds=60, limit=10 + ) + + assert again == [] + + +def test_an_expired_lease_is_reclaimable(sessions: sessionmaker[Session]) -> None: + """ + a worker that died holding a delivery must not strand it forever + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_due_deliveries(setup, 1, now=NOW)[0] + with sessions() as abandoned: + storage.claim_due_deliveries(abandoned, now=NOW, lease_seconds=60, limit=10) + + with sessions() as recovering: + reclaimed = storage.claim_due_deliveries( + recovering, now=NOW + timedelta(seconds=61), lease_seconds=60, limit=10 + ) + + assert [job.id for job in reclaimed] == [delivery_id] + + +def test_two_workers_racing_to_reclaim_an_expired_lease_do_not_both_win( + sessions: sessionmaker[Session], +) -> None: + with sessions() as setup: + seed_endpoint(setup) + seed_due_deliveries(setup, 1, now=NOW) + with sessions() as abandoned: + storage.claim_due_deliveries(abandoned, now=NOW, lease_seconds=60, limit=10) + + later = NOW + timedelta(seconds=61) + barrier = threading.Barrier(2) + + def reclaim() -> list[str]: + barrier.wait() + with sessions() as session: + return [ + job.id + for job in storage.claim_due_deliveries( + session, now=later, lease_seconds=60, limit=10 + ) + ] + + with ThreadPoolExecutor(max_workers=2) as pool: + first, second = (future.result() for future in [pool.submit(reclaim) for _ in range(2)]) + + assert len(first) + len(second) == 1, "both workers reclaimed the same expired delivery" + + +def test_a_terminal_delivery_is_never_claimed(sessions: sessionmaker[Session]) -> None: + with sessions() as setup: + seed_endpoint(setup) + delivery_id = seed_due_deliveries(setup, 1, now=NOW)[0] + delivery = setup.get(models.WebhookDelivery, delivery_id) + assert delivery is not None + delivery.state = DeliveryState.DELIVERED + delivery.completed_at = NOW + setup.commit() + + with sessions() as worker: + claimed = storage.claim_due_deliveries( + worker, now=NOW + timedelta(days=1), lease_seconds=60, limit=10 + ) + + assert claimed == [] diff --git a/tests/integration/test_constraints_postgres.py b/tests/integration/test_constraints_postgres.py new file mode 100644 index 0000000..03e5f81 --- /dev/null +++ b/tests/integration/test_constraints_postgres.py @@ -0,0 +1,193 @@ +""" +the database invariants, enforced by PostgreSQL rather than by Python + +The application leans on these: idempotency, one delivery per submission, and +the paired-column checks are all written as constraints on purpose, because a +concurrent request can slip past any check the application makes for itself. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms import models +from hymical_forms.webhooks import DeliveryState +from integration.support import seed_endpoint + +NOW = datetime(2026, 8, 24, 12, 0, tzinfo=UTC) + + +def a_submission( + submission_id: str = "sub_one", + *, + idempotency_key: str | None = None, + payload_fingerprint: str | None = None, +) -> models.Submission: + """ + build a submission row for the default endpoint + :param submission_id: the identifier to give it + :param idempotency_key: the retry key, if any + :param payload_fingerprint: the content digest, if any + :returns: an unsaved submission row + """ + return models.Submission( + id=submission_id, + endpoint_id="contact-form", + received_at=NOW, + fields={"email": ["dev@example.com"]}, + idempotency_key=idempotency_key, + payload_fingerprint=payload_fingerprint, + ) + + +def a_delivery(delivery_id: str, submission_id: str) -> models.WebhookDelivery: + """ + build a pending delivery row + :param delivery_id: the identifier to give it + :param submission_id: the submission it belongs to + :returns: an unsaved delivery row + """ + return models.WebhookDelivery( + id=delivery_id, + submission_id=submission_id, + destination_url="https://example.invalid/hook", + signing_secret="whsec_" + "a" * 64, + state=DeliveryState.PENDING, + attempts=0, + next_attempt_at=NOW, + created_at=NOW, + ) + + +def test_an_endpoint_id_is_unique(sessions: sessionmaker[Session]) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(models.Endpoint(id="contact-form", name="Another", is_active=True)) + with pytest.raises(IntegrityError): + session.commit() + + +def test_an_idempotency_key_is_unique_per_endpoint(sessions: sessionmaker[Session]) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(a_submission("sub_one", idempotency_key="k" * 16, payload_fingerprint="f" * 64)) + session.commit() + + session.add(a_submission("sub_two", idempotency_key="k" * 16, payload_fingerprint="f" * 64)) + with pytest.raises(IntegrityError): + session.commit() + + +def test_the_same_key_is_allowed_on_another_endpoint(sessions: sessionmaker[Session]) -> None: + with sessions() as session: + seed_endpoint(session) + seed_endpoint(session, "waitlist") + session.add(a_submission("sub_one", idempotency_key="k" * 16, payload_fingerprint="f" * 64)) + second = a_submission("sub_two", idempotency_key="k" * 16, payload_fingerprint="f" * 64) + second.endpoint_id = "waitlist" + session.add(second) + + session.commit() + + assert session.get(models.Submission, "sub_two") is not None + + +def test_submissions_without_a_key_are_unrestricted(sessions: sessionmaker[Session]) -> None: + """ + PostgreSQL treats NULLs in a unique constraint as distinct, which is relied on + :param sessions: factory handing out independent connections + """ + with sessions() as session: + seed_endpoint(session) + session.add(a_submission("sub_one")) + session.add(a_submission("sub_two")) + session.add(a_submission("sub_three")) + + session.commit() + + assert session.get(models.Submission, "sub_three") is not None + + +def test_a_submission_owes_at_most_one_delivery(sessions: sessionmaker[Session]) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(a_submission("sub_one")) + session.add(a_delivery("whd_one", "sub_one")) + session.commit() + + session.add(a_delivery("whd_two", "sub_one")) + with pytest.raises(IntegrityError): + session.commit() + + +def test_a_webhook_url_cannot_exist_without_its_secret(sessions: sessionmaker[Session]) -> None: + with sessions() as session: + session.add( + models.Endpoint( + id="contact-form", + name="Contact form", + is_active=True, + webhook_url="https://example.invalid/hook", + webhook_secret=None, + ) + ) + with pytest.raises(IntegrityError): + session.commit() + + +def test_half_an_idempotency_identity_is_refused(sessions: sessionmaker[Session]) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(a_submission("sub_one", idempotency_key="k" * 16)) + with pytest.raises(IntegrityError): + session.commit() + + +def test_a_terminal_delivery_must_have_a_completion_time( + sessions: sessionmaker[Session], +) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(a_submission("sub_one")) + delivery = a_delivery("whd_one", "sub_one") + delivery.state = DeliveryState.DELIVERED + delivery.completed_at = None + session.add(delivery) + with pytest.raises(IntegrityError): + session.commit() + + +def test_a_pending_delivery_must_not_have_a_completion_time( + sessions: sessionmaker[Session], +) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(a_submission("sub_one")) + delivery = a_delivery("whd_one", "sub_one") + delivery.completed_at = NOW + session.add(delivery) + with pytest.raises(IntegrityError): + session.commit() + + +def test_a_submission_cannot_reference_a_missing_endpoint( + sessions: sessionmaker[Session], +) -> None: + with sessions() as session: + session.add(a_submission("sub_one")) + with pytest.raises(IntegrityError): + session.commit() + + +def test_a_delivery_cannot_reference_a_missing_submission( + sessions: sessionmaker[Session], +) -> None: + with sessions() as session: + seed_endpoint(session) + session.add(a_delivery("whd_one", "sub_missing")) + with pytest.raises(IntegrityError): + session.commit() diff --git a/tests/integration/test_migrations_postgres.py b/tests/integration/test_migrations_postgres.py new file mode 100644 index 0000000..7b65e42 --- /dev/null +++ b/tests/integration/test_migrations_postgres.py @@ -0,0 +1,138 @@ +""" +migrations against a real PostgreSQL database + +Each test here owns a database of its own, created and dropped around it, so +migrating from genuinely nothing is what is being tested rather than migrating +from whatever a previous test happened to leave behind. +""" + +from __future__ import annotations + +from alembic import command +from alembic.autogenerate import compare_metadata +from alembic.migration import MigrationContext +from sqlalchemy import inspect, text + +from hymical_forms.db import create_engine_from_url +from hymical_forms.models import Base +from hymical_forms.schema import alembic_config, current_revision, head_revision +from integration.support import temporary_database + +EXPECTED_TABLES = {"endpoints", "submissions", "webhook_deliveries", "delivery_attempts"} + + +def test_an_empty_database_upgrades_to_head(postgres_url: str) -> None: + with temporary_database(postgres_url) as url: + engine = create_engine_from_url(url) + try: + assert current_revision(engine) is None + + command.upgrade(alembic_config(url), "head") + + assert current_revision(engine) == head_revision() + assert set(inspect(engine).get_table_names()) >= EXPECTED_TABLES + finally: + engine.dispose() + + +def test_the_migrated_schema_matches_the_models(postgres_url: str) -> None: + """ + the migration and the models must not be allowed to drift apart + :param postgres_url: a URL on the PostgreSQL server to work against + """ + # This is what makes it safe for the fast suite to build its schema with + # create_all instead of replaying migrations: the two are the same schema. + with temporary_database(postgres_url) as url: + command.upgrade(alembic_config(url), "head") + engine = create_engine_from_url(url) + try: + with engine.connect() as connection: + difference = compare_metadata(MigrationContext.configure(connection), Base.metadata) + finally: + engine.dispose() + + assert difference == [], f"migrated schema differs from the models: {difference}" + + +def test_the_migration_creates_the_constraints_the_application_relies_on( + postgres_url: str, +) -> None: + with temporary_database(postgres_url) as url: + command.upgrade(alembic_config(url), "head") + engine = create_engine_from_url(url) + try: + with engine.connect() as connection: + names = set( + connection.scalars( + text( + "select conname from pg_constraint c " + "join pg_class t on t.oid = c.conrelid " + "where t.relnamespace = 'public'::regnamespace" + ) + ) + ) + finally: + engine.dispose() + + assert { + "uq_submissions_endpoint_idempotency_key", + "uq_webhook_deliveries_submission", + "ck_endpoints_webhook_configuration", + "ck_submissions_idempotency_identity", + "ck_webhook_deliveries_completion", + "fk_submissions_endpoint_id_endpoints", + "fk_webhook_deliveries_submission_id_submissions", + "fk_delivery_attempts_delivery_id_webhook_deliveries", + "fk_delivery_attempts_submission_id_submissions", + } <= names + + +def test_timestamps_are_stored_with_a_timezone(postgres_url: str) -> None: + """ + the delivery queue compares instants, so a naive column would be a real bug + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with temporary_database(postgres_url) as url: + command.upgrade(alembic_config(url), "head") + engine = create_engine_from_url(url) + try: + with engine.connect() as connection: + rows = connection.execute( + text( + "select column_name, data_type from information_schema.columns " + "where table_name = 'webhook_deliveries'" + ) + ).all() + types = {str(row[0]): str(row[1]) for row in rows} + finally: + engine.dispose() + + assert types["next_attempt_at"] == "timestamp with time zone" + assert types["claim_expires_at"] == "timestamp with time zone" + assert types["completed_at"] == "timestamp with time zone" + + +def test_the_migration_round_trips(postgres_url: str) -> None: + """ + downgrading to base and upgrading again must leave the same schema + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with temporary_database(postgres_url) as url: + config = alembic_config(url) + engine = create_engine_from_url(url) + try: + command.upgrade(config, "head") + + command.downgrade(config, "base") + remaining = set(inspect(engine).get_table_names()) + assert remaining & EXPECTED_TABLES == set() + + command.upgrade(config, "head") + assert set(inspect(engine).get_table_names()) >= EXPECTED_TABLES + assert current_revision(engine) == head_revision() + + with engine.connect() as connection: + difference = compare_metadata(MigrationContext.configure(connection), Base.metadata) + assert difference == [] + finally: + engine.dispose() diff --git a/tests/integration/test_persistence_postgres.py b/tests/integration/test_persistence_postgres.py new file mode 100644 index 0000000..cd55a70 --- /dev/null +++ b/tests/integration/test_persistence_postgres.py @@ -0,0 +1,172 @@ +""" +representative application flows against a migrated PostgreSQL database + +Not a second copy of the API suite. The point is that the whole path works on +the schema Alembic produced, on the database this service is actually meant to +run on. +""" + +from __future__ import annotations + +from fastapi.testclient import TestClient +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms import models +from hymical_forms.webhooks import DeliveryState + +ENDPOINT = "/f/contact-form" +KEY = "b8f1c2d4e5a67890b8f1c2d4e5a67890" + + +def create_endpoint(client: TestClient, *, webhook: bool = True) -> dict[str, str]: + """ + register an endpoint through the API + :param client: the client to register through + :param webhook: whether to configure a webhook destination + :returns: the created endpoint as the API returned it + """ + body: dict[str, object] = {"id": "contact-form", "name": "Contact form"} + if webhook: + body["webhook_url"] = "https://example.invalid/hook" + response = client.post("/endpoints", json=body) + assert response.status_code == 201, response.text + return dict(response.json()) + + +def test_the_whole_ingestion_flow_works_on_postgresql( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + """ + endpoint, submission and queued delivery all land on the migrated schema + :param pg_client: an API client backed by PostgreSQL + :param sessions: factory handing out independent connections + """ + endpoint = create_endpoint(pg_client) + + response = pg_client.post( + ENDPOINT, data={"email": "dev@example.com", "topics": ["billing", "api"]} + ) + + assert response.status_code == 202 + body = response.json() + assert body["delivery"] == {"queued": True} + + with sessions() as session: + submission = session.get(models.Submission, body["submission_id"]) + assert submission is not None + assert submission.endpoint_id == "contact-form" + # Repeated values must survive the PostgreSQL json column intact. + assert submission.fields == { + "email": ["dev@example.com"], + "topics": ["billing", "api"], + } + assert submission.received_at.tzinfo is not None + + delivery = session.scalars(select(models.WebhookDelivery)).one() + assert delivery.submission_id == submission.id + assert delivery.state == DeliveryState.PENDING + assert delivery.destination_url == endpoint["webhook_url"] + assert delivery.signing_secret == endpoint["webhook_secret"] + + +def test_field_order_survives_the_postgresql_json_column( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + """ + the column is json rather than jsonb precisely so key order is preserved + :param pg_client: an API client backed by PostgreSQL + :param sessions: factory handing out independent connections + """ + create_endpoint(pg_client, webhook=False) + + pg_client.post( + ENDPOINT, + content=b"zebra=1&apple=2&mango=3", + headers={"content-type": "application/x-www-form-urlencoded"}, + ) + + with sessions() as session: + submission = session.scalars(select(models.Submission)).one() + assert list(submission.fields) == ["zebra", "apple", "mango"] + + +def test_an_idempotent_replay_duplicates_nothing_on_postgresql( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + create_endpoint(pg_client) + headers = {"Idempotency-Key": KEY} + + first = pg_client.post(ENDPOINT, data={"email": "dev@example.com"}, headers=headers) + second = pg_client.post(ENDPOINT, data={"email": "dev@example.com"}, headers=headers) + + assert first.status_code == second.status_code == 202 + assert second.json()["submission_id"] == first.json()["submission_id"] + assert second.json()["idempotent_replay"] is True + with sessions() as session: + assert len(list(session.scalars(select(models.Submission)))) == 1 + assert len(list(session.scalars(select(models.WebhookDelivery)))) == 1 + + +def test_an_idempotency_conflict_is_refused_on_postgresql( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + create_endpoint(pg_client) + headers = {"Idempotency-Key": KEY} + pg_client.post(ENDPOINT, data={"email": "dev@example.com"}, headers=headers) + + response = pg_client.post(ENDPOINT, data={"email": "other@example.com"}, headers=headers) + + assert response.status_code == 409 + assert response.json()["error"]["code"] == "idempotency_conflict" + with sessions() as session: + assert len(list(session.scalars(select(models.Submission)))) == 1 + + +def test_an_endpoint_without_a_webhook_queues_nothing_on_postgresql( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + create_endpoint(pg_client, webhook=False) + + response = pg_client.post(ENDPOINT, data={"email": "dev@example.com"}) + + assert response.json()["delivery"] == {"queued": False} + with sessions() as session: + assert list(session.scalars(select(models.WebhookDelivery))) == [] + + +def test_a_rejected_submission_persists_nothing_on_postgresql( + pg_client: TestClient, sessions: sessionmaker[Session] +) -> None: + create_endpoint(pg_client) + + response = pg_client.post( + ENDPOINT, content=b"", headers={"content-type": "application/x-www-form-urlencoded"} + ) + + assert response.status_code == 422 + with sessions() as session: + assert list(session.scalars(select(models.Submission))) == [] + assert list(session.scalars(select(models.WebhookDelivery))) == [] + + +def test_the_application_refuses_to_start_against_an_unmigrated_database( + postgres_url: str, +) -> None: + """ + startup checks the revision rather than quietly creating what it is missing + :param postgres_url: a URL on the PostgreSQL server to work against + """ + from hymical_forms.app import create_app + from hymical_forms.schema import SchemaNotReady + from integration.support import IsolatedSettings, temporary_database + + with temporary_database(postgres_url) as url: + app = create_app(IsolatedSettings(database_url=url)) + try: + with TestClient(app): + raise AssertionError("the application started against an empty database") + except SchemaNotReady as exc: + assert "alembic upgrade head" in str(exc) + finally: + app.state.engine.dispose() diff --git a/tests/test_errors.py b/tests/test_errors.py index f0a30cf..e4c21e9 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -9,6 +9,7 @@ from conftest import URLENCODED_HEADERS, build_settings from hymical_forms.app import create_app +from hymical_forms.schema import create_all ENDPOINT = "/f/contact-form" @@ -107,6 +108,7 @@ def test_wrong_methods_use_the_envelope(client: TestClient) -> None: def test_unexpected_errors_do_not_leak_internals() -> None: app = create_app(build_settings()) + create_all(app.state.engine) @app.get("/boom") async def boom() -> None: