From 1c2028501355add8d61f409da06c6534638f3d12 Mon Sep 17 00:00:00 2001 From: Josh Caponigro Date: Mon, 31 Aug 2026 23:03:58 -0500 Subject: [PATCH 1/2] fix: reject empty and oversized device tokens at validation --- models/device.py | 6 ++++-- tests/test_device_api.py | 24 ++++++++++++++++++++++-- 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/models/device.py b/models/device.py index a32d975..ae9d59b 100644 --- a/models/device.py +++ b/models/device.py @@ -1,6 +1,6 @@ from datetime import datetime -from pydantic import BaseModel +from pydantic import BaseModel, Field class DeviceRegistration(BaseModel): @@ -16,7 +16,9 @@ class DeviceRegistration(BaseModel): localizedModel (str): The localized model of the device. """ - token: str + # Bounded to the devices.token column (String(255)) so oversized input + # fails validation instead of raising a database error. + token: str = Field(min_length=1, max_length=255) name: str | None = None systemName: str | None = None systemVersion: str | None = None diff --git a/tests/test_device_api.py b/tests/test_device_api.py index a837949..e1ad3b9 100644 --- a/tests/test_device_api.py +++ b/tests/test_device_api.py @@ -1,5 +1,7 @@ """Tests for the /devices endpoints.""" +import pytest +from fastapi.testclient import TestClient from sqlalchemy import event from sqlalchemy.orm import Session @@ -30,10 +32,28 @@ def test_register_device_stores_optional_fields(client): assert body["systemVersion"] == "17.0" -def test_register_device_without_token_is_rejected(client): - response = client.post("/devices/register", json={"name": "no token"}) +@pytest.mark.parametrize( + "registration", + [ + {"name": "no token"}, + {"token": ""}, + {"token": "a" * 256}, + ], + ids=["missing token", "empty token", "token over column limit"], +) +def test_register_device_with_invalid_token_is_rejected( + client: TestClient, registration: dict[str, str] +) -> None: + response = client.post("/devices/register", json=registration) assert response.status_code == 422 + assert client.get("/devices/all").json() == [] + + +def test_register_device_accepts_token_at_column_limit(client: TestClient) -> None: + response = client.post("/devices/register", json={"token": "a" * 255}) + + assert response.status_code == 200 def test_get_all_devices_returns_registered_devices(client): From b5117eb70a64cf553acb3236fa17435ef30fbf21 Mon Sep 17 00:00:00 2001 From: Josh Caponigro Date: Mon, 31 Aug 2026 23:03:58 -0500 Subject: [PATCH 2/2] test: expand integration suite with lifecycle, auth, validation, and concurrency coverage --- tests/integration/conftest.py | 89 +++++++ tests/integration/server_harness.py | 162 ++++++++++++ tests/integration/test_lifecycle.py | 92 +++++++ tests/integration/test_server.py | 392 ++++++++++++++++------------ 4 files changed, 573 insertions(+), 162 deletions(-) create mode 100644 tests/integration/conftest.py create mode 100644 tests/integration/server_harness.py create mode 100644 tests/integration/test_lifecycle.py diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..6308db3 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,89 @@ +"""Fixtures for integration tests: the real server against real Postgres. + +One server (booted by server_harness.boot_server) is shared by the whole +session; tests that need a different environment or a restart boot their own +via the harness. Tests require a reachable Postgres instance configured via +the INTEGRATION_DB_* environment variables and are skipped when +INTEGRATION_DB_HOST is not set. +""" + +import os +from collections.abc import Iterator +from pathlib import Path + +import httpx +import pytest + +from tests.integration.server_harness import ( + API_KEY, + boot_server, + drop_devices_table, + integration_env, +) + + +def pytest_collection_modifyitems(items: list[pytest.Item]) -> None: + # The hook receives every collected item, not just this directory's. + integration_dir = Path(__file__).parent + skip = pytest.mark.skipif( + not os.getenv("INTEGRATION_DB_HOST"), + reason="INTEGRATION_DB_HOST not set; see README for running integration tests", + ) + for item in items: + if integration_dir in item.path.parents: + item.add_marker(pytest.mark.integration) + item.add_marker(skip) + + +@pytest.fixture(scope="session") +def server_tmp_dir(tmp_path_factory: pytest.TempPathFactory) -> Path: + return tmp_path_factory.mktemp("integration-servers") + + +@pytest.fixture(scope="session") +def server(server_tmp_dir: Path) -> Iterator[str]: + """One real server shared by the whole session, on a fresh database.""" + env = integration_env() + drop_devices_table(env) + with boot_server(server_tmp_dir, env) as running: + yield running.base_url + + +@pytest.fixture +def clean_devices(server: str) -> None: + """Start every test from an empty devices table, so tests stay + order-independent as the suite grows.""" + response = httpx.delete( + f"{server}/devices", headers={"Authorization": f"Bearer {API_KEY}"}, timeout=5 + ) + assert response.status_code == 200 + + +@pytest.fixture +def api(server: str, clean_devices: None) -> Iterator[httpx.Client]: + """HTTP client authenticated with the server's API key.""" + with httpx.Client(base_url=server, headers={"Authorization": f"Bearer {API_KEY}"}) as client: + yield client + + +@pytest.fixture +def anon_api(server: str, clean_devices: None) -> Iterator[httpx.Client]: + """HTTP client with no credentials.""" + with httpx.Client(base_url=server) as client: + yield client + + +BAD_AUTHORIZATIONS: dict[str, str] = { + "wrong key": "Bearer wrong-key", + "truncated key": f"Bearer {API_KEY[:-1]}", + "wrong case key": f"Bearer {API_KEY.upper()}", + "wrong scheme": f"Basic {API_KEY}", + "scheme only": "Bearer", + "key without scheme": API_KEY, +} + + +@pytest.fixture(params=BAD_AUTHORIZATIONS.values(), ids=BAD_AUTHORIZATIONS.keys()) +def bad_authorization(request: pytest.FixtureRequest) -> str: + """Authorization header values that must never authenticate.""" + return request.param diff --git a/tests/integration/server_harness.py b/tests/integration/server_harness.py new file mode 100644 index 0000000..b2a9e6d --- /dev/null +++ b/tests/integration/server_harness.py @@ -0,0 +1,162 @@ +"""Machinery for booting the real server as a subprocess during integration tests. + +The server is launched exactly as the README documents (`python main.py`), so +the entrypoint, startup schema creation, and graceful shutdown are all under +test. `boot_server` waits for `/health` and asserts a clean SIGINT shutdown on +exit; `boot_expecting_startup_failure` asserts a fast non-zero exit instead, +for tests that boot deliberately misconfigured servers. +""" + +import contextlib +import os +import secrets +import signal +import socket +import subprocess +import sys +import time +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path + +import httpx +import pytest +import sqlalchemy + +REPO_ROOT = Path(__file__).parent.parent.parent +API_KEY = secrets.token_hex(16) + +STARTUP_TIMEOUT = 20 +SHUTDOWN_TIMEOUT = 15 + + +def integration_env() -> dict[str, str]: + env = os.environ.copy() + env.update( + { + "DB_HOST": os.environ["INTEGRATION_DB_HOST"], + "DB_PORT": os.environ.get("INTEGRATION_DB_PORT", "5432"), + "DB_NAME": os.environ.get("INTEGRATION_DB_NAME", "postgres"), + "DB_USERNAME": os.environ.get("INTEGRATION_DB_USERNAME", "postgres"), + "DB_PASSWORD": os.environ.get("INTEGRATION_DB_PASSWORD", "postgres"), + "API_KEY": API_KEY, + "APNS_KEY_ID": "TESTKEY123", + "APNS_TEAM_ID": "TESTTEAM12", + "APNS_APP_BUNDLE_ID": "com.example.test", + "APNS_AUTH_KEY_PATH": str(REPO_ROOT / "tests" / "fixtures" / "apns_test_key.p8"), + "APNS_USE_SANDBOX": "true", + } + ) + env.pop("CORS_ORIGINS", None) + return env + + +def drop_devices_table(env: dict[str, str]) -> None: + url = ( + f"postgresql+psycopg2://{env['DB_USERNAME']}:{env['DB_PASSWORD']}" + f"@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}" + ) + engine = sqlalchemy.create_engine(url) + with engine.begin() as connection: + connection.execute(sqlalchemy.text("DROP TABLE IF EXISTS devices")) + engine.dispose() + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +@dataclass +class Server: + """A running server subprocess and how to reach it.""" + + process: subprocess.Popen[bytes] + base_url: str + log_path: Path + + def logs(self) -> str: + return self.log_path.read_text() + + def stop(self) -> int: + """Request graceful shutdown (SIGINT) and return the exit code.""" + if self.process.poll() is None: + self.process.send_signal(signal.SIGINT) + try: + self.process.wait(timeout=SHUTDOWN_TIMEOUT) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait() + pytest.fail(f"server did not shut down within {SHUTDOWN_TIMEOUT}s:\n{self.logs()}") + return self.process.returncode + + +@contextlib.contextmanager +def boot_server(tmp_dir: Path, env: dict[str, str] | None = None) -> Iterator[Server]: + """Boot `python main.py` and wait until `/health` answers. + + Yields a Server; on exit the server is shut down with SIGINT and a clean + exit code is asserted, so graceful shutdown is verified on every boot. + """ + env = env or integration_env() + port = _free_port() + env["PORT"] = str(port) + + # Server output goes to a file, not a pipe: an undrained pipe blocks the + # server once its buffer fills, which would hang the suite as it grows. + log_path = tmp_dir / f"server-{port}.log" + with open(log_path, "w") as log_file: + process = subprocess.Popen( + [sys.executable, "main.py"], + cwd=REPO_ROOT, + env=env, + stdout=log_file, + stderr=subprocess.STDOUT, + ) + server = Server(process, f"http://127.0.0.1:{port}", log_path) + try: + deadline = time.monotonic() + STARTUP_TIMEOUT + while time.monotonic() < deadline: + if process.poll() is not None: + pytest.fail(f"server exited during startup:\n{server.logs()}") + try: + if httpx.get(f"{server.base_url}/health", timeout=1).status_code == 200: + break + except httpx.TransportError: + pass + time.sleep(0.2) + else: + pytest.fail(f"server did not become healthy within {STARTUP_TIMEOUT}s") + + yield server + finally: + returncode = server.stop() + assert returncode == 0, f"unclean shutdown (exit {returncode}):\n{server.logs()}" + + +def boot_expecting_startup_failure(tmp_dir: Path, env: dict[str, str]) -> str: + """Boot the server with a broken environment and return its log output. + + Fails the test if the process serves traffic or survives past the startup + window instead of exiting with a non-zero code. + """ + port = _free_port() + env["PORT"] = str(port) + log_path = tmp_dir / f"server-fail-{port}.log" + with open(log_path, "w") as log_file: + process = subprocess.Popen( + [sys.executable, "main.py"], + cwd=REPO_ROOT, + env=env, + stdout=log_file, + stderr=subprocess.STDOUT, + ) + try: + process.wait(timeout=STARTUP_TIMEOUT) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + pytest.fail(f"server kept running despite broken config:\n{log_path.read_text()}") + assert process.returncode != 0, "expected startup to fail with a non-zero exit code" + return log_path.read_text() diff --git a/tests/integration/test_lifecycle.py b/tests/integration/test_lifecycle.py new file mode 100644 index 0000000..797eed6 --- /dev/null +++ b/tests/integration/test_lifecycle.py @@ -0,0 +1,92 @@ +"""Integration tests for server lifecycle: restarts, startup failures, CORS. + +Each test here boots its own server subprocess (or several) rather than using +the session-wide one, because they exercise boot-time behavior: schema +creation against an existing database, refusing to start on broken +configuration, and CORS middleware wiring from the environment. +""" + +from pathlib import Path + +import httpx + +from tests.integration.server_harness import ( + API_KEY, + boot_expecting_startup_failure, + boot_server, + integration_env, +) + + +def test_data_survives_a_server_restart(server_tmp_dir: Path) -> None: + # Cleared through the API rather than dropped: the session-wide server + # shares this database and must keep its table. + env = integration_env() + + with boot_server(server_tmp_dir, dict(env)) as first: + auth = {"Authorization": f"Bearer {API_KEY}"} + assert httpx.delete(f"{first.base_url}/devices", headers=auth).status_code == 200 + response = httpx.post( + f"{first.base_url}/devices/register", + json={"token": "durable-token", "name": "Survivor"}, + ) + assert response.status_code == 200 + + with boot_server(server_tmp_dir, dict(env)) as second: + devices = httpx.get( + f"{second.base_url}/devices/all", + headers={"Authorization": f"Bearer {API_KEY}"}, + ).json() + assert [(device["token"], device["name"]) for device in devices] == [ + ("durable-token", "Survivor") + ] + + +def test_startup_fails_fast_without_api_key(server_tmp_dir: Path) -> None: + env = integration_env() + del env["API_KEY"] + + logs = boot_expecting_startup_failure(server_tmp_dir, env) + + assert "API_KEY environment variable must be set" in logs + + +def test_startup_rejects_wildcard_cors_origins(server_tmp_dir: Path) -> None: + env = integration_env() + env["CORS_ORIGINS"] = "https://app.example.com, *" + + logs = boot_expecting_startup_failure(server_tmp_dir, env) + + assert "CORS_ORIGINS must list explicit origins" in logs + + +def test_cors_configured_server_only_allows_listed_origins(server_tmp_dir: Path) -> None: + env = integration_env() + env["CORS_ORIGINS"] = "https://app.example.com" + + with boot_server(server_tmp_dir, env) as running: + preflight = httpx.options( + f"{running.base_url}/devices/all", + headers={ + "Origin": "https://app.example.com", + "Access-Control-Request-Method": "GET", + "Access-Control-Request-Headers": "authorization", + }, + ) + assert preflight.status_code == 200 + assert preflight.headers["access-control-allow-origin"] == "https://app.example.com" + assert preflight.headers["access-control-allow-credentials"] == "true" + + cross_origin_get = httpx.get( + f"{running.base_url}/health", headers={"Origin": "https://app.example.com"} + ) + assert cross_origin_get.headers["access-control-allow-origin"] == "https://app.example.com" + + rejected_preflight = httpx.options( + f"{running.base_url}/devices/all", + headers={ + "Origin": "https://evil.example.com", + "Access-Control-Request-Method": "GET", + }, + ) + assert "access-control-allow-origin" not in rejected_preflight.headers diff --git a/tests/integration/test_server.py b/tests/integration/test_server.py index 76fc452..5978659 100644 --- a/tests/integration/test_server.py +++ b/tests/integration/test_server.py @@ -1,189 +1,257 @@ """Integration tests: the real server process against a real Postgres database. -These boot `uvicorn main:app` as a subprocess and drive it over HTTP, so they -cover startup schema creation, authentication, the full device lifecycle, and -graceful shutdown with no test doubles. They require a reachable Postgres -instance, configured via INTEGRATION_DB_* environment variables (see README), -and are skipped when INTEGRATION_DB_HOST is not set. +These drive the shared server (see conftest.py) over HTTP with no test +doubles, covering the device lifecycle, authentication, input validation, +and concurrent access. """ -import os -import secrets -import signal -import socket -import subprocess -import sys -import time -from pathlib import Path +from concurrent.futures import ThreadPoolExecutor +from typing import Any import httpx import pytest -import sqlalchemy - -pytestmark = [ - pytest.mark.integration, - pytest.mark.skipif( - not os.getenv("INTEGRATION_DB_HOST"), - reason="INTEGRATION_DB_HOST not set; see README for running integration tests", - ), -] - -REPO_ROOT = Path(__file__).parent.parent.parent -API_KEY = secrets.token_hex(16) - - -def _integration_env() -> dict[str, str]: - env = os.environ.copy() - env.update( - { - "DB_HOST": os.environ["INTEGRATION_DB_HOST"], - "DB_PORT": os.environ.get("INTEGRATION_DB_PORT", "5432"), - "DB_NAME": os.environ.get("INTEGRATION_DB_NAME", "postgres"), - "DB_USERNAME": os.environ.get("INTEGRATION_DB_USERNAME", "postgres"), - "DB_PASSWORD": os.environ.get("INTEGRATION_DB_PASSWORD", "postgres"), - "API_KEY": API_KEY, - "APNS_KEY_ID": "TESTKEY123", - "APNS_TEAM_ID": "TESTTEAM12", - "APNS_APP_BUNDLE_ID": "com.example.test", - "APNS_AUTH_KEY_PATH": str(REPO_ROOT / "tests" / "fixtures" / "apns_test_key.p8"), - "APNS_USE_SANDBOX": "true", - } - ) - return env -def _drop_devices_table(env: dict[str, str]) -> None: - url = ( - f"postgresql+psycopg2://{env['DB_USERNAME']}:{env['DB_PASSWORD']}" - f"@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}" - ) - engine = sqlalchemy.create_engine(url) - with engine.begin() as connection: - connection.execute(sqlalchemy.text("DROP TABLE IF EXISTS devices")) - engine.dispose() - - -def _free_port() -> int: - with socket.socket() as sock: - sock.bind(("127.0.0.1", 0)) - return sock.getsockname()[1] - - -@pytest.fixture(scope="module") -def server(tmp_path_factory): - """The real application booted as a subprocess against real Postgres.""" - env = _integration_env() - _drop_devices_table(env) - port = _free_port() - env["PORT"] = str(port) - - # Server output goes to a file, not a pipe: an undrained pipe blocks the - # server once its buffer fills, which would hang the suite as it grows. - log_path = tmp_path_factory.mktemp("server") / "server.log" - with open(log_path, "w") as log_file: - # Launched exactly as the README documents, so the entrypoint itself - # (including HOST/PORT handling) is part of what these tests cover. - process = subprocess.Popen( - [sys.executable, "main.py"], - cwd=REPO_ROOT, - env=env, - stdout=log_file, - stderr=subprocess.STDOUT, +class TestHealthAndRouting: + def test_health_and_root_are_public(self, anon_api: httpx.Client) -> None: + assert anon_api.get("/health").json() == {"message": "OK"} + assert anon_api.get("/").json() == {"message": "Hello World"} + + def test_openapi_schema_is_served(self, anon_api: httpx.Client) -> None: + schema = anon_api.get("/openapi.json") + assert schema.status_code == 200 + paths = schema.json()["paths"] + assert "/devices/register" in paths + assert "/push/send" in paths + + def test_unknown_route_is_404(self, api: httpx.Client) -> None: + assert api.get("/devices/nope").status_code == 404 + + def test_wrong_method_is_405(self, api: httpx.Client) -> None: + assert api.get("/push/send").status_code == 405 + + def test_old_clear_route_is_gone(self, api: httpx.Client) -> None: + assert api.get("/devices/clear").status_code in (404, 405) + + def test_cors_headers_absent_when_cors_not_configured(self, anon_api: httpx.Client) -> None: + response = anon_api.get("/health", headers={"Origin": "http://evil.example"}) + assert "access-control-allow-origin" not in response.headers + + +class TestDeviceLifecycle: + def test_startup_creates_schema_and_serves_devices(self, api: httpx.Client) -> None: + assert api.get("/devices/all").status_code == 200 + + def test_device_lifecycle(self, api: httpx.Client, anon_api: httpx.Client) -> None: + registered = anon_api.post( + "/devices/register", + json={"token": "integration-token", "name": "Integration Phone"}, ) - base_url = f"http://127.0.0.1:{port}" - try: - deadline = time.monotonic() + 20 - while time.monotonic() < deadline: - if process.poll() is not None: - pytest.fail(f"server exited during startup:\n{log_path.read_text()}") - try: - if httpx.get(f"{base_url}/health", timeout=1).status_code == 200: - break - except httpx.TransportError: - pass - time.sleep(0.2) - else: - pytest.fail("server did not become healthy within 20s") - - yield base_url - finally: - process.send_signal(signal.SIGINT) - try: - process.wait(timeout=15) - except subprocess.TimeoutExpired: - process.kill() - pytest.fail("server did not shut down gracefully within 15s") - assert process.returncode == 0, f"unclean shutdown:\n{log_path.read_text()}" - - -@pytest.fixture -def api(server): - """HTTP client authenticated with the server's API key.""" - with httpx.Client(base_url=server, headers={"Authorization": f"Bearer {API_KEY}"}) as client: - yield client - - -@pytest.fixture -def anon_api(server): - """HTTP client with no credentials.""" - with httpx.Client(base_url=server) as client: - yield client - - -def test_startup_creates_schema_and_serves_health(api): - assert api.get("/health").json() == {"message": "OK"} - assert api.get("/devices/all").status_code == 200 - - -def test_device_lifecycle(api, anon_api): - registered = anon_api.post( - "/devices/register", - json={"token": "integration-token", "name": "Integration Phone"}, - ) - assert registered.status_code == 200 - body = registered.json() - assert body["id"] is not None - assert body["created_at"] is not None + assert registered.status_code == 200 + body = registered.json() + assert body["id"] is not None + assert body["created_at"] is not None - reregistered = anon_api.post( - "/devices/register", json={"token": "integration-token", "systemVersion": "18.0"} + reregistered = anon_api.post( + "/devices/register", json={"token": "integration-token", "systemVersion": "18.0"} + ) + assert reregistered.status_code == 200 + assert reregistered.json()["id"] == body["id"] + assert reregistered.json()["name"] == "Integration Phone" + assert reregistered.json()["systemVersion"] == "18.0" + + tokens = {device["token"] for device in api.get("/devices/all").json()} + assert "integration-token" in tokens + + assert api.delete("/devices").status_code == 200 + assert api.get("/devices/all").json() == [] + + def test_reregistration_preserves_created_at_and_identity( + self, anon_api: httpx.Client + ) -> None: + first = anon_api.post("/devices/register", json={"token": "stable-token"}).json() + second = anon_api.post( + "/devices/register", json={"token": "stable-token", "name": "Named later"} + ).json() + + assert second["id"] == first["id"] + assert second["created_at"] == first["created_at"] + assert second["name"] == "Named later" + + def test_full_device_metadata_round_trips( + self, api: httpx.Client, anon_api: httpx.Client + ) -> None: + registration = { + "token": "full-token", + "name": "Josh’s iPhone \N{MOBILE PHONE}", + "systemName": "iOS", + "systemVersion": "18.4.1", + "model": "iPhone", + "localizedModel": "iPhone", + } + registered = anon_api.post("/devices/register", json=registration) + assert registered.status_code == 200 + + (stored,) = api.get("/devices/all").json() + for field, value in registration.items(): + assert stored[field] == value + + def test_many_devices_are_all_listed( + self, api: httpx.Client, anon_api: httpx.Client + ) -> None: + for i in range(25): + response = anon_api.post("/devices/register", json={"token": f"bulk-token-{i}"}) + assert response.status_code == 200 + + tokens = {device["token"] for device in api.get("/devices/all").json()} + assert tokens == {f"bulk-token-{i}" for i in range(25)} + + def test_clear_devices_is_idempotent(self, api: httpx.Client) -> None: + assert api.delete("/devices").status_code == 200 + assert api.delete("/devices").status_code == 200 + assert api.get("/devices/all").json() == [] + + +class TestRegistrationValidation: + @pytest.mark.parametrize( + "registration", + [ + {"name": "no token"}, + {"token": ""}, + {"token": 12345}, + {"token": "a" * 256}, + ], + ids=["missing token", "empty token", "non-string token", "token over column limit"], ) - assert reregistered.status_code == 200 - assert reregistered.json()["id"] == body["id"] - assert reregistered.json()["name"] == "Integration Phone" - assert reregistered.json()["systemVersion"] == "18.0" + def test_invalid_registration_is_rejected_and_stores_nothing( + self, api: httpx.Client, anon_api: httpx.Client, registration: dict[str, Any] + ) -> None: + response = anon_api.post("/devices/register", json=registration) + + assert response.status_code == 422 + assert api.get("/devices/all").json() == [] + + def test_malformed_json_is_rejected(self, anon_api: httpx.Client) -> None: + response = anon_api.post( + "/devices/register", + content=b'{"token": ', + headers={"Content-Type": "application/json"}, + ) + assert response.status_code == 422 - tokens = {device["token"] for device in api.get("/devices/all").json()} - assert "integration-token" in tokens + def test_unknown_fields_are_ignored_not_stored(self, anon_api: httpx.Client) -> None: + response = anon_api.post( + "/devices/register", json={"token": "extra-token", "id": 999999, "isAdmin": True} + ) + assert response.status_code == 200 + assert response.json()["id"] != 999999 + assert "isAdmin" not in response.json() + + def test_token_at_column_limit_round_trips( + self, api: httpx.Client, anon_api: httpx.Client + ) -> None: + token = "a" * 255 - assert api.delete("/devices").status_code == 200 - assert api.get("/devices/all").json() == [] + assert anon_api.post("/devices/register", json={"token": token}).status_code == 200 + assert [device["token"] for device in api.get("/devices/all").json()] == [token] -def test_registration_rejects_missing_token(anon_api): - assert anon_api.post("/devices/register", json={"name": "no token"}).status_code == 422 +class TestAuthentication: + def test_protected_routes_require_credentials( + self, anon_api: httpx.Client, protected_route: tuple[str, str] + ) -> None: + method, path = protected_route + response = anon_api.request(method, path, json={"recipients": [], "body": "x"}) -def test_protected_routes_require_credentials(anon_api, protected_route): - method, path = protected_route + assert response.status_code == 401 + assert response.headers["WWW-Authenticate"] == "Bearer" - response = anon_api.request(method, path, json={"recipients": [], "body": "x"}) + def test_rejected_credentials( + self, anon_api: httpx.Client, protected_route: tuple[str, str], bad_authorization: str + ) -> None: + method, path = protected_route - assert response.status_code == 401 - assert response.headers["WWW-Authenticate"] == "Bearer" + response = anon_api.request( + method, + path, + json={"recipients": [], "body": "x"}, + headers={"Authorization": bad_authorization}, + ) + assert response.status_code == 401 -def test_protected_routes_reject_wrong_key(server): - with httpx.Client(base_url=server, headers={"Authorization": "Bearer wrong-key"}) as client: - assert client.get("/devices/all").status_code == 401 + def test_registration_does_not_require_credentials(self, anon_api: httpx.Client) -> None: + response = anon_api.post("/devices/register", json={"token": "anon-token"}) + assert response.status_code == 200 + def test_health_ignores_bad_credentials(self, anon_api: httpx.Client) -> None: + response = anon_api.get("/health", headers={"Authorization": "Bearer wrong"}) + assert response.status_code == 200 -def test_push_send_with_no_recipients_succeeds_authenticated(api): - response = api.post("/push/send", json={"recipients": [], "body": "hello"}) - assert response.status_code == 200 - assert response.json() == {} +class TestPush: + def test_push_send_with_no_recipients_succeeds_authenticated(self, api: httpx.Client) -> None: + response = api.post("/push/send", json={"recipients": [], "body": "hello"}) + assert response.status_code == 200 + assert response.json() == {} -def test_old_clear_route_is_gone(api): - assert api.get("/devices/clear").status_code in (404, 405) + @pytest.mark.parametrize( + "message", + [ + {"body": "hello"}, + {"recipients": []}, + {"recipients": "not-a-list", "body": "x"}, + ], + ids=["missing recipients", "missing body", "non-list recipients"], + ) + def test_invalid_push_message_is_rejected( + self, api: httpx.Client, message: dict[str, Any] + ) -> None: + assert api.post("/push/send", json=message).status_code == 422 + + +class TestConcurrency: + def test_concurrent_registration_of_same_token_yields_one_device( + self, api: httpx.Client, anon_api: httpx.Client + ) -> None: + def register(i: int) -> httpx.Response: + return anon_api.post( + "/devices/register", json={"token": "race-token", "name": f"racer-{i}"} + ) + + with ThreadPoolExecutor(max_workers=16) as pool: + responses = list(pool.map(register, range(16))) + + assert all(response.status_code == 200 for response in responses) + assert len({response.json()["id"] for response in responses}) == 1 + devices = api.get("/devices/all").json() + assert [device["token"] for device in devices] == ["race-token"] + + def test_concurrent_registration_of_distinct_tokens_stores_all( + self, api: httpx.Client, server: str + ) -> None: + def register(i: int) -> httpx.Response: + with httpx.Client(base_url=server, timeout=10) as client: + return client.post("/devices/register", json={"token": f"parallel-token-{i}"}) + + with ThreadPoolExecutor(max_workers=8) as pool: + responses = list(pool.map(register, range(24))) + + assert all(response.status_code == 200 for response in responses) + tokens = {device["token"] for device in api.get("/devices/all").json()} + assert tokens == {f"parallel-token-{i}" for i in range(24)} + + def test_concurrent_reads_during_writes_stay_consistent( + self, api: httpx.Client, anon_api: httpx.Client + ) -> None: + def hammer(i: int) -> httpx.Response: + if i % 2: + return anon_api.post("/devices/register", json={"token": f"mixed-token-{i}"}) + return api.get("/devices/all") + + with ThreadPoolExecutor(max_workers=8) as pool: + responses = list(pool.map(hammer, range(20))) + + assert all(response.status_code == 200 for response in responses)