FabledCurator/tests/test_worker_lanes.py

"""Worker-lane definitions and the derived ceiling (milestone 422 step 1).

Pure — no DB, no broker. The ceiling reads files under /sys/fs/cgroup, so the
tests point it at temporary files instead of at whatever the CI container
happens to have.
"""

from __future__ import annotations

import pytest

from backend.app.services import worker_lanes as wl

# --- the lane definitions ----------------------------------------------------


def test_lane_names_are_unique_and_stable():
    names = [lane.name for lane in wl.LANES]
    assert len(names) == len(set(names))
    # The key every settings row and every API call uses. Renaming one orphans
    # its row silently, so the set is pinned rather than merely counted.
    assert set(names) == {"worker", "scheduler", "maintenance_long", "ml"}


def test_queue_sets_are_unique():
    """Two lanes sharing a queue set would be indistinguishable to
    `celery inspect`, which groups by exactly that (see service_roster)."""
    keys = [lane.queue_key for lane in wl.LANES]
    assert len(keys) == len(set(keys))


def test_every_routed_queue_has_a_lane_that_serves_it():
    """The check that actually matters: a queue celery routes tasks to, with
    no lane consuming it, is work that queues forever and never runs.

    Read from `celery_app`'s real routing table rather than a list written
    here, so adding a route without a lane fails this test.
    """
    from backend.app.celery_app import celery as celery_app

    routes = celery_app.conf.task_routes or {}
    routed = {spec["queue"] for spec in routes.values() if "queue" in spec}
    served = {q for lane in wl.LANES for q in lane.queues}
    assert routed - served == set(), (
        f"queues with no lane to consume them: {sorted(routed - served)}"
    )


def test_defaults_are_one_of_each_with_ml_off():
    """Operator, 2026-09-22: 'that starting value should be one of each.'

    ML at zero AND disabled is milestone 422 step 6's requirement arriving
    early: enabling the lane is what triggers the SigLIP download, and rule
    164 allows a runtime fetch only for a feature that is optional and clearly
    off. A default of 1 here would make every fresh install reach HuggingFace.
    """
    by_name = wl.LANES_BY_NAME
    for name in ("worker", "scheduler", "maintenance_long"):
        assert by_name[name].default_slots == 1
        assert by_name[name].default_enabled is True
    assert by_name["ml"].default_slots == 0
    assert by_name["ml"].default_enabled is False


def test_default_caps_leave_room_but_are_not_the_ceiling():
    """A cap that starts at the ceiling is a rubber stamp. Each lane's cap
    must be at least its default slots (or the CHECK constraint rejects the
    seeded row) and must leave somewhere to grow."""
    for lane in wl.LANES:
        assert lane.default_slots_cap >= lane.default_slots
        assert lane.default_slots_cap >= 1


def test_only_ml_is_memory_bound():
    """Which lane is memory-bound decides which ceiling formula applies. If
    another lane ever becomes so, it needs its own per-slot cost — the ML
    figure is about a SigLIP copy and means nothing for a thumbnailer."""
    assert [lane.name for lane in wl.LANES if lane.memory_bound] == ["ml"]


# --- reading the container's limits ------------------------------------------


def _point_memory_at(monkeypatch, tmp_path, contents: str | None, *, v2=True):
    target = tmp_path / "memory.max"
    if contents is not None:
        target.write_text(contents)
    attr = "_CGROUP_V2_MEMORY" if v2 else "_CGROUP_V1_MEMORY"
    monkeypatch.setattr(wl, attr, target)
    # Neutralise the other tier so the test controls exactly one source.
    other = "_CGROUP_V1_MEMORY" if v2 else "_CGROUP_V2_MEMORY"
    monkeypatch.setattr(wl, other, tmp_path / "absent")


def test_memory_read_from_cgroup_v2(monkeypatch, tmp_path):
    _point_memory_at(monkeypatch, tmp_path, str(8 * wl.GIB))
    assert wl.container_memory_bytes() == 8 * wl.GIB


def test_memory_read_from_cgroup_v1(monkeypatch, tmp_path):
    target = tmp_path / "limit_in_bytes"
    target.write_text(str(6 * wl.GIB))
    monkeypatch.setattr(wl, "_CGROUP_V2_MEMORY", tmp_path / "absent")
    monkeypatch.setattr(wl, "_CGROUP_V1_MEMORY", target)
    assert wl.container_memory_bytes() == 6 * wl.GIB


def test_v2_max_means_no_limit_not_zero(monkeypatch, tmp_path):
    """'max' is a container with no memory limit set. Falling through to host
    RAM is right; reading it as 0 or as an error would cap a large machine at
    the unknown-ceiling."""
    _point_memory_at(monkeypatch, tmp_path, "max")
    value = wl.container_memory_bytes()
    assert value is not None and value > wl.GIB


def test_v1_sentinel_is_recognised_as_unlimited(monkeypatch, tmp_path):
    """cgroup v1 spells 'no limit' as a PAGE_SIZE-aligned LONG_MAX rather than
    a word, so it is recognised by magnitude. Taken literally it would be
    petabytes and the ML ceiling would be nonsense."""
    target = tmp_path / "limit_in_bytes"
    target.write_text(str(9223372036854771712))
    monkeypatch.setattr(wl, "_CGROUP_V2_MEMORY", tmp_path / "absent")
    monkeypatch.setattr(wl, "_CGROUP_V1_MEMORY", target)
    value = wl.container_memory_bytes()
    assert value is not None
    assert value < wl._UNLIMITED_ABOVE


# --- the derived ceiling -----------------------------------------------------


def test_ml_ceiling_is_memory_divided_by_per_slot_cost(monkeypatch, tmp_path):
    # 2 GiB reserved for web and the other lanes, then 4 GiB per model copy.
    _point_memory_at(monkeypatch, tmp_path, str(14 * wl.GIB))
    assert wl.derived_ceiling(wl.LANES_BY_NAME["ml"]) == 3


def test_a_small_box_is_told_it_cannot_run_tagging(monkeypatch, tmp_path):
    """Honestly zero rather than a floor of one. A 4GB box cannot hold a model
    alongside the web process, and offering a slot that OOMs the container the
    first time it is used is precisely what the ceiling exists to prevent —
    the operator's 'I don't want it to kill their servers'."""
    _point_memory_at(monkeypatch, tmp_path, str(4 * wl.GIB))
    assert wl.derived_ceiling(wl.LANES_BY_NAME["ml"]) == 0


def test_unreadable_memory_limit_yields_a_low_ceiling_not_an_unlimited_one(
    monkeypatch, tmp_path
):
    """The failure DIRECTION is the point. Not knowing how much memory there
    is must never read as 'plenty' — an unswept absence is not a verdict."""
    _point_memory_at(monkeypatch, tmp_path, None)
    monkeypatch.setattr(wl.os, "sysconf", lambda _: (_ for _ in ()).throw(OSError))
    assert wl.derived_ceiling(wl.LANES_BY_NAME["ml"]) == wl.UNKNOWN_CEILING


def test_cpu_lanes_are_bounded_by_cores_not_memory(monkeypatch, tmp_path):
    """A CPU-bound lane on a memory-starved box must still get slots — its
    slots are processes, not model copies."""
    _point_memory_at(monkeypatch, tmp_path, str(2 * wl.GIB))
    monkeypatch.setattr(wl, "container_cpu_count", lambda: 8)
    assert wl.derived_ceiling(wl.LANES_BY_NAME["worker"]) == 8


def test_cpu_quota_beats_host_core_count(monkeypatch, tmp_path):
    """`os.cpu_count()` reports the HOST's cores from inside a container, so a
    4-core quota on a 32-core host would otherwise offer 32 slots. The
    operator's own stack sets `cpus: '4.0'` on ml-worker, so this is real
    configuration rather than a hypothetical."""
    quota = tmp_path / "cpu.max"
    quota.write_text("400000 100000")
    monkeypatch.setattr(wl, "_CGROUP_V2_CPU", quota)
    monkeypatch.setattr(wl.os, "cpu_count", lambda: 32)
    assert wl.container_cpu_count() == 4


def test_a_single_core_box_still_gets_a_slot(monkeypatch):
    monkeypatch.setattr(wl, "container_cpu_count", lambda: 1)
    assert wl.derived_ceiling(wl.LANES_BY_NAME["worker"]) >= wl.MIN_CEILING


def test_the_ceiling_is_computed_not_stored(monkeypatch, tmp_path):
    """The property the design rests on: the same lane yields a different
    ceiling when the container's limits change, with no row edit. A stored
    ceiling would keep authorising what the box no longer has."""
    lane = wl.LANES_BY_NAME["ml"]
    _point_memory_at(monkeypatch, tmp_path, str(34 * wl.GIB))
    big = wl.derived_ceiling(lane)
    (tmp_path / "memory.max").write_text(str(10 * wl.GIB))
    small = wl.derived_ceiling(lane)
    assert big > small


def test_ceilings_covers_every_lane():
    assert set(wl.ceilings()) == {lane.name for lane in wl.LANES}


# --- the duplicate this step collapsed ---------------------------------------


def test_role_names_is_derived_from_the_lane_table():
    """`service_roster.ROLE_NAMES` was a hand-kept second copy of 'queue set
    -> display name' and had already drifted: maintenance_long is a live lane
    with four task routes and a dedicated worker, and the roster did not know
    its name, so the System tab rendered `Worker (maintenance_long)`.

    Asserting the derivation rather than the contents — a test listing the
    names again would be a third copy.
    """
    from backend.app.services.service_roster import ROLE_NAMES, role_display_name

    assert ROLE_NAMES == {lane.queue_key: lane.display_name for lane in wl.LANES}
    assert role_display_name(("maintenance_long",)) == "Long maintenance"


def test_an_unrecognised_queue_set_still_gets_a_true_label():
    """A deployment slicing CELERY_QUEUES differently must not be handed a
    name this code invented for it."""
    from backend.app.services.service_roster import role_display_name

    assert role_display_name(("nonsense",)) == "Worker (nonsense)"


# --- the model's invariant ---------------------------------------------------


@pytest.mark.parametrize(
    "slots,cap,ok",
    [
        (0, 0, True),
        (1, 1, True),
        (1, 4, True),
        (5, 4, False),   # slots above its own cap
        (-1, 1, False),  # negative slots
        (1, -1, False),  # negative cap
    ],
)
def test_worker_lane_check_constraints(slots, cap, ok):
    """The constraints live in the database, not only in the service, because
    a row violating `slots <= slots_cap` is not a rejected request — it is a
    lane the reconcile (step 3) will drive UP to a number the operator
    capped."""
    from backend.app.models import WorkerLane

    constraints = {
        c.name: str(c.sqltext) for c in WorkerLane.__table__.constraints
        if hasattr(c, "sqltext")
    }
    # The names carry the convention's `ck_worker_lane_` prefix ALREADY — the
    # model declares them bare and Base.metadata's naming_convention applies it.
    # Asserting the prefixed form is what pins the thing that actually went
    # wrong once: alembic 0088 had to rename four constraints that shipped as
    # `ck_x_ck_x_name`, because the migration pre-prefixed a name the
    # convention then prefixed again (#3275). A bare-name assertion here would
    # pass just as happily against a doubled one.
    assert constraints == {
        "ck_worker_lane_slots_non_negative": "slots >= 0",
        "ck_worker_lane_cap_non_negative": "slots_cap >= 0",
        "ck_worker_lane_slots_within_cap": "slots <= slots_cap",
    }
    for name in constraints:
        assert not name.startswith("ck_worker_lane_ck_"), f"doubled prefix: {name}"

    # Evaluate the same predicates the database will, so the parametrize table
    # documents what is accepted rather than restating the SQL.
    satisfied = slots >= 0 and cap >= 0 and slots <= cap
    assert satisfied is ok