turnstone/tests/test_child_source.py

"""Unit tests for :mod:`turnstone.core.child_source`.

Covers both strategies in isolation against fakes — no live collector,
no live SessionManager. Adapter-level integration coverage continues to
live in ``test_coordinator_adapter.py``.
"""

from __future__ import annotations

import contextlib
import time
from typing import TYPE_CHECKING, Any

from turnstone.core.child_source import ClusterChildSource, SameNodeChildSource
from turnstone.core.children_registry import ChildrenRegistry
from turnstone.core.workstream import WorkstreamState

if TYPE_CHECKING:
    import queue


# ---------------------------------------------------------------------------
# SameNodeChildSource
# ---------------------------------------------------------------------------


class _FakeManager:
    """Minimal SessionManager stand-in implementing the subscribe API."""

    def __init__(self) -> None:
        self.subscribers: list[Any] = []

    def subscribe_to_state(self, callback: Any) -> None:
        self.subscribers.append(callback)

    def unsubscribe_from_state(self, callback: Any) -> None:
        with contextlib.suppress(ValueError):
            self.subscribers.remove(callback)

    def fire(self, ws_id: str, state: WorkstreamState) -> None:
        for cb in self.subscribers:
            cb(ws_id, state)


class TestSameNodeChildSource:
    def test_start_subscribes_to_manager(self) -> None:
        mgr = _FakeManager()
        registry = ChildrenRegistry()
        src = SameNodeChildSource(mgr, registry)
        sink_calls: list[dict[str, Any]] = []
        src.start(sink=sink_calls.append)
        assert len(mgr.subscribers) == 1

    def test_state_change_for_known_child_pushes_to_sink(self) -> None:
        mgr = _FakeManager()
        registry = ChildrenRegistry()
        registry.install("p1", object())
        registry.add_child("p1", "c1")
        src = SameNodeChildSource(mgr, registry)
        sink_calls: list[dict[str, Any]] = []
        src.start(sink=sink_calls.append)

        mgr.fire("c1", WorkstreamState.RUNNING)

        assert len(sink_calls) == 1
        ev = sink_calls[0]
        assert ev["type"] == "cluster_state"
        assert ev["ws_id"] == "c1"
        assert ev["state"] == "running"

    def test_state_change_for_unknown_workstream_is_dropped(self) -> None:
        mgr = _FakeManager()
        registry = ChildrenRegistry()
        src = SameNodeChildSource(mgr, registry)
        sink_calls: list[dict[str, Any]] = []
        src.start(sink=sink_calls.append)

        # No registry entry — pre-filter drops the event without
        # invoking the sink.
        mgr.fire("ws-unknown", WorkstreamState.IDLE)
        assert sink_calls == []

    def test_shutdown_unsubscribes(self) -> None:
        mgr = _FakeManager()
        registry = ChildrenRegistry()
        src = SameNodeChildSource(mgr, registry)
        src.start(sink=lambda ev: None)
        assert len(mgr.subscribers) == 1
        src.shutdown()
        assert mgr.subscribers == []

    def test_start_is_idempotent(self) -> None:
        mgr = _FakeManager()
        registry = ChildrenRegistry()
        src = SameNodeChildSource(mgr, registry)
        src.start(sink=lambda ev: None)
        src.start(sink=lambda ev: None)
        # Second start is a no-op; only one subscription.
        assert len(mgr.subscribers) == 1

    def test_sink_exception_does_not_propagate(self) -> None:
        mgr = _FakeManager()
        registry = ChildrenRegistry()
        registry.install("p1", object())
        registry.add_child("p1", "c1")
        src = SameNodeChildSource(mgr, registry)

        def bad_sink(ev: dict[str, Any]) -> None:
            raise RuntimeError("sink boom")

        src.start(sink=bad_sink)
        # Should not raise — the strategy catches sink failures and logs.
        mgr.fire("c1", WorkstreamState.RUNNING)


# ---------------------------------------------------------------------------
# ClusterChildSource
# ---------------------------------------------------------------------------


class _FakeCollector:
    """Minimal ClusterCollector stand-in providing the listener API."""

    def __init__(self, snapshot: dict[str, Any] | None = None) -> None:
        self._snapshot = snapshot or {"nodes": []}
        self.queues: list[queue.Queue[dict[str, Any]]] = []
        self.unregistered: list[queue.Queue[dict[str, Any]]] = []

    def get_snapshot_and_register(self, q: queue.Queue[dict[str, Any]]) -> dict[str, Any]:
        self.queues.append(q)
        return self._snapshot

    def unregister_listener(self, q: queue.Queue[dict[str, Any]]) -> None:
        self.unregistered.append(q)

    def emit(self, event: dict[str, Any]) -> None:
        """Push an event to all registered listener queues."""
        for q in self.queues:
            q.put(event)


class TestClusterChildSource:
    def test_start_subscribes_to_collector(self) -> None:
        coll = _FakeCollector()
        registry = ChildrenRegistry()
        src = ClusterChildSource(
            collector=coll,
            registry=registry,
            parents_provider=list,
        )
        try:
            src.start(sink=lambda ev: None)
            assert len(coll.queues) == 1
        finally:
            src.shutdown()

    def test_start_primes_registry_from_snapshot(self) -> None:
        snapshot = {
            "nodes": [
                {
                    "workstreams": [
                        {"id": "c1", "parent_ws_id": "p1"},
                        {"id": "c2", "parent_ws_id": "p1"},
                        # Unknown parent — dropped
                        {"id": "x", "parent_ws_id": "p-unknown"},
                    ],
                },
            ],
        }
        coll = _FakeCollector(snapshot)
        registry = ChildrenRegistry()
        registry.install("p1", object())
        src = ClusterChildSource(
            collector=coll,
            registry=registry,
            parents_provider=lambda: ["p1"],
        )
        try:
            src.start(sink=lambda ev: None)
            assert set(registry.children_of("p1")) == {"c1", "c2"}
            assert registry.parent_for("x") is None
        finally:
            src.shutdown()

    def test_event_dispatched_to_sink(self) -> None:
        coll = _FakeCollector()
        registry = ChildrenRegistry()
        src = ClusterChildSource(
            collector=coll,
            registry=registry,
            parents_provider=list,
        )
        sink_calls: list[dict[str, Any]] = []

        try:
            src.start(sink=sink_calls.append)
            coll.emit({"type": "cluster_state", "ws_id": "c1", "state": "running"})
            # Daemon thread loop has 1.0s queue timeout; poll briefly.
            for _ in range(20):
                if sink_calls:
                    break
                time.sleep(0.05)
            assert len(sink_calls) == 1
            assert sink_calls[0]["ws_id"] == "c1"
        finally:
            src.shutdown()

    def test_shutdown_unregisters_and_joins_thread(self) -> None:
        coll = _FakeCollector()
        registry = ChildrenRegistry()
        src = ClusterChildSource(
            collector=coll,
            registry=registry,
            parents_provider=list,
        )
        src.start(sink=lambda ev: None)
        src.shutdown()
        assert coll.unregistered == coll.queues
        # Second shutdown is a no-op (idempotent).
        src.shutdown()

    def test_start_is_idempotent(self) -> None:
        coll = _FakeCollector()
        registry = ChildrenRegistry()
        src = ClusterChildSource(
            collector=coll,
            registry=registry,
            parents_provider=list,
        )
        try:
            src.start(sink=lambda ev: None)
            src.start(sink=lambda ev: None)
            assert len(coll.queues) == 1
        finally:
            src.shutdown()

    def test_sink_exception_does_not_kill_thread(self) -> None:
        coll = _FakeCollector()
        registry = ChildrenRegistry()
        src = ClusterChildSource(
            collector=coll,
            registry=registry,
            parents_provider=list,
        )
        survived_calls: list[dict[str, Any]] = []
        call_count = [0]

        def flaky_sink(ev: dict[str, Any]) -> None:
            call_count[0] += 1
            if call_count[0] == 1:
                raise RuntimeError("first one boom")
            survived_calls.append(ev)

        try:
            src.start(sink=flaky_sink)
            coll.emit({"type": "cluster_state", "ws_id": "c1", "state": "x"})
            coll.emit({"type": "cluster_state", "ws_id": "c2", "state": "y"})
            for _ in range(40):
                if survived_calls:
                    break
                time.sleep(0.05)
            assert len(survived_calls) == 1
            assert survived_calls[0]["ws_id"] == "c2"
        finally:
            src.shutdown()


# Multi-subscriber observer tests for ``SessionManager.subscribe_to_state``
# / ``unsubscribe_from_state`` live in ``test_session_manager.py`` where
# the proper FakeAdapter / FakeStorage construction helpers already exist.