svcforge/tests/integration/test_day2.py

"""Day 2 against a real Postgres: the work list, the halt, and the window.

The work-list query is the entire rollout, so it is tested where it runs. `order by
team = %s desc` and `not exists (... halted)` are SQL semantics — a fake repo asserting
them would only prove the fake agrees with itself.
"""

from __future__ import annotations

from datetime import UTC, datetime
from uuid import UUID

import pytest

from svcforge_core.domain.models import TaskKind
from svcforge_core.domain.states import InstanceState
from svcforge_core.domain.windows import parse_window, schedule_upgrade_at
from svcforge_core.repo.db import DictPool
from svcforge_core.repo.instances import InstanceRepo
from svcforge_core.repo.tasks import TaskRepo
from tests.integration.helpers import make_instance

OLD = "21.3.19"  # what is deployed
PINNED = "21.3.20"  # what catalog.yaml now says
OWN_TEAM = "platform"


async def _set_window(pool: DictPool, instance_id: UUID, spec: str | None) -> None:
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute(
            "update instances set maintenance_window = %s where id = %s",
            (spec, instance_id),
        )


async def _halt(pool: DictPool, service_type: str) -> None:
    """What `handle_verify` does on a failed probe, and what you undo by hand with SQL."""
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute(
            """insert into catalog_versions (service_type, rollout_state) values (%s, 'halted')
               on conflict (service_type) do update set rollout_state = 'halted'""",
            (service_type,),
        )


@pytest.fixture
async def fleet(pool: DictPool) -> list[UUID]:
    """Three ready instances on the old version. The own-team one is created LAST.

    Created last on purpose: `created_at` is the tiebreak, so if the `team = %s desc` sort
    were dropped this fixture makes the test fail instead of passing by luck.
    """
    ids = [
        await make_instance(pool, team="tenant-a", state=InstanceState.READY, chart_version=OLD),
        await make_instance(pool, team="tenant-b", state=InstanceState.READY, chart_version=OLD),
        await make_instance(pool, team=OWN_TEAM, state=InstanceState.READY, chart_version=OLD),
    ]
    return ids


async def test_work_list_returns_one_row_and_it_is_the_own_team_row(
    pool: DictPool, fleet: list[UUID]
) -> None:
    """max_in_flight=1 means one instance moves at a time, and yours is the guinea pig."""
    repo = InstanceRepo(pool)

    rows = await repo.list_upgradable(
        service_type="elasticsearch",
        catalog_version=PINNED,
        own_team=OWN_TEAM,
        max_in_flight=1,
    )

    assert len(rows) == 1
    assert rows[0].instance.team == OWN_TEAM
    assert rows[0].instance.id == fleet[2]
    assert rows[0].instance.chart_version == OLD


async def test_work_list_skips_instances_already_on_the_pinned_version(
    pool: DictPool, fleet: list[UUID]
) -> None:
    """The query is the progress bar: as instances land on PINNED, the list drains to empty."""
    repo = InstanceRepo(pool)
    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("update instances set chart_version = %s", (PINNED,))

    rows = await repo.list_upgradable(
        service_type="elasticsearch", catalog_version=PINNED, own_team=OWN_TEAM, max_in_flight=10
    )

    assert rows == []


async def test_halted_rollout_returns_zero_rows(pool: DictPool, fleet: list[UUID]) -> None:
    """One column stops the fleet. This is the whole stop button."""
    repo = InstanceRepo(pool)
    await _halt(pool, "elasticsearch")

    rows = await repo.list_upgradable(
        service_type="elasticsearch",
        catalog_version=PINNED,
        own_team=OWN_TEAM,
        max_in_flight=10,  # generous on purpose: it is the halt returning 0, not the limit
    )

    assert rows == []


async def test_halt_is_scoped_to_one_service_type(pool: DictPool) -> None:
    """A broken redis chart must not freeze elasticsearch upgrades."""
    repo = InstanceRepo(pool)
    await make_instance(
        pool, team=OWN_TEAM, service_type="redis", state=InstanceState.READY, chart_version=OLD
    )
    await make_instance(
        pool, team=OWN_TEAM, service_type="elasticsearch", state=InstanceState.READY, chart_version=OLD
    )
    await _halt(pool, "redis")

    assert await repo.list_upgradable("redis", PINNED, OWN_TEAM, 10) == []
    assert len(await repo.list_upgradable("elasticsearch", PINNED, OWN_TEAM, 10)) == 1


async def test_windowed_upgrade_is_scheduled_in_the_future_and_security_bypasses_it(
    pool: DictPool,
) -> None:
    """The window lands in `tasks.run_after`, and `security: true` ignores it.

    Both paths go through the real enqueue, so this also pins the `timestamptz` round-trip:
    an aware UTC datetime must come back out of Postgres still aware and still that instant.
    """
    repo = InstanceRepo(pool)
    tasks = TaskRepo(pool)
    iid = await make_instance(pool, team=OWN_TEAM, state=InstanceState.READY, chart_version=OLD)
    await _set_window(pool, iid, "0 3 * * 0|Asia/Ho_Chi_Minh")

    (candidate,) = await repo.list_upgradable("elasticsearch", PINNED, OWN_TEAM, 1)
    window = parse_window(candidate.maintenance_window)
    assert window is not None

    now = datetime.now(UTC)

    routine = await tasks.enqueue_standalone(
        iid, TaskKind.UPGRADE, schedule_upgrade_at(window, security=False, now=now)
    )
    urgent = await tasks.enqueue_standalone(
        iid, TaskKind.UPGRADE, schedule_upgrade_at(window, security=True, now=now)
    )

    async with pool.connection() as conn, conn.cursor() as cur:
        await cur.execute("select id, run_after from tasks where id = any(%s)", ([routine, urgent],))
        run_after = {r["id"]: r["run_after"] for r in await cur.fetchall()}

    assert run_after[routine] > now  # waits for 03:00 Sunday, Vietnam time
    assert run_after[urgent] <= now  # a public exploit does not wait
    assert run_after[routine].tzinfo is not None