diff --git a/api/docs/adr/000-vertical-slice-architecture.md b/api/docs/adr/000-vertical-slice-architecture.md index 5f6e7547..42b8a520 100644 --- a/api/docs/adr/000-vertical-slice-architecture.md +++ b/api/docs/adr/000-vertical-slice-architecture.md @@ -79,7 +79,7 @@ damnit_api/ │ ├── runs/ # the core domain: runs, variables, previews │ ├── models.py # plain dataclasses -│ ├── repository.py +│ ├── repository.py # `DamnitRepository` interface + backends; see ADR-005 │ ├── serialization.py, preview.py │ └── gql.py # Strawberry types + Query/Subscription contributions │ diff --git a/api/docs/adr/001-error-classes.md b/api/docs/adr/001-error-classes.md index 128326f6..98e629e5 100644 --- a/api/docs/adr/001-error-classes.md +++ b/api/docs/adr/001-error-classes.md @@ -81,4 +81,4 @@ Subclasses are added per _caller-distinguishable situation_, not per call site. - Permission denials use the Strawberry permission mechanism (message on the permission class); `ForbiddenError` is the same concept for non-GraphQL paths. 3. **Boundaries translate exceptions.** Infrastructure exceptions do not cross layer boundaries raw: - MyMdC client errors become `UpstreamServiceError` - - Locator/repository/filesystem failures become `ProposalNotFoundError`/`DataUnavailableError`, logged with context at the point of translation (log-then-raise). + - Locator/repository/filesystem failures become `ProposalNotFoundError`/`DataUnavailableError`, logged with context at the point of translation (log-then-raise). See [ADR-005](005-repository-pattern.md) for the repository boundary. diff --git a/api/docs/adr/005-repository-pattern.md b/api/docs/adr/005-repository-pattern.md new file mode 100644 index 00000000..dc67c3b5 --- /dev/null +++ b/api/docs/adr/005-repository-pattern.md @@ -0,0 +1,36 @@ +--- +date: 2026-07-07 +--- + +# ADR-005 - Repository pattern for DAMNIT run data + +## Context and Problem Statement + +Run and variable data lives in per-proposal SQLite files (`runs.sqlite`) produced by DAMNIT itself. The schema is external: the API reads it but does not own it, must reflect it at runtime, and there is one database per proposal on a network filesystem. + +Reading that data directly with SQLAlchemy from the GraphQL resolvers welds the transport layer to the DAMNIT schema, makes resolver logic untestable without real SQLite fixtures, and scatters session handling and caching across call sites. The data-access layer needs a seam: a narrow interface the resolvers depend on, with the SQL, reflection, and caching behind it. + +## Considered Options + +- Direct SQLAlchemy access from the resolvers (the status quo). +- A `DamnitRepository` interface the resolvers depend on, with SQL, reflection, and caching behind it. + +## Decision Outcome + +Chosen option: the repository interface, because it decouples resolvers from SQLAlchemy and the DAMNIT schema, lets them be tested against a lightweight backend, and gives caching and session handling a single home. + +### Consequences + +- Good: resolvers depend only on the interface and plain domain models, so they are testable against the CSV backend without SQLite fixtures. +- Good: a new data source (a DAMNIT HTTP API, Parquet exports) is a new `runs//` package implementing the same ABC. +- Bad: the registry accumulates one repository, and one engine, per accessed proposal for the process lifetime; eviction is available but not automatic (acceptable at current scale). + +## Details + +### The rules + +1. **`DamnitRepository`** (`runs/repository.py`, ABC) is the only way application code reads DAMNIT run data. One instance per proposal. Contract: `get_runs`, `get_latest_runs`, `get_metadata`, `get_extracted_data` (must not block the event loop, see [ADR-004](004-proposal-path-locator.md)), and `invalidate_metadata_cache`. +2. **Return types are plain domain dataclasses** (`runs/models.py`: `RunRecord`, `VariableValue`, `MetadataSnapshot`, `VariableInfo`, `TagInfo`, `KnownVariable`) - no framework or SQLAlchemy types. Serialisation to transport shapes happens outside the repository. +3. **A `DamnitRepositoryRegistry`** (held on `AppState`, see [ADR-002](002-no-global-mutable-state.md)) lazily creates and caches one repository per `ProposalNumber` via an injected factory, and supports eviction (`pop`, `clear`). +4. **Implementations.** `runs/sqlite/` is production: async SQLAlchemy over `runs.sqlite`, read-only sessions (`PRAGMA query_only = ON`), `NullPool`, connection timeouts, and a per-repo table-reflection cache plus a metadata TTL cache (the approved cache pattern of [ADR-002](002-no-global-mutable-state.md)). `runs/csv/` is dev/test: it reads `runs.csv` / `run_variables.csv` / `variables.csv` and exercises resolver and subscription logic without SQLite fixtures. +5. Caching lives inside the implementations; callers never wrap repository calls in their own module-level caches. diff --git a/api/docs/architecture.md b/api/docs/architecture.md index 32d44162..c290efba 100644 --- a/api/docs/architecture.md +++ b/api/docs/architecture.md @@ -18,7 +18,7 @@ For more information, see [ADR-000](adr/000-vertical-slice-architecture.md). | Package | Capability | What belongs there | Status | Today | | --------------------------------- | --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------- | ------- | ------------------------------------------------------------------------- | -| `runs/` | Run/variable data - the core domain | Domain models, repository interface + implementations, serialisation, preview extraction, its GraphQL types and resolvers | Planned | `db.py` + `data.py`; resolvers in `graphql/queries.py`/`subscriptions.py` | +| `runs/` | Run/variable data - the core domain | Domain models, repository interface + implementations (see [ADR-005](adr/005-repository-pattern.md)), serialisation, preview extraction, its GraphQL types and resolvers | Partial | `runs/` (repository, models, sqlite + csv backends); resolvers still in `graphql/queries.py`/`subscriptions.py` | | `proposals/` | Proposal metadata and lookup | Proposal models, MyMdC-backed metadata services, path locator (see [ADR-004](adr/004-proposal-path-locator.md)) | Planned | `metadata/` | | `auth/` | Authentication and authorisation | OAuth flow, sessions, token store, `User`, permission classes, the membership policy | Partial | Policy still in `metadata/services.py` | | `contextfile/` | Context-file viewing | File reading, watching, its routes | Done | As-is | diff --git a/api/src/damnit_api/auth/gql.py b/api/src/damnit_api/auth/gql.py index 7faa1e75..97ec2cc8 100644 --- a/api/src/damnit_api/auth/gql.py +++ b/api/src/damnit_api/auth/gql.py @@ -35,7 +35,7 @@ async def proposals( if settings.is_local: from ..metadata.services import _local_proposal_meta, _local_proposal_number - proposal_number = await _local_proposal_number(info.context.damnit_registry) + proposal_number = await _local_proposal_number(info.context.repositories) if proposal_number is None: return [] return [ProposalMeta.from_pydantic(_local_proposal_meta(proposal_number))] diff --git a/api/src/damnit_api/auth/routers.py b/api/src/damnit_api/auth/routers.py index 81ee43d4..dd616f4c 100644 --- a/api/src/damnit_api/auth/routers.py +++ b/api/src/damnit_api/auth/routers.py @@ -148,7 +148,7 @@ async def noauth_userinfo(request: Request): proposals = {} proposal_number = await _local_proposal_number( - get_app_state(request).damnit_registry + get_app_state(request).repositories ) if proposal_number: proposals = {LOCAL_CYCLE: [proposal_number]} diff --git a/api/src/damnit_api/graphql/metadata.py b/api/src/damnit_api/graphql/metadata.py deleted file mode 100644 index d83278db..00000000 --- a/api/src/damnit_api/graphql/metadata.py +++ /dev/null @@ -1,47 +0,0 @@ -import asyncio - -from async_lru import alru_cache - -from ..runs import sqlite as db -from ..runs.types import DamnitRun -from ..utils import create_map - - -@alru_cache(ttl=10) -async def fetch_metadata(registry: db.DamnitDBRegistry, proposal=db.DEFAULT_PROPOSAL): - """Fetch the per-proposal metadata snapshot from SQLite. - - Returns a dict with `runs`, `variables`, `tags`, and `timestamp`. Result - is TTL-cached; the `latest_data` subscription invalidates this cache when - it observes new data so subsequent reads stay fresh. - """ - tags, variables, variable_tags, runs, max_timestamp = await asyncio.gather( - db.async_all_tags(registry, proposal), - db.async_variables(registry, proposal), - db.async_variable_tags(registry, proposal), - db.async_column(registry, proposal, table="run_info", name="run"), - db.async_max(registry, proposal, table="run_variables", column="timestamp"), - ) - - for name, var in variables.items(): - var["tags"] = [tags[tag]["name"] for tag in variable_tags.get(name, [])] - - variables = {**DamnitRun.known_variables(), **variables} - - for name, var_tags in variable_tags.items(): - for tag in var_tags: - tags[tag].setdefault("variables", []).append(name) - - untagged = { - "id": 0, - "name": "(Untagged)", - "variables": [name for name, var in variables.items() if not var.get("tags")], - } - tags = create_map([untagged, *tags.values()], key="name") - - return { - "runs": sorted(runs or []), - "variables": variables, - "tags": tags, - "timestamp": max_timestamp or 0, - } diff --git a/api/src/damnit_api/graphql/queries.py b/api/src/damnit_api/graphql/queries.py index 8692fda0..67ff590b 100644 --- a/api/src/damnit_api/graphql/queries.py +++ b/api/src/damnit_api/graphql/queries.py @@ -1,5 +1,6 @@ +import dataclasses + import strawberry -from sqlalchemy import and_, func, select from strawberry.scalars import JSON from strawberry.types import Info from strawberry.types.nodes import SelectedField @@ -7,19 +8,12 @@ from .. import get_logger from ..auth.permissions import PROPOSAL_PERMISSIONS from ..metadata.services import _get_proposal_meta, _update_proposal_meta -from ..runs.preview import get_preview_data -from ..runs.sqlite import async_table, get_session -from ..runs.types import KNOWN_DTYPES, DamnitRun +from ..runs.types import DamnitRun from ..shared.models import ProposalNumber -from .metadata import fetch_metadata -from .utils import DatabaseInput, fetch_info +from .utils import DatabaseInput logger = get_logger() -# Names that only `fetch_info` provides; `proposal` and `run` already come -# from `fetch_variables`. -RUN_INFO_NAMES = frozenset(KNOWN_DTYPES) - {"proposal", "run"} - async def _ensure_damnit_path(info: Info, proposal: ProposalNumber) -> None: """Ensure the proposal has a DAMNIT path, refreshing from MyMdC if needed. @@ -45,92 +39,6 @@ async def _ensure_damnit_path(info: Info, proposal: ProposalNumber) -> None: raise ValueError(msg) -def group_by_run(record): - grouped = {} - - for entry in record: - key = (entry["proposal"], entry["run"]) - if key not in grouped: - grouped[key] = { - "proposal": {"value": entry["proposal"]}, - "run": {"value": entry["run"]}, - } - # Outer-join placeholder for a run with no matching variables. - if entry["name"] is None: - continue - grouped[key][entry["name"]] = { - "value": entry["value"], - "summary_type": entry["summary_type"], - "attributes": entry["attributes"], - } - - return list(grouped.values()) - - -async def fetch_variables(registry, proposal, *, limit, offset, names=None): - table = await async_table(registry, proposal, name="run_variables") - if table is None: - return [] - - runs_subquery = ( - select(table.c.proposal, table.c.run) - .distinct() - .order_by(table.c.run) - .limit(limit) - .offset(offset) - .subquery() - ) - - latest_timestamp_subquery = select( - table.c.proposal, - table.c.run, - table.c.name, - func.max(table.c.timestamp).label("latest_timestamp"), - ).where(table.c.run.in_(select(runs_subquery.c.run))) - if names is not None: - latest_timestamp_subquery = latest_timestamp_subquery.where( - table.c.name.in_(names) - ) - latest_timestamp_subquery = latest_timestamp_subquery.group_by( - table.c.run, table.c.name - ).subquery() - - # Outer-join from `runs_subquery` so a `names` filter that excludes every - # row of a paginated run still keeps the run in the page. - query = ( - select( - runs_subquery.c.proposal, - runs_subquery.c.run, - latest_timestamp_subquery.c.name, - table.c.value, - table.c.summary_type, - table.c.attributes, - ) - .select_from(runs_subquery) - .outerjoin( - latest_timestamp_subquery, - runs_subquery.c.run == latest_timestamp_subquery.c.run, - ) - .outerjoin( - table, - and_( - table.c.proposal == latest_timestamp_subquery.c.proposal, - table.c.run == latest_timestamp_subquery.c.run, - table.c.name == latest_timestamp_subquery.c.name, - table.c.timestamp == latest_timestamp_subquery.c.latest_timestamp, - ), - ) - .order_by(runs_subquery.c.run) - ) - - async with get_session(registry, proposal) as session: - result = await session.execute(query) - if not result: - raise ValueError # TODO: Better error handling - - return group_by_run(result.mappings().all()) # type: ignore[assignment] - - def _selected_variable_names(info: Info) -> list[str] | None: """Union the `names` arguments across every `variables` sub-selection. Returns None if any selection omits the argument (forces a full fetch). @@ -149,12 +57,6 @@ def _selected_variable_names(info: Info) -> list[str] | None: return sorted(union) if found else None -def _wants_run_info(names: list[str] | None) -> bool: - if names is None: - return True - return bool(set(names) & RUN_INFO_NAMES) - - @strawberry.type class Query: """ @@ -179,45 +81,37 @@ async def runs( await _ensure_damnit_path(info, proposal) names = _selected_variable_names(info) - variables = await fetch_variables( - info.context.damnit_registry, - proposal, - limit=per_page, - offset=(page - 1) * per_page, - names=names, - ) - - if not len(variables): - return [] - - if _wants_run_info(names): - info_rows = await fetch_info( - info.context.damnit_registry, - proposal, - runs=[v["run"]["value"] for v in variables], + try: + repo = info.context.repositories.get(proposal) + records = await repo.get_runs( + limit=per_page, + offset=(page - 1) * per_page, + variable_names=names, ) - else: - info_rows = [{} for _ in variables] - - return [ - DamnitRun.from_db({**v, **i}) - for v, i in zip(variables, info_rows, strict=True) - ] + except Exception: + logger.exception("Failed to get runs", proposal=proposal) + raise + return [DamnitRun.from_record(r) for r in records] @strawberry.field(permission_classes=PROPOSAL_PERMISSIONS) async def metadata( self, info: Info, database: DatabaseInput, - ) -> JSON: # FIX: # pyright: ignore[reportInvalidTypeForm] + ) -> JSON: proposal = database.proposal await _ensure_damnit_path(info, proposal) - snapshot = await fetch_metadata(info.context.damnit_registry, proposal) - return { - **snapshot, - "timestamp": snapshot["timestamp"] * 1000, # ms for JS - } # pyright: ignore[reportReturnType] + try: + repo = info.context.repositories.get(proposal) + snapshot = await repo.get_metadata() + except Exception: + logger.exception("Failed to get metadata", proposal=proposal) + raise + result = dataclasses.asdict(snapshot) + result["runs"] = list(result["runs"]) + result["timestamp"] = snapshot.timestamp * 1000 # ms for JS + return result @strawberry.field(permission_classes=PROPOSAL_PERMISSIONS) async def extracted_data( @@ -226,12 +120,12 @@ async def extracted_data( database: DatabaseInput, run: int, variable: str, - ) -> JSON: # FIX: # pyright: ignore[reportInvalidTypeForm] - await _ensure_damnit_path(info, database.proposal) - # TODO: Convert to Strawberry type - # and make it analogous to DamitVariable; e.g. `data` - return get_preview_data( # FIX: # pyright: ignore[reportReturnType] - proposal_number=database.proposal, - run=run, - variable=variable, - ) + ) -> JSON: + proposal = database.proposal + await _ensure_damnit_path(info, proposal) + try: + repo = info.context.repositories.get(proposal) + return await repo.get_extracted_data(run=run, variable=variable) + except Exception: + logger.exception("Failed to get extracted data", proposal=proposal) + raise diff --git a/api/src/damnit_api/graphql/subscriptions.py b/api/src/damnit_api/graphql/subscriptions.py index d16e4555..1b5f79c8 100644 --- a/api/src/damnit_api/graphql/subscriptions.py +++ b/api/src/damnit_api/graphql/subscriptions.py @@ -1,17 +1,21 @@ import asyncio +import dataclasses from collections.abc import AsyncGenerator +from typing import Any import strawberry from async_lru import alru_cache from strawberry.scalars import JSON from strawberry.types import Info +from .. import get_logger from ..auth.permissions import PROPOSAL_PERMISSIONS -from ..runs.sqlite import async_latest_rows, async_max, async_table +from ..runs.repository import DamnitRepository from ..runs.types import DamnitRun, Timestamp -from ..utils import create_map -from .metadata import fetch_metadata -from .utils import DatabaseInput, LatestData, fetch_info +from ..shared.models import ProposalNumber +from .utils import DatabaseInput + +logger = get_logger() POLLING_INTERVAL = 1 # seconds @@ -22,94 +26,80 @@ class SubscriptionCursors: for alru_cache.""" def __init__(self) -> None: - self._data: dict[str, float] = {} + self._data: dict[ProposalNumber, float] = {} + + def __contains__(self, proposal_number: ProposalNumber) -> bool: + return proposal_number in self._data - def __contains__(self, proposal: str) -> bool: - return proposal in self._data + def __getitem__(self, proposal_number: ProposalNumber) -> float: + return self._data[proposal_number] - def __getitem__(self, proposal: str) -> float: - return self._data[proposal] + def __setitem__(self, proposal_number: ProposalNumber, value: float) -> None: + self._data[proposal_number] = value - def __setitem__(self, proposal: str, value: float) -> None: - self._data[proposal] = value + def clear(self) -> None: + self._data.clear() # Per-client cursor is deliberately omitted from the cache key so that # concurrent subscribers coalesce into a single DB read per tick. @alru_cache(maxsize=32, ttl=POLLING_INTERVAL) -async def poll_proposal(registry, proposal, cursors: SubscriptionCursors): - table = await async_table(registry, proposal, name="run_variables") - if table is None: - return None - - if proposal not in cursors: - max_timestamp = await async_max( - registry, proposal, table="run_variables", column="timestamp" - ) - cursors[proposal] = max_timestamp or 0 - - rows = await async_latest_rows( - registry, - proposal, - table=table, - by="timestamp", - start_at=cursors[proposal], - ) - if not rows: +async def poll_proposal( + proposal_number: ProposalNumber, + cursors: SubscriptionCursors, + repo: DamnitRepository, +) -> dict[str, Any] | None: + # Initialise cursor from the current max timestamp on first visit. + if proposal_number not in cursors: + try: + metadata = await repo.get_metadata() + cursors[proposal_number] = metadata.timestamp + except Exception: + return None + + start_at = cursors[proposal_number] + records = await repo.get_latest_runs(start_at=start_at) + if not records: return None - latest_data = LatestData.from_list(rows) - - latest_runs = await fetch_info( - registry, proposal, runs=list(latest_data.runs.keys()) - ) - latest_runs = create_map(latest_runs, key="run") - - fetch_metadata.cache_invalidate(registry, proposal) - metadata = await fetch_metadata(registry, proposal) - - runs = {} - run_timestamps = {} - for run, variables in latest_data.runs.items(): - run_values = { - name: { - "value": data.value, - "summary_type": data.summary_type, - "attributes": data.attributes, - } - for name, data in variables.items() - } - run_values.setdefault("run", {"value": run}) - - if run_info := latest_runs.get(run): - run_values.update(run_info) - - runs[run] = DamnitRun.resolve(run_values) - run_timestamps[run] = max(data.timestamp for data in variables.values()) + runs: dict[int, Any] = {} + run_timestamps: dict[int, float] = {} + for record in records: + runs[record.run] = DamnitRun.resolve_record(record) + if record.variables: + run_timestamps[record.run] = max( + vv.timestamp for vv in record.variables.values() + ) if not runs: return None - if latest_data.timestamp is None: - msg = "Latest data has no timestamp." - raise ValueError(msg) + latest_ts = max(run_timestamps.values(), default=start_at) - cursors[proposal] = latest_data.timestamp + repo.invalidate_metadata_cache() + try: + metadata = await repo.get_metadata() + except Exception: + return None + cursors[proposal_number] = latest_ts + meta_dict = dataclasses.asdict(metadata) - metadata = { - "runs": sorted(set(metadata["runs"]) | set(runs.keys())), - "variables": metadata["variables"], - "timestamp": latest_data.timestamp * 1000, # ms for JS - } return { "runs": runs, "run_timestamps": run_timestamps, - "max_timestamp": max(run_timestamps.values()), - "metadata": metadata, + "max_timestamp": latest_ts, + "metadata": { + "runs": sorted(set(metadata.runs) | set(runs.keys())), + "variables": meta_dict["variables"], + "timestamp": latest_ts * 1000, # ms for JS + }, } -def filter_for_client(snapshot, since): +def filter_for_client( + snapshot: dict[str, Any] | None, + since: float | None, +) -> dict[str, Any] | None: if snapshot is None or not since or snapshot["max_timestamp"] <= since: return None runs = { @@ -130,15 +120,23 @@ async def latest_data( info: Info, database: DatabaseInput, timestamp: Timestamp, - ) -> AsyncGenerator[JSON]: # FIX: # pyright: ignore[reportInvalidTypeForm] + ) -> AsyncGenerator[JSON]: + cursors: SubscriptionCursors = info.context.subscription_cursors while True: await asyncio.sleep(POLLING_INTERVAL) - snapshot = await poll_proposal( - info.context.damnit_registry, - proposal=database.proposal, - cursors=info.context.subscription_cursors, - ) + proposal = database.proposal + try: + repo = info.context.repositories.get(proposal) + snapshot = await poll_proposal( + proposal_number=proposal, + cursors=cursors, + repo=repo, + ) + except Exception: + logger.exception("Subscription poll failed", proposal=proposal) + raise + result = filter_for_client(snapshot, timestamp) if result is not None: - yield result # FIX: # pyright: ignore[reportReturnType] + yield result # ty: ignore[invalid-yield] diff --git a/api/src/damnit_api/graphql/utils.py b/api/src/damnit_api/graphql/utils.py index 71d6c99b..3d7a159a 100644 --- a/api/src/damnit_api/graphql/utils.py +++ b/api/src/damnit_api/graphql/utils.py @@ -1,11 +1,5 @@ -from collections import defaultdict -from dataclasses import dataclass -from typing import Any - import strawberry -from sqlalchemy import or_, select -from ..runs.sqlite import DamnitDBRegistry, async_table, get_session from ..shared.const import DEFAULT_PROPOSAL from ..shared.models import ProposalNumber @@ -16,67 +10,3 @@ class DatabaseInput: default=ProposalNumber(DEFAULT_PROPOSAL) ) path: str | None = strawberry.field(default=strawberry.UNSET) - - -@dataclass -class MetaData: - timestamp: float = 0 - - -@dataclass -class Data(MetaData): - value: Any = None - summary_type: str | None = None - attributes: str | None = None - - -class LatestData: - def __init__(self): - self.runs = defaultdict(lambda: defaultdict(Data)) - self.variables = defaultdict(MetaData) - - def add(self, data): - timestamp = data["timestamp"] - - # Bookkeep by runs - run = self.runs[data["run"]] - if run[data["name"]].timestamp < timestamp: - run[data["name"]] = Data( - value=data["value"], - summary_type=data.get("summary_type"), - attributes=data.get("attributes"), - timestamp=timestamp, - ) - - # Bookkeep by variables - variable = self.variables[data["name"]] - if variable.timestamp < timestamp: - variable.timestamp = timestamp - - @property - def timestamp(self): - timestamps = [data.timestamp for data in self.variables.values()] - return max(timestamps) if len(timestamps) else None - - @classmethod - def from_list(cls, sequence): - instance = cls() - for seq in sequence: - instance.add(seq) - - return instance - - -async def fetch_info(registry: DamnitDBRegistry, proposal, *, runs): - table = await async_table(registry, proposal, name="run_info") - if table is None: - return [] - conditions = [table.c.run == run for run in runs] - query = select(table).where(or_(*conditions)).order_by(table.c.run) - - async with get_session(registry, proposal) as session: - result = await session.execute(query) - if not result: - raise ValueError # TODO: Better error handling - - return result.mappings().all() diff --git a/api/src/damnit_api/main.py b/api/src/damnit_api/main.py index 9e1fa233..ffe97b8b 100644 --- a/api/src/damnit_api/main.py +++ b/api/src/damnit_api/main.py @@ -17,11 +17,11 @@ def create_app(): from .shared.settings import settings from .state import ( AppState, - create_damnit_registry, create_db_engine, create_db_sessionmaker, create_mymdc_client, create_oauth_client, + create_repositories, create_subscription_cursors, create_token_store, ) @@ -49,7 +49,7 @@ async def lifespan(app: FastAPI): mymdc_client=create_mymdc_client(settings), oauth_client=oauth_client, token_store=create_token_store(), - damnit_registry=create_damnit_registry(), + repositories=create_repositories(), subscription_cursors=create_subscription_cursors(), ) diff --git a/api/src/damnit_api/metadata/services.py b/api/src/damnit_api/metadata/services.py index f8798dd1..d76b45d2 100644 --- a/api/src/damnit_api/metadata/services.py +++ b/api/src/damnit_api/metadata/services.py @@ -19,7 +19,7 @@ from .._db.dependencies import DBSession from .._mymdc.clients import MyMdCClient from ..auth.dependencies import User - from ..runs.sqlite import DamnitDBRegistry + from ..runs.repository import DamnitRepositoryRegistry LOCAL_CYCLE = "197001" @@ -43,24 +43,29 @@ def _local_proposal_meta(proposal_number: ProposalNumber) -> ProposalMeta: ) -async def _local_proposal_number(registry: "DamnitDBRegistry") -> ProposalNumber | None: - from sqlalchemy import select +async def _local_proposal_number( + repositories: "DamnitRepositoryRegistry", +) -> ProposalNumber | None: + from ..runs.sqlite.repository import SQLiteDamnitRepository - from ..runs.sqlite import async_table, get_session - from ..shared.const import DEFAULT_PROPOSAL - - proposal = ProposalNumber(DEFAULT_PROPOSAL) - table = await async_table(registry, proposal, name="metameta") - if table is None: + # In local mode every session uses the proposal at settings.damnit_path + # regardless of proposal number, so any ProposalNumber works as the key. + repo = repositories.get(ProposalNumber(1)) + if not isinstance(repo, SQLiteDamnitRepository): + # get_proposal_number() reads the DAMNIT-sqlite-specific `metameta` + # table; it has no equivalent on other repository backends. return None - async with get_session(registry, proposal) as session: - result = await session.execute( - select(table.c.value).where(table.c.key == "proposal") + value = await repo.get_proposal_number() + if not value: + return None + try: + return ProposalNumber(value) + except ValueError: + await logger.aerror( + "Invalid proposal number in metameta table", raw_value=value ) - value = result.scalar() - - return ProposalNumber(int(value)) if value else None + return None async def _fetch_proposal_meta( diff --git a/api/src/damnit_api/runs/csv/__init__.py b/api/src/damnit_api/runs/csv/__init__.py new file mode 100644 index 00000000..db01d50b --- /dev/null +++ b/api/src/damnit_api/runs/csv/__init__.py @@ -0,0 +1,3 @@ +from .repository import CsvDamnitRepository + +__all__ = ["CsvDamnitRepository"] diff --git a/api/src/damnit_api/runs/csv/repository.py b/api/src/damnit_api/runs/csv/repository.py new file mode 100644 index 00000000..51ac537d --- /dev/null +++ b/api/src/damnit_api/runs/csv/repository.py @@ -0,0 +1,252 @@ +"""CSV-backed DamnitRepository for local development and testing.""" + +from __future__ import annotations + +import asyncio +import csv +from collections import defaultdict +from datetime import datetime +from pathlib import Path +from typing import Any + +from ... import get_logger +from ...shared.models import ProposalNumber +from ..models import ( + KNOWN_VARIABLES, + MetadataSnapshot, + RunRecord, + TagInfo, + VariableInfo, + VariableValue, +) +from ..repository import DamnitRepository + +logger = get_logger() + + +class CsvDamnitRepository(DamnitRepository): + """CSV-backed DamnitRepository for local development and testing. + + Reads from three CSV files in *csv_dir*: + + - ``runs.csv`` — one row per run: ``run,start_time,added_at`` + - ``run_variables.csv`` — one row per variable value: + ``run,name,value,summary_type,timestamp`` + - ``variables.csv`` (optional) — variable metadata: ``name,title,tags`` + where ``tags`` is a semicolon-separated list of tag names. + + ``get_extracted_data`` always returns ``None``; there is no binary preview + data in the CSV format. + """ + + def __init__(self, proposal_number: ProposalNumber, csv_dir: Path | str) -> None: + self._proposal = ProposalNumber(proposal_number) + self._csv_dir = Path(csv_dir) + + @property + def proposal(self) -> ProposalNumber: + return self._proposal + + def _path(self, filename: str) -> Path: + return self._csv_dir / filename + + def _load_run_info(self) -> dict[int, dict[str, float | None]]: + path = self._path("runs.csv") + if not path.exists(): + return {} + result: dict[int, dict[str, float | None]] = {} + with path.open(newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + try: + run = int(row["run"]) + result[run] = { + "start_time": float(row["start_time"]) + if row.get("start_time") + else None, + "added_at": float(row["added_at"]) + if row.get("added_at") + else None, + } + except (ValueError, KeyError): + logger.warning( + "Invalid row in runs.csv; skipping", + csv_dir=self._csv_dir, + row=dict(row), + ) + return result + + def _load_run_variables(self) -> dict[int, dict[str, VariableValue]]: + """Load run_variables.csv, keeping only the latest value per (run, name).""" + path = self._path("run_variables.csv") + if not path.exists(): + return {} + result: dict[int, dict[str, VariableValue]] = defaultdict(dict) + with path.open(newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + try: + run = int(row["run"]) + ts = float(row.get("timestamp") or 0.0) + except (ValueError, KeyError): + logger.warning( + "Invalid row in run_variables.csv; skipping", + csv_dir=self._csv_dir, + row=dict(row), + ) + continue + name = row["name"] + value: Any = row.get("value") or None + summary_type = row.get("summary_type") or None + existing = result[run].get(name) + if existing is None or ts > existing.timestamp: + result[run][name] = VariableValue( + value=value, + summary_type=summary_type, + timestamp=ts, + ) + return dict(result) + + def _load_latest_run_variables( + self, start_at: float + ) -> dict[int, dict[str, VariableValue]]: + """Load run_variables.csv, keeping only rows newer than start_at.""" + path = self._path("run_variables.csv") + if not path.exists(): + return {} + result: dict[int, dict[str, VariableValue]] = defaultdict(dict) + with path.open(newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + try: + ts = float(row.get("timestamp") or 0.0) + if ts <= start_at: + continue + run = int(row["run"]) + except (ValueError, KeyError): + logger.warning( + "Invalid row in run_variables.csv; skipping", + csv_dir=self._csv_dir, + row=dict(row), + ) + continue + name = row["name"] + existing = result[run].get(name) + if existing is None or ts > existing.timestamp: + result[run][name] = VariableValue( + value=row.get("value") or None, + summary_type=row.get("summary_type") or None, + timestamp=ts, + ) + return dict(result) + + def _load_variables_meta(self) -> dict[str, VariableInfo]: + """Load variables.csv; returns an empty dict if the file is absent.""" + path = self._path("variables.csv") + if not path.exists(): + return {} + result: dict[str, VariableInfo] = {} + with path.open(newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + name = row["name"] + title: str | None = row.get("title") or None + tags_raw = row.get("tags") or "" + tags = [t.strip() for t in tags_raw.split(";") if t.strip()] + result[name] = VariableInfo(name=name, title=title, tags=tags) + return result + + async def get_runs( + self, + *, + limit: int, + offset: int, + variable_names: list[str] | None = None, + ) -> list[RunRecord]: + loop = asyncio.get_running_loop() + run_vars = await loop.run_in_executor(None, self._load_run_variables) + run_info = await loop.run_in_executor(None, self._load_run_info) + + page_run_ids = sorted(run_vars.keys())[offset : offset + limit] + records = [] + for run in page_run_ids: + variables = run_vars[run] + if variable_names is not None: + variables = {k: v for k, v in variables.items() if k in variable_names} + info = run_info.get(run, {}) + records.append( + RunRecord( + proposal=self._proposal, + run=run, + start_time=info.get("start_time"), + added_at=info.get("added_at"), + variables=variables, + ) + ) + return records + + async def get_latest_runs( + self, + *, + start_at: float | None = None, + ) -> list[RunRecord]: + if start_at is None: + start_at = datetime.now().astimezone().timestamp() + + loop = asyncio.get_running_loop() + run_vars = await loop.run_in_executor( + None, self._load_latest_run_variables, start_at + ) + if not run_vars: + return [] + + run_info = await loop.run_in_executor(None, self._load_run_info) + return [ + RunRecord( + proposal=self._proposal, + run=run, + start_time=run_info.get(run, {}).get("start_time"), + added_at=run_info.get(run, {}).get("added_at"), + variables=run_vars[run], + ) + for run in sorted(run_vars.keys()) + ] + + async def get_metadata(self) -> MetadataSnapshot: + loop = asyncio.get_running_loop() + run_info = await loop.run_in_executor(None, self._load_run_info) + db_variables = await loop.run_in_executor(None, self._load_variables_meta) + + # Known variables first; db variables take precedence (they carry titles/tags) + variables: dict[str, VariableInfo] = { + v.name: VariableInfo(name=v.name, title=v.title) + for v in KNOWN_VARIABLES + } + variables.update(db_variables) + + # Build tags from variable metadata + tag_to_vars: dict[str, list[str]] = defaultdict(list) + for name, vi in db_variables.items(): + for tag in vi.tags: + tag_to_vars[tag].append(name) + + untagged = [n for n, v in variables.items() if not v.tags] + tags: dict[str, TagInfo] = { + "(Untagged)": TagInfo(id=0, name="(Untagged)", variables=untagged), + } + for i, (tag_name, var_names) in enumerate(tag_to_vars.items(), start=1): + tags[tag_name] = TagInfo(id=i, name=tag_name, variables=var_names) + + # Max timestamp across all run variables + run_vars = await loop.run_in_executor(None, self._load_run_variables) + max_ts = 0.0 + for vars_dict in run_vars.values(): + for vv in vars_dict.values(): + if vv.timestamp > max_ts: + max_ts = vv.timestamp + + return MetadataSnapshot( + runs=tuple(sorted(run_info.keys())), + variables=variables, + tags=tags, + timestamp=max_ts, + ) + + async def get_extracted_data(self, *, run: int, variable: str) -> Any: + return None diff --git a/api/src/damnit_api/runs/dependencies.py b/api/src/damnit_api/runs/dependencies.py new file mode 100644 index 00000000..4f24a86c --- /dev/null +++ b/api/src/damnit_api/runs/dependencies.py @@ -0,0 +1,17 @@ +"""FastAPI dependency helpers for the runs repository registry.""" + +from typing import Annotated + +from fastapi import Depends, Request + +from ..state import get_app_state +from .repository import DamnitRepositoryRegistry + + +def get_repositories(request: Request) -> DamnitRepositoryRegistry: + """Provide the per-proposal repository registry from the application state.""" + return get_app_state(request).repositories + + +Repositories = Annotated[DamnitRepositoryRegistry, Depends(get_repositories)] +"""Type alias for the DAMNIT repository registry dependency.""" diff --git a/api/src/damnit_api/runs/models.py b/api/src/damnit_api/runs/models.py new file mode 100644 index 00000000..ff5f0b3a --- /dev/null +++ b/api/src/damnit_api/runs/models.py @@ -0,0 +1,84 @@ +"""Domain models for runs. + +All are implemented as plain stdlib dataclasses, it is assumed that +any required validation has already been done. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any + +from ..shared.const import DamnitType + +if TYPE_CHECKING: + from ..shared.models import ProposalNumber + + +@dataclass(frozen=True) +class VariableValue: + """A single variable's latest summarised value for one run.""" + + value: Any + summary_type: str | None + timestamp: float + attributes: str | None = None + """Raw JSON `attributes` blob from `run_variables`; carries variable errors.""" + + +@dataclass(frozen=True) +class RunRecord: + """All variable values for a single run.""" + + proposal: ProposalNumber + run: int + start_time: float | None + added_at: float | None + variables: dict[str, VariableValue] = field(default_factory=dict) + + +@dataclass +class VariableInfo: + """Static metadata about one variable (from the `variables` table).""" + + name: str + title: str | None + tags: list[str] = field(default_factory=list) + + +@dataclass +class TagInfo: + """A tag and the variables it is applied to.""" + + id: int + name: str + variables: list[str] = field(default_factory=list) + + +@dataclass(frozen=True) +class KnownVariable: + """A hard-coded variable that every DAMNIT proposal exposes.""" + + name: str + title: str + dtype: DamnitType + + +KNOWN_VARIABLES: tuple[KnownVariable, ...] = ( + KnownVariable(name="proposal", title="Proposal", dtype=DamnitType.NUMBER), + KnownVariable(name="run", title="Run", dtype=DamnitType.NUMBER), + KnownVariable(name="start_time", title="Timestamp", dtype=DamnitType.TIMESTAMP), + KnownVariable(name="added_at", title="Added at", dtype=DamnitType.TIMESTAMP), +) + + +@dataclass(frozen=True) +class MetadataSnapshot: + """Full proposal-level metadata: runs list, variable catalogue, tags.""" + + runs: tuple[int, ...] + variables: dict[str, VariableInfo] + """All variables, keyed by variable name""" + tags: dict[str, TagInfo] + """All tags, keyed by tag name""" + timestamp: float diff --git a/api/src/damnit_api/runs/repository.py b/api/src/damnit_api/runs/repository.py new file mode 100644 index 00000000..9b536ccf --- /dev/null +++ b/api/src/damnit_api/runs/repository.py @@ -0,0 +1,130 @@ +"""Repository interface for DAMNIT run/variable data.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Any + +from .. import get_logger + +logger = get_logger() + +if TYPE_CHECKING: + from collections.abc import Callable + + from ..shared.models import ProposalNumber + from .models import MetadataSnapshot, RunRecord + + +class DamnitRepository(ABC): + """Per-proposal data-access interface. + + One instance per proposal. Implementations own their own session management + and caching (e.g. metadata TTL). + """ + + @property + @abstractmethod + def proposal(self) -> ProposalNumber: ... + + @abstractmethod + async def get_runs( + self, + *, + limit: int, + offset: int, + variable_names: list[str] | None = None, + ) -> list[RunRecord]: + """Return paginated runs with their latest variable values. + + Args: + limit: Maximum number of runs to return. + offset: Number of runs to skip (for pagination). + variable_names: If given, only include these variables in each + run's `variables` dict. `None` means all variables. + """ + + @abstractmethod + async def get_latest_runs( + self, + *, + start_at: float | None = None, + ) -> list[RunRecord]: + """Return runs with timestamps newer than `start_at`. + + Used by the `latest_data` subscription to stream incremental updates. `start_at` + defaults to *now* when `None`. + """ + + @abstractmethod + async def get_metadata(self) -> MetadataSnapshot: + """Return the full proposal-level metadata snapshot. + + Implementations are expected to cache this with a short TTL (e.g. ~10s) because + it is fetched on every subscription tick. + """ + + @abstractmethod + async def get_extracted_data( + self, + *, + run: int, + variable: str, + ) -> Any: + """Return the preview data for one (run, variable) pair. + + The return type is intentionally `Any` as the concrete type depends on the + variable's dtype (scalar, PNG bytes, numpy array, etc..). + + Implementations must **not** block the event loop. + """ + + def invalidate_metadata_cache(self) -> None: + """Invalidate any cached metadata so the next `get_metadata` call re-fetches. + + The default implementation is a no-op; override in implementations that cache. + """ + return + + +class DamnitRepositoryRegistry: + """Lazily creates and caches one `DamnitRepository` per proposal. + + `factory` callable receives a `ProposalNumber`, must return a fully initialised + `DamnitRepository`. + """ + + def __init__( + self, + factory: Callable[[ProposalNumber], DamnitRepository], + ) -> None: + self._factory = factory + self._repos: dict[ProposalNumber, DamnitRepository] = {} + + def __contains__(self, proposal_number: ProposalNumber) -> bool: + return proposal_number in self._repos + + def get(self, proposal_number: ProposalNumber) -> DamnitRepository: + """Return the cached repository, creating it on first access.""" + if proposal_number not in self._repos: + try: + self._repos[proposal_number] = self._factory(proposal_number) + except Exception: + logger.exception( + "Failed to create repository for proposal", + proposal=proposal_number, + ) + raise + return self._repos[proposal_number] + + def pop( + self, + proposal_number: ProposalNumber, + default: DamnitRepository | None = None, + ) -> DamnitRepository | None: + """Remove and return the repository for *proposal_number*, if present.""" + return self._repos.pop(proposal_number, default) + + def clear(self) -> None: + """Evict all cached repositories.""" + self._repos.clear() diff --git a/api/src/damnit_api/runs/sqlite/__init__.py b/api/src/damnit_api/runs/sqlite/__init__.py index ecbb9438..9459ff39 100644 --- a/api/src/damnit_api/runs/sqlite/__init__.py +++ b/api/src/damnit_api/runs/sqlite/__init__.py @@ -1,33 +1,15 @@ from ...shared.const import DEFAULT_PROPOSAL -from .repository import ( - async_all_tags, - async_column, - async_latest_rows, - async_max, - async_table, - async_variable_tags, - async_variables, -) +from .repository import SQLiteDamnitRepository from .session import ( DAMNIT_PATH, - DamnitDBRegistry, DatabaseSessionManager, get_damnit_path, - get_session, ) __all__ = [ "DAMNIT_PATH", "DEFAULT_PROPOSAL", - "DamnitDBRegistry", "DatabaseSessionManager", - "async_all_tags", - "async_column", - "async_latest_rows", - "async_max", - "async_table", - "async_variable_tags", - "async_variables", + "SQLiteDamnitRepository", "get_damnit_path", - "get_session", ] diff --git a/api/src/damnit_api/runs/sqlite/dependencies.py b/api/src/damnit_api/runs/sqlite/dependencies.py deleted file mode 100644 index 277e9fd7..00000000 --- a/api/src/damnit_api/runs/sqlite/dependencies.py +++ /dev/null @@ -1,17 +0,0 @@ -"""FastAPI dependency helpers for the DAMNIT database registry.""" - -from typing import Annotated - -from fastapi import Depends, Request - -from ...state import get_app_state -from .session import DamnitDBRegistry - - -def get_damnit_registry(request: Request) -> DamnitDBRegistry: - """Provide the DAMNIT database registry from the application state.""" - return get_app_state(request).damnit_registry - - -DamnitRegistry = Annotated[DamnitDBRegistry, Depends(get_damnit_registry)] -"""Type alias for the DAMNIT database registry dependency.""" diff --git a/api/src/damnit_api/runs/sqlite/repository.py b/api/src/damnit_api/runs/sqlite/repository.py index 0a9bfe14..0346ae3a 100644 --- a/api/src/damnit_api/runs/sqlite/repository.py +++ b/api/src/damnit_api/runs/sqlite/repository.py @@ -1,105 +1,442 @@ +from __future__ import annotations + +import asyncio +import time from collections import defaultdict +from contextlib import asynccontextmanager from datetime import datetime +from typing import TYPE_CHECKING, Any from sqlalchemy import ( + MetaData, Table, + and_, desc, func, select, + text, +) +from sqlalchemy.exc import NoSuchTableError, SQLAlchemyError + +from ... import get_logger +from ...shared.models import ProposalNumber +from ..models import ( + KNOWN_VARIABLES, + MetadataSnapshot, + RunRecord, + TagInfo, + VariableInfo, + VariableValue, ) +from ..repository import DamnitRepository +from .session import DatabaseSessionManager + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from sqlalchemy.ext.asyncio import AsyncSession + +logger = get_logger() + + +class SQLiteDamnitRepository(DamnitRepository): + """SQLite-backed DamnitRepository for a single DAMNIT proposal. + + One instance per proposal. Holds its own `DatabaseSessionManager` and a short-lived + metadata cache. + + Pass the class directly as the factory argument of `DamnitRepositoryRegistry` + (bound to a `metadata_ttl` via `functools.partial`, e.g. in + `state.create_repositories`). Its constructor accepts a single `ProposalNumber`. + """ + + def __init__( + self, proposal_number: ProposalNumber, *, metadata_ttl: float = 10.0 + ) -> None: + self._proposal = proposal_number + self._db = DatabaseSessionManager(proposal_number) + # Per-repo table-reflection cache (invalidateable, no global TTL) + self._tables: dict[str, Table | None] = {} + # Metadata TTL cache + self._metadata_ttl = metadata_ttl + self._metadata_cache: MetadataSnapshot | None = None + self._metadata_cache_at: float = 0.0 + + @asynccontextmanager + async def _session(self) -> AsyncIterator[AsyncSession]: + """Yield a read-only async session (PRAGMA query_only = ON).""" + async with self._db.session() as session: + await session.execute(text("PRAGMA query_only = ON")) + yield session + + async def _get_table(self, name: str) -> Table | None: + """Return the reflected SQLAlchemy `Table`, or `None` if absent.""" + if name not in self._tables: + async with self._db.connect() as conn: + try: + table = await conn.run_sync( + lambda c: Table(name, MetaData(), autoload_with=c) + ) + self._tables[name] = table + except NoSuchTableError: + logger.warning( + "Table not found in proposal database", + proposal=self._proposal, + table=name, + ) + self._tables[name] = None + return None + return self._tables[name] + + @property + def proposal(self) -> ProposalNumber: + return self._proposal + + def invalidate_metadata_cache(self) -> None: + """Discard cached metadata; next `get_metadata` call re-fetches.""" + self._metadata_cache = None + self._metadata_cache_at = 0.0 + + async def get_proposal_number(self) -> str | None: + """Read the raw ``proposal`` value out of this database's `metameta` table. + + Local/dev mode has no MyMdC to ask "which proposal is this?", so + `metadata.services._local_proposal_number` reads it back out of the + DAMNIT database itself. Not part of the `DamnitRepository` ABC: the + `metameta` table is a DAMNIT/sqlite-specific artefact with no CSV + (or future backend) equivalent. Returns the raw string value (or + `None` if the table/row is absent) — parsing into a `ProposalNumber` + and handling invalid values is the caller's job. + """ + table = await self._get_table("metameta") + if table is None: + return None + async with self._session() as session: + result = await session.execute( + select(table.c.value).where(table.c.key == "proposal") + ) + return result.scalar() + + async def _all_tags(self) -> dict[int, dict[str, Any]]: + table = await self._get_table("tags") + if table is None: + return {} + async with self._session() as session: + result = await session.execute(select(table.c.id, table.c.name)) + return { + row["id"]: {"id": row["id"], "name": row["name"]} + for row in result.mappings().all() + } + + async def _variables_meta(self) -> dict[str, dict[str, Any]]: + table = await self._get_table("variables") + if table is None: + return {} + async with self._session() as session: + result = await session.execute(select(table.c.name, table.c.title)) + return { + row["name"]: {"name": row["name"], "title": row["title"]} + for row in result.mappings().all() + } + + async def _variable_tags(self) -> dict[str, list[int]]: + table = await self._get_table("variable_tags") + if table is None: + return {} + async with self._session() as session: + result = await session.execute( + select(table.c.variable_name, table.c.tag_id) + ) + out: dict[str, list[int]] = defaultdict(list) + for row in result.mappings().all(): + out[row["variable_name"]].append(row["tag_id"]) + return dict(out) + + async def _run_ids(self) -> list[int]: + table = await self._get_table("run_info") + if table is None: + return [] + async with self._session() as session: + result = await session.execute(select(table.c.run)) + return list(result.scalars().all()) + + async def _max_timestamp(self) -> float | None: + table = await self._get_table("run_variables") + if table is None: + return None + async with self._session() as session: + result = await session.execute(select(func.max(table.c.timestamp))) + return result.scalar() -from ...utils import create_map -from .session import DamnitDBRegistry, get_session + async def _fetch_run_info(self, run_ids: list[int]) -> dict[int, Any]: + info_map: dict[int, Any] = {} + if not run_ids: + return info_map + info_table = await self._get_table("run_info") + if info_table is None: + return info_map + async with self._session() as session: + result = await session.execute( + select(info_table) + .where(info_table.c.run.in_(run_ids)) + .order_by(info_table.c.run) + ) + for row in result.mappings().all(): + info_map[row["run"]] = row + return info_map + async def get_runs( + self, + *, + limit: int, + offset: int, + variable_names: list[str] | None = None, + ) -> list[RunRecord]: + table = await self._get_table("run_variables") + if table is None: + return [] -async def async_table( - registry: DamnitDBRegistry, proposal, name: str = "runs" -) -> Table | None: - return await registry.get(proposal).get_table(name) + # Paginated set of distinct (proposal, run) pairs + runs_subquery = ( + select(table.c.proposal, table.c.run) + .distinct() + .order_by(table.c.run) + .limit(limit) + .offset(offset) + .subquery() + ) + # Latest timestamp per (run, variable), optionally filtered by name + latest_ts_subquery = select( + table.c.proposal, + table.c.run, + table.c.name, + func.max(table.c.timestamp).label("latest_timestamp"), + ).where(table.c.run.in_(select(runs_subquery.c.run))) + if variable_names is not None: + latest_ts_subquery = latest_ts_subquery.where( + table.c.name.in_(variable_names) + ) + latest_ts_subquery = latest_ts_subquery.group_by( + table.c.run, table.c.name + ).subquery() -async def async_variables(registry: DamnitDBRegistry, proposal): - variables = await async_table(registry, proposal, name="variables") - if variables is None: - return {} - selection_variables = select(variables.c.name, variables.c.title) - async with get_session(registry, proposal) as session: - result = await session.execute(selection_variables) + # Outer-join from runs_subquery so runs with no matching variables still appear + query = ( + select( + runs_subquery.c.proposal, + runs_subquery.c.run, + latest_ts_subquery.c.name, + table.c.value, + table.c.summary_type, + table.c.attributes, + table.c.timestamp, + ) + .select_from(runs_subquery) + .outerjoin( + latest_ts_subquery, + runs_subquery.c.run == latest_ts_subquery.c.run, + ) + .outerjoin( + table, + and_( + table.c.proposal == latest_ts_subquery.c.proposal, + table.c.run == latest_ts_subquery.c.run, + table.c.name == latest_ts_subquery.c.name, + table.c.timestamp == latest_ts_subquery.c.latest_timestamp, + ), + ) + .order_by(runs_subquery.c.run) + ) - variable_rows = result.mappings().all() + async with self._session() as session: + result = await session.execute(query) + variable_rows = result.mappings().all() - return create_map(variable_rows, key="name") + # Group rows by run + run_variable_map: dict[int, dict[str, VariableValue]] = defaultdict(dict) + run_proposals: dict[int, int] = {} + for row in variable_rows: + run = row["run"] + run_proposals[run] = row["proposal"] + if row["name"] is not None: + run_variable_map[run][row["name"]] = VariableValue( + value=row["value"], + summary_type=row["summary_type"], + attributes=row.get("attributes"), + timestamp=row["timestamp"] or 0.0, + ) + run_ids = list(run_proposals.keys()) + info_map = await self._fetch_run_info(run_ids) -async def async_latest_rows( - registry: DamnitDBRegistry, - proposal, - *, - table: Table, - by: str, - start_at=None, - descending=True, -) -> dict: - if start_at is None: - start_at = datetime.now().astimezone().timestamp() - order_by = desc(by) if descending else by + return [ + RunRecord( + proposal=ProposalNumber(run_proposals[run]), + run=run, + start_time=info_map[run]["start_time"] if run in info_map else None, + added_at=info_map[run]["added_at"] if run in info_map else None, + variables=run_variable_map.get(run, {}), + ) + for run in sorted(run_ids) + ] - selection = select(table).where(table.c.get(by) > start_at).order_by(order_by) + async def get_latest_runs( + self, *, start_at: float | None = None + ) -> list[RunRecord]: + if start_at is None: + start_at = datetime.now().astimezone().timestamp() - async with get_session(registry, proposal) as session: - result = await session.execute(selection) - return result.mappings().all() # FIX: # pyright: ignore[reportReturnType] + table = await self._get_table("run_variables") + if table is None: + return [] + selection = ( + select(table) + .where(table.c.timestamp > start_at) + .order_by(desc(table.c.timestamp)) + ) + async with self._session() as session: + result = await session.execute(selection) + rows = result.mappings().all() -async def async_max(registry: DamnitDBRegistry, proposal, *, table: str, column: str): - table = await async_table(registry, proposal, name=table) - if table is None: - return None - selection = select(func.max(table.c.get(column))) - async with get_session(registry, proposal) as session: - result = await session.execute(selection) - return result.scalar() + # Keep only the latest value per (run, variable) + run_variable_map: dict[int, dict[str, VariableValue]] = defaultdict(dict) + run_proposals: dict[int, int] = {} + for row in rows: + run = row["run"] + run_proposals[run] = row["proposal"] + name = row["name"] + ts = row["timestamp"] + existing = run_variable_map[run].get(name) + if existing is None or ts > existing.timestamp: + run_variable_map[run][name] = VariableValue( + value=row["value"], + summary_type=row.get("summary_type"), + attributes=row.get("attributes"), + timestamp=ts, + ) + run_ids = list(run_proposals.keys()) + info_map = await self._fetch_run_info(run_ids) -async def async_column(registry: DamnitDBRegistry, proposal, *, table: str, name: str): - table = await async_table(registry, proposal, name=table) - if table is None: - return [] - selection = select(table.c.get(name)) + return [ + RunRecord( + proposal=ProposalNumber(run_proposals[run]), + run=run, + start_time=info_map[run]["start_time"] if run in info_map else None, + added_at=info_map[run]["added_at"] if run in info_map else None, + variables=run_variable_map[run], + ) + for run in sorted(run_ids) + ] - async with get_session(registry, proposal) as session: - result = await session.execute(selection) + async def get_metadata(self) -> MetadataSnapshot: + now = time.monotonic() + if ( + self._metadata_cache is not None + and now - self._metadata_cache_at < self._metadata_ttl + ): + return self._metadata_cache - return result.scalars().all() + snapshot = await self._fetch_metadata() + self._metadata_cache = snapshot + self._metadata_cache_at = now + return snapshot + async def _fetch_metadata(self) -> MetadataSnapshot: + try: + ( + tags_raw, + variables_raw, + variable_tags_raw, + run_ids_raw, + max_ts, + ) = await asyncio.gather( + self._all_tags(), + self._variables_meta(), + self._variable_tags(), + self._run_ids(), + self._max_timestamp(), + ) + except SQLAlchemyError: + logger.exception( + "Failed to fetch metadata for proposal", + proposal=self._proposal, + ) + raise -async def async_all_tags(registry: DamnitDBRegistry, proposal): - tags_table = await async_table(registry, proposal, name="tags") - if tags_table is None: - return {} - selection = select( - tags_table.c.id, - tags_table.c.name, - ) - async with get_session(registry, proposal) as session: - result = await session.execute(selection) + # Build VariableInfo for DB-defined variables (with tag names resolved) + db_variables: dict[str, VariableInfo] = {} + for name, row in variables_raw.items(): + tag_names = [] + for tid in variable_tags_raw.get(name, []): + tag_row = tags_raw.get(tid) + if tag_row is None: + logger.warning( + "Tag id not found in tags table; skipping", + proposal=self._proposal, + tag_id=tid, + ) + continue + tag_names.append(tag_row["name"]) + db_variables[name] = VariableInfo( + name=name, title=row["title"], tags=tag_names + ) - return create_map(result.mappings().all(), key="id") + # Known variables first; DB variables take precedence + # (they carry resolved tag names) + variables: dict[str, VariableInfo] = { + v.name: VariableInfo(name=v.name, title=v.title) for v in KNOWN_VARIABLES + } + variables.update(db_variables) + # Build TagInfo (prepend the "(Untagged)" tag) + named_tags: dict[str, TagInfo] = {} + for tid, tag_row in tags_raw.items(): + tag_name = tag_row["name"] + var_names = [ + v_name for v_name, tids in variable_tags_raw.items() if tid in tids + ] + named_tags[tag_name] = TagInfo(id=tid, name=tag_name, variables=var_names) -async def async_variable_tags(registry: DamnitDBRegistry, proposal): - variable_tags_table = await async_table(registry, proposal, name="variable_tags") - if variable_tags_table is None: - return {} + untagged_vars = [n for n, v in variables.items() if not v.tags] + tags: dict[str, TagInfo] = { + "(Untagged)": TagInfo(id=0, name="(Untagged)", variables=untagged_vars), + **named_tags, + } - selection = select( - variable_tags_table.c.variable_name, variable_tags_table.c.tag_id - ) - async with get_session(registry, proposal) as session: - result = await session.execute(selection) + return MetadataSnapshot( + runs=tuple(sorted(run_ids_raw)), + variables=variables, + tags=tags, + timestamp=max_ts or 0.0, + ) - variable_tags: dict[str, list[int]] = defaultdict(list) - for row in result.mappings().all(): - variable_tags[row["variable_name"]].append(row["tag_id"]) + async def get_extracted_data(self, *, run: int, variable: str) -> Any: + from ..preview import get_preview_data - return variable_tags + loop = asyncio.get_running_loop() + try: + return await asyncio.wait_for( + loop.run_in_executor( + None, get_preview_data, self._proposal, run, variable + ), + timeout=30.0, + ) + except TimeoutError: + logger.exception( + "Timed out fetching extracted data", + proposal=self._proposal, + run=run, + variable=variable, + ) + raise + except Exception: + logger.exception( + "Failed to fetch extracted data", + proposal=self._proposal, + run=run, + variable=variable, + ) + raise diff --git a/api/src/damnit_api/runs/sqlite/session.py b/api/src/damnit_api/runs/sqlite/session.py index 18cee099..122968cf 100644 --- a/api/src/damnit_api/runs/sqlite/session.py +++ b/api/src/damnit_api/runs/sqlite/session.py @@ -1,10 +1,7 @@ from collections.abc import AsyncIterator -from contextlib import AbstractAsyncContextManager, asynccontextmanager +from contextlib import asynccontextmanager from pathlib import Path -from async_lru import alru_cache -from sqlalchemy import MetaData, Table -from sqlalchemy.exc import NoSuchTableError from sqlalchemy.ext.asyncio import ( AsyncConnection, AsyncSession, @@ -22,11 +19,12 @@ _DEFAULT_PROPOSAL = ProposalNumber(DEFAULT_PROPOSAL) -# ----------------------------------------------------------------------------- -# Asynchronous +class DatabaseSessionManager: + """Async SQLAlchemy session manager for a single DAMNIT proposal DB. + One instance per proposal, held by `SQLiteDamnitRepository`. + """ -class DatabaseSessionManager: def __init__(self, proposal: ProposalNumber = _DEFAULT_PROPOSAL): self.proposal = proposal self.root_path = get_damnit_path(proposal) @@ -34,11 +32,9 @@ def __init__(self, proposal: ProposalNumber = _DEFAULT_PROPOSAL): self.db_path, isolation_level="AUTOCOMMIT", poolclass=NullPool, + connect_args={"timeout": 30}, ) self._sessionmaker = async_sessionmaker(autocommit=False, bind=self._engine) - # Owned by this manager, not the module: - # one reflection cache per proposal, cleared when the manager is. - self._table_cache = alru_cache(ttl=300)(self._reflect_table) @property def db_path(self): @@ -81,42 +77,6 @@ async def session(self) -> AsyncIterator[AsyncSession]: finally: await session.close() - async def _reflect_table(self, name: str) -> Table | None: - async with self.connect() as conn: - try: - return await conn.run_sync( - lambda conn: Table(name, MetaData(), autoload_with=conn) - ) - except NoSuchTableError: - # Don't cache misses; the table may appear shortly. - self._table_cache.cache_invalidate(name) - return None - - async def get_table(self, name: str = "runs") -> Table | None: - return await self._table_cache(name) - - -class DamnitDBRegistry: - """Per-proposal DAMNIT database registry.""" - - def __init__(self) -> None: - self._managers: dict[ProposalNumber, DatabaseSessionManager] = {} - - def get(self, proposal: ProposalNumber) -> DatabaseSessionManager: - if proposal not in self._managers: - self._managers[proposal] = DatabaseSessionManager(proposal) - return self._managers[proposal] - - -def get_session( - registry: DamnitDBRegistry, proposal: ProposalNumber -) -> AbstractAsyncContextManager[AsyncSession]: - return registry.get(proposal).session() - - -# ----------------------------------------------------------------------------- -# Etc. - def get_damnit_path(proposal: ProposalNumber = _DEFAULT_PROPOSAL) -> str: """Returns the directory of the given proposal.""" diff --git a/api/src/damnit_api/runs/types.py b/api/src/damnit_api/runs/types.py index 21381d36..cc92b3d4 100644 --- a/api/src/damnit_api/runs/types.py +++ b/api/src/damnit_api/runs/types.py @@ -1,6 +1,6 @@ import json -from dataclasses import asdict, dataclass -from typing import NewType +from dataclasses import asdict +from typing import TYPE_CHECKING, NewType import strawberry @@ -12,24 +12,13 @@ python_type_to_damnit_type, summary_type_to_damnit_type, ) +from .models import KNOWN_VARIABLES from .serialization import serialize -logger = get_logger() - - -@dataclass(frozen=True) -class KnownVariable: - name: str - title: str - dtype: DamnitType +if TYPE_CHECKING: + from .models import RunRecord - -KNOWN_VARIABLES = ( - KnownVariable(name="proposal", title="Proposal", dtype=DamnitType.NUMBER), - KnownVariable(name="run", title="Run", dtype=DamnitType.NUMBER), - KnownVariable(name="start_time", title="Timestamp", dtype=DamnitType.TIMESTAMP), - KnownVariable(name="added_at", title="Added at", dtype=DamnitType.TIMESTAMP), -) +logger = get_logger() KNOWN_DTYPES = {v.name: v.dtype for v in KNOWN_VARIABLES} @@ -118,10 +107,35 @@ def _iter_variables(cls, record): error = DamnitVariableError.from_attrs(entry.get("attributes")) yield DamnitVariable(name=name, value=Any(value), dtype=dtype, error=error) + @classmethod + def _flatten_record(cls, record: "RunRecord") -> dict: + """Flatten a `RunRecord` into the legacy `{name: {value, summary_type}}` + dict shape `_iter_variables` expects. Shared by `from_record` (queries) + and `resolve_record` (subscriptions) so there is one RunRecord -> dict path. + """ + flat: dict = { + "proposal": record.proposal, + "run": record.run, + "start_time": record.start_time, + "added_at": record.added_at, + } + for name, vv in record.variables.items(): + flat[name] = { + "value": vv.value, + "summary_type": vv.summary_type, + "attributes": vv.attributes, + } + return flat + @classmethod def from_db(cls, record): return cls(_variables=list(cls._iter_variables(record))) + @classmethod + def from_record(cls, record: "RunRecord") -> "DamnitRun": + """Build a `DamnitRun` from a `RunRecord` (repository layer).""" + return cls.from_db(cls._flatten_record(record)) + @classmethod def resolve(cls, record): out: dict[str, object | None] = { @@ -140,6 +154,14 @@ def resolve(cls, record): out[v.name] = resolved return out + @classmethod + def resolve_record(cls, record: "RunRecord") -> dict: + """Build the raw JSON-scalar payload shape for a `RunRecord`. + + Subscriptions stream `JSON` rather than the `DamnitRun` GraphQL type. + """ + return cls.resolve(cls._flatten_record(record)) + @staticmethod def known_variables(): return create_map( diff --git a/api/src/damnit_api/shared/gql.py b/api/src/damnit_api/shared/gql.py index 1ff184c8..8786c791 100644 --- a/api/src/damnit_api/shared/gql.py +++ b/api/src/damnit_api/shared/gql.py @@ -17,7 +17,7 @@ from ..graphql.dependencies import SubscriptionCursorsDep from ..metadata import gql as metadata from ..runs import types as run_types -from ..runs.sqlite.dependencies import DamnitRegistry +from ..runs.dependencies import Repositories SUBSCRIPTION_PROTOCOLS = [ GRAPHQL_TRANSPORT_WS_PROTOCOL, @@ -57,7 +57,7 @@ class Context(BaseContext): mymdc: MyMdCClient oauth_user: OAuthUserInfo session: DBSession - damnit_registry: DamnitRegistry + repositories: Repositories subscription_cursors: SubscriptionCursorsDep _user: User | None = None @@ -74,14 +74,14 @@ async def get_context( # noqa: RUF029 oauth_user: OAuthUserInfo, mymdc: MyMdCClient, session: DBSession, - damnit_registry: DamnitRegistry, + repositories: Repositories, subscription_cursors: SubscriptionCursorsDep, ): return Context( oauth_user=oauth_user, mymdc=mymdc, session=session, - damnit_registry=damnit_registry, + repositories=repositories, subscription_cursors=subscription_cursors, ) diff --git a/api/src/damnit_api/state.py b/api/src/damnit_api/state.py index 92039ceb..e362a803 100644 --- a/api/src/damnit_api/state.py +++ b/api/src/damnit_api/state.py @@ -23,7 +23,7 @@ from ._mymdc.clients import MyMdCClient from .auth.token_store import TokenStore from .graphql.subscriptions import SubscriptionCursors - from .runs.sqlite.session import DamnitDBRegistry + from .runs.repository import DamnitRepositoryRegistry from .shared.settings import Settings @@ -34,7 +34,7 @@ class AppState: mymdc_client: MyMdCClient oauth_client: StarletteOAuth2App | None # None when auth is disabled token_store: TokenStore - damnit_registry: DamnitDBRegistry + repositories: DamnitRepositoryRegistry subscription_cursors: SubscriptionCursors @@ -85,10 +85,16 @@ def create_token_store() -> TokenStore: return InMemoryTokenStore() -def create_damnit_registry() -> DamnitDBRegistry: - from .runs.sqlite.session import DamnitDBRegistry +def create_repositories() -> DamnitRepositoryRegistry: + """Registry of per-proposal `DamnitRepository` objects (ADR-005). - return DamnitDBRegistry() + The SQLite backend is passed directly as the factory; its `metadata_ttl` + keeps its default (see ADR-005 - TTLs are an implementation concern). + """ + from .runs.repository import DamnitRepositoryRegistry + from .runs.sqlite.repository import SQLiteDamnitRepository + + return DamnitRepositoryRegistry(SQLiteDamnitRepository) def create_subscription_cursors() -> SubscriptionCursors: diff --git a/api/tests/conftest.py b/api/tests/conftest.py deleted file mode 100644 index 9c374349..00000000 --- a/api/tests/conftest.py +++ /dev/null @@ -1,9 +0,0 @@ -import pytest - -from damnit_api.runs.sqlite import DamnitDBRegistry - - -@pytest.fixture -def damnit_registry() -> DamnitDBRegistry: - """A fresh per-test registry; nothing module-level to clear between tests.""" - return DamnitDBRegistry() diff --git a/api/tests/graphql/conftest.py b/api/tests/graphql/conftest.py index 6380d78f..9977d43c 100644 --- a/api/tests/graphql/conftest.py +++ b/api/tests/graphql/conftest.py @@ -1,3 +1,4 @@ +from pathlib import Path from types import SimpleNamespace import pytest @@ -5,28 +6,48 @@ from strawberry.schema.config import StrawberryConfig from damnit_api.graphql.directives import lightweight -from damnit_api.graphql.metadata import fetch_metadata from damnit_api.graphql.queries import Query from damnit_api.graphql.subscriptions import ( Subscription, SubscriptionCursors, poll_proposal, ) +from damnit_api.runs.csv import CsvDamnitRepository +from damnit_api.runs.repository import DamnitRepositoryRegistry from damnit_api.runs.types import SCALAR_MAP, DamnitVariable -from .const import ( - EXAMPLE_TAGS, - EXAMPLE_VARIABLE_TAGS, - EXAMPLE_VARIABLES, - RUNS, -) + +class SchemaWithContext: + """Wraps a Strawberry Schema and injects a default context_value. + + An explicit `context_value` at the call site still wins. + """ + + def __init__(self, schema, default_context) -> None: + self._schema = schema + self._context = default_context + + async def execute(self, query, *, context_value=None, variable_values=None): + ctx = context_value if context_value is not None else self._context + return await self._schema.execute( + query, context_value=ctx, variable_values=variable_values + ) + + async def subscribe(self, query, *, context_value=None, variable_values=None): + ctx = context_value if context_value is not None else self._context + return await self._schema.subscribe( + query, context_value=ctx, variable_values=variable_values + ) + + +@pytest.fixture +def subscription_cursors() -> SubscriptionCursors: + return SubscriptionCursors() @pytest.fixture(autouse=True) def reset_caches(): - fetch_metadata.cache_clear() poll_proposal.cache_clear() - return def _patch_permissions(mocker, *, authenticated: bool, member: bool) -> None: @@ -49,109 +70,59 @@ def bypass_proposal_permission(mocker): @pytest.fixture -def mocked_metadata_variables(mocker): - mocker.patch( - "damnit_api.graphql.metadata.db.async_variables", - return_value=EXAMPLE_VARIABLES, - ) - - -@pytest.fixture -def mocked_metadata_all_tags(mocker): - mocker.patch( - "damnit_api.graphql.metadata.db.async_all_tags", - return_value=EXAMPLE_TAGS, - ) - - -@pytest.fixture -def mocked_metadata_variable_tags(mocker): +def mocked_ensure_damnit_path(mocker): + """Bypass the damnit_path validation; tests run without a request context.""" mocker.patch( - "damnit_api.graphql.metadata.db.async_variable_tags", - return_value=EXAMPLE_VARIABLE_TAGS, + "damnit_api.graphql.queries._ensure_damnit_path", + return_value=None, ) @pytest.fixture -def mocked_metadata_column(mocker): - mocker.patch( - "damnit_api.graphql.metadata.db.async_column", - return_value=RUNS, - ) +def csv_fixture_dir() -> Path: + """Path to the graphql test CSV fixture files.""" + return Path(__file__).parent / "fixtures" @pytest.fixture -def mocked_metadata_max(mocker): - mocker.patch( - "damnit_api.graphql.metadata.db.async_max", - return_value=0, +def mock_repositories(csv_fixture_dir): + """A repository registry backed by CsvDamnitRepository over the fixtures.""" + return DamnitRepositoryRegistry( + lambda proposal: CsvDamnitRepository(proposal, csv_fixture_dir) ) @pytest.fixture -def mocked_ensure_damnit_path(mocker): - """Bypass the damnit_path validation; tests run without a request context.""" - mocker.patch( - "damnit_api.graphql.queries._ensure_damnit_path", - return_value=None, +def graphql_context(mock_repositories): + return SimpleNamespace( + repositories=mock_repositories, + subscription_cursors=SubscriptionCursors(), + oauth_user=None, ) @pytest.fixture -def graphql_schema_no_auth( - mocked_metadata_variables, - mocked_metadata_column, - mocked_metadata_all_tags, - mocked_metadata_variable_tags, -): - """Schema without the bypass_proposal_permission fixture, so permission - checks run normally (and fail since there is no real request context).""" - return strawberry.Schema( +def graphql_schema_no_auth(graphql_context): + """Schema without permission bypass, so permission checks run normally.""" + schema = strawberry.Schema( query=Query, subscription=Subscription, types=[DamnitVariable], directives=[lightweight], config=StrawberryConfig(auto_camel_case=False, scalar_map=SCALAR_MAP), ) - - -class _SchemaWithDefaultContext: - """Wraps a strawberry `Schema` so resolvers see a default context (with - a fresh `damnit_registry`) without every test passing `context_value`; - an explicit `context_value` at the call site still wins.""" - - def __init__(self, schema, context): - self._schema = schema - self._context = context - - def execute(self, query, **kwargs): - kwargs.setdefault("context_value", self._context) - return self._schema.execute(query, **kwargs) - - def subscribe(self, query, **kwargs): - kwargs.setdefault("context_value", self._context) - return self._schema.subscribe(query, **kwargs) - - -@pytest.fixture -def graphql_context(damnit_registry): - return SimpleNamespace( - damnit_registry=damnit_registry, - subscription_cursors=SubscriptionCursors(), - ) + return SchemaWithContext(schema, graphql_context) @pytest.fixture def graphql_schema( bypass_proposal_permission, mocked_ensure_damnit_path, - mocked_metadata_max, graphql_schema_no_auth, - graphql_context, ): """Same schema as graphql_schema_no_auth, with permission and damnit-path checks bypassed so tests exercise resolver logic only.""" - return _SchemaWithDefaultContext(graphql_schema_no_auth, graphql_context) + return graphql_schema_no_auth @pytest.fixture diff --git a/api/tests/graphql/fixtures/run_variables.csv b/api/tests/graphql/fixtures/run_variables.csv new file mode 100644 index 00000000..efcdc85c --- /dev/null +++ b/api/tests/graphql/fixtures/run_variables.csv @@ -0,0 +1,5 @@ +run,name,value,summary_type,timestamp +348,n_trains,3641,,1000.0 +348,run_length,0:06:03,,1000.0 +348,xgm_intensity,2.073,,1000.0 +348,etof_settings.ret0,-77.0,,1000.0 diff --git a/api/tests/graphql/fixtures/runs.csv b/api/tests/graphql/fixtures/runs.csv new file mode 100644 index 00000000..8183e1d0 --- /dev/null +++ b/api/tests/graphql/fixtures/runs.csv @@ -0,0 +1,4 @@ +run,start_time,added_at +348,1740154563.795096,1775038442.775598 +349,, +350,, diff --git a/api/tests/graphql/fixtures/variables.csv b/api/tests/graphql/fixtures/variables.csv new file mode 100644 index 00000000..b4c1991f --- /dev/null +++ b/api/tests/graphql/fixtures/variables.csv @@ -0,0 +1,7 @@ +name,title,tags +n_trains,Trains, +run_length,Run length, +xgm_intensity,XGM intensity [uJ], +etof_settings.ret0,"eTOF settings/Retardation, sector 0",eTOF setting +etof.eTOF_calibration,eTOF calib./eTOF calibration,eTOF +etof.eTOF_response_width,eTOF calib./eTOF response FWHM,eTOF diff --git a/api/tests/graphql/test_queries.py b/api/tests/graphql/test_queries.py index adc7431e..e445443d 100644 --- a/api/tests/graphql/test_queries.py +++ b/api/tests/graphql/test_queries.py @@ -1,367 +1,153 @@ -import pytest -import pytest_asyncio -from sqlalchemy import text - -from damnit_api.runs.sqlite import DAMNIT_PATH, DatabaseSessionManager -from damnit_api.runs.types import DamnitRun -from damnit_api.shared.models import ProposalNumber - -from .const import ( - EXAMPLE_DATA, - EXAMPLE_VARIABLES, - KNOWN_DATA, - PROPOSAL, - RUNS, - get_values, -) - - -@pytest.fixture -def mocked_fetch_variables(mocker): - # fetch_variables returns wrapped values: {"run": {"value": 348}, ...} - values = get_values(EXAMPLE_DATA) - wrapped = { - "proposal": {"value": values["proposal"]}, - "run": {"value": values["run"]}, - **{ - name: {"value": value, "summary_type": None} - for name, value in values.items() - if name not in ("proposal", "run") - }, - } - return mocker.patch( - "damnit_api.graphql.queries.fetch_variables", - return_value=[wrapped], - ) - - -@pytest.fixture -def mocked_fetch_info(mocker): - return mocker.patch( - "damnit_api.graphql.queries.fetch_info", - return_value=[get_values(KNOWN_DATA)], - ) - - -@pytest.mark.asyncio -async def test_runs_query(graphql_schema, mocked_fetch_variables, mocked_fetch_info): - query = f""" - query TableDataQuery($per_page: Int = 2) {{ - runs(database: {{proposal: "{PROPOSAL}"}}, per_page: $per_page) {{ - variables {{ - name - value - }} - }} - }} - """ - result = await graphql_schema.execute(query) +"""Resolver tests for the runs/metadata/extracted_data queries. - assert result.errors is None +Resolvers are exercised against a `CsvDamnitRepository` (see conftest), so no +SQLite fixtures are needed. The CSV fixtures define runs 348/349/350, but only +run 348 has variable rows, so `runs` returns just run 348 while `metadata` +lists every run. +""" - runs = result.data["runs"] - assert len(runs) == 1 +import pytest - variables = runs[0]["variables"] - variable_names = {v["name"] for v in variables} - assert "run" in variable_names - assert "n_trains" in variable_names +from .const import PROPOSAL - assert mocked_fetch_variables.call_args.kwargs["names"] is None - assert mocked_fetch_info.called +FIXTURE_VARIABLES = {"n_trains", "run_length", "xgm_intensity", "etof_settings.ret0"} @pytest.mark.asyncio -async def test_runs_query_filters_variables_by_name( - graphql_schema, mocked_fetch_variables, mocked_fetch_info -): +async def test_runs_returns_runs_with_variables(graphql_schema): query = f""" query {{ - runs(database: {{proposal: "{PROPOSAL}"}}, per_page: 2) {{ - variables(names: ["n_trains"]) {{ - name - }} + runs(database: {{proposal: {PROPOSAL}}}, per_page: 10) {{ + variables {{ name value dtype }} }} }} """ result = await graphql_schema.execute(query) assert result.errors is None - - variables = result.data["runs"][0]["variables"] - assert [v["name"] for v in variables] == ["n_trains"] - - assert mocked_fetch_variables.call_args.kwargs["names"] == ["n_trains"] - assert not mocked_fetch_info.called + runs = result.data["runs"] + assert len(runs) == 1 + names = {v["name"] for v in runs[0]["variables"]} + assert names >= FIXTURE_VARIABLES + assert {"proposal", "run"} <= names @pytest.mark.asyncio -async def test_runs_query_filters_multiple_names( - graphql_schema, mocked_fetch_variables, mocked_fetch_info -): +async def test_runs_variable_shape(graphql_schema): query = f""" query {{ - runs(database: {{proposal: "{PROPOSAL}"}}, per_page: 2) {{ - variables(names: ["n_trains", "etof_settings.ret0"]) {{ - name - }} + runs(database: {{proposal: {PROPOSAL}}}, per_page: 10) {{ + variables {{ name value dtype }} }} }} """ result = await graphql_schema.execute(query) assert result.errors is None - - names = {v["name"] for v in result.data["runs"][0]["variables"]} - assert names == {"n_trains", "etof_settings.ret0"} - - forwarded = mocked_fetch_variables.call_args.kwargs["names"] - assert set(forwarded) == {"n_trains", "etof_settings.ret0"} - - -@pytest_asyncio.fixture -async def real_damnit_db(mocker, tmp_path): - """Real on-disk sqlite with a populated `run_variables` table, wired - via `find_proposal`. Yields the proposal id. - """ - proposal = ProposalNumber(999999) - proposal_root = tmp_path / "proposal" - (proposal_root / DAMNIT_PATH).mkdir(parents=True) - - mocker.patch( - "damnit_api.runs.sqlite.session.find_proposal", - return_value=str(proposal_root), - ) - - manager = DatabaseSessionManager(proposal) - async with manager.connect() as conn: - await conn.execute( - text( - "CREATE TABLE run_variables (" - " proposal INTEGER NOT NULL," - " run INTEGER NOT NULL," - " name TEXT NOT NULL," - " value BLOB," - " summary_type TEXT," - " attributes BLOB," - " timestamp REAL NOT NULL," - " PRIMARY KEY (proposal, run, name, timestamp)" - ")" - ) - ) - rows = [ - # run 1: has both vars - (int(proposal), 1, "alpha", "a1", None, 1000.0), - (int(proposal), 1, "beta", "b1", None, 1000.0), - # run 2: has only `alpha` - (int(proposal), 2, "alpha", "a2", None, 1100.0), - # run 3: has only `beta` (so a filter on `alpha` excludes it) - (int(proposal), 3, "beta", "b3", None, 1200.0), - ] - await conn.execute( - text( - "INSERT INTO run_variables" - " (proposal, run, name, value, summary_type, timestamp)" - " VALUES (:proposal, :run, :name, :value, :summary_type," - " :timestamp)" - ), - [ - { - "proposal": p, - "run": r, - "name": n, - "value": v, - "summary_type": s, - "timestamp": t, - } - for p, r, n, v, s, t in rows - ], - ) - - yield proposal - - await manager.close() + for variable in result.data["runs"][0]["variables"]: + assert set(variable.keys()) == {"name", "value", "dtype"} @pytest.mark.asyncio -async def test_runs_query_unknown_name(graphql_schema, real_damnit_db): - """Filtering by an unknown name returns every paginated run with an - empty `variables` list (regression: an inner join used to drop them). - """ - proposal = real_damnit_db +async def test_runs_variable_name_filter(graphql_schema): query = f""" query {{ - runs(database: {{proposal: "{proposal}"}}, per_page: 10) {{ - variables(names: ["does.not.exist"]) {{ - name - }} + runs(database: {{proposal: {PROPOSAL}}}, per_page: 10) {{ + variables(names: ["n_trains"]) {{ name }} }} }} """ result = await graphql_schema.execute(query) assert result.errors is None - runs = result.data["runs"] - assert [r["variables"] for r in runs] == [[], [], []] + names = [v["name"] for v in result.data["runs"][0]["variables"]] + assert names == ["n_trains"] @pytest.mark.asyncio -async def test_runs_query_partial_name_match(graphql_schema, real_damnit_db): - """Runs with no rows for the filtered name still appear in the page; - run 3 has no `alpha`, so only the implicit `run` value comes back. - """ - proposal = real_damnit_db - query = f""" - query {{ - runs(database: {{proposal: "{proposal}"}}, per_page: 10) {{ - variables(names: ["alpha", "run"]) {{ - name - value - }} - }} - }} - """ +async def test_metadata_lists_all_runs(graphql_schema): + query = f"query {{ metadata(database: {{proposal: {PROPOSAL}}}) }}" result = await graphql_schema.execute(query) assert result.errors is None - runs = result.data["runs"] - by_run = [{v["name"]: v["value"] for v in r["variables"]} for r in runs] - assert by_run == [ - {"alpha": "a1", "run": 1}, - {"alpha": "a2", "run": 2}, - {"run": 3}, - ] + metadata = result.data["metadata"] + assert set(metadata.keys()) == {"runs", "variables", "tags", "timestamp"} + assert metadata["runs"] == [348, 349, 350] + assert set(metadata["variables"].keys()) >= FIXTURE_VARIABLES + assert "(Untagged)" in metadata["tags"] @pytest.mark.asyncio -async def test_runs_query_fetches_run_info_when_metadata_requested( - graphql_schema, mocked_fetch_variables, mocked_fetch_info -): - query = f""" - query {{ - runs(database: {{proposal: "{PROPOSAL}"}}, per_page: 2) {{ - variables(names: ["start_time"]) {{ - name - }} - }} - }} - """ +async def test_metadata_timestamp_is_milliseconds(graphql_schema): + query = f"query {{ metadata(database: {{proposal: {PROPOSAL}}}) }}" result = await graphql_schema.execute(query) assert result.errors is None - assert mocked_fetch_info.called - assert [v["name"] for v in result.data["runs"][0]["variables"]] == ["start_time"] - - -@pytest.mark.asyncio -async def test_metadata_query(graphql_schema): - query = """ - query TableMetadataQuery($proposal: ProposalNo!) { - metadata(database: { proposal: $proposal }) - } - """ - result = await graphql_schema.execute( - query, - variable_values={"proposal": str(PROPOSAL)}, - ) - - assert result.errors is None - - metadata = result.data["metadata"] - assert set(metadata.keys()) == {"runs", "variables", "timestamp", "tags"} - assert metadata["runs"] == RUNS - assert metadata["variables"] == { - **DamnitRun.known_variables(), - **EXAMPLE_VARIABLES, - } - assert "(Untagged)" in metadata["tags"] - assert "eTOF" in metadata["tags"] + # Max fixture timestamp is 1000.0s; the resolver reports it in ms. + assert result.data["metadata"]["timestamp"] == pytest.approx(1000.0 * 1000) @pytest.mark.asyncio -async def test_runs_forbidden(graphql_schema_authenticated_non_member): +async def test_extracted_data_csv_has_no_preview(graphql_schema): + """The CSV backend has no binary preview data, so `get_extracted_data` + returns None; the non-nullable `extracted_data: JSON!` field surfaces that + as a null-field error rather than a value.""" query = f""" query {{ - runs(database: {{proposal: "{PROPOSAL}"}}) {{ - variables {{ name }} - }} + extracted_data( + database: {{proposal: {PROPOSAL}}}, run: 348, variable: "n_trains" + ) }} """ - result = await graphql_schema_authenticated_non_member.execute(query) + result = await graphql_schema.execute(query) assert result.errors is not None - assert result.errors[0].message == "Access to this proposal is forbidden." - + assert result.data is None -@pytest.mark.asyncio -async def test_extracted_data_forbidden(graphql_schema_authenticated_non_member): - query = f""" - query {{ - extracted_data(database: {{proposal: "{PROPOSAL}"}}, run: 1, variable: "x") - }} - """ - result = await graphql_schema_authenticated_non_member.execute(query) - assert result.errors is not None - assert result.errors[0].message == "Access to this proposal is forbidden." +# ----------------------------------------------------------------------------- +# ProposalNo scalar boundary: out-of-range proposals are rejected at coercion. @pytest.mark.asyncio -async def test_runs_unauthorized(graphql_schema_no_auth): - query = f""" - query {{ - runs(database: {{proposal: "{PROPOSAL}"}}) {{ - variables {{ name }} - }} - }} +async def test_runs_rejects_out_of_range_proposal(graphql_schema): + query = """ + query { runs(database: {proposal: 1000000}) { variables { name } } } """ - result = await graphql_schema_no_auth.execute(query) - + result = await graphql_schema.execute(query) assert result.errors is not None - assert result.errors[0].message == "Authentication required." @pytest.mark.asyncio -async def test_extracted_data_unauthorized(graphql_schema_no_auth): - query = f""" - query {{ - extracted_data(database: {{proposal: "{PROPOSAL}"}}, run: 1, variable: "x") - }} +async def test_runs_rejects_negative_proposal(graphql_schema): + query = """ + query { runs(database: {proposal: -1}) { variables { name } } } """ - result = await graphql_schema_no_auth.execute(query) - + result = await graphql_schema.execute(query) assert result.errors is not None - assert result.errors[0].message == "Authentication required." # ----------------------------------------------------------------------------- -# ProposalNo scalar boundary: out-of-range proposals are rejected at input -# coercion, before any resolver or permission runs. +# Permission boundaries. @pytest.mark.asyncio -async def test_runs_rejects_out_of_range_proposal(graphql_schema): - query = """ - query { - runs(database: {proposal: 1000000}) { - variables { name } - } - } +async def test_runs_forbidden_for_non_member(graphql_schema_authenticated_non_member): + query = f""" + query {{ runs(database: {{proposal: {PROPOSAL}}}) {{ variables {{ name }} }} }} """ - result = await graphql_schema.execute(query) + result = await graphql_schema_authenticated_non_member.execute(query) assert result.errors is not None + assert result.errors[0].message == "Access to this proposal is forbidden." @pytest.mark.asyncio -async def test_runs_rejects_negative_proposal(graphql_schema): - query = """ - query { - runs(database: {proposal: -1}) { - variables { name } - } - } +async def test_runs_unauthorized(graphql_schema_no_auth): + query = f""" + query {{ runs(database: {{proposal: {PROPOSAL}}}) {{ variables {{ name }} }} }} """ - result = await graphql_schema.execute(query) + result = await graphql_schema_no_auth.execute(query) assert result.errors is not None + assert result.errors[0].message == "Authentication required." diff --git a/api/tests/graphql/test_subscriptions.py b/api/tests/graphql/test_subscriptions.py index 5e271ec9..b0f93fc1 100644 --- a/api/tests/graphql/test_subscriptions.py +++ b/api/tests/graphql/test_subscriptions.py @@ -1,285 +1,95 @@ -import asyncio -from datetime import UTC, datetime -from unittest.mock import patch +"""Tests for the latest_data subscription polling logic. -import pytest +`poll_proposal` and `filter_for_client` are exercised directly against a +`CsvDamnitRepository` (see conftest). The CSV fixtures put run 348's variables +at timestamp 1000.0, so a cursor seeded below that surfaces the run. +""" -from damnit_api.graphql.subscriptions import POLLING_INTERVAL, filter_for_client -from damnit_api.runs.types import DamnitRun -from damnit_api.shared.const import DamnitType +import pytest -from .const import ( - EXAMPLE_VARIABLES, - KNOWN_DATA, - NEW_DATA, - PROPOSAL, - RUNS, - DatabaseVariable, - get_values, +from damnit_api.graphql.subscriptions import ( + SubscriptionCursors, + filter_for_client, + poll_proposal, ) -from .utils import create_run_variables - -NEW_RUN = 400 - - -patched_sleep = patch.object(asyncio, "sleep", return_value=None) - - -@pytest.fixture(scope="module") -def current_timestamp(): - return datetime.now(tz=UTC).timestamp() - - -@pytest.fixture -def mocked_latest_rows(mocker, current_timestamp): - table_sentinel = mocker.sentinel.run_variables_table - mocker.patch( - "damnit_api.graphql.subscriptions.async_table", - return_value=table_sentinel, - ) - mocker.patch( - "damnit_api.graphql.subscriptions.async_max", - return_value=0, - ) - - def mocked_returns(*args, table, **kwargs): - if table is table_sentinel: - return create_run_variables( - get_values(NEW_DATA), - proposal=PROPOSAL, - run=NEW_RUN, - timestamp=current_timestamp, - ) - return None +from damnit_api.shared.models import ProposalNumber - return mocker.patch( - "damnit_api.graphql.subscriptions.async_latest_rows", - side_effect=mocked_returns, - ) +from .const import PROPOSAL - -@pytest.fixture -def mocked_fetch_info(mocker): - return mocker.patch( - "damnit_api.graphql.subscriptions.fetch_info", - return_value=[{**get_values(KNOWN_DATA), "run": NEW_RUN}], - ) +_PROPOSAL = ProposalNumber(PROPOSAL) @pytest.mark.asyncio -async def test_latest_data( - graphql_schema, - current_timestamp, - mocked_latest_rows, - mocked_fetch_info, -): - subscription = await graphql_schema.subscribe( - """ - subscription LatestDataSubscription( - $proposal: ProposalNo!, - $timestamp: Timestamp!) { - latest_data(database: { proposal: $proposal }, timestamp: $timestamp) - } - """, - variable_values={ - "proposal": str(PROPOSAL), - "timestamp": (current_timestamp - 1) * 1000, # before the new row - }, - ) - - try: - result = await asyncio.wait_for(anext(subscription), timeout=2) - assert not result.errors - - data = { - **KNOWN_DATA, - **NEW_DATA, - "run": DatabaseVariable( - value=NEW_RUN, - damnit_dtype=DamnitType.NUMBER, - ), - } - - latest_data = result.data["latest_data"] - assert set(latest_data.keys()) == {"runs", "metadata"} - assert latest_data["runs"] == { - NEW_RUN: { - name: { - "value": var.damnit_value, - "dtype": var.damnit_dtype.value, - } - for name, var in data.items() - } - } - - metadata = latest_data["metadata"] - assert set(metadata.keys()) == {"runs", "timestamp", "variables"} - assert metadata["runs"] == [*RUNS, NEW_RUN] - assert metadata["timestamp"] == current_timestamp * 1000 - assert metadata["variables"] == { - **DamnitRun.known_variables(), - **EXAMPLE_VARIABLES, - } - finally: - await subscription.aclose() - - -@pytest.mark.asyncio -async def test_latest_data_with_concurrent_subscriptions( - graphql_schema, - current_timestamp, - mocked_latest_rows, - mocked_fetch_info, -): - query = """ - subscription LatestDataSubscription( - $proposal: ProposalNo!, - $timestamp: Timestamp!) { - latest_data(database: { proposal: $proposal }, timestamp: $timestamp) - } - """ - variables = { - "proposal": str(PROPOSAL), - "timestamp": (current_timestamp - 1) * 1000, # before the new row +async def test_poll_proposal_surfaces_new_runs(mock_repositories): + cursors = SubscriptionCursors() + cursors[_PROPOSAL] = 0.0 # seed below the fixture timestamps + repo = mock_repositories.get(_PROPOSAL) + + snapshot = await poll_proposal(_PROPOSAL, cursors, repo) + + assert snapshot is not None + assert set(snapshot.keys()) == { + "runs", + "run_timestamps", + "max_timestamp", + "metadata", } + assert 348 in snapshot["runs"] + assert snapshot["max_timestamp"] == pytest.approx(1000.0) + run_payload = snapshot["runs"][348] + for variable in run_payload.values(): + assert set(variable.keys()) == {"value", "dtype"} - first_sub = await graphql_schema.subscribe( - query, - variable_values=variables, - ) - second_sub = await graphql_schema.subscribe( - query, - variable_values=variables, - ) - try: - with patched_sleep: - result = await asyncio.wait_for(anext(first_sub), timeout=2) - assert not result.errors - mocked_latest_rows.assert_called() +@pytest.mark.asyncio +async def test_poll_proposal_metadata_shape(mock_repositories): + cursors = SubscriptionCursors() + cursors[_PROPOSAL] = 0.0 + repo = mock_repositories.get(_PROPOSAL) - mocked_latest_rows.reset_mock() + snapshot = await poll_proposal(_PROPOSAL, cursors, repo) - with patched_sleep: - result = await asyncio.wait_for(anext(second_sub), timeout=2) - assert not result.errors - mocked_latest_rows.assert_not_called() - finally: - await first_sub.aclose() - await second_sub.aclose() + assert snapshot is not None + metadata = snapshot["metadata"] + assert set(metadata.keys()) == {"runs", "variables", "timestamp"} + assert 348 in metadata["runs"] + # ms-timestamp, matching the metadata query's serialization. + assert metadata["timestamp"] == pytest.approx(1000.0 * 1000) @pytest.mark.asyncio -async def test_latest_data_with_nonconcurrent_subscriptions( - graphql_schema, - current_timestamp, - mocked_latest_rows, - mocked_fetch_info, -): - query = """ - subscription LatestDataSubscription( - $proposal: ProposalNo!, - $timestamp: Timestamp!) { - latest_data(database: { proposal: $proposal }, timestamp: $timestamp) - } - """ - variables = { - "proposal": str(PROPOSAL), - "timestamp": (current_timestamp - 1) * 1000, # before the new row - } - - first_sub = await graphql_schema.subscribe( - query, - variable_values=variables, - ) - second_sub = await graphql_schema.subscribe( - query, - variable_values=variables, - ) - - try: - with patched_sleep: - result = await asyncio.wait_for(anext(first_sub), timeout=2) - assert not result.errors - mocked_latest_rows.assert_called() +async def test_poll_proposal_no_new_data_returns_none(mock_repositories): + """A cursor at/above the newest row yields nothing.""" + cursors = SubscriptionCursors() + cursors[_PROPOSAL] = 1000.0 + repo = mock_repositories.get(_PROPOSAL) - await asyncio.sleep(POLLING_INTERVAL * 3) # give enough time to clear the cache - mocked_latest_rows.reset_mock() + assert await poll_proposal(_PROPOSAL, cursors, repo) is None - with patched_sleep: - result = await asyncio.wait_for(anext(second_sub), timeout=2) - assert not result.errors - mocked_latest_rows.assert_called() - finally: - await first_sub.aclose() - await second_sub.aclose() - -# ----------------------------------------------------------------------------- -# filter_for_client +def test_filter_for_client_drops_stale_snapshot(): + snapshot = { + "runs": {348: {"x": {"value": 1, "dtype": "number"}}}, + "run_timestamps": {348: 1000.0}, + "max_timestamp": 1000.0, + "metadata": {"runs": [348], "variables": {}, "timestamp": 1_000_000.0}, + } + # since >= max_timestamp -> nothing new + assert filter_for_client(snapshot, since=1000.0) is None -def _snapshot(run_timestamps): - runs = {run: {"value": run} for run in run_timestamps} - return { - "runs": runs, - "run_timestamps": run_timestamps, - "max_timestamp": max(run_timestamps.values()), - "metadata": {"runs": list(runs), "variables": {}, "timestamp": 0}, +def test_filter_for_client_returns_fresh_runs(): + snapshot = { + "runs": {348: {"x": {"value": 1, "dtype": "number"}}}, + "run_timestamps": {348: 1000.0}, + "max_timestamp": 1000.0, + "metadata": {"runs": [348], "variables": {}, "timestamp": 1_000_000.0}, } + result = filter_for_client(snapshot, since=500.0) + assert result is not None + assert set(result.keys()) == {"runs", "metadata"} + assert 348 in result["runs"] def test_filter_for_client_none_snapshot(): - assert filter_for_client(None, since=0) is None - - -def test_filter_for_client_since_zero_returns_none(): - snapshot = _snapshot({1: 100.0, 2: 200.0}) - assert filter_for_client(snapshot, since=0) is None - - -def test_filter_for_client_excludes_equal_timestamp(): - snapshot = _snapshot({1: 100.0, 2: 200.0}) - result = filter_for_client(snapshot, since=100.0) - assert set(result["runs"].keys()) == {2} - - -def test_filter_for_client_since_above_all_returns_none(): - snapshot = _snapshot({1: 100.0, 2: 200.0}) - assert filter_for_client(snapshot, since=300.0) is None - - -# ----------------------------------------------------------------------------- -# Authorization - - -@pytest.mark.asyncio -async def test_latest_data_unauthorized(graphql_schema_no_auth, current_timestamp): - gen = await graphql_schema_no_auth.subscribe( - """ - subscription { - latest_data(database: { proposal: "999999" }, timestamp: 0) - } - """, - ) - # Subscription permission failures surface as a PreExecutionError on the - # first iteration rather than immediately from subscribe(). - first = await gen.__anext__() - assert first.errors is not None - assert first.errors[0].message == "Authentication required." - - -@pytest.mark.asyncio -async def test_latest_data_forbidden(graphql_schema_authenticated_non_member): - gen = await graphql_schema_authenticated_non_member.subscribe( - f""" - subscription {{ - latest_data(database: {{ proposal: "{PROPOSAL}" }}, timestamp: 0) - }} - """, - ) - # Subscription permission failures surface as a PreExecutionError on the - # first iteration rather than immediately from subscribe(). - first = await gen.__anext__() - assert first.errors is not None - assert first.errors[0].message == "Access to this proposal is forbidden." + assert filter_for_client(None, since=500.0) is None diff --git a/api/tests/graphql/test_utils.py b/api/tests/graphql/test_utils.py deleted file mode 100644 index 049bc880..00000000 --- a/api/tests/graphql/test_utils.py +++ /dev/null @@ -1,53 +0,0 @@ -from damnit_api.graphql.utils import LatestData - -from .const import EXAMPLE_DATA, NEW_DATA, get_values - - -def to_row(values, run=1, timestamp=1): - return [ - {"run": run, "name": name, "value": value, "timestamp": timestamp} - for name, value in values.items() - ] - - -def test_latest_data_update_run(): - example_values = get_values(EXAMPLE_DATA) - new_values = get_values(NEW_DATA) - - first = to_row(example_values) - second = to_row(new_values, timestamp=2) - - latest_data = LatestData.from_list(first + second) - assert len(latest_data.runs) == 1 - - run, variables = next(iter(latest_data.runs.items())) - assert run == 1 - - updated_values = {**example_values, **new_values} - assert variables.keys() == updated_values.keys() - for name, data in variables.items(): - assert data.value == updated_values[name] - assert data.timestamp == (2 if name in new_values else 1) - - -def test_latest_data_multiple_runs(): - example_values = get_values(EXAMPLE_DATA) - new_values = get_values(NEW_DATA) - - first = to_row(example_values, run=1) - second = to_row(new_values, run=2) - - latest_data = LatestData.from_list(first + second) - assert list(latest_data.runs.keys()) == [1, 2] - - run_1 = latest_data.runs[1] - assert run_1.keys() == example_values.keys() - for name, data in run_1.items(): - assert data.value == example_values[name] - assert data.timestamp == 1 - - run_2 = latest_data.runs[2] - assert run_2.keys() == new_values.keys() - for name, data in run_2.items(): - assert data.value == new_values[name] - assert data.timestamp == 1 diff --git a/api/tests/refactor/conftest.py b/api/tests/refactor/conftest.py index cdf5dd2f..5d9e8c98 100644 --- a/api/tests/refactor/conftest.py +++ b/api/tests/refactor/conftest.py @@ -6,14 +6,11 @@ from ..graphql.conftest import ( # noqa: F401 bypass_proposal_permission, + csv_fixture_dir, graphql_context, graphql_schema, graphql_schema_no_auth, + mock_repositories, mocked_ensure_damnit_path, - mocked_metadata_all_tags, - mocked_metadata_column, - mocked_metadata_max, - mocked_metadata_variable_tags, - mocked_metadata_variables, reset_caches, ) diff --git a/api/tests/refactor/test_gql_parity.py b/api/tests/refactor/test_gql_parity.py index a46537f9..ad9d9840 100644 --- a/api/tests/refactor/test_gql_parity.py +++ b/api/tests/refactor/test_gql_parity.py @@ -6,12 +6,11 @@ `Schema` object, not through the FastAPI transport), so it keeps working once the transport underneath is swapped. - The wire-shape of a few representative queries/subscriptions, so a resolver rewrite - that changes response *shape* (not just SDL) is caught too. + that changes response *shape* (not just SDL) is caught too. These run against the + CSV-backed repository from `tests/graphql/conftest.py` (fixture run 348). """ -import asyncio import os -from datetime import UTC, datetime from pathlib import Path import pytest @@ -20,21 +19,14 @@ from damnit_api.graphql import directives as gql_directives from damnit_api.runs import types as run_types -from damnit_api.runs.types import DamnitRun from damnit_api.shared.gql import Query, Subscription -from ..graphql.const import ( - EXAMPLE_DATA, - EXAMPLE_VARIABLES, - KNOWN_DATA, - NEW_DATA, - PROPOSAL, - get_values, -) -from ..graphql.utils import create_run_variables +from ..graphql.const import PROPOSAL SNAPSHOT_PATH = Path(__file__).parent / "snapshots" / "schema.graphql" -NEW_RUN = 400 + +FIXTURE_RUN = 348 +FIXTURE_VARIABLES = {"n_trains", "run_length", "xgm_intensity"} # ----------------------------------------------------------------------------- @@ -78,47 +70,16 @@ def test_public_sdl_unchanged(full_schema): # ----------------------------------------------------------------------------- -# Wire-shape parity: the exact JSON structure the frontend consumes. -# -# These use `graphql_schema` (reused from tests/graphql/conftest.py), unlike the SDL -# parity above, these assert on *values* too, since the wire shape (key names, nesting, -# ms-timestamps) is what a resolver rewrite could silently change without touching the -# SDL at all - - -@pytest.fixture -def mocked_fetch_variables(mocker): - values = get_values(EXAMPLE_DATA) - wrapped = { - "proposal": {"value": values["proposal"]}, - "run": {"value": values["run"]}, - **{ - name: {"value": value, "summary_type": None} - for name, value in values.items() - if name not in ("proposal", "run") - }, - } - return mocker.patch( - "damnit_api.graphql.queries.fetch_variables", - return_value=[wrapped], - ) - - -@pytest.fixture -def mocked_fetch_info(mocker): - return mocker.patch( - "damnit_api.graphql.queries.fetch_info", - return_value=[get_values(KNOWN_DATA)], - ) +# Wire-shape parity: the exact JSON structure the frontend consumes (key names, +# nesting, ms-timestamps) - what a resolver rewrite could change without +# touching the SDL. Exercised against the CSV-backed repository. @pytest.mark.asyncio -async def test_runs_query_wire_shape_unchanged( - graphql_schema, mocked_fetch_variables, mocked_fetch_info -): +async def test_runs_query_wire_shape_unchanged(graphql_schema): query = f""" query {{ - runs(database: {{proposal: "{PROPOSAL}"}}, per_page: 1) {{ + runs(database: {{proposal: {PROPOSAL}}}, per_page: 1) {{ variables {{ name value @@ -137,12 +98,11 @@ async def test_runs_query_wire_shape_unchanged( assert set(runs[0].keys()) == {"variables"} variables = {v["name"]: v for v in runs[0]["variables"]} - assert set(variables) >= {"proposal", "run", "n_trains", "start_time"} + assert set(variables) >= {"proposal", "run", "start_time"} | FIXTURE_VARIABLES for variable in variables.values(): assert set(variable.keys()) == {"name", "value", "dtype"} - # `start_time` is a timestamp: the frontend expects milliseconds - assert variables["start_time"]["value"] == KNOWN_DATA["start_time"].damnit_value + # `start_time` is a timestamp variable. assert variables["start_time"]["dtype"] == "timestamp" @@ -155,7 +115,7 @@ async def test_metadata_query_wire_shape_unchanged(graphql_schema): """ result = await graphql_schema.execute( query, - variable_values={"proposal": str(PROPOSAL)}, + variable_values={"proposal": PROPOSAL}, ) assert result.errors is None @@ -163,95 +123,40 @@ async def test_metadata_query_wire_shape_unchanged(graphql_schema): metadata = result.data["metadata"] assert set(metadata.keys()) == {"runs", "variables", "timestamp", "tags"} assert isinstance(metadata["runs"], list) - assert metadata["variables"] == { - **DamnitRun.known_variables(), - **EXAMPLE_VARIABLES, - } + assert isinstance(metadata["variables"], dict) + for variable in metadata["variables"].values(): + assert set(variable.keys()) == {"name", "title", "tags"} assert isinstance(metadata["timestamp"], int | float) -@pytest.fixture -def current_timestamp(): - return datetime.now(tz=UTC).timestamp() - - -@pytest.fixture -def mocked_latest_rows(mocker, current_timestamp): - table_sentinel = mocker.sentinel.run_variables_table - mocker.patch( - "damnit_api.graphql.subscriptions.async_table", - return_value=table_sentinel, - ) - mocker.patch( - "damnit_api.graphql.subscriptions.async_max", - return_value=0, +@pytest.mark.asyncio +async def test_latest_data_subscription_wire_shape_unchanged(mock_repositories): + """The subscription's client payload shape, exercised through the same + poll_proposal/filter_for_client path the resolver drives.""" + from damnit_api.graphql.subscriptions import ( + SubscriptionCursors, + filter_for_client, + poll_proposal, ) + from damnit_api.shared.models import ProposalNumber - def mocked_returns(*args, table, **kwargs): - if table is table_sentinel: - return create_run_variables( - get_values(NEW_DATA), - proposal=PROPOSAL, - run=NEW_RUN, - timestamp=current_timestamp, - ) - return None - - return mocker.patch( - "damnit_api.graphql.subscriptions.async_latest_rows", - side_effect=mocked_returns, - ) + proposal = ProposalNumber(PROPOSAL) + cursors = SubscriptionCursors() + cursors[proposal] = 0.0 # seed below the fixture timestamps so a run surfaces + repo = mock_repositories.get(proposal) + snapshot = await poll_proposal(proposal, cursors, repo) + result = filter_for_client(snapshot, since=500.0) -@pytest.fixture -def mocked_subscription_fetch_info(mocker): - return mocker.patch( - "damnit_api.graphql.subscriptions.fetch_info", - return_value=[{**get_values(KNOWN_DATA), "run": NEW_RUN}], - ) + assert result is not None + assert set(result.keys()) == {"runs", "metadata"} + runs = result["runs"] + assert FIXTURE_RUN in runs + for variable in runs[FIXTURE_RUN].values(): + assert set(variable.keys()) == {"value", "dtype"} -@pytest.mark.asyncio -async def test_latest_data_subscription_wire_shape_unchanged( - graphql_schema, - current_timestamp, - mocked_latest_rows, - mocked_subscription_fetch_info, -): - subscription = await graphql_schema.subscribe( - """ - subscription( - $proposal: ProposalNo!, - $timestamp: Timestamp!) { - latest_data( - database: { proposal: $proposal }, - timestamp: $timestamp - ) - } - """, - variable_values={ - "proposal": str(PROPOSAL), - "timestamp": (current_timestamp - 1) * 1000, - }, - ) - - try: - result = await asyncio.wait_for(anext(subscription), timeout=2) - assert not result.errors - - payload = result.data["latest_data"] - assert set(payload.keys()) == {"runs", "metadata"} - - runs = payload["runs"] - assert set(runs.keys()) == {NEW_RUN} - run_data = runs[NEW_RUN] - assert set(run_data.keys()) >= {"n_trains", "run_length", "xgm_intensity"} - for variable in run_data.values(): - assert set(variable.keys()) == {"value", "dtype"} - - metadata = payload["metadata"] - assert set(metadata.keys()) == {"runs", "timestamp", "variables"} - # ms-timestamp, matching the `metadata` query's serialization - assert metadata["timestamp"] == current_timestamp * 1000 - finally: - await subscription.aclose() + metadata = result["metadata"] + assert set(metadata.keys()) == {"runs", "timestamp", "variables"} + # ms-timestamp, matching the `metadata` query's serialization. + assert isinstance(metadata["timestamp"], int | float) diff --git a/api/tests/runs/__init__.py b/api/tests/runs/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/api/tests/runs/csv/__init__.py b/api/tests/runs/csv/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/api/tests/runs/csv/test_repository.py b/api/tests/runs/csv/test_repository.py new file mode 100644 index 00000000..e38a9117 --- /dev/null +++ b/api/tests/runs/csv/test_repository.py @@ -0,0 +1,268 @@ +"""Tests for CsvDamnitRepository.""" + +from __future__ import annotations + +import pytest + +from damnit_api.runs.csv import CsvDamnitRepository +from damnit_api.runs.models import KNOWN_VARIABLES +from damnit_api.shared.models import ProposalNumber + +_TEST_PROPOSAL = ProposalNumber(999997) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def csv_dir(tmp_path): + """Populate tmp_path with a representative set of CSV fixture files.""" + (tmp_path / "runs.csv").write_text( + "run,start_time,added_at\n1,1000.0,1001.0\n2,1100.0,1101.0\n3,1200.0,1201.0\n", + encoding="utf-8", + ) + (tmp_path / "run_variables.csv").write_text( + "run,name,value,summary_type,timestamp\n" + "1,alpha,a1,,1000.0\n" + "1,beta,b1,,1000.0\n" + "2,alpha,a2,,1100.0\n" + "3,beta,b3,,1200.0\n" + # Extra timestamp for run 1 / alpha (older — should be ignored) + "1,alpha,a1_old,,900.0\n", + encoding="utf-8", + ) + (tmp_path / "variables.csv").write_text( + "name,title,tags\nalpha,Alpha Variable,GroupA\nbeta,Beta Variable,\n", + encoding="utf-8", + ) + return tmp_path + + +@pytest.fixture +def repo(csv_dir): + return CsvDamnitRepository(_TEST_PROPOSAL, csv_dir) + + +# --------------------------------------------------------------------------- +# get_runs +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_runs_returns_all_runs(repo): + records = await repo.get_runs(limit=10, offset=0) + assert [r.run for r in records] == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_get_runs_picks_latest_timestamp(repo): + """run 1 / alpha has two timestamps; only the newer value should appear.""" + records = await repo.get_runs(limit=10, offset=0) + run1 = next(r for r in records if r.run == 1) + assert run1.variables["alpha"].value == "a1" + assert run1.variables["alpha"].timestamp == pytest.approx(1000.0) + + +@pytest.mark.asyncio +async def test_get_runs_pagination(repo): + page1 = await repo.get_runs(limit=2, offset=0) + page2 = await repo.get_runs(limit=2, offset=2) + assert [r.run for r in page1] == [1, 2] + assert [r.run for r in page2] == [3] + + +@pytest.mark.asyncio +async def test_get_runs_filter_by_variable_name(repo): + """Filtering by 'alpha' keeps only variables named alpha; other variables + are absent but all runs are still returned.""" + records = await repo.get_runs(limit=10, offset=0, variable_names=["alpha"]) + assert [r.run for r in records] == [1, 2, 3] + run1 = next(r for r in records if r.run == 1) + assert "alpha" in run1.variables + assert "beta" not in run1.variables + run3 = next(r for r in records if r.run == 3) + assert run3.variables == {} + + +@pytest.mark.asyncio +async def test_get_runs_unknown_variable_name_returns_empty_variables(repo): + records = await repo.get_runs(limit=10, offset=0, variable_names=["nonexistent"]) + assert [r.run for r in records] == [1, 2, 3] + assert all(r.variables == {} for r in records) + + +@pytest.mark.asyncio +async def test_get_runs_includes_run_info(repo): + records = await repo.get_runs(limit=10, offset=0) + run2 = next(r for r in records if r.run == 2) + assert run2.start_time == pytest.approx(1100.0) + assert run2.added_at == pytest.approx(1101.0) + + +@pytest.mark.asyncio +async def test_get_runs_missing_run_variables_csv(tmp_path): + """If run_variables.csv is absent, return an empty list.""" + (tmp_path / "runs.csv").write_text("run,start_time,added_at\n1,1000.0,1001.0\n") + repo = CsvDamnitRepository(_TEST_PROPOSAL, tmp_path) + records = await repo.get_runs(limit=10, offset=0) + assert records == [] + + +# --------------------------------------------------------------------------- +# get_latest_runs +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_latest_runs_returns_rows_after_cutoff(repo): + records = await repo.get_latest_runs(start_at=1050.0) + assert [r.run for r in records] == [2, 3] + + +@pytest.mark.asyncio +async def test_get_latest_runs_future_cutoff_returns_empty(repo): + records = await repo.get_latest_runs(start_at=9999.0) + assert records == [] + + +@pytest.mark.asyncio +async def test_get_latest_runs_deduplicates_per_variable(repo): + """run 1 / alpha has timestamps 900 and 1000; start_at=0 includes both + rows but keeps only the latest value.""" + records = await repo.get_latest_runs(start_at=0.0) + run1 = next(r for r in records if r.run == 1) + assert run1.variables["alpha"].timestamp == pytest.approx(1000.0) + assert run1.variables["alpha"].value == "a1" + + +@pytest.mark.asyncio +async def test_get_latest_runs_includes_run_info(repo): + records = await repo.get_latest_runs(start_at=0.0) + run3 = next(r for r in records if r.run == 3) + assert run3.start_time == pytest.approx(1200.0) + assert run3.added_at == pytest.approx(1201.0) + + +@pytest.mark.asyncio +async def test_get_latest_runs_start_at_none_returns_empty(repo): + """start_at=None substitutes current time, so old rows are excluded.""" + records = await repo.get_latest_runs(start_at=None) + assert records == [] + + +@pytest.mark.asyncio +async def test_get_latest_runs_missing_csv(tmp_path): + repo = CsvDamnitRepository(_TEST_PROPOSAL, tmp_path) + records = await repo.get_latest_runs(start_at=0.0) + assert records == [] + + +# --------------------------------------------------------------------------- +# get_metadata +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_metadata_runs_list(repo): + snap = await repo.get_metadata() + assert snap.runs == (1, 2, 3) + + +@pytest.mark.asyncio +async def test_get_metadata_includes_known_variables(repo): + snap = await repo.get_metadata() + known_names = {v.name for v in KNOWN_VARIABLES} + assert known_names.issubset(snap.variables.keys()) + + +@pytest.mark.asyncio +async def test_get_metadata_db_variables_have_titles(repo): + snap = await repo.get_metadata() + assert snap.variables["alpha"].title == "Alpha Variable" + assert snap.variables["beta"].title == "Beta Variable" + + +@pytest.mark.asyncio +async def test_get_metadata_tags(repo): + snap = await repo.get_metadata() + assert "(Untagged)" in snap.tags + assert "GroupA" in snap.tags + assert "alpha" in snap.tags["GroupA"].variables + + +@pytest.mark.asyncio +async def test_get_metadata_untagged_contains_beta(repo): + snap = await repo.get_metadata() + assert "beta" in snap.tags["(Untagged)"].variables + + +@pytest.mark.asyncio +async def test_get_metadata_timestamp(repo): + snap = await repo.get_metadata() + assert snap.timestamp == pytest.approx(1200.0) + + +@pytest.mark.asyncio +async def test_get_metadata_missing_variables_csv(tmp_path): + """Without variables.csv only known variables appear; no named tags.""" + (tmp_path / "runs.csv").write_text( + "run,start_time,added_at\n1,1000.0,1001.0\n", encoding="utf-8" + ) + (tmp_path / "run_variables.csv").write_text( + "run,name,value,summary_type,timestamp\n1,alpha,a1,,1000.0\n", + encoding="utf-8", + ) + repo = CsvDamnitRepository(_TEST_PROPOSAL, tmp_path) + snap = await repo.get_metadata() + known_names = {v.name for v in KNOWN_VARIABLES} + assert known_names.issubset(snap.variables.keys()) + assert "alpha" not in snap.variables + assert list(snap.tags.keys()) == ["(Untagged)"] + + +@pytest.mark.asyncio +async def test_get_metadata_empty_runs_csv(tmp_path): + """Empty runs.csv gives an empty runs list and zero timestamp.""" + (tmp_path / "runs.csv").write_text("run,start_time,added_at\n", encoding="utf-8") + repo = CsvDamnitRepository(_TEST_PROPOSAL, tmp_path) + snap = await repo.get_metadata() + assert snap.runs == () + assert snap.timestamp == pytest.approx(0.0) + + +# --------------------------------------------------------------------------- +# get_extracted_data +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_extracted_data_returns_none(repo): + result = await repo.get_extracted_data(run=1, variable="alpha") + assert result is None + + +# --------------------------------------------------------------------------- +# Multi-tag (T4) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_metadata_multi_tag_variable(tmp_path): + """Semicolon-separated tags in a variables.csv row produce multiple tags.""" + (tmp_path / "runs.csv").write_text( + "run,start_time,added_at\n1,1.0,2.0\n", encoding="utf-8" + ) + (tmp_path / "run_variables.csv").write_text( + "run,name,value,summary_type,timestamp\n1,gamma,g1,,1.0\n", encoding="utf-8" + ) + (tmp_path / "variables.csv").write_text( + "name,title,tags\ngamma,Gamma Variable,TagA;TagB\n", encoding="utf-8" + ) + repo = CsvDamnitRepository(_TEST_PROPOSAL, tmp_path) + snap = await repo.get_metadata() + assert "TagA" in snap.tags + assert "TagB" in snap.tags + assert "gamma" in snap.tags["TagA"].variables + assert "gamma" in snap.tags["TagB"].variables diff --git a/api/tests/runs/sqlite/__init__.py b/api/tests/runs/sqlite/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/api/tests/runs/sqlite/test_repository.py b/api/tests/runs/sqlite/test_repository.py new file mode 100644 index 00000000..b8a4b28e --- /dev/null +++ b/api/tests/runs/sqlite/test_repository.py @@ -0,0 +1,184 @@ +"""Integration tests for SQLiteDamnitRepository against a real on-disk DB. + +The repository resolves its path through `DatabaseSessionManager` -> +`get_damnit_path`; the fixtures mock that to point at a tmp `runs.sqlite`. +""" + +import sqlite3 + +import pytest +import pytest_asyncio + +from damnit_api.runs.models import KNOWN_VARIABLES +from damnit_api.runs.sqlite.repository import SQLiteDamnitRepository +from damnit_api.shared.models import ProposalNumber + +_TEST_PROPOSAL = ProposalNumber(999998) + +_SCHEMA = """ +CREATE TABLE run_variables ( + proposal INTEGER NOT NULL, + run INTEGER NOT NULL, + name TEXT NOT NULL, + value BLOB, + summary_type TEXT, + attributes BLOB, + timestamp REAL NOT NULL, + PRIMARY KEY (proposal, run, name, timestamp) +); +CREATE TABLE run_info (run INTEGER PRIMARY KEY, start_time REAL, added_at REAL); +CREATE TABLE variables (name TEXT PRIMARY KEY, title TEXT); +CREATE TABLE tags (id INTEGER PRIMARY KEY, name TEXT NOT NULL); +CREATE TABLE variable_tags ( + variable_name TEXT NOT NULL, tag_id INTEGER NOT NULL, + PRIMARY KEY (variable_name, tag_id) +); +CREATE TABLE metameta (key TEXT PRIMARY KEY, value TEXT); +""" + + +def _seed(db_file): + conn = sqlite3.connect(str(db_file)) + try: + conn.executescript(_SCHEMA) + p = int(_TEST_PROPOSAL) + conn.executemany( + "INSERT INTO run_variables" + " (proposal, run, name, value, summary_type, timestamp)" + " VALUES (?, ?, ?, ?, ?, ?)", + [ + (p, 1, "alpha", "a1", None, 1000.0), + (p, 1, "beta", "b1", None, 1000.0), + (p, 2, "alpha", "a2", None, 1100.0), + (p, 3, "beta", "b3", None, 1200.0), + # Older timestamp for run 1 / alpha - must be ignored. + (p, 1, "alpha", "a1_old", None, 900.0), + ], + ) + conn.executemany( + "INSERT INTO run_info (run, start_time, added_at) VALUES (?, ?, ?)", + [(1, 1000.0, 1001.0), (2, 1100.0, 1101.0), (3, 1200.0, 1201.0)], + ) + conn.executemany( + "INSERT INTO variables (name, title) VALUES (?, ?)", + [("alpha", "Alpha Variable"), ("beta", "Beta Variable")], + ) + conn.execute("INSERT INTO tags (id, name) VALUES (1, 'GroupA')") + conn.execute( + "INSERT INTO variable_tags (variable_name, tag_id) VALUES ('alpha', 1)" + ) + conn.execute( + "INSERT INTO metameta (key, value) VALUES ('proposal', ?)", (str(p),) + ) + conn.commit() + finally: + conn.close() + + +@pytest_asyncio.fixture +async def repo(mocker, tmp_path): + _seed(tmp_path / "runs.sqlite") + mocker.patch( + "damnit_api.runs.sqlite.session.get_damnit_path", + return_value=str(tmp_path), + ) + instance = SQLiteDamnitRepository(_TEST_PROPOSAL) + yield instance + await instance._db.close() + + +# ----------------------------------------------------------------------------- +# get_runs + + +@pytest.mark.asyncio +async def test_get_runs_returns_all_runs(repo): + records = await repo.get_runs(limit=10, offset=0) + assert [r.run for r in records] == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_get_runs_picks_latest_timestamp(repo): + records = await repo.get_runs(limit=10, offset=0) + run1 = next(r for r in records if r.run == 1) + assert run1.variables["alpha"].value == "a1" + assert run1.variables["alpha"].timestamp == pytest.approx(1000.0) + + +@pytest.mark.asyncio +async def test_get_runs_pagination(repo): + page1 = await repo.get_runs(limit=2, offset=0) + page2 = await repo.get_runs(limit=2, offset=2) + assert [r.run for r in page1] == [1, 2] + assert [r.run for r in page2] == [3] + + +@pytest.mark.asyncio +async def test_get_runs_name_filter(repo): + records = await repo.get_runs(limit=10, offset=0, variable_names=["alpha"]) + run1 = next(r for r in records if r.run == 1) + assert set(run1.variables) == {"alpha"} + + +@pytest.mark.asyncio +async def test_get_runs_includes_run_info(repo): + records = await repo.get_runs(limit=10, offset=0) + run2 = next(r for r in records if r.run == 2) + assert run2.start_time == pytest.approx(1100.0) + assert run2.added_at == pytest.approx(1101.0) + + +# ----------------------------------------------------------------------------- +# get_latest_runs + + +@pytest.mark.asyncio +async def test_get_latest_runs_after_cutoff(repo): + records = await repo.get_latest_runs(start_at=1050.0) + assert [r.run for r in records] == [2, 3] + + +@pytest.mark.asyncio +async def test_get_latest_runs_future_cutoff_empty(repo): + assert await repo.get_latest_runs(start_at=9999.0) == [] + + +# ----------------------------------------------------------------------------- +# get_metadata + + +@pytest.mark.asyncio +async def test_get_metadata_runs_and_timestamp(repo): + snap = await repo.get_metadata() + assert snap.runs == (1, 2, 3) + assert snap.timestamp == pytest.approx(1200.0) + + +@pytest.mark.asyncio +async def test_get_metadata_variables_and_tags(repo): + snap = await repo.get_metadata() + known = {v.name for v in KNOWN_VARIABLES} + assert known <= set(snap.variables) + assert snap.variables["alpha"].title == "Alpha Variable" + assert "GroupA" in snap.tags + assert "alpha" in snap.tags["GroupA"].variables + assert "beta" in snap.tags["(Untagged)"].variables + + +@pytest.mark.asyncio +async def test_get_metadata_ttl_cache(repo): + """Within the TTL, get_metadata returns the same cached object; invalidating + forces a fresh snapshot.""" + first = await repo.get_metadata() + assert await repo.get_metadata() is first + repo.invalidate_metadata_cache() + assert await repo.get_metadata() is not first + + +# ----------------------------------------------------------------------------- +# get_proposal_number (local-mode metameta lookup) + + +@pytest.mark.asyncio +async def test_get_proposal_number_reads_metameta(repo): + assert await repo.get_proposal_number() == str(int(_TEST_PROPOSAL)) diff --git a/api/tests/test_db.py b/api/tests/runs/sqlite/test_session.py similarity index 83% rename from api/tests/test_db.py rename to api/tests/runs/sqlite/test_session.py index 601b307b..6ecbad5a 100644 --- a/api/tests/test_db.py +++ b/api/tests/runs/sqlite/test_session.py @@ -13,14 +13,13 @@ from pathlib import Path import pytest +from sqlalchemy import text from sqlalchemy.pool import NullPool from damnit_api.runs.sqlite import ( DAMNIT_PATH, DatabaseSessionManager, - async_table, get_damnit_path, - get_session, ) from damnit_api.shared.errors import ProposalNotFoundError from damnit_api.shared.models import ProposalNumber @@ -78,6 +77,7 @@ def _open_file_descriptors_to(db_file: Path): def test_engine_uses_nullpool_and_autocommit(damnit_db): mgr = DatabaseSessionManager(damnit_db) + assert mgr._engine is not None assert isinstance(mgr._engine.pool, NullPool) # Private attribute: the public get_execution_options() does not # surface the engine-level isolation_level for async engines. @@ -88,10 +88,6 @@ def test_engine_uses_nullpool_and_autocommit(damnit_db): # File descriptor lifetime -# asyncio.run() creates and tears down a fresh event loop, which is what -# this test verifies (no file descriptors leak after the loop dies). -# alru_cached async_table sees that loop change; warning is intrinsic. -@pytest.mark.filterwarnings("ignore::async_lru.AlruCacheLoopResetWarning") def test_get_damnit_path_raises_when_proposal_not_found(mocker): """A proposal with no resolvable directory raises ProposalNotFoundError.""" mocker.patch("damnit_api.runs.sqlite.session.find_proposal", return_value="") @@ -102,13 +98,16 @@ def test_get_damnit_path_raises_when_proposal_not_found(mocker): get_damnit_path(_TEST_PROPOSAL) -def test_no_lingering_file_descriptor_after_read(damnit_db, damnit_registry, tmp_path): +# asyncio.run() creates and tears down a fresh event loop; this test verifies +# no file descriptors leak after the loop dies (NullPool disposes per checkout). +def test_no_lingering_file_descriptor_after_read(damnit_db, tmp_path): db_file = tmp_path / DAMNIT_PATH / "runs.sqlite" async def do_read(): - table = await async_table(damnit_registry, damnit_db, name="runs") - async with get_session(damnit_registry, damnit_db) as session: - await session.execute(table.select()) + manager = DatabaseSessionManager(damnit_db) + async with manager.session() as session: + await session.execute(text("SELECT * FROM runs")) + await manager.close() assert _open_file_descriptors_to(db_file) == [] asyncio.run(do_read()) diff --git a/api/tests/test_data.py b/api/tests/runs/test_preview.py similarity index 100% rename from api/tests/test_data.py rename to api/tests/runs/test_preview.py diff --git a/api/tests/test_state.py b/api/tests/test_state.py index a0b7386f..59931daf 100644 --- a/api/tests/test_state.py +++ b/api/tests/test_state.py @@ -4,7 +4,7 @@ from pathlib import Path from damnit_api.auth.token_store import InMemoryTokenStore -from damnit_api.runs.sqlite import DamnitDBRegistry +from damnit_api.runs.repository import DamnitRepositoryRegistry from damnit_api.shared.models import ProposalNumber from damnit_api.shared.settings import Settings from damnit_api.state import create_oauth_client @@ -39,19 +39,15 @@ def test_create_oauth_client_returns_none_when_auth_disabled(tmp_path): assert create_oauth_client(settings) is None -def test_registry_memoizes_managers_per_proposal(monkeypatch): +def test_repository_registry_memoizes_per_proposal(): created = [] - class DummyManager: + class DummyRepo: def __init__(self, proposal): self.proposal = proposal created.append(proposal) - monkeypatch.setattr( - "damnit_api.runs.sqlite.session.DatabaseSessionManager", DummyManager - ) - - registry = DamnitDBRegistry() + registry = DamnitRepositoryRegistry(DummyRepo) # ty: ignore[invalid-argument-type] first = registry.get(ProposalNumber(1234)) assert registry.get(ProposalNumber(1234)) is first assert registry.get(ProposalNumber(5678)) is not first