Close coverage gaps and add a CI coverage gate

.github/workflows/test.yml

@@ -48,12 +48,12 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         pip install -r requirements.txt
-        pip install ruff mypy
+        pip install "ruff==0.16.3" "mypy==2.3.1"
     - name: Lint with ruff
       run: |
         ruff check .
         ruff format --check .
     - name: Type-check with mypy
       run: mypy apis entities models push services utils main.py database.py auth.py
     - name: Test with pytest
-      run: pytest
+      run: pytest --cov=. --cov-fail-under=95

main.py

@@ -6,6 +6,7 @@
 import os
 from contextlib import asynccontextmanager
 
+import uvicorn
 from fastapi import APIRouter, FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
@@ -68,6 +69,4 @@ async def root():
 app = create_app()
 
 if __name__ == "__main__":
-    import uvicorn
-
     uvicorn.run(app, host=getenv("HOST", "127.0.0.1"), port=int(getenv("PORT", "8000")))

pyproject.toml

@@ -37,3 +37,13 @@ check_untyped_defs = true
 [[tool.mypy.overrides]]
 module = "tests.*"
 disable_error_code = ["method-assign"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
+markers = [
+    "integration: boots the real server against real Postgres (needs INTEGRATION_DB_HOST)",
+]
+
+[tool.coverage.run]
+omit = [".venv/*", "tests/*"]

requirements.txt

@@ -6,3 +6,4 @@ python-dotenv >=1.0.0, <2.0.0
 pydantic >=2.13.0, <3.0.0
 PyJWT[crypto] >=2.10.0, <3.0.0
 psycopg2-binary >=2.9.0, <2.10.0
+pytest-cov >=7.0.0, <8.0.0

tests/test_apns_client.py

@@ -6,7 +6,14 @@
 import pytest
 
 from push.apn_handler import APNsClient, Payload, TokenCredentials
-from push.apn_handler.errors import APNsException, BadDeviceToken, Unregistered
+from push.apn_handler.client import NotificationPriority
+from push.apn_handler.errors import (
+    APNsException,
+    BadDeviceToken,
+    ExpiredToken,
+    InvalidPushType,
+    Unregistered,
+)
 
 KEY_PATH = str(Path(__file__).parent / "fixtures" / "apns_test_key.p8")
 
@@ -137,8 +144,6 @@ def test_non_dict_json_error_body_maps_to_status_marker(monkeypatch):
 
 
 def test_new_apns_reasons_map_to_typed_exceptions(monkeypatch):
-    from push.apn_handler.errors import InvalidPushType
-
     client = make_client(monkeypatch, 400, {"reason": "InvalidPushType"})
 
     with pytest.raises(InvalidPushType):
@@ -156,3 +161,59 @@ def test_live_activity_topic_infers_liveactivity_push_type(monkeypatch):
     )
 
     assert requests[0].headers["apns-push-type"] == "liveactivity"
+
+
+def test_optional_headers_are_sent_when_specified(monkeypatch):
+    requests = []
+    client = make_client(monkeypatch, 200, {}, requests)
+
+    client.send_notification(
+        "device-token",
+        Payload(alert="hello"),
+        topic="com.example.test",
+        priority=NotificationPriority.Delayed,
+        expiration=0,
+        collapse_id="thread-1",
+    )
+
+    headers = requests[0].headers
+    assert headers["apns-priority"] == "5"
+    assert headers["apns-expiration"] == "0"
+    assert headers["apns-collapse-id"] == "thread-1"
+
+
+def test_default_priority_sends_no_priority_header(monkeypatch):
+    requests = []
+    client = make_client(monkeypatch, 200, {}, requests)
+
+    client.send_notification("device-token", Payload(alert="hello"), topic="com.example.test")
+
+    assert "apns-priority" not in requests[0].headers
+
+
+def test_silent_payload_infers_background_push_type(monkeypatch):
+    requests = []
+    client = make_client(monkeypatch, 200, {}, requests)
+
+    client.send_notification(
+        "device-token", Payload(content_available=True), topic="com.example.test"
+    )
+
+    assert requests[0].headers["apns-push-type"] == "background"
+
+
+def test_no_topic_sends_no_topic_or_push_type_headers(monkeypatch):
+    requests = []
+    client = make_client(monkeypatch, 200, {}, requests)
+
+    client.send_notification("device-token", Payload(alert="hello"))
+
+    assert "apns-topic" not in requests[0].headers
+    assert "apns-push-type" not in requests[0].headers
+
+
+def test_expired_token_maps_to_typed_exception(monkeypatch):
+    client = make_client(monkeypatch, 410, {"reason": "ExpiredToken", "timestamp": "1700000000"})
+
+    with pytest.raises(ExpiredToken):
+        client.send_notification("dead-token", Payload(alert="hello"), topic="com.example.test")

tests/test_database.py

@@ -0,0 +1,23 @@
+"""Tests for the database session dependency."""
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.pool import QueuePool
+
+import database
+
+
+def test_db_session_yields_working_session_and_releases_connection(monkeypatch):
+    engine = create_engine("sqlite://", poolclass=QueuePool)
+    monkeypatch.setattr(database, "engine", engine)
+
+    generator = database.db_session()
+    session = next(generator)
+    assert session.execute(text("SELECT 1")).scalar() == 1
+    assert engine.pool.checkedout() == 1
+
+    with pytest.raises(StopIteration):
+        next(generator)
+
+    assert engine.pool.checkedout() == 0
+    engine.dispose()

tests/test_device_api.py

@@ -1,5 +1,6 @@
 """Tests for the /devices endpoints."""
 
+from sqlalchemy import event
 from sqlalchemy.orm import Session
 
 from models import DeviceRegistration
@@ -130,3 +131,27 @@ def test_register_ignores_client_supplied_server_fields(client):
     assert body["id"] != first["id"]
     assert body["created_at"] != "2000-01-01T00:00:00"
     assert len(client.get("/devices/all").json()) == 2
+
+
+def test_root_endpoint(client):
+    response = client.get("/")
+
+    assert response.status_code == 200
+    assert response.json() == {"message": "Hello World"}
+
+
+def test_remove_devices_with_no_tokens_issues_no_sql(test_engine):
+
+    statements = []
+
+    def record(conn, cursor, statement, parameters, context, executemany):
+        statements.append(statement)
+
+    event.listen(test_engine, "before_cursor_execute", record)
+    try:
+        with Session(test_engine) as session:
+            DeviceService(session=session).remove_devices([])
+    finally:
+        event.remove(test_engine, "before_cursor_execute", record)
+
+    assert statements == []

tests/test_env.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from utils import getenv
+from utils import getenv, getenv_bool
 
 
 def test_getenv_returns_value_when_set(monkeypatch):
@@ -26,34 +26,26 @@ def test_getenv_honors_falsy_default(monkeypatch):
 
 @pytest.mark.parametrize("value", ["1", "true", "YES ", " on"])
 def test_getenv_bool_parses_truthy_values(monkeypatch, value):
-    from utils import getenv_bool
-
     monkeypatch.setenv("SOME_TEST_VAR", value)
 
     assert getenv_bool("SOME_TEST_VAR") is True
 
 
 @pytest.mark.parametrize("value", ["0", "false", "NO", "off", ""])
 def test_getenv_bool_parses_falsy_values(monkeypatch, value):
-    from utils import getenv_bool
-
     monkeypatch.setenv("SOME_TEST_VAR", value)
 
     assert getenv_bool("SOME_TEST_VAR") is False
 
 
 def test_getenv_bool_returns_default_when_unset(monkeypatch):
-    from utils import getenv_bool
-
     monkeypatch.delenv("SOME_TEST_VAR", raising=False)
 
     assert getenv_bool("SOME_TEST_VAR") is False
     assert getenv_bool("SOME_TEST_VAR", default=True) is True
 
 
 def test_getenv_bool_rejects_unrecognized_values(monkeypatch):
-    from utils import getenv_bool
-
     monkeypatch.setenv("SOME_TEST_VAR", "banana")
 
     with pytest.raises(ValueError):

tests/test_payload.py

@@ -0,0 +1,100 @@
+"""Tests for APNs payload serialization.
+
+Key names and nesting must match Apple's payload schema exactly; a wrong key
+is silently ignored by APNs rather than rejected, so these tests are the only
+guard against that class of bug.
+"""
+
+from push.apn_handler import Payload
+from push.apn_handler.payload import PayloadAlert
+
+
+def test_string_alert():
+    assert Payload(alert="hello").dict() == {"aps": {"alert": "hello"}}
+
+
+def test_empty_payload_has_empty_aps():
+    assert Payload().dict() == {"aps": {}}
+
+
+def test_alert_with_sound_and_badge():
+    result = Payload(alert="hello", sound="default", badge=3).dict()
+
+    assert result == {"aps": {"alert": "hello", "sound": "default", "badge": 3}}
+
+
+def test_badge_zero_is_included():
+    # badge=0 is meaningful: it clears the app icon badge.
+    assert Payload(badge=0).dict()["aps"]["badge"] == 0
+
+
+def test_background_push():
+    result = Payload(content_available=True).dict()
+
+    assert result == {"aps": {"content-available": 1}}
+
+
+def test_mutable_content_flag():
+    assert Payload(alert="hi", mutable_content=True).dict()["aps"]["mutable-content"] == 1
+
+
+def test_category_and_thread_id():
+    result = Payload(alert="hi", category="MESSAGE", thread_id="chat-42").dict()
+
+    assert result["aps"]["category"] == "MESSAGE"
+    assert result["aps"]["thread-id"] == "chat-42"
+
+
+def test_url_args():
+    assert Payload(url_args=["a", "b"]).dict()["aps"]["url-args"] == ["a", "b"]
+
+
+def test_custom_data_merges_at_top_level():
+    result = Payload(alert="hi", custom={"conversation_id": 7}).dict()
+
+    assert result["conversation_id"] == 7
+    assert "conversation_id" not in result["aps"]
+
+
+def test_payload_alert_full_fields():
+    alert = PayloadAlert(
+        title="Title",
+        title_localized_key="TITLE_KEY",
+        title_localized_args=["t1"],
+        subtitle="Subtitle",
+        subtitle_localized_key="SUBTITLE_KEY",
+        subtitle_localized_args=["s1"],
+        body="Body",
+        body_localized_key="BODY_KEY",
+        body_localized_args=["b1"],
+        action_localized_key="ACTION_KEY",
+        action="View",
+        launch_image="launch.png",
+    )
+
+    assert alert.dict() == {
+        "title": "Title",
+        "title-loc-key": "TITLE_KEY",
+        "title-loc-args": ["t1"],
+        "subtitle": "Subtitle",
+        "subtitle-loc-key": "SUBTITLE_KEY",
+        "subtitle-loc-args": ["s1"],
+        "body": "Body",
+        "loc-key": "BODY_KEY",
+        "loc-args": ["b1"],
+        "action-loc-key": "ACTION_KEY",
+        "action": "View",
+        "launch-image": "launch.png",
+    }
+
+
+def test_payload_alert_omits_unset_fields():
+    assert PayloadAlert(title="Title").dict() == {"title": "Title"}
+
+
+def test_structured_alert_nests_inside_aps():
+    alert = PayloadAlert(title="Title", body="Body")
+
+    result = Payload(alert=alert).dict()
+
+    assert result["aps"]["alert"] == {"title": "Title", "body": "Body"}