FabledCurator/tests/test_worker_control.py

"""Changing a lane's slots on a running system (milestone 422 step 2).

The celery control calls are stubbed: what is being tested is the DELTA
ARITHMETIC and the persist/push split, not that celery can resize its own
pool. A test that asserted celery's behaviour would be testing celery.
"""

from __future__ import annotations

import pytest

from backend.app.services import worker_control as wc
from backend.app.services.worker_lanes import LANES_BY_NAME

# --- what inspect reports ----------------------------------------------------


class _Control:
    """Records the control messages that were sent."""

    def __init__(self):
        self.grew: list[tuple[int, list[str]]] = []
        self.shrank: list[tuple[int, list[str]]] = []
        self.added: list[tuple[str, list[str]]] = []
        self.cancelled: list[tuple[str, list[str]]] = []

    def pool_grow(self, n, destination=None):
        self.grew.append((n, destination))

    def pool_shrink(self, n, destination=None):
        self.shrank.append((n, destination))

    def add_consumer(self, queue, destination=None):
        self.added.append((queue, destination))

    def cancel_consumer(self, queue, destination=None):
        self.cancelled.append((queue, destination))


def _stub_live(
    monkeypatch, lane_name, *, pools, present=True, reserved=0, consuming=None,
):
    # `consuming` defaults to the lane's full queue set — i.e. an ENABLED
    # lane. Tests for the disabled case pass an empty set explicitly.
    if consuming is None:
        consuming = set(LANES_BY_NAME[lane_name].queues)
    state = wc.LaneLiveState(
        present=present,
        replicas=len(pools),
        hostnames=sorted(pools),
        pools=dict(pools),
        reserved=reserved,
        consuming=set(consuming),
    )
    monkeypatch.setattr(
        wc, "inspect_lanes_sync",
        lambda: {name: (state if name == lane_name else wc.LaneLiveState())
                 for name in LANES_BY_NAME},
    )
    return state


def _stub_control(monkeypatch):
    control = _Control()

    class _Celery:
        pass

    celery = _Celery()
    celery.control = control
    import sys
    import types
    mod = types.ModuleType("backend.app.celery_app")
    mod.celery = celery
    monkeypatch.setitem(sys.modules, "backend.app.celery_app", mod)
    return control


def test_pool_property_is_max_not_sum():
    """`slots` means the pool size of ONE process, so the aggregate shown to
    the operator is the largest replica — not the total. A sum would report 8
    for two replicas of 4 and invite them to 'reduce it to 4', which would
    halve the lane."""
    state = wc.LaneLiveState(pools={"a": 4, "b": 4})
    assert state.pool == 4


def test_pool_is_none_when_nothing_reported():
    """Unknown, never zero — the distinction step 3's reconcile depends on."""
    assert wc.LaneLiveState(present=True).pool is None


# --- the delta arithmetic ----------------------------------------------------


def test_each_replica_gets_its_own_delta(monkeypatch):
    """The bug this exists to prevent: one delta computed from an aggregate
    and applied to every replica. With replicas at 2 and 6 and a target of 4,
    a shared delta moves both the same way and leaves them at 4 and 8 — or 0
    and 4 — depending on which aggregate was used. Per-replica deltas
    converge both on 4.
    """
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={"host-a": 2, "host-b": 6})

    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)

    assert (applied, err) == (True, None)
    assert control.grew == [(2, ["host-a"])]
    assert control.shrank == [(2, ["host-b"])]


def test_a_replica_already_at_the_target_is_sent_nothing(monkeypatch):
    """The fixed point step 3's reconcile needs. An enforcer that re-issues a
    grow of zero every tick never converges and re-does its own work forever
    (lesson #4183)."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={"host-a": 4})

    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)

    assert (applied, err) == (True, None)
    assert control.grew == []
    assert control.shrank == []


def test_resizing_an_absent_lane_reports_rather_than_raises(monkeypatch):
    _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={}, present=False)

    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)

    assert applied is False
    assert "not running" in err


def test_a_replica_with_no_reported_pool_is_named_not_skipped_silently(
    monkeypatch,
):
    """Resize what can be resized, then say which could not. Silence would
    leave a replica running at a size the UI claims it is not."""
    control = _stub_control(monkeypatch)
    state = _stub_live(monkeypatch, "worker", pools={"host-a": 2})
    state.hostnames = ["host-a", "host-b"]  # b answered inspect, not stats
    state.replicas = 2

    applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4)

    assert control.grew == [(2, ["host-a"])]
    assert applied is False
    assert "host-b" in err


# --- enabling and disabling --------------------------------------------------


def test_disabling_cancels_consumers_rather_than_killing_the_worker(monkeypatch):
    """A cancelled consumer keeps the process alive and answering inspect, so
    a disabled lane stays visible. A killed worker reads as ABSENT, which is
    the same signal as a crash — and milestone 365 exists precisely so those
    two do not look alike."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "ml", pools={"host-a": 1})

    applied, err = wc.set_lane_enabled_sync(LANES_BY_NAME["ml"], False)

    assert (applied, err) == (True, None)
    assert control.cancelled == [("ml", ["host-a"])]
    assert control.added == []


def test_enabling_adds_a_consumer_for_every_queue_in_the_lane(monkeypatch):
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={"host-a": 1})

    wc.set_lane_enabled_sync(LANES_BY_NAME["worker"], True)

    assert [q for q, _ in control.added] == list(LANES_BY_NAME["worker"].queues)


# --- lane identity -----------------------------------------------------------


def test_queue_sets_map_back_to_their_lane_in_any_order():
    """celery does not guarantee the order it lists a worker's queues in, so
    the lookup sorts. Unsorted, a lane would intermittently fail to match and
    read as absent."""
    lane = LANES_BY_NAME["worker"]
    assert wc._lane_for_queues(tuple(reversed(lane.queues))) is lane


def test_an_unknown_queue_set_maps_to_no_lane():
    """A deployment slicing CELERY_QUEUES differently has no lane row to
    control. Honest rather than an error — the roster still reports it."""
    assert wc._lane_for_queues(("something", "else")) is None


# --- the reconcile (step 3) --------------------------------------------------


def test_a_settled_system_sends_no_control_messages(monkeypatch):
    """THE property this sweep lives or dies on. It runs every 5 minutes
    forever, so a converged tick must be silent — one broker round trip and
    nothing else. An enforcer that re-issues a grow of zero churns forever and
    buries a real correction in its own noise (lesson #4183).
    """
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={"host-a": 4})

    result = wc.reconcile_lanes_sync({"worker": (4, True)})

    # EVERY control family, not just the pool ones. The first version of this
    # test asserted only grew/shrank and would have passed while the reconcile
    # re-sent add_consumer for all four queues on every tick — harmless per
    # call, unbounded churn in aggregate, and invisible.
    assert control.grew == []
    assert control.shrank == []
    assert control.added == []
    assert control.cancelled == []
    assert result["changed"] == []


def test_a_worker_back_at_its_env_concurrency_is_corrected(monkeypatch):
    """The failure this exists for: a restart drops the pool to CELERY_
    CONCURRENCY, silently below what the operator set."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={"host-a": 2})

    result = wc.reconcile_lanes_sync({"worker": (6, True)})

    assert control.grew == [(4, ["host-a"])]
    assert result["changed"] == ["worker"]


def test_an_absent_lane_is_skipped_not_corrected(monkeypatch):
    """`present=False` is 'nothing answered', not 'zero slots'. Correcting it
    would be a conclusion drawn from an unswept read (snippet #3969) — and
    there is nothing to send the message to anyway."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={}, present=False)

    result = wc.reconcile_lanes_sync({"worker": (6, True)})

    assert result["skipped"] == ["worker"]
    assert result["changed"] == []
    assert control.grew == []
    assert control.shrank == []


def test_one_lane_failing_does_not_stop_the_others(monkeypatch):
    """A broker blip on one lane must not leave the rest un-reconciled for
    another five minutes."""
    control = _stub_control(monkeypatch)
    present = wc.LaneLiveState(
        present=True, replicas=1, hostnames=["host-a"], pools={"host-a": 1},
    )
    absent = wc.LaneLiveState()
    monkeypatch.setattr(
        wc, "inspect_lanes_sync",
        lambda: {"worker": absent, "scheduler": present,
                 "maintenance_long": absent, "ml": absent},
    )

    result = wc.reconcile_lanes_sync({
        "worker": (4, True), "scheduler": (3, True),
    })

    assert result["skipped"] == ["worker"]
    assert result["changed"] == ["scheduler"]
    assert control.grew == [(2, ["host-a"])]


def test_a_lane_disabled_in_settings_but_still_consuming_is_stopped(monkeypatch):
    """The case that made `consuming` necessary: a lane turned off while its
    worker was down comes back consuming, and must be stopped when it
    returns."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "ml", pools={"host-a": 0}, consuming={"ml"})

    result = wc.reconcile_lanes_sync({"ml": (0, False)})

    assert control.cancelled == [("ml", ["host-a"])]
    assert result["changed"] == ["ml"]


def test_an_already_disabled_lane_is_not_cancelled_again(monkeypatch):
    """The other half of the fixed point. `cancel_consumer` on a queue that is
    not being consumed succeeds and does nothing, so an unconditional
    reconcile would churn here forever with no symptom."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "ml", pools={"host-a": 0}, consuming=set())

    result = wc.reconcile_lanes_sync({"ml": (0, False)})

    assert control.cancelled == []
    assert control.added == []
    assert result["changed"] == []


def test_a_lane_with_no_stored_row_is_left_alone(monkeypatch):
    """A lane in LANES but not in `desired` is one whose row has not been
    seeded. Inventing a target here would let the sweep enforce a number that
    disagrees with the migration it is supposed to be upholding."""
    control = _stub_control(monkeypatch)
    _stub_live(monkeypatch, "worker", pools={"host-a": 2})

    result = wc.reconcile_lanes_sync({})

    assert result["changed"] == []
    assert control.grew == []


def test_the_reconcile_task_is_registered_and_scheduled():
    """A task name only enters `celery.tasks` when its module is imported, and
    a beat entry naming a task that is not registered fails at tick time
    rather than at import — silently, every five minutes."""
    import backend.app.tasks.maintenance  # noqa: F401
    from backend.app.celery_app import celery

    name = "backend.app.tasks.maintenance.reconcile_worker_lanes"
    assert name in celery.tasks
    scheduled = {e["task"] for e in celery.conf.beat_schedule.values()}
    assert name in scheduled