test(db): raise interloper-db coverage from 87% to 98%

packages/interloper-db/tests/models/test_columns.py

@@ -0,0 +1,52 @@
+"""Tests for ``interloper_db.models.columns``.
+
+The column primitives carry the Postgres/SQLite portability the whole
+schema depends on, so the variance is asserted directly rather than
+inferred from a table that happens to work.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import JSON
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import dialect as postgresql_dialect
+from sqlalchemy.dialects.sqlite import dialect as sqlite_dialect
+
+from interloper_db.models.columns import PortableJSON, TZDateTime, timestamp_column
+
+
+def test_datetimes_are_timezone_aware() -> None:
+    """TIMESTAMPTZ, so SQLAlchemy hands back aware values rather than naive ones."""
+    assert TZDateTime.timezone is True
+
+
+def test_json_is_jsonb_on_postgres() -> None:
+    """JSONB where it exists, for indexing and containment operators."""
+    assert isinstance(PortableJSON.dialect_impl(postgresql_dialect()), JSONB)
+
+
+def test_json_falls_back_to_plain_json_elsewhere() -> None:
+    """The in-memory SQLite test databases get plain JSON."""
+    assert not isinstance(PortableJSON.dialect_impl(sqlite_dialect()), JSONB)
+    assert isinstance(PortableJSON, JSON)
+
+
+class TestTimestampColumn:
+    """The nullable, insert-stamped timestamp every table reuses."""
+
+    def test_it_defaults_to_the_insert_time(self) -> None:
+        # CURRENT_TIMESTAMP rather than now(), so SQLite works identically.
+        column = timestamp_column().sa_column
+
+        assert column.server_default is not None
+        assert "CURRENT_TIMESTAMP" in str(column.server_default.arg)
+
+    def test_it_is_a_timezone_aware_datetime(self) -> None:
+        column = timestamp_column().sa_column
+
+        assert column.type.timezone is True
+
+    def test_extra_kwargs_reach_the_column(self) -> None:
+        column = timestamp_column(index=True).sa_column
+
+        assert column.index is True

packages/interloper-db/tests/models/test_components.py

@@ -0,0 +1,99 @@
+"""Tests for the behaviour on ``interloper_db.models.components``.
+
+The row carries two pieces of logic beyond its declarations: resolving an
+owned asset's parent key, and merging machine-owned state without touching
+the spec.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+from uuid import uuid4
+
+import pydantic
+import pytest
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from interloper_db.models import Component
+
+_ORG = uuid4()
+
+
+class TestParentKey:
+    """A source-owned asset needs its parent's catalog key to resolve."""
+
+    def test_a_parentless_row_has_none(self, component_db: Engine):
+        row = Component(org_id=_ORG, kind="source", key="demo_source")
+
+        with Session(component_db) as session:
+            assert row.parent_key(session) is None
+
+    def test_an_owned_asset_reports_its_sources_key(self, component_db: Engine):
+        with Session(component_db) as session:
+            source = Component(org_id=_ORG, kind="source", key="demo_source")
+            session.add(source)
+            session.commit()
+            asset = Component(org_id=_ORG, kind="asset", key="a", parent_id=source.id)
+            session.add(asset)
+            session.commit()
+
+            assert asset.parent_key(session) == "demo_source"
+
+    def test_a_vanished_parent_reports_none(self, component_db: Engine):
+        # Defensive: the foreign key makes this unreachable in practice.
+        row = Component(org_id=_ORG, kind="asset", key="a", parent_id=uuid4())
+
+        with Session(component_db) as session:
+            assert row.parent_key(session) is None
+
+
+class TestStampState:
+    """State is machine-owned and merges over what is already there."""
+
+    def test_it_merges_rather_than_replaces(self):
+        row = Component(org_id=_ORG, kind="job", key="cron_job", state={"next_run_at": "x"})
+
+        row.stamp_state(last_run_at="y")
+
+        assert row.state == {"next_run_at": "x", "last_run_at": "y"}
+
+    def test_a_later_stamp_wins(self):
+        row = Component(org_id=_ORG, kind="job", key="cron_job", state={"next_run_at": "x"})
+
+        row.stamp_state(next_run_at="y")
+
+        assert row.state == {"next_run_at": "y"}
+
+    def test_datetimes_are_stored_as_aware_iso_strings(self):
+        # Lexicographic comparison in SQL then stays chronological.
+        row = Component(org_id=_ORG, kind="job", key="cron_job")
+        stamp = dt.datetime(2026, 6, 1, 12, 30, tzinfo=dt.timezone.utc)
+
+        row.stamp_state(next_run_at=stamp)
+
+        assert row.state is not None
+        stored = row.state["next_run_at"]
+        assert isinstance(stored, str)
+        assert dt.datetime.fromisoformat(stored) == stamp
+
+    def test_none_clears_a_field(self):
+        row = Component(org_id=_ORG, kind="job", key="cron_job", state={"next_run_at": "x"})
+
+        row.stamp_state(next_run_at=None)
+
+        assert row.state is not None
+        assert row.state["next_run_at"] is None
+
+    def test_the_spec_is_untouched(self):
+        row = Component(org_id=_ORG, kind="job", key="cron_job", config={"cron": "0 2 * * *"})
+
+        row.stamp_state(next_run_at=None)
+
+        assert row.config == {"cron": "0 2 * * *"}
+
+    def test_a_shape_the_kind_does_not_declare_is_rejected(self):
+        row = Component(org_id=_ORG, kind="job", key="cron_job")
+
+        with pytest.raises(pydantic.ValidationError):
+            row.stamp_state(next_run_at=object())

packages/interloper-db/tests/store/quotas/test_base.py

@@ -16,6 +16,7 @@
 from interloper_db.models import Component, Quota, Run
 from interloper_db.store import Store
 from interloper_db.store.quotas import METRIC_SUCCESSFUL_RUNS, UsageLedger
+from interloper_db.store.quotas.definitions import QUOTA_MAX_SUCCESSFUL_RUNS_PER_MONTH
 
 UsageRows = Callable[[], dict[dt.date, tuple[int, int]]]
 RunFactory = Callable[..., Run]
@@ -246,3 +247,40 @@ def test_rejects_unknown_and_negative(self, store: Store, org_id: UUID):
             store.quotas.set_overrides(org_id, {"max_bananas": 1})
         with pytest.raises(ValueError, match=">= 0"):
             store.quotas.set_overrides(org_id, {"max_sources": -1})
+
+
+class TestRunStatus:
+    """``run_status`` reports the committed count against the effective limit."""
+
+    def test_an_unlimited_quota_reads_the_ledger_not_at_all(self, store: Store, org_id: UUID):
+        # No default and no override, so the ledger is never touched.
+        assert store.quotas.run_status(org_id) == (0, None)
+
+    def test_a_limited_quota_reports_the_committed_count(
+        self, store: Store, org_id: UUID, make_run
+    ):
+        store.quotas.set_overrides(org_id, {QUOTA_MAX_SUCCESSFUL_RUNS_PER_MONTH: 5})
+        run = make_run()
+        store.quotas.try_reserve_run(run)
+
+        used, limit = store.quotas.run_status(org_id)
+
+        assert (used, limit) == (1, 5)
+
+
+class TestTryReserveRunUnlimited:
+    """An unlimited quota reserves without writing a ledger row."""
+
+    def test_it_succeeds_without_metering(self, store: Store, make_run, usage_rows):
+        run = make_run()
+
+        assert store.quotas.try_reserve_run(run) is True
+        assert usage_rows() == {}
+
+
+class TestCheckWithLock:
+    """A lock-requiring quota short-circuits when unlimited."""
+
+    def test_an_unlimited_lock_quota_passes_without_a_session(self, store: Store, org_id: UUID):
+        # effective_limit returns None, so the gate returns before opening one.
+        store.quotas.check(org_id, QUOTA_MAX_SUCCESSFUL_RUNS_PER_MONTH, subject="run")