tgit/tests/runner/test_deploy.py

"""Bot side of #13: the SSH client, the deployer and the push → deploy flow, down to the real tgit-runner script."""

import asyncio
import base64
import json
import subprocess
from collections.abc import Callable
from dataclasses import replace
from datetime import UTC, datetime
from pathlib import Path

import pytest
from aiogram import Bot, Dispatcher
from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendMessage
from aiogram.types import InlineKeyboardMarkup

from app.access.service import AccessService
from app.alerts.broadcast import TelegramNotifier
from app.alerts.detectors import load_alerts_config
from app.alerts.service import AlertService
from app.bot.routers.commit import Act, CommitCb
from app.bot.routers.status import StatusAct, StatusCb, render_card, runner_service
from app.config import RunnerSettings
from app.db.models import Alert, AlertKind
from app.db.session import SessionMaker
from app.main import build_dispatcher
from app.repo.git import UploadsRepo
from app.runner.client import RunnerClient, RunnerResult
from app.runner.deployer import Deployer, Outcome, changed_services
from app.texts import Texts
from tests.alerts.test_detectors import RUNNING
from tests.conftest import ADMIN, ROOT
from tests.fakes import TgClient
from tests.runner.test_tgit_runner import FAKE, RUNNER

HERE = Path(__file__).parent


@pytest.fixture
def fake_host(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """ssh → tgit-runner → fake docker, all on PATH."""
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    (bin_dir / "ssh").symlink_to(HERE / "fake_ssh.py")
    (bin_dir / "docker").symlink_to(FAKE)
    monkeypatch.setenv("PATH", f"{bin_dir}:/usr/bin:/bin")
    monkeypatch.setenv("FAKE_RUNNER", str(RUNNER))
    monkeypatch.setenv("FAKE_DOCKER_STATE", str(tmp_path / "state.json"))
    monkeypatch.setenv("FAKE_DOCKER_LOG", str(tmp_path / "calls.log"))
    monkeypatch.setenv("TGIT_RUNNER_STATE", str(tmp_path / "runner"))
    monkeypatch.setenv("TGIT_HEALTH_TIMEOUT", "2")
    monkeypatch.setenv("TGIT_SETTLE", "0")
    return tmp_path


def docker_state(host: Path) -> dict[str, object]:
    return json.loads((host / "state.json").read_text())


def client(key: bytes | None = None) -> RunnerClient:
    b64 = base64.b64encode(key).decode() if key else None
    return RunnerClient(RunnerSettings(host="host.example", ssh_key_b64=b64, timeout=30))  # type: ignore[arg-type]


def test_changed_services() -> None:
    paths = ["services/hello/main.py", "services/hello/x/y.py", "services/Bad/main.py", "README.md", "services/a.py"]
    assert changed_services(paths) == ["hello"]


async def test_client_validates_before_ssh(fake_host: Path) -> None:
    result = await client().deploy("../etc", "abc1234", b"")

    assert (result.ok, result.code) == (False, 2)
    assert not (fake_host / "calls.log").exists()


def archive_of(remote: Path) -> bytes:
    return subprocess.run(
        ["git", "archive", "--format=tar", "HEAD:services/hello"], cwd=remote, capture_output=True, check=True
    ).stdout


async def until(condition: Callable[[], bool]) -> None:
    """The deploy runs in a background task: wait for its visible effect."""
    async with asyncio.timeout(15):
        while not condition():  # noqa: ASYNC110 - polling a fake Telegram session, no event to wait on
            await asyncio.sleep(0.05)


async def test_client_through_forced_command(fake_host: Path, remote: Path) -> None:
    c = client(b"-----BEGIN KEY-----")
    archive = archive_of(remote)

    ok = await c.deploy("hello", "abc1234", archive)
    restart = await c.restart("hello")
    bad = await c.run("deploy", "hello;", "rm")

    assert ok.ok and ok.summary.startswith("deployed hello abc1234")
    assert restart.ok
    assert (bad.ok, bad.code) == (False, 2)
    assert "tgit-svc-hello" in docker_state(fake_host)["containers"]  # type: ignore[operator]


class Recorder:
    def __init__(self) -> None:
        self.outcomes: list[Outcome] = []

    async def __call__(self, o: Outcome) -> None:
        self.outcomes.append(o)


async def test_deployer_deploys_and_removes(fake_host: Path, repo: UploadsRepo, remote: Path, tmp_path: Path) -> None:
    await repo.ensure()
    sha = await repo.head()
    rec = Recorder()
    deployer = Deployer(repo, client())

    await deployer.deploy(sha, ["hello", "gone"], rec)

    hello, gone = rec.outcomes
    assert hello.result.ok and not hello.removed
    assert gone.removed and gone.result.ok
    calls = [json.loads(line) for line in (fake_host / "calls.log").read_text().splitlines()]
    assert any(c[0] == "run" for c in calls)


async def test_refused_deploy_raises_alert(
    fake_host: Path, repo: UploadsRepo, sessions: SessionMaker, monkeypatch: pytest.MonkeyPatch
) -> None:
    await repo.ensure()
    monkeypatch.setenv("FAKE_BUILD_FAIL", "1")
    events: list[Alert] = []

    class Notes:
        async def opened(self, alert: Alert, admins_only: bool) -> None:
            events.append(alert)

        async def repeated(self, alert: Alert) -> None: ...

        async def resolved(self, alert: Alert) -> None: ...

    alerts = AlertService(sessions, Notes(), load_alerts_config(ROOT / "config"))
    rec = Recorder()
    await Deployer(repo, client(), alerts).deploy(await repo.head(), ["hello"], rec)

    [o] = rec.outcomes
    assert (o.result.ok, o.result.code, o.result.summary) == (False, 5, "build failed")
    [alert] = events
    assert alert.kind is AlertKind.DEPLOY_REFUSED and alert.service == "tgit-svc-hello"
    assert "could not install" in alert.payload["log"]


async def test_one_broken_service_does_not_stop_others(repo: UploadsRepo) -> None:
    class Flaky(RunnerClient):
        async def run(self, *args: str, stdin: bytes = b"") -> RunnerResult:
            if args[1] == "aaa":
                raise RuntimeError("ssh exploded")
            return RunnerResult(True, 0, f"deployed {args[1]}", "OK")

    await repo.ensure()
    rec = Recorder()
    await Deployer(repo, Flaky(RunnerSettings(host="h"))).deploy(await repo.head(), ["aaa", "hello"], rec)

    assert [(o.service, o.result.ok) for o in rec.outcomes] == [("aaa", False), ("hello", True)]


# ---- in the bot -------------------------------------------------------------------------------------------------


@pytest.fixture
def tg(texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, fake_host: Path) -> TgClient:
    def dispatcher(bot: Bot) -> Dispatcher:
        alerts = AlertService(
            sessions, TelegramNotifier(bot, access, sessions, texts), load_alerts_config(ROOT / "config")
        )
        return build_dispatcher(texts, sessions, access, repo, alerts=alerts, runner=client())

    return TgClient(dispatcher)


async def test_push_deploys_touched_service(tg: TgClient, texts: Texts, fake_host: Path, remote: Path) -> None:
    await tg.send(ADMIN, texts.menu.commit)
    await tg.press(ADMIN, CommitCb(a=Act.OPEN, i=2).pack())  # services
    await tg.press(ADMIN, CommitCb(a=Act.OPEN, i=0).pack())  # services/hello
    await tg.press(ADMIN, CommitCb(a=Act.NEW_FILE).pack())
    await tg.send(ADMIN, "util.py")
    await tg.send(ADMIN, "X = 1\n")
    await tg.press(ADMIN, CommitCb(a=Act.MESSAGE).pack())
    await tg.send(ADMIN, "add util")
    await tg.press(ADMIN, CommitCb(a=Act.PUSH).pack())
    await tg.press(ADMIN, CommitCb(a=Act.PUSH_GO).pack())
    await until(lambda: any("работает" in (m.text or "") for m in tg.session.sent(SendMessage)))

    pushed = [e for e in tg.session.sent(EditMessageText) if "Запушено" in (e.text or "")]
    assert pushed and "🚀 Разворачиваю: hello…" in pushed[-1].text
    [done] = [m for m in tg.session.sent(SendMessage) if "работает" in (m.text or "")]
    assert done.text.startswith("🚀 <b>hello</b> работает · ")
    assert "tgit-svc-hello" in docker_state(fake_host)["containers"]  # type: ignore[operator]


def buttons(markup: InlineKeyboardMarkup) -> list[str]:
    return [b.text for row in markup.inline_keyboard for b in row]


def test_restart_button_only_for_runner_services(texts: Texts) -> None:
    service = replace(RUNNING, labels={"tgit.service": "hello"})
    now = datetime.now(UTC)

    assert texts.status.restart in buttons(render_card(texts, service, None, now, can_restart=True)[1])
    assert texts.status.restart not in buttons(render_card(texts, service, None, now)[1])  # not an admin
    assert texts.status.restart not in buttons(render_card(texts, RUNNING, None, now, can_restart=True)[1])  # the bot
    assert runner_service(service) == "hello" and runner_service(RUNNING) is None


async def test_restart_through_status(
    texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, monkeypatch: pytest.MonkeyPatch
) -> None:
    from tests.runtime.fake_docker import FakeDocker

    fake = FakeDocker()
    svc = fake.add("tgit-svc-hello", labels={"tgit.service": "hello"})
    calls: list[tuple[str, ...]] = []

    async def run(self: RunnerClient, *args: str, stdin: bytes = b"") -> RunnerResult:
        calls.append(args)
        return RunnerResult(True, 0, "restarted hello", "OK restarted hello")

    monkeypatch.setattr(RunnerClient, "run", run)
    async with fake.client() as docker:
        tg = TgClient(build_dispatcher(texts, sessions, access, repo, docker=docker, runner=client()))
        await tg.press(ADMIN, StatusCb(a=StatusAct.RESTART, id=svc.id[:12]).pack())

    assert calls == [("restart", "hello")]
    assert tg.session.sent(AnswerCallbackQuery)[0].text == texts.status.restarted