Commit e2a9508577
Verified · cmc
Layout: unified · split
src/srht_contrib/main.py +12
| @@ -2,6 +2,7 @@ from __future__ import annotations | ||
| 2 | 2 | |
| 3 | 3 | from collections.abc import AsyncIterator |
| 4 | 4 | from contextlib import asynccontextmanager |
| 5 | import logging | |
| 5 | 6 | |
| 6 | 7 | from apscheduler.schedulers.background import BackgroundScheduler |
| 7 | 8 | from fastapi import FastAPI |
| @@ -21,6 +22,9 @@ from srht_contrib.services.todo import TodoIngestionService | ||
| 21 | 22 | from srht_contrib.utils.identity import ActorIdentityResolver |
| 22 | 23 | |
| 23 | 24 | |
| 25 | logger = logging.getLogger(__name__) | |
| 26 | ||
| 27 | ||
| 24 | 28 | def build_poller(settings: Settings) -> PollerService: |
| 25 | 29 | todo_client = SourceHutGraphQLClient(settings.todo_srht_endpoint, settings.srht_token) |
| 26 | 30 | git_client = SourceHutGraphQLClient(settings.git_srht_endpoint, settings.srht_token) |
| @@ -68,6 +72,7 @@ def create_app( | ||
| 68 | 72 | replace_existing=True, |
| 69 | 73 | ) |
| 70 | 74 | scheduler.start() |
| 75 | _run_startup_poll(app) | |
| 71 | 76 | app.state.scheduler = scheduler |
| 72 | 77 | try: |
| 73 | 78 | yield |
| @@ -96,4 +101,11 @@ def _scheduled_poll(app: FastAPI) -> None: | ||
| 96 | 101 | db.close() |
| 97 | 102 | |
| 98 | 103 | |
| 104 | def _run_startup_poll(app: FastAPI) -> None: | |
| 105 | try: | |
| 106 | _scheduled_poll(app) | |
| 107 | except Exception: | |
| 108 | logger.exception("Initial scheduled poll failed during application startup") | |
| 109 | ||
| 110 | ||
| 99 | 111 | app = create_app() |
tests/test_polling_api.py +31
| @@ -19,6 +19,7 @@ class InsertingPoller: | ||
| 19 | 19 | service = type("Service", (), {"client": _Closable()})() |
| 20 | 20 | self.todo_service = service |
| 21 | 21 | self.git_service = service |
| 22 | self.tracked_poll_calls: list[str] = [] | |
| 22 | 23 | |
| 23 | 24 | def track_actor_request(self, db, actor: str, *, update_last_requested: bool = True): |
| 24 | 25 | tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor)) |
| @@ -47,6 +48,12 @@ class InsertingPoller: | ||
| 47 | 48 | db.commit() |
| 48 | 49 | return 1 |
| 49 | 50 | |
| 51 | def poll_tracked_actors(self, db, default_actor: str | None = None) -> dict[str, int]: | |
| 52 | if default_actor is not None: | |
| 53 | self.tracked_poll_calls.append(default_actor) | |
| 54 | return {default_actor: self.poll_all(db, default_actor)} | |
| 55 | return {} | |
| 56 | ||
| 50 | 57 | |
| 51 | 58 | class FailingPoller: |
| 52 | 59 | def __init__(self) -> None: |
| @@ -65,6 +72,9 @@ class FailingPoller: | ||
| 65 | 72 | def poll_all(self, db, actor: str) -> int: |
| 66 | 73 | raise SourceHutClientError("boom") |
| 67 | 74 | |
| 75 | def poll_tracked_actors(self, db, default_actor: str | None = None) -> dict[str, int]: | |
| 76 | raise SourceHutClientError("boom") | |
| 77 | ||
| 68 | 78 | |
| 69 | 79 | def test_manual_poll_uses_same_database_session(settings: Settings, db_engine, session_factory) -> None: |
| 70 | 80 | app = create_app(settings, engine=db_engine, session_factory=session_factory, poller=InsertingPoller()) |
| @@ -89,3 +99,24 @@ def test_manual_poll_maps_sourcehut_failures_to_502(settings: Settings, db_engin | ||
| 89 | 99 | |
| 90 | 100 | assert response.status_code == 502 |
| 91 | 101 | assert "SourceHut polling failed" in response.json()["detail"] |
| 102 | ||
| 103 | ||
| 104 | def test_scheduler_runs_initial_poll_on_startup(settings: Settings, db_engine, session_factory) -> None: | |
| 105 | scheduler_settings = settings.model_copy(update={"enable_scheduler": True}) | |
| 106 | poller = InsertingPoller() | |
| 107 | app = create_app(scheduler_settings, engine=db_engine, session_factory=session_factory, poller=poller) | |
| 108 | ||
| 109 | with TestClient(app): | |
| 110 | pass | |
| 111 | ||
| 112 | assert poller.tracked_poll_calls == ["~ccleberg"] | |
| 113 | ||
| 114 | ||
| 115 | def test_startup_poll_failure_does_not_block_app_start(settings: Settings, db_engine, session_factory) -> None: | |
| 116 | scheduler_settings = settings.model_copy(update={"enable_scheduler": True}) | |
| 117 | app = create_app(scheduler_settings, engine=db_engine, session_factory=session_factory, poller=FailingPoller()) | |
| 118 | ||
| 119 | with TestClient(app) as client: | |
| 120 | response = client.get("/health") | |
| 121 | ||
| 122 | assert response.status_code == 200 | |