tgit/tests/pipeline/test_watcher.py

"""PipelineWatcher with a scripted CI, a fake deployer, an offline Telegram and a hand-turned clock."""

import asyncio
from collections.abc import Callable
from datetime import datetime, timedelta
from typing import Any

import pytest
from aiogram import Bot
from aiogram.methods import EditMessageText, SendDocument, SendMessage
from sqlalchemy import select

from app.access.service import AccessService
from app.db.models import PipelineRun, PipelineStatus, Push, Stage, utcnow
from app.db.session import SessionMaker
from app.pipeline.gitea import GiteaError
from app.pipeline.provider import CIRun
from app.pipeline.watcher import PipelineWatcher, initial_status
from app.runner.client import RunnerResult
from app.runner.deployer import Deployer, Outcome, Report
from app.texts import Texts
from tests.conftest import ADMIN
from tests.fakes import BOT_TOKEN, FakeSession

S = PipelineStatus
URL = "https://ci.example/runs/1"


class FakeCI:
    def __init__(self) -> None:
        self.runs: dict[str, CIRun | None] = {}
        self.log = "ruff check .\nF821 undefined name 'x'"
        self.errors = 0  # the next N calls fail like a flaky API

    def set(self, sha: str, status: PipelineStatus, step: str | None = None) -> None:
        self.runs[sha] = CIRun("1", status, URL, step)

    async def find(self, sha: str) -> CIRun | None:
        if self.errors:
            self.errors -= 1
            raise GiteaError("HTTP 502")
        return self.runs.get(sha)

    async def failure_log(self, run: CIRun) -> str:
        return self.log

    async def aclose(self) -> None:
        pass


class FakeDeployer(Deployer):
    def __init__(self, fail: frozenset[str] = frozenset()) -> None:  # no repo, no runner
        self.calls: list[tuple[str, list[str]]] = []
        self.fail = fail

    async def deploy(self, sha: str, services: list[str], report: Report) -> None:
        self.calls.append((sha, services))
        for svc in services:
            ok = svc not in self.fail
            await report(Outcome(svc, RunnerResult(ok, 0 if ok else 3, "ok" if ok else "not healthy", "boom\nFAIL")))


class Clock:
    def __init__(self) -> None:
        self.now = utcnow()

    def __call__(self) -> datetime:
        return self.now


@pytest.fixture
def tg() -> tuple[Bot, FakeSession]:
    session = FakeSession()
    return Bot(BOT_TOKEN, session=session), session


@pytest.fixture
def ci() -> FakeCI:
    return FakeCI()


@pytest.fixture
def clock() -> Clock:
    return Clock()


Make = Callable[..., PipelineWatcher]


@pytest.fixture
def make(sessions: SessionMaker, texts: Texts, tg: tuple[Bot, FakeSession], ci: FakeCI, clock: Clock) -> Make:
    def build(deployer: Deployer | None = None, with_ci: bool = True) -> PipelineWatcher:
        return PipelineWatcher(
            sessions, tg[0], texts, ci if with_ci else None, deployer, poll_interval=0.01, clock=clock
        )

    return build


async def add_push(
    sessions: SessionMaker, sha: str, paths: list[str], ci: PipelineStatus = S.PENDING, cd: PipelineStatus = S.PENDING
) -> Push:
    push = Push(
        user_id=ADMIN, sha=sha, message="feat: x", paths=paths, chat_id=ADMIN, message_id=7, ci_status=ci, cd_status=cd
    )
    async with sessions.begin() as s:
        s.add(push)
    return push


async def state(sessions: SessionMaker, sha: str) -> tuple[PipelineStatus, PipelineStatus]:
    async with sessions() as s:
        push = await s.scalar(select(Push).where(Push.sha == sha))
        assert push is not None
        return push.ci_status, push.cd_status


async def settle(watcher: PipelineWatcher, sessions: SessionMaker, sha: str) -> tuple[PipelineStatus, PipelineStatus]:
    """CD runs in a background task. Wait for it, not for the DB: in tests SQLite is one shared connection, and a
    read racing the watcher's write could roll that write back."""
    async with asyncio.timeout(5):
        await watcher.drain()
    return await state(sessions, sha)


def texts_of(session: FakeSession, method: Any) -> list[str]:
    return [m.text if hasattr(m, "text") else m.caption for m in session.sent(method)]


SHA_A = "a" * 40
SHA_B = "b" * 40
HELLO = ["services/hello/main.py"]


def test_initial_status() -> None:
    assert initial_status(True, True, HELLO) == (S.PENDING, S.PENDING)
    assert initial_status(True, True, ["README.md"]) == (S.PENDING, S.SKIPPED)  # no service touched
    assert initial_status(False, True, HELLO) == (S.SKIPPED, S.PENDING)
    assert initial_status(True, False, HELLO) == (S.PENDING, S.SKIPPED)


async def test_ci_green_then_deploy(
    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
) -> None:
    t, session = texts.pipeline, tg[1]
    deployer = FakeDeployer()
    watcher = make(deployer)
    await add_push(sessions, SHA_A, HELLO)

    assert await watcher.tick() is True  # no run yet: nothing to say, keep polling
    assert session.requests == []

    ci.set(SHA_A, S.RUNNING)
    await watcher.tick()
    assert texts_of(session, EditMessageText)[-1].endswith(
        f"{t.ci_running}\n{URL}\n{t.cd_waiting.format(services='hello')}"
    )
    await watcher.tick()  # same status — no edit
    assert len(session.sent(EditMessageText)) == 1

    ci.set(SHA_A, S.SUCCESS)
    assert await watcher.tick() is False
    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.SUCCESS)

    assert deployer.calls == [(SHA_A, ["hello"])]
    edits = texts_of(session, EditMessageText)
    assert edits[-1].endswith(f"{t.ci_success}\n{URL}\n{t.cd_success.format(services='hello')}")
    assert any(t.cd_running.format(services="hello") in e for e in edits)
    sent = texts_of(session, SendMessage)
    assert sent == [
        t.notify_ci_ok.format(sha="aaaaaaa", url=URL),
        t.notify_cd_ok.format(services="hello", sha="aaaaaaa"),
    ]
    assert all(m.reply_parameters.message_id == 7 for m in session.sent(SendMessage))
    assert await watcher.tick() is False  # nothing left: the loop would sleep until wake()


async def test_ci_red_reports_step_and_log_and_skips_deploy(
    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
) -> None:
    t, session = texts.pipeline, tg[1]
    deployer = FakeDeployer()
    watcher = make(deployer)
    await add_push(sessions, SHA_A, HELLO)
    ci.set(SHA_A, S.FAILED, "check / lint")

    await watcher.tick()

    assert await state(sessions, SHA_A) == (S.FAILED, S.SKIPPED)
    assert deployer.calls == []
    assert texts_of(session, EditMessageText)[-1].endswith(
        f"{t.ci_failed.format(step='check / lint')}\n{URL}\n{t.cd_not_run}"
    )
    [note] = texts_of(session, SendMessage)
    assert note.startswith("❌ CI упал на шаге <b>check / lint</b> · aaaaaaa")
    assert "<pre>ruff check .\nF821 undefined name &#x27;x&#x27;</pre>" in note
    assert session.sent(SendDocument) == []  # short log: inline only
    async with sessions() as s:
        row = await s.scalar(select(PipelineRun).where(PipelineRun.stage == Stage.CI))
        assert row is not None and (row.run_slug, row.status, row.failed_cube, row.url) == (
            "1",
            S.FAILED,
            "check / lint",
            URL,
        )


async def test_long_log_also_goes_as_a_file(
    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession]
) -> None:
    ci.log = "\n".join(f"line {i}" for i in range(2000))
    await add_push(sessions, SHA_A, HELLO)
    ci.set(SHA_A, S.FAILED, "check / tests")

    await make(FakeDeployer()).tick()

    [note] = texts_of(tg[1], SendMessage)
    assert "line 1999" in note and "line 0\n" not in note and len(note) < 4096
    [doc] = tg[1].sent(SendDocument)
    assert doc.document.filename == "ci-aaaaaaa.log"


async def test_no_run_for_too_long_times_out(
    access: AccessService, sessions: SessionMaker, make: Make, clock: Clock, tg: tuple[Bot, FakeSession], texts: Texts
) -> None:
    deployer = FakeDeployer()
    watcher = make(deployer)
    await add_push(sessions, SHA_A, HELLO)
    clock.now += timedelta(minutes=29)
    await watcher.tick()
    assert await state(sessions, SHA_A) == (S.PENDING, S.PENDING)

    clock.now += timedelta(minutes=2)
    await watcher.tick()

    assert await state(sessions, SHA_A) == (S.TIMEOUT, S.SKIPPED)
    assert deployer.calls == []
    assert texts_of(tg[1], SendMessage) == [texts.pipeline.notify_ci_timeout.format(sha="aaaaaaa", minutes=30)]


async def test_failed_deploy(
    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
) -> None:
    t = texts.pipeline
    paths = ["services/hello/main.py", "services/weather/main.py"]
    await add_push(sessions, SHA_A, paths)
    ci.set(SHA_A, S.SUCCESS)

    watcher = make(FakeDeployer(fail=frozenset({"weather"})))
    await watcher.tick()

    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.FAILED)
    assert texts_of(tg[1], EditMessageText)[-1].endswith(t.cd_failed.format(services="weather"))
    sent = texts_of(tg[1], SendMessage)
    assert sent[1].startswith("⛔ <b>weather</b> не развёрнут: not healthy") and "<pre>boom" in sent[1]
    assert sent[2] == t.notify_cd_ok.format(services="hello", sha="aaaaaaa")


async def test_newer_push_wins_when_ci_finishes_out_of_order(
    access: AccessService, sessions: SessionMaker, make: Make, ci: FakeCI, tg: tuple[Bot, FakeSession], texts: Texts
) -> None:
    deployer = FakeDeployer()
    watcher = make(deployer)
    await add_push(sessions, SHA_A, HELLO)
    await add_push(sessions, SHA_B, [*HELLO, "services/weather/main.py"])
    ci.set(SHA_B, S.SUCCESS)
    await watcher.tick()
    await settle(watcher, sessions, SHA_B)

    ci.set(SHA_A, S.SUCCESS)  # the older push turns green later: deploying it would roll hello back
    await watcher.tick()

    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.SKIPPED)
    assert deployer.calls == [(SHA_B, ["hello", "weather"])]
    assert any(texts.pipeline.cd_superseded in e for e in texts_of(tg[1], EditMessageText))


async def test_restart_picks_up_an_interrupted_deploy(
    access: AccessService, sessions: SessionMaker, make: Make
) -> None:
    await add_push(sessions, SHA_A, HELLO, ci=S.SUCCESS, cd=S.RUNNING)  # the bot died mid-deploy
    deployer = FakeDeployer()
    watcher = make(deployer)

    await watcher.tick()

    assert await settle(watcher, sessions, SHA_A) == (S.SUCCESS, S.SUCCESS)
    assert deployer.calls == [(SHA_A, ["hello"])]


async def test_without_ci_deploys_right_away(access: AccessService, sessions: SessionMaker, make: Make) -> None:
    await add_push(sessions, SHA_A, HELLO, ci=S.SKIPPED)
    deployer = FakeDeployer()

    watcher = make(deployer, with_ci=False)

    assert await watcher.tick() is False
    assert await settle(watcher, sessions, SHA_A) == (S.SKIPPED, S.SUCCESS)


async def test_ci_switched_off_does_not_leave_pushes_waiting(
    access: AccessService, sessions: SessionMaker, make: Make
) -> None:
    await add_push(sessions, SHA_A, ["README.md"], cd=S.SKIPPED)

    await make(FakeDeployer(), with_ci=False).tick()

    assert await state(sessions, SHA_A) == (S.SKIPPED, S.SKIPPED)


async def test_loop_survives_api_errors_and_wakes_on_push(
    access: AccessService,
    sessions: SessionMaker,
    make: Make,
    ci: FakeCI,
    tg: tuple[Bot, FakeSession],
    texts: Texts,
) -> None:
    watcher = make(FakeDeployer())
    done = texts.pipeline.notify_cd_ok.format(services="hello", sha="aaaaaaa")
    await add_push(sessions, SHA_B, ["README.md"], ci=S.SKIPPED, cd=S.SKIPPED)  # finished: nothing to watch
    loop = asyncio.create_task(watcher.run())
    try:
        await asyncio.sleep(0.05)
        assert tg[1].requests == []  # idle until woken
        ci.errors = 2
        ci.set(SHA_A, S.SUCCESS)
        await add_push(sessions, SHA_A, HELLO)
        watcher.wake()
        async with asyncio.timeout(5):
            while done not in texts_of(tg[1], SendMessage):  # noqa: ASYNC110 - fake Telegram, no event
                await asyncio.sleep(0.01)
            await watcher.drain()
        assert ci.errors == 0
    finally:
        loop.cancel()
    assert await state(sessions, SHA_A) == (S.SUCCESS, S.SUCCESS)