FabledCurator/tests/test_api_workers.py

"""/api/system/workers — the lane dial (milestone 422 step 2).

Exercises the real endpoint against the real database. Only `celery inspect`
is stubbed, and only to keep the suite fast: an unstubbed inspect blocks for
its full 2s timeout per call with no workers to answer, which several writes
would turn into most of the lane's runtime.
"""

import pytest
import pytest_asyncio
from sqlalchemy import select

from backend.app.models import WorkerLane
from backend.app.services import worker_control as wc
from backend.app.services.worker_lanes import LANES

pytestmark = pytest.mark.integration


@pytest_asyncio.fixture
async def no_live_workers(monkeypatch):
    """Nothing is running — which is the CI lane's actual truth, asserted
    rather than waited for. Makes every push fail, which is the interesting
    half: the setting must still be stored."""
    monkeypatch.setattr(
        wc, "inspect_lanes_sync",
        lambda: {lane.name: wc.LaneLiveState() for lane in LANES},
    )


async def _lane_row(db, name: str) -> WorkerLane:
    return (await db.execute(
        select(WorkerLane).where(WorkerLane.name == name)
    )).scalar_one()


# --- reading -----------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_lists_every_lane_with_its_ceiling(client, no_live_workers):
    resp = await client.get("/api/system/workers")
    assert resp.status_code == 200
    body = await resp.get_json()

    by_name = {lane["name"]: lane for lane in body["lanes"]}
    assert set(by_name) == {"worker", "scheduler", "maintenance_long", "ml"}
    for lane in body["lanes"]:
        assert lane["ceiling"] >= 0
        assert lane["slots_cap"] <= lane["ceiling"]
        # Nothing is running, so live state must say so rather than report
        # zeroes that read like a healthy idle lane.
        assert lane["live"]["present"] is False


@pytest.mark.asyncio
async def test_enabled_is_derived_from_the_cap_and_never_stored(
    client, no_live_workers,
):
    """The reshape of 2026-09-23. "Off" and "may use no workers" were two
    spellings of one fact, stored separately and free to disagree."""
    body = await (await client.get("/api/system/workers")).get_json()
    for lane in body["lanes"]:
        assert lane["enabled"] == (lane["slots_cap"] > 0), lane["name"]


@pytest.mark.asyncio
async def test_ml_ships_off(client, no_live_workers):
    """Rule 164's carve-out and the weak-hardware default in one row: raising
    the cap is what triggers the SigLIP download, so a fresh install must not
    find it above zero."""
    body = await (await client.get("/api/system/workers")).get_json()
    ml = next(lane for lane in body["lanes"] if lane["name"] == "ml")
    assert ml["slots_cap"] == 0
    assert ml["enabled"] is False


@pytest.mark.asyncio
async def test_the_other_lanes_ship_at_one(client, no_live_workers):
    """Operator: *"the cap defaults should be 1 and 0 for the ml-worker."*"""
    body = await (await client.get("/api/system/workers")).get_json()
    caps = {lane["name"]: lane["slots_cap"] for lane in body["lanes"]}
    assert caps == {
        "worker": 1, "scheduler": 1, "maintenance_long": 1, "ml": 0,
    }


# --- the persist / push split ------------------------------------------------


@pytest.mark.asyncio
async def test_raising_a_cap_stores_it_and_pushes_nothing(
    client, db, no_live_workers,
):
    """A cap is PERMISSION, not a request. Raising it must not grow the pool
    here — that would put workers on a lane with nothing to do — so there is
    nothing to push and `applied` is vacuously true.

    This asserted `applied is False` until run 7367, carried over from when
    the number meant "run this many". The code was right and the test was
    describing the control it replaced.
    """
    resp = await client.post("/api/system/workers/worker", json={"slots_cap": 3})

    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["slots_cap"] == 3
    assert body["applied"] is True
    assert (await _lane_row(db, "worker")).slots_cap == 3


@pytest.mark.asyncio
async def test_turning_a_lane_off_is_stored_even_when_it_cannot_be_pushed(
    client, db, no_live_workers,
):
    """The direction that DOES push. Consumers follow the cap immediately in
    both directions — off must take effect when it is asked for — so with
    nothing answering, the push fails.

    That is NOT a failed setting: the value is saved and the sizing pass
    carries it within a minute (lesson #4202 — a live change that does not
    survive, with nothing saying so). The UI says "saved, not yet live"
    rather than "that didn't work", which is the distinction `applied`
    exists to carry.
    """
    resp = await client.post("/api/system/workers/worker", json={"slots_cap": 0})

    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["applied"] is False
    assert "not running" in body["apply_error"]
    assert (await _lane_row(db, "worker")).slots_cap == 0


# --- the cap is the switch ---------------------------------------------------


@pytest.mark.asyncio
async def test_a_cap_of_zero_turns_the_lane_off(client, db, no_live_workers):
    await client.post("/api/system/workers/worker", json={"slots_cap": 0})

    row = await _lane_row(db, "worker")
    assert row.slots_cap == 0
    body = await (await client.get("/api/system/workers")).get_json()
    worker = next(lane for lane in body["lanes"] if lane["name"] == "worker")
    assert worker["enabled"] is False


@pytest.mark.asyncio
async def test_raising_it_off_zero_turns_the_lane_on(client, db, no_live_workers):
    await client.post("/api/system/workers/ml", json={"slots_cap": 1})

    assert (await _lane_row(db, "ml")).slots_cap == 1
    body = await (await client.get("/api/system/workers")).get_json()
    ml = next(lane for lane in body["lanes"] if lane["name"] == "ml")
    assert ml["enabled"] is True


@pytest.mark.asyncio
async def test_the_model_fetch_fires_on_the_transition_not_on_every_write(
    client, db, no_live_workers, monkeypatch,
):
    """Raising the cap off zero downloads SigLIP, once. A second nudge of the
    same dial must not re-enqueue a multi-GB download — and the trigger must
    be the TRANSITION rather than "a field was sent", which is what it tested
    before the UI stopped sending `enabled` at all."""
    monkeypatch.setattr(
        wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None),
    )
    monkeypatch.setattr(
        wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None),
    )
    fired = []
    monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True)

    first = await (await client.post(
        "/api/system/workers/ml", json={"slots_cap": 1},
    )).get_json()
    second = await (await client.post(
        "/api/system/workers/ml", json={"slots_cap": 2},
    )).get_json()

    assert first["fetching_models"] is True
    assert second["fetching_models"] is False
    assert fired == [1]


# --- what is refused ---------------------------------------------------------


@pytest.mark.asyncio
async def test_a_cap_above_the_derived_ceiling_is_refused(
    client, db, no_live_workers,
):
    """The ceiling is the machine's, not the operator's, and it is the one
    bound they cannot lower themselves past. The detail is written to be read
    by a person — a refused control with no reason reads as a bug."""
    before = (await _lane_row(db, "ml")).slots_cap

    resp = await client.post(
        "/api/system/workers/ml", json={"slots_cap": 10_000},
    )

    assert resp.status_code == 400
    body = await resp.get_json()
    assert "container can hold" in body["detail"]
    await _refreshed(db, "ml", before)


@pytest.mark.asyncio
async def test_a_negative_cap_is_refused(client, db, no_live_workers):
    resp = await client.post("/api/system/workers/worker", json={"slots_cap": -1})
    assert resp.status_code == 400


@pytest.mark.asyncio
async def test_a_boolean_is_not_accepted_as_a_cap(client, no_live_workers):
    """`True` is an int in Python. Reading it as a cap of 1 would be a control
    that appears to work and sets something nobody asked for."""
    resp = await client.post("/api/system/workers/worker", json={"slots_cap": True})
    assert resp.status_code == 400


@pytest.mark.asyncio
async def test_the_retired_fields_are_no_longer_accepted(client, no_live_workers):
    """`slots`, `enabled` and `autoscale` are gone. A client still sending one
    must be told, not silently ignored — a POST that returns 200 having
    changed nothing is the worst of the three outcomes."""
    for field in ("slots", "enabled", "autoscale"):
        resp = await client.post(
            "/api/system/workers/worker", json={field: 2},
        )
        assert resp.status_code == 400, field


@pytest.mark.asyncio
async def test_an_unknown_lane_is_refused_and_names_the_known_ones(
    client, no_live_workers,
):
    resp = await client.post("/api/system/workers/nope", json={"slots_cap": 1})
    assert resp.status_code == 400
    body = await resp.get_json()
    assert "worker" in body["known"]


@pytest.mark.asyncio
async def test_an_empty_body_is_refused_rather_than_treated_as_a_no_op(
    client, no_live_workers,
):
    resp = await client.post("/api/system/workers/worker", json={})
    assert resp.status_code == 400


async def _refreshed(db, name: str, expected: int) -> None:
    row = await _lane_row(db, name)
    await db.refresh(row)
    assert row.slots_cap == expected, "a refused write must store nothing"


# --- no database connection is held across a broker round trip ---------------


@pytest.mark.asyncio
async def test_the_lane_read_holds_no_session_while_it_inspects(monkeypatch):
    """Operator, 2026-09-23: *"something about changing the cap number is
    blocking to the website... it shouldn't be"*.

    Nothing here was slow in itself. A Postgres connection was held across a
    celery inspect whose budget is eleven seconds, on a page that polls every
    fifteen — so with a lane not answering, each poll pinned a connection for
    most of the interval. SQLAlchemy's default pool is five plus ten overflow;
    two browser tabs, the health endpoint doing the same, and a cap change
    adding more inspects exhausts it, and every OTHER request then waits on a
    connection. It surfaced as the whole site stalling rather than as one slow
    page, which is why it took a screenshot to find.

    Asserted STRUCTURALLY rather than by timing: `lane_view` must not accept a
    session at all. A timing test would be flaky, and a mock-call-order test
    would pass against a version that took the session and merely used it
    early — the property that matters is that it CANNOT.
    """
    import inspect as _inspect

    from backend.app.services.worker_control import lane_settings, lane_view

    assert "session" not in _inspect.signature(lane_view).parameters, (
        "lane_view takes a session again; the broker work must run with none held"
    )
    # And the DB half still exists, so the split did not simply lose the reads.
    assert "session" in _inspect.signature(lane_settings).parameters


@pytest.mark.asyncio
async def test_the_cap_write_holds_no_session_while_it_pushes():
    """The same property on the write path, where it was worse: a cap change
    could make three broker round trips, each with a connection held."""
    import inspect as _inspect

    from backend.app.services.worker_control import push_lane_cap, store_lane_cap

    assert "session" in _inspect.signature(store_lane_cap).parameters
    assert "session" not in _inspect.signature(push_lane_cap).parameters, (
        "push_lane_cap takes a session again; the push must run with none held"
    )


@pytest.mark.asyncio
async def test_raising_a_cap_costs_no_broker_round_trip_at_all(
    client, db, no_live_workers, monkeypatch,
):
    """The common case must be instant. Raising a cap is permission, not a
    request — the sizing pass spends it — so there is nothing to tell the
    broker, and the operator's `+` should answer immediately rather than
    waiting out an inspect."""
    from backend.app.services import worker_control as wc

    calls = []
    monkeypatch.setattr(
        wc, "inspect_lanes_sync", lambda: calls.append("inspect") or {},
    )
    monkeypatch.setattr(
        wc, "set_lane_slots_sync",
        lambda *a, **k: calls.append("resize") or (True, None),
    )
    monkeypatch.setattr(
        wc, "set_lane_enabled_sync",
        lambda *a, **k: calls.append("consumers") or (True, None),
    )

    await client.post("/api/system/workers/worker", json={"slots_cap": 1})
    calls.clear()
    resp = await client.post("/api/system/workers/worker", json={"slots_cap": 6})

    assert resp.status_code == 200
    assert calls == [], f"raising a cap talked to the broker: {calls}"