chore(tests):

- update tests with daemon context fixture
- update mypy config for apscheduler
- fix incidental type errors
- refresh lock file

Author: tcjennings

packages/cm-models/src/lsst/cmservice/models/db/schedules.py

@@ -10,7 +10,8 @@
 
 from ..enums import ManifestKind
 from ..types import KindField
-from .campaigns import BaseSQLModel, jsonb_column
+from .base import BaseSQLModel
+from .campaigns import jsonb_column
 
 
 class ScheduleBase(BaseSQLModel):
@@ -64,7 +65,6 @@ class ManifestTemplateBase(BaseSQLModel):
     kind: KindField = Field(
         sa_column=Column("kind", Enum(ManifestKind, length=20, native_enum=False, create_constraint=False)),
     )
-    schedule_id: UUID4 = Field(foreign_key="schedules_v2.id", ondelete="CASCADE")
     manifest: dict = jsonb_column("manifest")
     metadata_: dict = jsonb_column("metadata", aliases=["metadata", "metadata_"])
 
@@ -74,12 +74,13 @@ class ManifestTemplate(ManifestTemplateBase, table=True):
 
     model_config = {"validate_assignment": True}
     __tablename__: str = "templates_v2"  # type: ignore[misc]
+    schedule_id: UUID4 = Field(foreign_key="schedules_v2.id", ondelete="CASCADE")
 
 
 class CreateManifestTemplate(ManifestTemplateBase):
     """A validating model for manifest templates in the new schedule API."""
 
-    # This model differs from its parent in that the schedule_id is an optional
-    # field instead of a mandatory FK constraint, allowing it to be used to
+    # This model differs from its sibling in that the schedule_id is optional
+    # rather than a mandatory FK constraint, allowing it to be used to
     # create new ManifestTemplate objects.
     schedule_id: UUID4 | None = None

pyproject.toml

@@ -25,6 +25,7 @@ dynamic = ["version"]
 dependencies = [
     "alembic==1.16.*",
     "anyio==4.10.*",
+    "apscheduler>=3.11.2,<4",
     "asgi-correlation-id>=4.3.4",
     "asyncpg==0.30.*",
     "click==8.1.*",
@@ -46,15 +47,14 @@ dependencies = [
     "structlog==25.*",
     "tabulate==0.9.*",
     "uvicorn[standard]==0.34.*",
-    "lsst-cmservice-models",
-    "apscheduler>=3.11.2",
 ]
 
 [dependency-groups]
 # The LSST dependencies are listed separately to more easily identify the first-
 # party dependencies. This dependency group is not optional and will always be
 # included in the default installation of the project environment.
 lsst = [
+    "lsst-cmservice-models",
     "lsst-ctrl-bps>=29.2025.4700",
     "lsst-ctrl-bps-htcondor>=29.2025.4700; sys_platform == 'linux'",
     "lsst-ctrl-bps-panda>=29.2025.4700",
@@ -256,7 +256,7 @@ explicit_package_bases = true
 namespace_packages = true
 
 [[tool.mypy.overrides]]
-module = ["lsst.ctrl.*", "testcontainers.*", "botocore.*", "pandaclient.*", "asyncpg.*"]
+module = ["lsst.ctrl.*", "testcontainers.*", "botocore.*", "pandaclient.*", "asyncpg.*", "apscheduler.*"]
 ignore_missing_imports = true
 disable_error_code = ["import-untyped"]
 

src/lsst/cmservice/common/daemon_v2.py

@@ -4,6 +4,7 @@
 from collections.abc import Awaitable, Mapping
 from dataclasses import dataclass, field
 from datetime import datetime
+from types import TracebackType
 from typing import TYPE_CHECKING, Self, cast
 from uuid import UUID, uuid5
 
@@ -44,7 +45,9 @@ async def __aenter__(self) -> Self:
         self.session = db_session_dependency.sessionmaker()
         return self
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+    async def __aexit__(
+        self, exc_type: type[BaseException], exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
         await self.session.close()
         return None
 
@@ -232,7 +235,7 @@ async def consider_nodes(context: DaemonContext) -> None:
             await session.commit()
 
 
-async def do_schedule_stuff(schedule_id: UUID) -> None:
+async def daemon_check_schedule(schedule_id: UUID) -> None:
     ...
     breakpoint()
     # with daemon context...
@@ -259,7 +262,7 @@ async def consider_schedules(context: DaemonContext) -> None:
             context.app.state.scheduler.scheduler.reschedule_job(job_id, job_trigger)
         else:
             context.app.state.scheduler.scheduler.add_job(
-                do_schedule_stuff,
+                daemon_check_schedule,
                 job_trigger,
                 id=job_id,
                 args=[schedule.id],

src/lsst/cmservice/common/templates.py

@@ -3,7 +3,7 @@
 """
 
 from collections import deque
-from collections.abc import MutableMapping, MutableSequence
+from collections.abc import Mapping, MutableMapping, MutableSequence
 from datetime import datetime, timedelta
 from typing import Any, Literal
 
@@ -26,7 +26,7 @@ def as_obs_day(value: datetime) -> str:
     return f"{value:%Y%m%d}"
 
 
-def as_exposure(value: datetime, exposure=0) -> str:
+def as_exposure(value: datetime, exposure: int = 0) -> str:
     return f"{value:%Y%m%d}{exposure:05d}"
 
 
@@ -36,12 +36,12 @@ def compile_user_expressions(expressions: MutableMapping) -> dict[str, Any]:
     expressed value is cast as a string before being returned as a mapping
     of expression name to expression result.
     """
-    whitelist_modules = {
+    whitelist_modules: Mapping[str, type] = {
         "datetime": datetime,
         "timedelta": timedelta,
     }
     sandbox = ImmutableSandboxedEnvironment()
-    sandbox.globals = whitelist_modules
+    sandbox.globals.update(whitelist_modules)
 
     # TODO exception handling here, should the expression blow up
     # Compile and evaluate the user expression in the sandbox environment
@@ -60,7 +60,7 @@ async def build_sandbox_and_render_templates(expressions: dict, templates: list[
     """
     # . expressions = schedule.expressions
     # . templates = schedule.templates
-    rendered_templates = deque(maxlen=len(templates))
+    rendered_templates: deque[str] = deque(maxlen=len(templates))
     compiled_expressions = compile_user_expressions(expressions)
 
     sandbox = ImmutableSandboxedEnvironment(

src/lsst/cmservice/main.py

@@ -62,7 +62,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator:
 
 # Add Exception Handlers
 @app.exception_handler(NoResultFound)
-async def notfound_error_handler(request: Request, exc: NoResultFound):
+async def notfound_error_handler(request: Request, exc: NoResultFound) -> None:
     """Raise a 404 when the `NoResultFound` exception is raised.
 
     The NoResultFound exception may be raised in a route that uses the
@@ -72,7 +72,7 @@ async def notfound_error_handler(request: Request, exc: NoResultFound):
 
 
 @app.exception_handler(IntegrityError)
-async def duplicate_error_handler(request: Request, exc: IntegrityError):
+async def duplicate_error_handler(request: Request, exc: IntegrityError) -> None:
     """Raise a 409 when the `IntegrityError` exception is raised."""
     raise HTTPException(status_code=status.HTTP_409_CONFLICT)
 

src/lsst/cmservice/routers/v2/schedules.py

@@ -218,7 +218,7 @@ async def update_schedule_resource(
         # TODO consider a custom validators? Might be overkill for this
         # The cron string is just a string when it comes back out of the db
         schedule_cron = CronStr(schedule.cron)
-        schedule.next_run_at = schedule_cron.next_run  # type: ignore[attr-assigned]
+        schedule.next_run_at = schedule_cron.next_run  # type: ignore
 
     await session.commit()
 

tests/v2/conftest.py

@@ -4,6 +4,7 @@
 import os
 from collections.abc import AsyncGenerator, Callable, Generator
 from typing import TYPE_CHECKING
+from unittest.mock import Mock
 from uuid import NAMESPACE_DNS, uuid4
 
 import pytest
@@ -12,8 +13,10 @@
 from sqlalchemy import insert
 from sqlalchemy.pool import NullPool
 from sqlalchemy.schema import CreateSchema, DropSchema
+from sqlmodel.ext.asyncio.session import AsyncSession
 from testcontainers.postgres import PostgresContainer
 
+from lsst.cmservice.common.daemon_v2 import DaemonContext
 from lsst.cmservice.common.flags import Features
 from lsst.cmservice.config import config
 from lsst.cmservice.db.session import DatabaseManager, db_session_dependency
@@ -171,6 +174,17 @@ async def async_client_fixture(
     app.dependency_overrides.clear()
 
 
+@pytest_asyncio.fixture(name="daemon_context", scope="module", loop_scope="module")
+async def daemon_context(session: AnyAsyncSession) -> AsyncGenerator[DaemonContext]:
+    if TYPE_CHECKING:
+        assert isinstance(session, AsyncSession)
+    dc = DaemonContext(
+        app=Mock(),
+    )
+    dc.session = session
+    yield dc
+
+
 @pytest_asyncio.fixture(scope="function", loop_scope="module")
 async def test_campaign(aclient: AsyncClient, manifest_fixtures: None) -> AsyncGenerator[str]:
     """Fixture managing a test campaign with three (additional) nodes, which

tests/v2/test_daemon.py

@@ -6,9 +6,8 @@
 
 import pytest
 from sqlmodel import select
-from sqlmodel.ext.asyncio.session import AsyncSession
 
-from lsst.cmservice.common.daemon_v2 import consider_campaigns, consider_nodes
+from lsst.cmservice.common.daemon_v2 import DaemonContext, consider_campaigns, consider_nodes
 from lsst.cmservice.common.launchers import LauncherCheckResponse
 from lsst.cmservice.models.db.campaigns import Campaign, Node, Task
 from lsst.cmservice.models.enums import StatusEnum
@@ -18,7 +17,7 @@
 
 
 async def test_daemon_campaign(
-    caplog: pytest.LogCaptureFixture, test_campaign: str, session: AsyncSession
+    caplog: pytest.LogCaptureFixture, test_campaign: str, daemon_context: DaemonContext
 ) -> None:
     """Tests the handling of campaigns during daemon iteration, which is
     primarily done by checking side effects. This test assesses a test campaign
@@ -27,11 +26,12 @@ async def test_daemon_campaign(
     continue to traverse the campaign graph and visit more nodes until the
     END node is reached.
     """
+    session = daemon_context.session
 
     # At first, the test_campaign in a waiting state is not subject to daemon
     # consideration
     caplog.clear()
-    await consider_campaigns(session)
+    await consider_campaigns(daemon_context)
 
     # extract the test campaign id from the fixture url
     campaign_id = urlparse(test_campaign).path.split("/")[-2:][0]
@@ -47,7 +47,7 @@ async def test_daemon_campaign(
 
     # now the daemon should consider the running campaign
     caplog.clear()
-    await consider_campaigns(session)
+    await consider_campaigns(daemon_context)
     found_log_messages = 0
     for r in caplog.records:
         if any(["considering campaign" in r.message, "considering node" in r.message]):
@@ -66,7 +66,7 @@ async def test_daemon_campaign(
     await session.commit()
 
     caplog.clear()
-    await consider_campaigns(session)
+    await consider_campaigns(daemon_context)
     tasks = (await session.exec(select(Task))).all()
     # One additional task should be in the table now
     assert len(tasks) == 2
@@ -76,7 +76,7 @@ async def test_daemon_campaign(
 
     # The next assessment should produce two nodes to be handled in parallel
     caplog.clear()
-    await consider_campaigns(session)
+    await consider_campaigns(daemon_context)
     tasks = (await session.exec(select(Task))).all()
     # Two additional tasks should be in the table now
     assert len(tasks) == 4
@@ -89,7 +89,7 @@ async def test_daemon_campaign(
 
     # The next assessment should produce the END node
     caplog.clear()
-    await consider_campaigns(session)
+    await consider_campaigns(daemon_context)
     tasks = (await session.exec(select(Task))).all()
     # One additional task should be in the table now
     assert len(tasks) == 5
@@ -114,8 +114,9 @@ async def test_daemon_node(
     mock_launch: Mock,
     caplog: pytest.LogCaptureFixture,
     test_campaign: str,
-    session: AsyncSession,
+    daemon_context: DaemonContext,
 ) -> None:
+    session = daemon_context.session
     # set the campaign to running (without involving a campaign machine)
     campaign_id = urlparse(url=test_campaign).path.split("/")[-2:][0]
     campaign = await session.get_one(Campaign, campaign_id)
@@ -124,8 +125,8 @@ async def test_daemon_node(
 
     # after the equivalent of a single iteration, the test campaign's START
     # Node will be on the task list with a transition from waiting->ready.
-    await consider_campaigns(session)
-    await consider_nodes(session)
+    await consider_campaigns(daemon_context)
+    await consider_nodes(daemon_context)
 
     # As we continue to iterate the daemon over the campaign's 5 nodes
     # (including its START and END), each node in the graph is evolved.
@@ -139,8 +140,8 @@ async def test_daemon_node(
     i = 12
     while end_node.status is not StatusEnum.accepted:
         i -= 1
-        await consider_campaigns(session)
-        await consider_nodes(session)
+        await consider_campaigns(daemon_context)
+        await consider_nodes(daemon_context)
         # the end node is expunged from the session as a side effect when the
         # graph is built
         session.add(end_node)

tests/v2/test_rpc.py

@@ -7,9 +7,8 @@
 import pytest
 from httpx import AsyncClient, Response
 from sqlmodel import select
-from sqlmodel.ext.asyncio.session import AsyncSession
 
-from lsst.cmservice.common.daemon_v2 import consider_nodes
+from lsst.cmservice.common.daemon_v2 import DaemonContext, consider_nodes
 from lsst.cmservice.common.launchers import LauncherCheckResponse
 from lsst.cmservice.models.db.campaigns import Task
 
@@ -28,11 +27,12 @@ async def test_rpc_process_node(
     caplog: pytest.LogCaptureFixture,
     test_campaign: str,
     aclient: AsyncClient,
-    session: AsyncSession,
+    daemon_context: DaemonContext,
 ) -> None:
     """Tests the manual step-through of a campaign node using the "process" RPC
     API.
     """
+    session = daemon_context.session
     campaign_id = urlparse(url=test_campaign).path.split("/")[-2:][0]
     node_id = uuid5(UUID(campaign_id), "START.1")
 
@@ -51,7 +51,7 @@ async def process_once() -> Response:
 
     # now the daemon should consider the prepared campaign
     caplog.clear()
-    await consider_nodes(session)
+    await consider_nodes(daemon_context)
     found_log_messages = 0
     for r in caplog.records:
         if any(["evolving node" in r.message]):
@@ -70,7 +70,7 @@ async def process_once() -> Response:
     # Use the RPC API to process the node into a "running" state
     x = await process_once()
     assert x.is_success
-    await consider_nodes(session)
+    await consider_nodes(daemon_context)
 
     x = await aclient.get(f"/v2/nodes/{node_id}")
     assert x.is_success
@@ -79,7 +79,7 @@ async def process_once() -> Response:
     # Use the RPC API to process the node into a "accepted" state
     x = await process_once()
     assert x.is_success
-    await consider_nodes(session)
+    await consider_nodes(daemon_context)
 
     x = await aclient.get(f"/v2/nodes/{node_id}")
     assert x.is_success
@@ -122,7 +122,7 @@ async def test_rpc_with_breakpoint_node(
     mock_launch: Mock,
     test_campaign: str,
     aclient: AsyncClient,
-    session: AsyncSession,
+    daemon_context: DaemonContext,
 ) -> None:
     """Tests the addition of a Breakpoint node in a campaign, and that process-
     ing the campaign does not proceed past the breakpoint until it is applied
@@ -167,15 +167,15 @@ async def process_once() -> Response:
     for _ in range(5):
         x = await process_once()
         assert x.is_success
-        await consider_nodes(session)
+        await consider_nodes(daemon_context)
 
     # confirm that the breakpoint node does not advance from "running"
     r = await aclient.get(breakpoint_url)
     assert r.json()["status"] == "running"
 
     x = await process_once()
     assert x.is_success
-    await consider_nodes(session)
+    await consider_nodes(daemon_context)
 
     r = await aclient.get(breakpoint_url)
     assert r.json()["status"] == "running"
@@ -195,4 +195,4 @@ async def process_once() -> Response:
     # confirm that the rpc proceeds to the next node
     x = await process_once()
     assert x.is_success
-    await consider_nodes(session)
+    await consider_nodes(daemon_context)