turnstone/tests/test_storage_notify.py

"""Tests for the storage layer's cross-process ``notify`` / ``listen`` API.

Covers SQLite (synthetic-sweep + in-process fan-out) and PostgreSQL
(real ``LISTEN``/``NOTIFY``).  The PG-only cases are gated on the
``--storage-backend=postgresql`` flag so they no-op on default CI runs.
"""

from __future__ import annotations

import threading
import time

import pytest


def _drain_until(stream, predicate, deadline_sec: float = 5.0):
    """Poll ``stream`` until ``predicate`` matches one of the drained notifies.

    Returns the matching notify or raises ``TimeoutError``.  Tests use
    this so timing flakes against the bounded-blocking ``poll`` shape
    don't masquerade as logic bugs.
    """
    deadline = time.monotonic() + deadline_sec
    while time.monotonic() < deadline:
        remaining = max(0.05, deadline - time.monotonic())
        for n in stream.poll(min(0.5, remaining)):
            if predicate(n):
                return n
    msg = "no matching notify drained before deadline"
    raise TimeoutError(msg)


class TestSqliteNotify:
    """SQLite path: in-process fan-out + synthetic sweep."""

    def test_notify_no_listeners_is_noop(self, storage):
        # No exception, no side effect — safe to always call from dispatch.
        storage.notify("services", '{"op": "INSERT"}')

    def test_notify_delivers_to_in_process_listener(self, storage):
        with storage.listen(["services"]) as stream:
            storage.notify("services", '{"op": "INSERT"}')
            got = _drain_until(stream, lambda n: n.payload == '{"op": "INSERT"}')
            assert got.channel == "services"
            # ``pid`` is 0 on the SQLite synthetic path and the sending
            # backend's PID on Postgres — both are valid notify shapes,
            # so don't assert on the value here.

    def test_notify_filters_by_channel(self, storage):
        with storage.listen(["services"]) as stream:
            storage.notify("other_channel", "ignored")
            storage.notify("services", "wanted")
            got = _drain_until(stream, lambda n: True)
            assert got.payload == "wanted"

    def test_multiple_listeners_each_get_event(self, storage):
        # Two streams open on the same channel; each gets its own copy.
        with storage.listen(["services"]) as s1, storage.listen(["services"]) as s2:
            storage.notify("services", "broadcast")
            got1 = _drain_until(s1, lambda n: True)
            got2 = _drain_until(s2, lambda n: True)
            assert got1.payload == "broadcast"
            assert got2.payload == "broadcast"

    def test_close_stops_stream(self, storage):
        with storage.listen(["services"]) as stream:
            pass
        # After context exit, the stream is closed; poll returns [] without
        # blocking.  A second close() is idempotent.
        assert stream.poll(0.05) == []
        stream.close()

    def test_synthetic_sweep_emits_after_interval(self, storage, _is_sqlite):
        # Synthetic sweep is fundamentally SQLite-specific — the PG path
        # uses real ``LISTEN``/``NOTIFY`` and has no sweep tick.  Gate
        # so the test doesn't false-fail by waiting for a "sweep" notify
        # that the PG stream will never produce.
        with storage.listen(["services"], sweep_interval=0.1) as stream:
            # First poll: not yet at the interval, so likely empty.
            stream.poll(0.05)
            # Wait past the interval, then poll again — should emit a
            # synthetic-sweep notify per declared channel.
            time.sleep(0.15)
            got = _drain_until(stream, lambda n: n.payload == "sweep")
            assert got.channel == "services"
            assert got.payload == "sweep"

    def test_empty_channel_list_yields_empty_stream(self, storage):
        with storage.listen([]) as stream:
            # No channels — poll returns [] regardless of how long we wait.
            assert stream.poll(0.05) == []


# ---------------------------------------------------------------------------
# PostgreSQL path — gated on --storage-backend=postgresql.
# ---------------------------------------------------------------------------


@pytest.fixture
def _is_postgres(storage):
    """Skip the wrapped test when the active backend isn't Postgres."""
    if storage.__class__.__name__ != "PostgreSQLBackend":
        pytest.skip("PostgreSQL-specific test")
    return True


@pytest.fixture
def _is_sqlite(storage):
    """Skip the wrapped test when the active backend isn't SQLite."""
    if storage.__class__.__name__ != "SQLiteBackend":
        pytest.skip("SQLite-specific test")
    return True


class TestPostgresNotify:
    def test_round_trip(self, storage, _is_postgres):
        # Open a listener, fire a notify on a regular pooled connection,
        # drain the listener within a reasonable bound (PG NOTIFY is
        # typically sub-100ms on a local socket).
        with storage.listen(["pytest_round_trip"]) as stream:
            # Tiny sleep so the LISTEN settles before the NOTIFY fires —
            # otherwise the notify can arrive on the connection before
            # the LISTEN is registered (race only visible in tests).
            time.sleep(0.05)
            storage.notify("pytest_round_trip", '{"hello": "world"}')
            got = _drain_until(stream, lambda n: True, deadline_sec=3.0)
            assert got.channel == "pytest_round_trip"
            assert got.payload == '{"hello": "world"}'
            assert got.pid > 0

    def test_concurrent_notifies_all_arrive(self, storage, _is_postgres):
        with storage.listen(["pytest_concurrent"]) as stream:
            time.sleep(0.05)
            for i in range(5):
                storage.notify("pytest_concurrent", str(i))
            seen: set[str] = set()
            deadline = time.monotonic() + 3.0
            while len(seen) < 5 and time.monotonic() < deadline:
                for n in stream.poll(0.2):
                    seen.add(n.payload)
            assert seen == {"0", "1", "2", "3", "4"}

    def test_close_aborts_blocked_poll(self, storage, _is_postgres):
        # poll() should return promptly once close() runs on another thread.
        with storage.listen(["pytest_close"]) as stream:
            done = threading.Event()
            result: list[list] = []

            def _poll_long():
                result.append(stream.poll(5.0))
                done.set()

            t = threading.Thread(target=_poll_long, daemon=True)
            t.start()
            time.sleep(0.1)
            stream.close()
            assert done.wait(2.0), "close() did not unblock poll()"
            # No notify arrived, so the polled batch is empty — but the
            # poll loop must have exited well under the 5 s timeout.
            assert result == [[]]


class TestServicesTriggerFilter:
    """Migration 053's trigger: fires on real changes, quiet on heartbeats.

    PG-only — the SQLite path has no trigger and is covered by
    :class:`TestSqliteNotify`.  Verifies the in-trigger ``IS NOT DISTINCT
    FROM`` filter — a heartbeat-only UPDATE (same url + same metadata,
    only ``last_heartbeat`` changed) must NOT emit a NOTIFY, since
    ``register_service`` runs the same UPSERT on every 30 s tick × N
    nodes and the channel would otherwise flood.
    """

    def test_insert_fires_notify(self, storage, _is_postgres):
        with storage.listen(["services"]) as stream:
            time.sleep(0.05)
            storage.register_service("server", "pytest-trigger-node", "http://127.0.0.1:1")
            got = _drain_until(stream, lambda n: True, deadline_sec=3.0)
            assert got.channel == "services"
            assert '"op": "INSERT"' in got.payload or "INSERT" in got.payload
            # Cleanup so concurrent suites don't pick up the row.
            storage.deregister_service("server", "pytest-trigger-node")

    def test_delete_fires_notify(self, storage, _is_postgres):
        storage.register_service("server", "pytest-trigger-node-del", "http://127.0.0.1:2")
        with storage.listen(["services"]) as stream:
            time.sleep(0.05)
            storage.deregister_service("server", "pytest-trigger-node-del")
            got = _drain_until(stream, lambda n: True, deadline_sec=3.0)
            assert "DELETE" in got.payload

    def test_url_change_update_fires_notify(self, storage, _is_postgres):
        storage.register_service("server", "pytest-trigger-node-url", "http://127.0.0.1:3")
        with storage.listen(["services"]) as stream:
            time.sleep(0.05)
            # UPSERT with different url — UPDATE path with url diff,
            # trigger must fire.
            storage.register_service("server", "pytest-trigger-node-url", "http://127.0.0.1:9")
            got = _drain_until(stream, lambda n: True, deadline_sec=3.0)
            assert "UPDATE" in got.payload
            storage.deregister_service("server", "pytest-trigger-node-url")

    def test_heartbeat_only_update_is_quiet(self, storage, _is_postgres):
        # Open the LISTEN session FIRST so PG delivers the INSERT NOTIFY
        # to this connection — pg_notify routes only to sessions that
        # have LISTENed at COMMIT time, so an INSERT committed before the
        # listen opens would be lost and the drain would time out instead
        # of exercising the heartbeat-quiet check below.
        with storage.listen(["services"]) as stream:
            time.sleep(0.05)
            storage.register_service("server", "pytest-trigger-node-hb", "http://127.0.0.1:4")
            # Drain the INSERT notify so subsequent polls see only what
            # heartbeats emit (if anything).
            _drain_until(stream, lambda n: True, deadline_sec=2.0)
            # Now fire a heartbeat tick — same url + same metadata,
            # only last_heartbeat updates.  Trigger must NOT emit.
            storage.heartbeat_service("server", "pytest-trigger-node-hb")
            # Poll long enough that any spurious notify would have
            # arrived; the channel must stay silent.
            spurious = stream.poll(0.5)
            assert spurious == [], f"heartbeat-only update emitted unexpected notify: {spurious}"
            storage.deregister_service("server", "pytest-trigger-node-hb")