diff --git a/packages/interloper-db/tests/models/test_columns.py b/packages/interloper-db/tests/models/test_columns.py new file mode 100644 index 000000000..b8edd1e0d --- /dev/null +++ b/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 diff --git a/packages/interloper-db/tests/models/test_components.py b/packages/interloper-db/tests/models/test_components.py new file mode 100644 index 000000000..73490ef2f --- /dev/null +++ b/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()) diff --git a/packages/interloper-db/tests/store/quotas/test_base.py b/packages/interloper-db/tests/store/quotas/test_base.py index e02cc2dd8..032546a2a 100644 --- a/packages/interloper-db/tests/store/quotas/test_base.py +++ b/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") diff --git a/packages/interloper-db/tests/store/test_auth.py b/packages/interloper-db/tests/store/test_auth.py index 31d1ac9f5..aedf30bba 100644 --- a/packages/interloper-db/tests/store/test_auth.py +++ b/packages/interloper-db/tests/store/test_auth.py @@ -2,11 +2,14 @@ from __future__ import annotations +from datetime import datetime, timedelta, timezone from uuid import uuid4 import pytest from interloper.errors import NotFoundError +from sqlmodel import Session, select +from interloper_db.models import AuthSession from interloper_db.store import Store @@ -99,3 +102,198 @@ def test_login_upsert_fills_an_empty_name(self, store: Store): filled = store.auth.upsert_profile(google_id="g-fill", email="fill@example.com", name="Google Name") assert filled.name == "Google Name" + + +class TestSetSuperAdmin: + """Platform-wide privilege, toggled by id.""" + + def test_promotes_a_profile(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + + promoted = store.auth.set_super_admin(profile.id) + + assert promoted.is_super_admin is True + assert store.auth.get_profile(profile.id).is_super_admin is True + + def test_demotes_when_asked(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + store.auth.set_super_admin(profile.id) + + store.auth.set_super_admin(profile.id, is_super_admin=False) + + assert store.auth.get_profile(profile.id).is_super_admin is False + + def test_a_missing_profile_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Profile {missing} not found"): + store.auth.set_super_admin(missing) + + +class TestUpsertProfileAvatar: + """The login upsert refreshes Google-managed fields.""" + + def test_a_new_avatar_replaces_the_old_one(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x", avatar_url="https://a/1.png") + + store.auth.upsert_profile(google_id="g1", email="ada@x", avatar_url="https://a/2.png") + + assert store.auth.get_profile(profile.id).avatar_url == "https://a/2.png" + + def test_an_omitted_avatar_leaves_the_stored_one(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x", avatar_url="https://a/1.png") + + store.auth.upsert_profile(google_id="g1", email="ada@x") + + assert store.auth.get_profile(profile.id).avatar_url == "https://a/1.png" + + def test_the_email_is_refreshed_on_every_login(self, store: Store): + # Google owns the address; a change there must land. + profile = store.auth.upsert_profile(google_id="g1", email="old@x") + + store.auth.upsert_profile(google_id="g1", email="new@x") + + assert store.auth.get_profile(profile.id).email == "new@x" + + +class TestSessions: + """Session creation, resolution, expiry and the active-organisation pointer.""" + + def test_a_created_session_resolves_to_its_profile(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + + token = store.auth.create_session(profile.id) + resolved = store.auth.resolve_session(token) + + assert resolved is not None + found, session_row = resolved + assert found.id == profile.id + assert session_row.user_id == profile.id + + def test_the_raw_token_is_not_stored(self, store: Store): + # Only its hash is persisted, so a database leak yields no usable session. + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + + token = store.auth.create_session(profile.id) + resolved = store.auth.resolve_session(token) + + assert resolved is not None + assert resolved[1].token_hash != token + + def test_an_unknown_token_resolves_to_nothing(self, store: Store): + assert store.auth.resolve_session("never-issued") is None + + def test_an_expired_session_resolves_to_nothing_and_is_swept(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + token = store.auth.create_session(profile.id) + with Session(store.engine) as session: + row = session.exec(select(AuthSession)).one() + row.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + session.add(row) + session.commit() + + assert store.auth.resolve_session(token) is None + # The expired row is deleted rather than left to accumulate. + with Session(store.engine) as session: + assert session.exec(select(AuthSession)).all() == [] + + def test_a_session_whose_profile_vanished_resolves_to_nothing(self, store: Store): + # The foreign key makes this unreachable through the store's own API, + # so the orphan is fabricated with constraints off — the guard exists + # for a row that should not be there, and this proves what it does. + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + token = store.auth.create_session(profile.id) + with store.engine.connect() as connection: + connection.exec_driver_sql("PRAGMA foreign_keys=OFF") + connection.exec_driver_sql("DELETE FROM profiles") + connection.commit() + + assert store.auth.resolve_session(token) is None + + def test_a_session_can_be_created_already_scoped(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + + token = store.auth.create_session(profile.id, organisation_id=org.id) + resolved = store.auth.resolve_session(token) + + assert resolved is not None + assert resolved[1].organisation_id == org.id + + +class TestSetSessionOrg: + """Switching the active organisation, and remembering it on the profile.""" + + def test_it_updates_the_session(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + token = store.auth.create_session(profile.id) + + store.auth.set_session_org(token, org.id) + resolved = store.auth.resolve_session(token) + + assert resolved is not None + assert resolved[1].organisation_id == org.id + + def test_a_user_id_also_records_the_preference(self, store: Store): + # So the next login lands in the org the user last used. + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + token = store.auth.create_session(profile.id) + + store.auth.set_session_org(token, org.id, user_id=profile.id) + + assert store.auth.get_profile(profile.id).last_organisation_id == org.id + + def test_without_a_user_id_the_profile_is_untouched(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + token = store.auth.create_session(profile.id) + + store.auth.set_session_org(token, org.id) + + assert store.auth.get_profile(profile.id).last_organisation_id is None + + def test_an_unknown_token_is_a_no_op(self, store: Store): + store.auth.set_session_org("never-issued", uuid4()) + + def test_an_unknown_user_id_leaves_the_session_updated(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + token = store.auth.create_session(profile.id) + + store.auth.set_session_org(token, org.id, user_id=uuid4()) + + resolved = store.auth.resolve_session(token) + assert resolved is not None + assert resolved[1].organisation_id == org.id + + +class TestDeleteUserSessions: + """A logout ends every browser, not just the one that asked.""" + + def test_all_of_the_users_sessions_go(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + first = store.auth.create_session(profile.id) + second = store.auth.create_session(profile.id) + + store.auth.delete_user_sessions(profile.id) + + assert store.auth.resolve_session(first) is None + assert store.auth.resolve_session(second) is None + + def test_another_users_sessions_survive(self, store: Store): + mine = store.auth.upsert_profile(google_id="g1", email="ada@x") + theirs = store.auth.upsert_profile(google_id="g2", email="bob@x") + my_token = store.auth.create_session(mine.id) + their_token = store.auth.create_session(theirs.id) + + store.auth.delete_user_sessions(mine.id) + + assert store.auth.resolve_session(my_token) is None + assert store.auth.resolve_session(their_token) is not None + + def test_a_user_with_no_sessions_is_a_no_op(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + + store.auth.delete_user_sessions(profile.id) diff --git a/packages/interloper-db/tests/store/test_base.py b/packages/interloper-db/tests/store/test_base.py index 090d19cbb..4d6e48be7 100644 --- a/packages/interloper-db/tests/store/test_base.py +++ b/packages/interloper-db/tests/store/test_base.py @@ -2,15 +2,18 @@ from __future__ import annotations +from types import SimpleNamespace from uuid import uuid4 import interloper as il import pytest +from cryptography.fernet import Fernet from sqlalchemy import Engine from sqlmodel import Session, select from interloper_db.models import Component from interloper_db.store import Store +from interloper_db.store import base as base_module _ORG = uuid4() @@ -90,3 +93,128 @@ def test_nested_block_joins_the_outer_one(self, store: Store, component_db: Engi _make(store, "b") raise RuntimeError("boom") assert _count(component_db) == 0 + + +class TestFromSettings: + """The canonical constructor every long-lived process uses.""" + + @pytest.fixture + def settings(self, monkeypatch: pytest.MonkeyPatch, component_db: Engine): + """Wire settings and the process engine for ``from_settings``. + + Args: + monkeypatch: Fixture used to swap the settings and engine lookups. + component_db: The in-memory engine ``from_settings`` should find. + + Returns: + A callable taking the encryption key and returning the quota + settings object it installed. + """ + + def install(encryption_key: str | None): + quota = SimpleNamespace(max_sources=7) + monkeypatch.setattr( + "interloper.settings.AppSettings.get", + classmethod( + lambda cls: SimpleNamespace( + secrets=SimpleNamespace(encryption_key=encryption_key), quota=quota + ) + ), + ) + monkeypatch.setattr(base_module, "engine_from_settings", lambda: component_db) + monkeypatch.setattr( + il.Catalog, "from_settings", classmethod(lambda cls: il.Catalog(components={})) + ) + return quota + + return install + + def test_a_configured_key_attaches_the_cipher(self, settings) -> None: + settings(Fernet.generate_key().decode()) + + store = Store.from_settings() + + assert store._encrypt is not None + assert store._decrypt is not None + + def test_the_cipher_round_trips(self, settings) -> None: + settings(Fernet.generate_key().decode()) + + store = Store.from_settings() + + encrypt, decrypt = store._encrypt, store._decrypt + assert encrypt is not None + assert decrypt is not None + assert decrypt(encrypt(b"secret")) == b"secret" + + def test_no_key_leaves_the_store_without_a_cipher( + self, settings, caplog: pytest.LogCaptureFixture + ) -> None: + # Resource persistence then fails closed rather than writing plaintext. + settings(None) + + with caplog.at_level("WARNING", logger="interloper_db.store.base"): + store = Store.from_settings() + + assert store._encrypt is None + assert store._decrypt is None + assert "INTERLOPER_ENCRYPTION_KEY is not configured" in caplog.text + assert "fail closed" in caplog.text + + def test_the_quota_defaults_are_carried_through(self, settings) -> None: + quota = settings(Fernet.generate_key().decode()) + + assert Store.from_settings()._quota_defaults is quota + + def test_the_quota_defaults_are_carried_through_without_a_key(self, settings) -> None: + quota = settings(None) + + assert Store.from_settings()._quota_defaults is quota + + def test_the_engine_comes_from_the_process(self, settings, component_db: Engine) -> None: + settings(None) + + assert Store.from_settings().engine is component_db + + def test_an_explicit_catalog_wins_over_the_configured_one(self, settings) -> None: + settings(None) + catalog = il.Catalog(components={}) + + assert Store.from_settings(catalog=catalog)._catalog is catalog + + def test_without_one_the_configured_catalog_is_used( + self, settings, monkeypatch: pytest.MonkeyPatch + ) -> None: + settings(None) + configured = il.Catalog(components={}) + monkeypatch.setattr(il.Catalog, "from_settings", classmethod(lambda cls: configured)) + + assert Store.from_settings()._catalog is configured + + +class TestEngineProperty: + """The engine is exposed so co-located SQL runs against the same one.""" + + def test_it_reports_the_stores_engine(self, store: Store, component_db: Engine) -> None: + assert store.engine is component_db + + def test_an_explicit_engine_wins_over_the_process_one(self, component_db: Engine) -> None: + store = Store(catalog=il.Catalog(components={}), engine=component_db) + + assert store.engine is component_db + + +class TestFacets: + """Every documented facet is wired at construction.""" + + @pytest.mark.parametrize( + "facet", + ["auth", "organisations", "tokens", "relations", "quotas", "events", "runs", "components"], + ) + def test_the_facet_is_present(self, store: Store, facet: str) -> None: + assert getattr(store, facet) is not None + + def test_every_facet_shares_the_stores_engine(self, store: Store, component_db: Engine) -> None: + # Otherwise a facet would read a different database than its siblings. + for facet in ("auth", "organisations", "relations", "events", "runs", "components"): + assert getattr(store, facet)._engine is component_db diff --git a/packages/interloper-db/tests/store/test_components.py b/packages/interloper-db/tests/store/test_components.py index 964f3e257..090a6f6e5 100644 --- a/packages/interloper-db/tests/store/test_components.py +++ b/packages/interloper-db/tests/store/test_components.py @@ -10,8 +10,17 @@ import interloper as il import pytest from cryptography.fernet import InvalidToken -from interloper.errors import ConfigError, HydrationError, InUseError, NotFoundError -from interloper_assets.demo.source import DemoSource +from interloper.errors import ( + CatalogKeyError, + ComponentDriftError, + ConfigError, + HydrationError, + InUseError, + NotFoundError, +) +from interloper.partitioning.time import TimeGranularity +from interloper_assets.demo.source import DemoMonthlySource, DemoSource +from interloper_assets.facebook_ads.connection import FacebookAdsConnection from sqlalchemy import Engine, create_engine from sqlmodel import Session, select @@ -763,3 +772,254 @@ def _wrong_key(_data: bytes) -> bytes: rotated = Store(catalog=catalog, encrypt=lambda b: b[::-1], decrypt=_wrong_key) with pytest.raises(HydrationError, match="does not decrypt"): rotated.components.load(row.id) + + +class TestMissingComponents: + """Every id-addressed operation names what it could not find.""" + + def test_update_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Component {missing} not found"): + store.components.update(missing, name="x") + + def test_delete_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Component {missing} not found"): + store.components.delete(missing) + + def test_load_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Component {missing} not found"): + store.components.load(missing) + + +class TestUpdateChildren: + """Only a source has children to update.""" + + def test_children_on_a_childless_kind_is_refused(self, store: Store): + row = store.components.create(_ORG, kind="destination", key="dest") + + with pytest.raises(ConfigError, match="Components of kind 'destination' have no children"): + store.components.update(row.id, children=["a"]) + + def test_a_source_accepts_a_child_selection(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = store.components.create(_ORG, kind="source", key="demo_source", children=["a", "b"]) + + store.components.update(row.id, children=["a"]) + + with Session(component_db) as session: + children = session.exec(select(Component).where(Component.parent_id == row.id)).all() + assert [child.key for child in children] == ["a"] + + +class TestLoadDrift: + """Hydration fails closed, with the remedy the failure implies. + + Drift is produced the way it arises in practice: the row is written by a + store whose catalog lists the key, then read by one whose catalog does + not — ``create`` itself rejects an unknown key. The class is still + importable, so the status is ``disabled`` rather than ``missing``. + """ + + def test_a_drifted_key_raises_component_drift(self, component_db: Engine): + writer = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = writer.components.create(_ORG, kind="source", key="demo_source") + reader = Store(catalog=il.Catalog(components={})) + + with pytest.raises(ComponentDriftError, match="its catalog key is disabled"): + reader.components.load(row.id) + + def test_an_unreadable_payload_raises_hydration_error(self, component_db: Engine): + # A different remedy from drift: the config needs re-entering or re-keying. + catalog = il.Catalog( + components={FacebookAdsConnection.key: FacebookAdsConnection.definition()} + ) + writer = Store(catalog=catalog, encrypt=lambda b: b, decrypt=lambda b: b) + row = writer.components.create( + _ORG, + kind="connection", + key=FacebookAdsConnection.key, + config={"access_token": "TOK", "app_id": "A", "app_secret": "S"}, + ) + + def broken_decrypt(payload: bytes) -> bytes: + raise InvalidToken + + reader = Store(catalog=catalog, encrypt=lambda b: b, decrypt=broken_decrypt) + + with pytest.raises(HydrationError, match="does not decrypt"): + reader.components.load(row.id) + + def test_an_asset_the_source_no_longer_declares_raises(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + source = store.components.create(_ORG, kind="source", key="demo_source", children=["a"]) + with Session(component_db) as session: + orphan = Component(org_id=_ORG, kind="asset", key="not_declared", parent_id=source.id) + session.add(orphan) + session.commit() + orphan_id = orphan.id + + with pytest.raises(ComponentDriftError, match="is no longer declared by source"): + store.components.load(orphan_id) + + +class TestPublicConfig: + """The ``x-public`` subset a collection response may disclose.""" + + def test_a_schema_without_public_fields_discloses_nothing(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = store.components.create(_ORG, kind="source", key="demo_source") + + assert store.components.public_config(row) == {} + + def test_a_drifted_key_discloses_nothing(self, component_db: Engine): + writer = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = writer.components.create(_ORG, kind="source", key="demo_source") + reader = Store(catalog=il.Catalog(components={})) + + assert reader.components.public_config(row) == {} + + +class TestDerivedName: + """The display name derived from a component's own config.""" + + def test_a_drifted_key_derives_no_name(self, component_db: Engine): + writer = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = writer.components.create(_ORG, kind="source", key="demo_source", config={}) + reader = Store(catalog=il.Catalog(components={})) + + assert reader.components._derived_name(row, row.config) is None + + def test_a_kind_mismatch_derives_no_name(self, component_db: Engine): + # The stored kind and the catalog class disagree, so nothing is derivable. + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = store.components.create(_ORG, kind="source", key="demo_source") + row.kind = "destination" + + assert store.components._derived_name(row, {}) is None + + def test_an_unconstructable_config_derives_no_name(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = store.components.create(_ORG, kind="source", key="demo_source") + + assert store.components._derived_name(row, {"random_failure_probability": "not-a-float"}) is None + + +class TestSourceCollision: + """Two instances of one source must not silently share a materialization target.""" + + def test_colliding_datasets_are_refused(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + store.components.create(_ORG, kind="source", key="demo_source", config={"dataset": "shared"}) + + with pytest.raises(ConfigError): + store.components.create(_ORG, kind="source", key="demo_source", config={"dataset": "shared"}) + + def test_distinct_datasets_are_allowed(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + store.components.create(_ORG, kind="source", key="demo_source", config={"dataset": "one"}) + + store.components.create(_ORG, kind="source", key="demo_source", config={"dataset": "two"}) + + def test_another_org_is_not_a_collision(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + store.components.create(_ORG, kind="source", key="demo_source", config={"dataset": "shared"}) + + store.components.create(uuid4(), kind="source", key="demo_source", config={"dataset": "shared"}) + + +class TestEnsureChildrenDrift: + """A source whose key no longer resolves cannot have its children reconciled.""" + + def test_a_drifted_source_key_raises(self, component_db: Engine): + writer = Store(catalog=il.Catalog.from_assets([DemoSource])) + row = writer.components.create(_ORG, kind="source", key="demo_source") + reader = Store(catalog=il.Catalog(components={})) + + with pytest.raises(CatalogKeyError, match="Unknown source key: demo_source"): + reader.components.update(row.id, children=["a"]) + + +class TestJobPartitionGranularity: + """The scheduler reads one granularity off a job's targets, or fails closed.""" + + def test_a_job_with_no_targets_has_none(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + job = store.components.create(_ORG, kind="job", key="cron_job") + + with Session(component_db) as session: + assert store.components.job_partition_granularity(session, job.id) is None + + def test_a_partitioned_source_target_reports_its_granularity(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + source = store.components.create(_ORG, kind="source", key="demo_source") + job = store.components.create( + _ORG, kind="job", key="cron_job", relations={"target": [(source.id, "")]} + ) + + with Session(component_db) as session: + assert store.components.job_partition_granularity(session, job.id) is TimeGranularity.DAY + + def test_a_partitioned_asset_target_reports_its_granularity(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + source = store.components.create(_ORG, kind="source", key="demo_source", children=["a"]) + with Session(component_db) as session: + asset = session.exec(select(Component).where(Component.parent_id == source.id)).one() + asset_id = asset.id + job = store.components.create( + _ORG, kind="job", key="cron_job", relations={"target": [(asset_id, "")]} + ) + + with Session(component_db) as session: + assert store.components.job_partition_granularity(session, job.id) is TimeGranularity.DAY + + def test_targets_disagreeing_on_granularity_fail_closed(self, component_db: Engine): + # Scheduling one window would be wrong for at least one target. + store = Store(catalog=il.Catalog.from_assets([DemoSource, DemoMonthlySource])) + daily = store.components.create(_ORG, kind="source", key="demo_source") + monthly = store.components.create(_ORG, kind="source", key="demo_monthly_source") + job = store.components.create( + _ORG, + kind="job", + key="cron_job", + relations={"target": [(daily.id, ""), (monthly.id, "")]}, + ) + + with Session(component_db) as session, pytest.raises( + ValueError, match="Job targets disagree on partition granularity" + ): + store.components.job_partition_granularity(session, job.id) + + def test_a_drifted_target_contributes_nothing(self, component_db: Engine): + # Drift is the run path's problem, not the scheduler's. + writer = Store(catalog=il.Catalog.from_assets([DemoSource])) + source = writer.components.create(_ORG, kind="source", key="demo_source") + job = writer.components.create( + _ORG, kind="job", key="cron_job", relations={"target": [(source.id, "")]} + ) + reader = Store(catalog=il.Catalog(components={})) + + with Session(component_db) as session: + assert reader.components.job_partition_granularity(session, job.id) is None + + +class TestCheckJobTargets: + """A job whose target drifted cannot be hydrated.""" + + def test_a_drifted_target_raises(self, component_db: Engine): + writer = Store(catalog=il.Catalog.from_assets([DemoSource])) + source = writer.components.create(_ORG, kind="source", key="demo_source") + job = writer.components.create( + _ORG, kind="job", key="cron_job", relations={"target": [(source.id, "")]} + ) + reader = Store(catalog=il.Catalog(components={})) + + with Session(component_db) as session: + db_job = session.get(Component, job.id) + assert db_job is not None + with pytest.raises(ComponentDriftError, match="is disabled"): + reader.components._check_job_targets(session, db_job) diff --git a/packages/interloper-db/tests/store/test_events.py b/packages/interloper-db/tests/store/test_events.py index c3d23e7cc..570af8edc 100644 --- a/packages/interloper-db/tests/store/test_events.py +++ b/packages/interloper-db/tests/store/test_events.py @@ -432,3 +432,70 @@ def test_executions_read_model_maps_the_view(store: Store) -> None: rows = store.events.list_executions(run_id) assert [(row.component_key, row.status) for row in rows] == [("a", "success")] assert store.events.list_executions(uuid4()) == [] + + +class TestEventValues: + """The row values derived from a framework event.""" + + def test_a_non_uuid_event_id_gets_a_fresh_one(self) -> None: + # Producer ids are normally uuid5-derived; anything else still persists + # rather than failing the write and dropping the event. + event = il.Event(type=il.EventType.RUN_STARTED, metadata={}) + event.id = "not-a-uuid" + + values = EventStore._event_values(event, _ORG_ID, _RUN_ID) + + assert isinstance(values["id"], UUID) + + def test_a_uuid_event_id_is_preserved(self) -> None: + # Identity survives end to end so the upsert dedups re-delivery. + event = il.Event(type=il.EventType.RUN_STARTED, metadata={}) + + values = EventStore._event_values(event, _ORG_ID, _RUN_ID) + + assert str(values["id"]) == event.id + + +class TestFilters: + """``list_all`` and ``count`` share one filter set.""" + + def test_the_org_filter_narrows_the_listing(self, store: Store) -> None: + _seed(_make_events(2)) + _seed([ + Event( + id=uuid4(), + org_id=uuid4(), + run_id=_OTHER_RUN_ID, + event_type="asset_completed", + timestamp=_BASE_TS, + ) + ]) + + assert len(store.events.list_all(org_id=_ORG_ID)) == 2 + assert store.events.count(org_id=_ORG_ID) == 2 + + def test_the_component_filter_narrows_the_listing(self, store: Store) -> None: + mine = uuid4() + _seed(_make_events(2, component_id=mine)) + _seed(_make_events(3, component_id=uuid4(), start=100)) + + assert len(store.events.list_all(component_ids=[mine])) == 2 + assert store.events.count(component_ids=[mine]) == 2 + + def test_several_component_ids_union(self, store: Store) -> None: + first, second = uuid4(), uuid4() + _seed(_make_events(2, component_id=first)) + _seed(_make_events(3, component_id=second, start=100)) + + assert store.events.count(component_ids=[first, second]) == 5 + + def test_the_type_filter_narrows_the_listing(self, store: Store) -> None: + _seed(_make_events(3)) + + assert store.events.count(run_id=_RUN_ID, event_types=["asset_completed"]) == 1 + + def test_no_filters_counts_everything(self, store: Store) -> None: + _seed(_make_events(2)) + _seed(_make_events(3, run_id=_OTHER_RUN_ID, start=100)) + + assert store.events.count() == 5 diff --git a/packages/interloper-db/tests/store/test_hydration.py b/packages/interloper-db/tests/store/test_hydration.py index f6059045d..2468bdc81 100644 --- a/packages/interloper-db/tests/store/test_hydration.py +++ b/packages/interloper-db/tests/store/test_hydration.py @@ -214,3 +214,52 @@ def raise_validation_error(spec): assert "Failed to hydrate source 'demo_source'" in message assert "app_secret: Field required" in message assert "s3cret-value" not in message + + +class TestDecodeData: + """The stored ``data`` payload, decrypted for sensitive kinds.""" + + def test_no_data_decodes_to_an_empty_dict(self, store: Store): + row = store.components.create(_ORG, kind="source", key="demo_source") + + assert store.components._hydrator.decode_data(row) == {} + + def test_an_encrypted_row_without_a_cipher_is_an_actionable_error(self, component_db: Engine): + writer = Store(catalog=_CATALOG, encrypt=lambda b: b, decrypt=lambda b: b) + row = writer.components.create( + _ORG, kind="connection", key="demo_connection", config={"token": "s3cret"} + ) + row.data = row.data or b"payload" + row.encrypted = True + reader = Store(catalog=_CATALOG) + + with pytest.raises(HydrationError, match="INTERLOPER_ENCRYPTION_KEY"): + reader.components._hydrator.decode_data(row) + + +class TestRelationsByType: + """The grouped relation lookup used while building a spec.""" + + def test_a_row_with_no_id_has_no_relations(self, store: Store): + from sqlmodel import Session + + with Session(store.engine) as session: + assert store.components._hydrator._relations_by_type(session, None) == {} + + +class TestResolvePath: + """A spec's import path comes from the catalog entry.""" + + def test_a_drifted_key_is_an_actionable_error(self, component_db: Engine): + from interloper.errors import CatalogKeyError + from sqlmodel import Session + + writer = Store(catalog=_CATALOG) + row = writer.components.create(_ORG, kind="source", key="demo_source") + reader = Store(catalog=il.Catalog(components={})) + + with Session(component_db) as session: + db_row = session.get(type(row), row.id) + assert db_row is not None + with pytest.raises(CatalogKeyError, match="Unknown source key: demo_source"): + reader.components._hydrator._resolve_path(session, db_row) diff --git a/packages/interloper-db/tests/store/test_organisations.py b/packages/interloper-db/tests/store/test_organisations.py index c924d6af4..0c50339ea 100644 --- a/packages/interloper-db/tests/store/test_organisations.py +++ b/packages/interloper-db/tests/store/test_organisations.py @@ -217,3 +217,251 @@ def test_expired_invitation_returns_false(self, store: Store, auth_db: Engine): session.commit() assert not store.organisations.has_pending_invitation("new@example.com") + + +class TestUpdate: + """Renaming, and the soft-delete guard on it.""" + + def test_renames_the_organisation(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + + renamed = store.organisations.update(org.id, name="Acme Corp") + + assert renamed.name == "Acme Corp" + assert store.organisations.get(org.id).name == "Acme Corp" + + def test_a_missing_organisation_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Organisation {missing} not found"): + store.organisations.update(missing, name="Acme") + + def test_a_soft_deleted_organisation_reads_as_missing(self, store: Store): + # The ledger row survives the delete, but it is not writable. + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + store.organisations.delete(org.id) + + with pytest.raises(NotFoundError, match=f"Organisation {org.id} not found"): + store.organisations.update(org.id, name="Acme Corp") + + +class TestListAll: + """The super-admin listing pairs each organisation with its member count.""" + + def test_counts_members_per_organisation(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + busy = store.organisations.create(name="Busy", creator_id=ada.id) + store.organisations.add_member(busy.id, bob.id, "viewer") + store.organisations.create(name="Quiet", creator_id=ada.id) + + counts = {org.name: count for org, count in store.organisations.list_all()} + + assert counts == {"Busy": 2, "Quiet": 1} + + def test_an_organisation_with_no_members_counts_zero(self, store: Store, auth_db: Engine): + # A soft-deleted org has its memberships purged but stays in the ledger. + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + store.organisations.delete(org.id) + + counts = {org.name: count for org, count in store.organisations.list_all()} + + assert counts == {"Acme": 0} + + def test_no_organisations_is_an_empty_list(self, store: Store): + assert store.organisations.list_all() == [] + + +class TestListForUser: + """Membership-scoped listing.""" + + def test_lists_only_the_users_organisations(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + mine = store.organisations.create(name="Mine", creator_id=ada.id) + store.organisations.create(name="Theirs", creator_id=bob.id) + + assert [org.id for org in store.organisations.list_for_user(ada.id)] == [mine.id] + + def test_a_user_with_no_memberships_gets_an_empty_list(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + + assert store.organisations.list_for_user(profile.id) == [] + + +class TestMembers: + """Roles, listing, and the two mutations.""" + + def test_the_creator_is_an_admin(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + + assert store.organisations.member_role(profile.id, org.id) == "admin" + + def test_a_non_member_has_no_role(self, store: Store): + profile = store.auth.upsert_profile(google_id="g1", email="ada@x") + other = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=profile.id) + + assert store.organisations.member_role(other.id, org.id) is None + + def test_members_are_listed_with_their_roles(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + store.organisations.add_member(org.id, bob.id, "editor") + + members = {profile.email: role for profile, role in store.organisations.list_members(org.id)} + + assert members == {"ada@x": "admin", "bob@x": "editor"} + + def test_an_organisation_with_no_members_lists_nothing(self, store: Store): + assert store.organisations.list_members(uuid4()) == [] + + def test_adding_a_member_reports_whether_it_was_new(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + + assert store.organisations.add_member(org.id, bob.id, "viewer") is True + assert store.organisations.add_member(org.id, bob.id, "viewer") is False + + def test_a_role_can_be_changed(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + store.organisations.add_member(org.id, bob.id, "viewer") + + store.organisations.update_member_role(org.id, bob.id, "admin") + + assert store.organisations.member_role(bob.id, org.id) == "admin" + + def test_changing_a_non_members_role_raises(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + stranger = uuid4() + + with pytest.raises(NotFoundError, match=f"User {stranger} is not a member"): + store.organisations.update_member_role(org.id, stranger, "admin") + + def test_a_member_can_be_removed(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + store.organisations.add_member(org.id, bob.id, "viewer") + + store.organisations.remove_member(org.id, bob.id) + + assert store.organisations.member_role(bob.id, org.id) is None + + def test_removing_a_non_member_raises(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + stranger = uuid4() + + with pytest.raises(NotFoundError, match=f"User {stranger} is not a member"): + store.organisations.remove_member(org.id, stranger) + + +class TestInvitations: + """Creation, listing, token lookup and deletion.""" + + def test_a_created_invitation_carries_a_token(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + + invitation = store.organisations.create_invitation( + org_id=org.id, email="new@x", role="editor", invited_by=ada.id + ) + + assert invitation.token + assert invitation.email == "new@x" + assert invitation.role == "editor" + + def test_invitations_are_listed_per_organisation(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + mine = store.organisations.create(name="Mine", creator_id=ada.id) + theirs = store.organisations.create(name="Theirs", creator_id=ada.id) + store.organisations.create_invitation(org_id=mine.id, email="a@x", role="viewer", invited_by=ada.id) + store.organisations.create_invitation(org_id=theirs.id, email="b@x", role="viewer", invited_by=ada.id) + + assert [inv.email for inv in store.organisations.list_invitations(mine.id)] == ["a@x"] + + def test_an_organisation_with_no_invitations_lists_nothing(self, store: Store): + assert store.organisations.list_invitations(uuid4()) == [] + + def test_an_invitation_resolves_by_token(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + invitation = store.organisations.create_invitation( + org_id=org.id, email="new@x", role="viewer", invited_by=ada.id + ) + + found = store.organisations.get_invitation_by_token(invitation.token) + + assert found is not None + assert found.id == invitation.id + + def test_an_unknown_token_resolves_to_nothing(self, store: Store): + assert store.organisations.get_invitation_by_token("never-issued") is None + + def test_an_invitation_can_be_deleted(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + invitation = store.organisations.create_invitation( + org_id=org.id, email="new@x", role="viewer", invited_by=ada.id + ) + + store.organisations.delete_invitation(invitation.id) + + assert store.organisations.list_invitations(org.id) == [] + + def test_deleting_a_missing_invitation_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Invitation {missing} not found"): + store.organisations.delete_invitation(missing) + + +class TestAcceptInvitationEdges: + """The arms of ``accept_invitation`` beyond the happy path.""" + + def test_an_expired_invitation_is_refused_and_swept(self, store: Store, auth_db: Engine): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + invitation = store.organisations.create_invitation( + org_id=org.id, email="bob@x", role="viewer", invited_by=ada.id + ) + with SQLSession(auth_db) as session: + row = session.get(Invitation, invitation.id) + assert row is not None + row.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + session.add(row) + session.commit() + + assert store.organisations.accept_invitation(invitation.token, bob.id) is None + # The stale row is deleted rather than left to accumulate. + assert store.organisations.list_invitations(org.id) == [] + + def test_accepting_twice_leaves_one_membership(self, store: Store): + ada = store.auth.upsert_profile(google_id="g1", email="ada@x") + bob = store.auth.upsert_profile(google_id="g2", email="bob@x") + org = store.organisations.create(name="Acme", creator_id=ada.id) + first = store.organisations.create_invitation( + org_id=org.id, email="bob@x", role="viewer", invited_by=ada.id + ) + store.organisations.accept_invitation(first.token, bob.id) + second = store.organisations.create_invitation( + org_id=org.id, email="bob@x", role="admin", invited_by=ada.id + ) + + store.organisations.accept_invitation(second.token, bob.id) + + members = store.organisations.list_members(org.id) + assert len([m for m in members if m[0].id == bob.id]) == 1 + # The existing membership is left as it was, not re-roled. + assert store.organisations.member_role(bob.id, org.id) == "viewer" diff --git a/packages/interloper-db/tests/store/test_relations.py b/packages/interloper-db/tests/store/test_relations.py index 133f7f5e9..a879c0c09 100644 --- a/packages/interloper-db/tests/store/test_relations.py +++ b/packages/interloper-db/tests/store/test_relations.py @@ -8,6 +8,8 @@ import interloper as il import pytest from interloper.errors import ConfigError, NotFoundError +from interloper_assets.demo.source import DemoSource +from interloper_assets.facebook_ads.source import FacebookAds from sqlalchemy import Engine from sqlmodel import Session, select @@ -298,3 +300,32 @@ def test_sync_repoint_of_required_dependency_allowed(self, dep_store: Store): dep_store.components.update(down.id, relations={"dependency": [(up_two.id, "up")]}) (edge,) = dep_store.relations.list_all(_ORG, type="dependency") assert edge.dst_id == up_two.id + + +class TestAddValidation: + """``add`` vets the source, the vocabulary and the slot before writing.""" + + def test_a_missing_source_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Component {missing} not found"): + store.relations.add(missing, type="resource", dst_id=uuid4(), slot="connection") + + def test_a_slot_on_an_unslotted_relation_is_refused(self, component_db: Engine): + store = Store(catalog=il.Catalog.from_assets([DemoSource])) + source = store.components.create(_ORG, kind="source", key="demo_source") + destination = store.components.create(_ORG, kind="destination", key="dest") + + with pytest.raises(ConfigError, match="is not slotted"): + store.relations.add(source.id, type="destination", dst_id=destination.id, slot="nope") + + def test_a_slot_expecting_another_key_is_refused(self, component_db: Engine): + # The slot declares the component key it accepts; anything else would + # be wired into a resource the source cannot use. + catalog = il.Catalog.from_assets([FacebookAds]) + store = Store(catalog=catalog, encrypt=lambda b: b, decrypt=lambda b: b) + source = store.components.create(_ORG, kind="source", key="facebook_ads") + wrong = store.components.create(_ORG, kind="destination", key="dest") + + with pytest.raises(ConfigError): + store.relations.add(source.id, type="resource", dst_id=wrong.id, slot="connection") diff --git a/packages/interloper-db/tests/store/test_runs.py b/packages/interloper-db/tests/store/test_runs.py index ae63a5e85..b8424c609 100644 --- a/packages/interloper-db/tests/store/test_runs.py +++ b/packages/interloper-db/tests/store/test_runs.py @@ -359,3 +359,229 @@ def test_unbounded_listing_keeps_every_run(self, store: Store): _timed_run(store, started_at=None, completed_at=None) assert len(store.runs.list_all(_ORG_ID, limit=100)) == 2 + + +class TestGetAndComplete: + """Id-addressed reads and the terminal transition.""" + + def test_get_returns_the_run(self, store: Store): + run = store.runs.create(_ORG_ID) + + assert store.runs.get(run.id).id == run.id + + def test_get_missing_run_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Run {missing} not found"): + store.runs.get(missing) + + def test_complete_records_success(self, store: Store): + run = store.runs.create(_ORG_ID) + + store.runs.complete(run.id, success=True) + + completed = store.runs.get(run.id) + assert completed.status == "success" + assert completed.completed_at is not None + + def test_complete_records_failure(self, store: Store): + run = store.runs.create(_ORG_ID) + + store.runs.complete(run.id, success=False) + + assert store.runs.get(run.id).status == "failed" + + def test_complete_missing_run_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Run {missing} not found"): + store.runs.complete(missing, success=True) + + def test_get_backfill_returns_it(self, store: Store): + backfill = _backfill(store) + + assert store.runs.get_backfill(backfill.id).id == backfill.id + + def test_get_missing_backfill_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Backfill {missing} not found"): + store.runs.get_backfill(missing) + + +class TestPartitionKeyValidation: + """A run's partition key must be a shape the framework recognises.""" + + def test_a_well_formed_key_is_accepted(self, store: Store): + run = store.runs.create(_ORG_ID, partition_key="2026-01-01") + + assert run.partition_key == "2026-01-01" + + def test_an_unrecognised_shape_is_rejected(self, store: Store): + with pytest.raises(ValueError): + store.runs.create(_ORG_ID, partition_key="not-a-key") + + +class TestRetryValidation: + """Only a failed run can be retried, and only with a known scope.""" + + def test_an_unknown_scope_is_rejected(self, store: Store): + run = store.runs.create(_ORG_ID) + + with pytest.raises(ValueError, match="Invalid retry scope: 'sideways'"): + store.runs.retry(run.id, scope="sideways") + + def test_a_missing_run_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Run {missing} not found"): + store.runs.retry(missing) + + def test_a_run_that_did_not_fail_is_rejected(self, store: Store): + run = store.runs.create(_ORG_ID) + + with pytest.raises(ValueError, match="is not failed"): + store.runs.retry(run.id) + + def test_the_failed_scope_is_accepted(self, store: Store): + run = store.runs.create(_ORG_ID) + store.runs.complete(run.id, success=False) + + retried = store.runs.retry(run.id, scope="failed") + + assert retried.retry_of == run.id + assert retried.retry_scope == "failed" + + +class TestBackfillGranularity: + """A backfill spans one granularity; mixed bounds fail closed.""" + + def test_mixed_granularity_bounds_are_rejected(self, store: Store): + with pytest.raises(ValueError, match="must share one granularity"): + store.runs.create_backfill(_ORG_ID, start_key="2026-01", end_key="2026-01-05") + + def test_a_monthly_span_is_accepted(self, store: Store): + backfill = store.runs.create_backfill(_ORG_ID, start_key="2026-01", end_key="2026-03") + + assert backfill.partitions == 3 + + +class TestListActiveBackfills: + """The active listing covers the two non-terminal statuses.""" + + def test_running_and_queued_are_listed(self, store: Store): + backfill = _backfill(store) + + active = store.runs.list_active_backfills(_ORG_ID) + + assert [row.id for row in active] == [backfill.id] + + def test_a_terminal_backfill_is_excluded(self, store: Store): + backfill = _backfill(store) + with Session(store.engine) as session: + row = session.get(Backfill, backfill.id) + assert row is not None + row.status = "success" + session.add(row) + session.commit() + + assert store.runs.list_active_backfills(_ORG_ID) == [] + + def test_another_orgs_backfills_are_excluded(self, store: Store): + _backfill(store) + + assert store.runs.list_active_backfills(uuid4()) == [] + + +class TestBackfillProgression: + """Completing a run advances the backfill, or terminates it.""" + + def test_the_backfill_succeeds_once_every_run_has(self, store: Store): + backfill = store.runs.create_backfill( + _ORG_ID, start_key="2026-01-01", end_key="2026-01-02", concurrency=2 + ) + for run_id in _run_statuses(store, backfill.id): + store.runs.complete(run_id, success=True) + + assert store.runs.get_backfill(backfill.id).status == "success" + + def test_one_failure_without_fail_fast_still_finishes_as_failed(self, store: Store): + backfill = store.runs.create_backfill( + _ORG_ID, start_key="2026-01-01", end_key="2026-01-02", concurrency=2, fail_fast=False + ) + run_ids = list(_run_statuses(store, backfill.id)) + store.runs.complete(run_ids[0], success=False) + store.runs.complete(run_ids[1], success=True) + + finished = store.runs.get_backfill(backfill.id) + assert finished.status == "failed" + assert finished.completed_at is not None + + def test_fail_fast_cancels_the_pending_runs(self, store: Store): + # The remaining partitions are not worth spending once one failed. + backfill = store.runs.create_backfill( + _ORG_ID, start_key="2026-01-01", end_key="2026-01-04", concurrency=1, fail_fast=True + ) + first = next( + run_id for run_id, status in _run_statuses(store, backfill.id).items() if status == "queued" + ) + + store.runs.complete(first, success=False) + + assert store.runs.get_backfill(backfill.id).status == "failed" + assert "pending" not in _run_statuses(store, backfill.id).values() + + def test_a_completion_promotes_the_next_partition(self, store: Store): + backfill = store.runs.create_backfill( + _ORG_ID, start_key="2026-01-01", end_key="2026-01-04", concurrency=1 + ) + first = next( + run_id for run_id, status in _run_statuses(store, backfill.id).items() if status == "queued" + ) + + store.runs.complete(first, success=True) + + statuses = _partition_statuses(store, backfill.id) + assert statuses["2026-01-04"] == "success" + # Newest-first, matching the initial dispatch order. + assert statuses["2026-01-03"] == "queued" + + def test_a_completion_outside_any_backfill_is_a_no_op(self, store: Store): + run = store.runs.create(_ORG_ID) + + store.runs.complete(run.id, success=True) + + assert store.runs.get(run.id).status == "success" + + +class TestRunFilters: + """``list_all``/``count`` narrow on component, backfill and status.""" + + def test_the_component_filter_narrows_the_listing(self, store: Store): + target = _component(store, kind="source") + store.runs.create(_ORG_ID, component_id=target) + store.runs.create(_ORG_ID) + + assert len(store.runs.list_all(_ORG_ID, component_id=target)) == 1 + assert store.runs.count(_ORG_ID, component_id=target) == 1 + + def test_the_backfill_filter_narrows_the_listing(self, store: Store): + backfill = _backfill(store, days=2) + store.runs.create(_ORG_ID) + + assert len(store.runs.list_all(_ORG_ID, backfill_id=backfill.id)) == 2 + assert store.runs.count(_ORG_ID, backfill_id=backfill.id) == 2 + + def test_the_status_filter_narrows_the_listing(self, store: Store): + succeeded = store.runs.create(_ORG_ID) + store.runs.complete(succeeded.id, success=True) + store.runs.create(_ORG_ID) + + assert [row.id for row in store.runs.list_all(_ORG_ID, status="success")] == [succeeded.id] + assert store.runs.count(_ORG_ID, status="success") == 1 + + def test_another_orgs_runs_are_never_listed(self, store: Store): + store.runs.create(_ORG_ID) + + assert store.runs.list_all(uuid4()) == [] + assert store.runs.count(uuid4()) == 0 diff --git a/packages/interloper-db/tests/store/test_status.py b/packages/interloper-db/tests/store/test_status.py index 0eec29ac5..5146057b3 100644 --- a/packages/interloper-db/tests/store/test_status.py +++ b/packages/interloper-db/tests/store/test_status.py @@ -3,6 +3,7 @@ from __future__ import annotations import interloper as il +import pytest from interloper_assets.demo.source import DemoSource from interloper_db.store.status import ComponentStatus, asset_status, source_status @@ -55,3 +56,22 @@ def test_owned_asset_status_cascades_missing_parent() -> None: def test_owned_asset_status_cascades_disabled_parent() -> None: status = asset_status(_EMPTY, _ASSET_KEY, source_key=_SOURCE_KEY, discovered=_ENABLED) assert status is ComponentStatus.DISABLED + + +def test_an_enabled_source_whose_class_vanished_reads_as_missing(monkeypatch: pytest.MonkeyPatch) -> None: + """The defensive arm: the catalog said ok but the class did not resolve. + + Unreachable through normal operation — the catalog is built from importable + classes — so the resolution is broken deliberately to prove the fallback + reports ``missing`` rather than raising into the caller. + """ + catalog = il.Catalog.from_assets([DemoSource]) + + def unresolvable(key: str, catalog: object) -> None: + raise ImportError("class went away") + + monkeypatch.setattr( + il.Source, "resolve_key", classmethod(lambda cls, key, catalog: unresolvable(key, catalog)) + ) + + assert asset_status(catalog, _ASSET_KEY, source_key=_SOURCE_KEY) is ComponentStatus.MISSING diff --git a/packages/interloper-db/tests/store/test_tokens.py b/packages/interloper-db/tests/store/test_tokens.py index 5f5e5c697..f1c8a4dbb 100644 --- a/packages/interloper-db/tests/store/test_tokens.py +++ b/packages/interloper-db/tests/store/test_tokens.py @@ -172,3 +172,36 @@ def test_revoke_is_idempotent(self, store: Store, member: tuple[Profile, Organis assert first.revoked_at is not None assert second.revoked_at == first.revoked_at + + +class TestGetToken: + """Id-addressed reads.""" + + def test_get_returns_the_token(self, store: Store, member: tuple[Profile, Organisation]): + profile, org = member + token, _raw = store.tokens.create(user_id=profile.id, organisation_id=org.id, name="ci") + + assert store.tokens.get(token.id).id == token.id + + def test_get_missing_token_raises(self, store: Store): + missing = uuid4() + + with pytest.raises(NotFoundError, match=f"Token {missing} not found"): + store.tokens.get(missing) + + +class TestResolveOrphanedToken: + """A token whose profile vanished resolves to nothing.""" + + def test_it_resolves_to_none(self, store: Store, member: tuple[Profile, Organisation], token_db: Engine): + # The foreign key makes this unreachable through the store's own API, + # so the orphan is fabricated with constraints off — the guard exists + # for a row that should not be there. + profile, org = member + _token, raw = store.tokens.create(user_id=profile.id, organisation_id=org.id, name="ci") + with token_db.connect() as connection: + connection.exec_driver_sql("PRAGMA foreign_keys=OFF") + connection.exec_driver_sql("DELETE FROM profiles") + connection.commit() + + assert store.tokens.resolve(raw) is None diff --git a/packages/interloper-db/tests/test_engine.py b/packages/interloper-db/tests/test_engine.py new file mode 100644 index 000000000..90df87b22 --- /dev/null +++ b/packages/interloper-db/tests/test_engine.py @@ -0,0 +1,104 @@ +"""Tests for ``interloper_db.engine``. + +The engine is a process-wide singleton, so every test restores whatever it +found — otherwise an engine installed here leaks into the suites that run +later. +""" + +from __future__ import annotations + +from collections.abc import Iterator + +import pytest +from interloper.errors import ConfigError +from sqlalchemy import Engine + +from interloper_db import engine as engine_module +from interloper_db.engine import engine_from_settings, get_engine, init_engine + + +@pytest.fixture(autouse=True) +def restore_engine() -> Iterator[None]: + """Put the module-level engine back the way the test found it. + + Yields: + ``None``; the teardown restores the saved engine. + """ + saved = engine_module._engine + engine_module._engine = None + yield + engine_module._engine = saved + + +class TestInitEngine: + """Explicit DSN, environment fallback, and the actionable failure.""" + + def test_an_explicit_dsn_builds_the_engine(self) -> None: + engine = init_engine("sqlite://") + + assert isinstance(engine, Engine) + assert engine.url.drivername == "sqlite" + + def test_the_engine_becomes_the_process_singleton(self) -> None: + engine = init_engine("sqlite://") + + assert get_engine() is engine + + def test_it_falls_back_to_the_database_url_variable(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DATABASE_URL", "sqlite://") + + assert init_engine().url.drivername == "sqlite" + + def test_no_dsn_anywhere_is_an_actionable_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("DATABASE_URL", raising=False) + + with pytest.raises(ConfigError, match="Database DSN required: pass dsn= or set DATABASE_URL"): + init_engine() + + def test_extra_kwargs_reach_create_engine(self) -> None: + engine = init_engine("sqlite://", echo=True) + + assert engine.echo is True + + def test_re_initializing_replaces_the_singleton(self) -> None: + first = init_engine("sqlite://") + second = init_engine("sqlite://") + + assert second is not first + assert get_engine() is second + + +class TestGetEngine: + """Reading the singleton before it exists is a programming error.""" + + def test_before_initialization_it_is_an_actionable_error(self) -> None: + with pytest.raises(RuntimeError, match=r"Database engine not initialized. Call init_engine\(\) first."): + get_engine() + + +class TestEngineFromSettings: + """The lazy accessor the store layer uses when nothing installed an engine.""" + + def test_an_existing_engine_is_reused_without_reading_settings( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + engine = init_engine("sqlite://") + monkeypatch.setattr( + "interloper.settings.AppSettings.get", + classmethod(lambda cls: pytest.fail("settings must not be read when an engine exists")), + ) + + assert engine_from_settings() is engine + + def test_without_an_engine_it_initializes_from_settings(self, monkeypatch: pytest.MonkeyPatch) -> None: + from types import SimpleNamespace + + monkeypatch.setattr( + "interloper.settings.AppSettings.get", + classmethod(lambda cls: SimpleNamespace(postgres=SimpleNamespace(dsn="sqlite://"))), + ) + + engine = engine_from_settings() + + assert engine.url.drivername == "sqlite" + assert get_engine() is engine diff --git a/packages/interloper-db/tests/test_provision.py b/packages/interloper-db/tests/test_provision.py new file mode 100644 index 000000000..a89262c04 --- /dev/null +++ b/packages/interloper-db/tests/test_provision.py @@ -0,0 +1,400 @@ +"""Tests for ``interloper_db.provision``. + +The routines here are Postgres-specific — an advisory lock, a maintenance +connection to ``postgres``, ``pg_terminate_backend`` — so the engine is +faked and the statements it is handed are asserted, matching the package's +SQLite-backed unit-test style rather than requiring a live server. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +import pytest +import sqlalchemy +from typing_extensions import Self + +from interloper_db import provision as provision_module +from interloper_db.provision import ( + _alembic_config, + create_all, + downgrade, + drop_database, + ensure_database, + upgrade, +) + + +class FakeConnection: + """Records every statement executed against it, plus commits.""" + + def __init__(self, rows: list[Any] | None = None) -> None: + """Set up the fake. + + Args: + rows: Results ``fetchone`` hands back in order; an exhausted + queue yields ``None``, standing in for "no such row". + """ + self.statements: list[tuple[str, dict[str, Any] | None]] = [] + self.commits = 0 + self._rows = list(rows or []) + + def execute(self, statement: Any, params: dict[str, Any] | None = None) -> FakeConnection: + """Record a statement and return self as its result proxy. + + Args: + statement: The SQLAlchemy text clause executed. + params: Bound parameters, when any. + + Returns: + This connection, standing in for the result. + """ + self.statements.append((str(statement), params)) + return self + + def fetchone(self) -> Any: + """Pop the next queued row. + + Returns: + The next row, or ``None`` once the queue is empty. + """ + return self._rows.pop(0) if self._rows else None + + def commit(self) -> None: + """Record a commit.""" + self.commits += 1 + + def __enter__(self) -> Self: + """Enter the context. + + Returns: + This connection. + """ + return self + + def __exit__(self, *args: object) -> None: + """Leave the context. + + Args: + *args: Exception triple, ignored. + """ + + +class FakeEngine: + """Engine stand-in handing out one recording connection.""" + + def __init__(self, connection: FakeConnection) -> None: + """Set up the fake. + + Args: + connection: The connection every ``connect`` call returns. + """ + self.connection = connection + self.disposed = 0 + + def connect(self) -> FakeConnection: + """Hand out the recording connection. + + Returns: + The connection. + """ + return self.connection + + def dispose(self) -> None: + """Record a dispose.""" + self.disposed += 1 + + +def _sql(connection: FakeConnection) -> str: + return " | ".join(statement for statement, _ in connection.statements) + + +class TestMigrations: + """``upgrade`` / ``downgrade`` delegate to Alembic with this package's config.""" + + @pytest.fixture + def alembic(self, monkeypatch: pytest.MonkeyPatch) -> dict[str, list[Any]]: + """Record the Alembic commands instead of running them. + + Args: + monkeypatch: Fixture used to swap the command module. + + Returns: + The recorded ``upgrade`` and ``downgrade`` calls. + """ + from alembic import command + + recorded: dict[str, list[Any]] = {"upgrade": [], "downgrade": []} + monkeypatch.setattr( + command, "upgrade", lambda config, revision: recorded["upgrade"].append((config, revision)) + ) + monkeypatch.setattr( + command, "downgrade", lambda config, revision: recorded["downgrade"].append((config, revision)) + ) + return recorded + + def test_upgrade_targets_head_by_default( + self, alembic: dict[str, list[Any]], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(provision_module, "get_engine", lambda: object()) + + upgrade() + + (_config, revision), = alembic["upgrade"] + assert revision == "head" + + def test_upgrade_takes_an_explicit_revision(self, alembic: dict[str, list[Any]]) -> None: + upgrade(engine=object(), revision="0007") # ty: ignore[invalid-argument-type] + + assert alembic["upgrade"][0][1] == "0007" + + def test_downgrade_steps_back_one_by_default( + self, alembic: dict[str, list[Any]], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(provision_module, "get_engine", lambda: object()) + + downgrade() + + assert alembic["downgrade"][0][1] == "-1" + + def test_downgrade_takes_an_explicit_revision(self, alembic: dict[str, list[Any]]) -> None: + downgrade(engine=object(), revision="base") # ty: ignore[invalid-argument-type] + + assert alembic["downgrade"][0][1] == "base" + + def test_without_an_engine_the_global_one_is_resolved( + self, alembic: dict[str, list[Any]], monkeypatch: pytest.MonkeyPatch + ) -> None: + # Resolved for its side effect only: alembic's env.py reads the + # global singleton, so an uninitialized engine must still surface. + resolved: list[bool] = [] + monkeypatch.setattr(provision_module, "get_engine", lambda: resolved.append(True)) + + upgrade() + + assert resolved == [True] + + +class TestAlembicConfig: + """The config resolves from the package layout, not the working directory.""" + + def test_the_script_location_is_the_packages_migrations(self) -> None: + config = _alembic_config() + + script_location = config.get_main_option("script_location") + assert script_location is not None + location = Path(script_location) + assert location.is_absolute() + assert location.name == "migrations" + assert (location / "env.py").is_file() + + +class TestCreateAll: + """Table creation is bracketed by an advisory lock, then hands off to Alembic.""" + + @pytest.fixture + def harness(self, monkeypatch: pytest.MonkeyPatch) -> Iterator[dict[str, Any]]: + """Fake the engine, the metadata create and the Alembic upgrade. + + Args: + monkeypatch: Fixture used to swap the collaborators. + + Yields: + The recording connection plus what ``create_all`` was handed. + """ + from sqlmodel import SQLModel + + connection = FakeConnection() + engine = FakeEngine(connection) + created: dict[str, Any] = {} + upgrades: list[Any] = [] + + monkeypatch.setattr( + SQLModel.metadata, + "create_all", + lambda bind, tables=None: created.update(bind=bind, tables=tables), + ) + monkeypatch.setattr(provision_module, "upgrade", lambda engine: upgrades.append(engine)) + yield {"connection": connection, "engine": engine, "created": created, "upgrades": upgrades} + + def test_the_advisory_lock_brackets_the_work(self, harness: dict[str, Any]) -> None: + # CREATE TABLE IF NOT EXISTS is not race-safe in Postgres, so + # concurrent provisioners must serialize on the lock. + create_all(harness["engine"]) + + sql = _sql(harness["connection"]) + assert "pg_advisory_lock" in sql + assert "pg_advisory_unlock" in sql + assert sql.index("pg_advisory_lock") < sql.index("pg_advisory_unlock") + + def test_the_lock_and_unlock_share_one_key(self, harness: dict[str, Any]) -> None: + create_all(harness["engine"]) + + keys = {params["k"] for _statement, params in harness["connection"].statements if params} + assert len(keys) == 1 + + def test_the_unlock_is_committed(self, harness: dict[str, Any]) -> None: + create_all(harness["engine"]) + + assert harness["connection"].commits == 1 + + def test_alembic_runs_after_the_tables_exist(self, harness: dict[str, Any]) -> None: + create_all(harness["engine"]) + + assert harness["created"]["bind"] is harness["engine"] + assert harness["upgrades"] == [harness["engine"]] + + def test_view_backed_models_are_left_to_alembic(self, harness: dict[str, Any]) -> None: + # A view cannot be created by metadata.create_all; the migrations own it. + create_all(harness["engine"]) + + tables = harness["created"]["tables"] + assert tables + assert all(not table.info.get("is_view") for table in tables) + + def test_a_view_backed_model_exists_to_be_excluded(self) -> None: + # Guards the filter above from silently becoming a no-op. + from sqlmodel import SQLModel + + assert any(table.info.get("is_view") for table in SQLModel.metadata.sorted_tables) + + def test_the_lock_is_released_even_when_the_work_fails( + self, harness: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + provision_module, "upgrade", lambda engine: (_ for _ in ()).throw(RuntimeError("migration broke")) + ) + + with pytest.raises(RuntimeError, match="migration broke"): + create_all(harness["engine"]) + + assert "pg_advisory_unlock" in _sql(harness["connection"]) + + def test_without_an_engine_the_global_one_is_used( + self, harness: dict[str, Any], monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(provision_module, "get_engine", lambda: harness["engine"]) + + create_all() + + assert harness["created"]["bind"] is harness["engine"] + + +@pytest.fixture +def maintenance(monkeypatch: pytest.MonkeyPatch) -> Any: + """Intercept the maintenance-database engine the provisioners build. + + Args: + monkeypatch: Fixture used to swap ``sqlalchemy.create_engine``. + + Returns: + A callable taking the rows ``fetchone`` should yield and returning + the recorded DSN, kwargs, connection and engine. + """ + + def install(rows: list[Any] | None = None) -> dict[str, Any]: + record: dict[str, Any] = {} + connection = FakeConnection(rows) + engine = FakeEngine(connection) + + def create_engine(dsn: str, **kwargs: Any) -> FakeEngine: + record.update(dsn=dsn, kwargs=kwargs) + return engine + + monkeypatch.setattr(sqlalchemy, "create_engine", create_engine) + record["connection"] = connection + record["engine"] = engine + return record + + return install + + +class TestEnsureDatabase: + """``CREATE DATABASE`` is issued through the ``postgres`` maintenance database.""" + + def test_it_connects_to_the_maintenance_database(self, maintenance: Any) -> None: + record = maintenance() + + ensure_database("postgresql://user:pw@host:5432/interloper") + + assert record["dsn"] == "postgresql://user:pw@host:5432/postgres" + # CREATE DATABASE cannot run inside a transaction. + assert record["kwargs"]["isolation_level"] == "AUTOCOMMIT" + + def test_a_missing_database_is_created(self, maintenance: Any) -> None: + record = maintenance(rows=[]) + + ensure_database("postgresql://host/interloper") + + sql = _sql(record["connection"]) + assert "SELECT 1 FROM pg_database" in sql + assert 'CREATE DATABASE "interloper"' in sql + + def test_an_existing_database_is_left_alone(self, maintenance: Any) -> None: + record = maintenance(rows=[(1,)]) + + ensure_database("postgresql://host/interloper") + + assert "CREATE DATABASE" not in _sql(record["connection"]) + + def test_the_engine_is_disposed(self, maintenance: Any) -> None: + record = maintenance() + + ensure_database("postgresql://host/interloper") + + assert record["engine"].disposed == 1 + + @pytest.mark.parametrize("dsn", ["postgresql://host", "postgresql://host/"]) + def test_a_dsn_naming_no_database_is_a_no_op(self, maintenance: Any, dsn: str) -> None: + record = maintenance() + + ensure_database(dsn) + + assert record == {} or "dsn" not in record + + +class TestDropDatabase: + """``DROP DATABASE`` first evicts the sessions that would block it.""" + + def test_active_connections_are_terminated_before_the_drop(self, maintenance: Any) -> None: + record = maintenance() + + drop_database("postgresql://host/interloper") + + sql = _sql(record["connection"]) + assert "pg_terminate_backend" in sql + assert 'DROP DATABASE IF EXISTS "interloper"' in sql + assert sql.index("pg_terminate_backend") < sql.index("DROP DATABASE") + + def test_the_terminate_spares_this_session(self, maintenance: Any) -> None: + # Otherwise the statement kills the connection issuing it. + record = maintenance() + + drop_database("postgresql://host/interloper") + + assert "pid <> pg_backend_pid()" in _sql(record["connection"]) + + def test_it_connects_to_the_maintenance_database(self, maintenance: Any) -> None: + record = maintenance() + + drop_database("postgresql://user:pw@host:5432/interloper") + + assert record["dsn"] == "postgresql://user:pw@host:5432/postgres" + assert record["kwargs"]["isolation_level"] == "AUTOCOMMIT" + + def test_the_engine_is_disposed(self, maintenance: Any) -> None: + record = maintenance() + + drop_database("postgresql://host/interloper") + + assert record["engine"].disposed == 1 + + @pytest.mark.parametrize("dsn", ["postgresql://host", "postgresql://host/"]) + def test_a_dsn_naming_no_database_is_a_no_op(self, maintenance: Any, dsn: str) -> None: + record = maintenance() + + drop_database(dsn) + + assert record == {} or "dsn" not in record diff --git a/packages/interloper-db/tests/test_session.py b/packages/interloper-db/tests/test_session.py new file mode 100644 index 000000000..6491fde85 --- /dev/null +++ b/packages/interloper-db/tests/test_session.py @@ -0,0 +1,77 @@ +"""Tests for ``interloper_db.session``. + +The session policy is exercised through the ``Store`` in +``tests/store/test_base.py``; this covers the primitives directly, including +the dialect dispatch that keeps every upsert portable. +""" + +from __future__ import annotations + +from uuid import uuid4 + +import pytest +from sqlalchemy import Engine +from sqlalchemy.dialects.postgresql import insert as postgresql_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert +from sqlmodel import Session, select + +from interloper_db.models import Component +from interloper_db.session import commit, dialect_insert, session_scope, transaction + +_ORG = uuid4() + + +class TestDialectInsert: + """``on_conflict_do_update`` is dialect-specific, so the constructor is chosen.""" + + def test_sqlite_gets_the_sqlite_constructor(self, component_db: Engine): + with Session(component_db) as session: + assert dialect_insert(session) is sqlite_insert + + def test_postgres_gets_the_postgres_constructor(self, component_db: Engine, monkeypatch): + # The store's upserts must compile to ON CONFLICT on the real backend. + with Session(component_db) as session: + bind = session.get_bind() + monkeypatch.setattr(type(bind.dialect), "name", "postgresql", raising=False) + + assert dialect_insert(session) is postgresql_insert + + +class TestSessionScope: + """A standalone call opens its own session and commits on its own.""" + + def test_a_write_lands(self, component_db: Engine): + with session_scope(component_db) as session: + session.add(Component(org_id=_ORG, kind="job", key="cron_job")) + commit(session) + + with Session(component_db) as session: + assert len(session.exec(select(Component)).all()) == 1 + + def test_nested_scopes_share_one_session(self, component_db: Engine): + with session_scope(component_db) as outer, session_scope(component_db) as inner: + assert inner is outer + + +class TestTransaction: + """An explicit block makes several calls one atomic unit.""" + + def test_the_block_commits_once_completed(self, component_db: Engine): + with transaction(component_db) as session: + session.add(Component(org_id=_ORG, kind="job", key="a")) + session.add(Component(org_id=_ORG, kind="job", key="b")) + + with Session(component_db) as session: + assert len(session.exec(select(Component)).all()) == 2 + + def test_an_error_rolls_the_whole_block_back(self, component_db: Engine): + with pytest.raises(RuntimeError, match="boom"), transaction(component_db) as session: + session.add(Component(org_id=_ORG, kind="job", key="a")) + raise RuntimeError("boom") + + with Session(component_db) as session: + assert session.exec(select(Component)).all() == [] + + def test_a_call_inside_the_block_joins_it(self, component_db: Engine): + with transaction(component_db) as outer, session_scope(component_db) as inner: + assert inner is outer