tgit/tests/bot/test_status.py

from collections.abc import AsyncIterator

import httpx
import pytest
from aiogram.methods import AnswerCallbackQuery, EditMessageText, SendMessage

from app.access.service import AccessService
from app.bot.routers.status import StatusAct, StatusCb, duration
from app.db.session import SessionMaker
from app.main import build_dispatcher
from app.repo.git import UploadsRepo
from app.runtime.checks import Checker, load_checks
from app.runtime.docker import DockerClient
from app.runtime.snapshots import SnapshotStore
from app.texts import Texts
from tests.conftest import ADMIN, ROOT
from tests.fakes import TgClient
from tests.runtime.fake_docker import FakeDocker


@pytest.fixture
def fake() -> FakeDocker:
    fake = FakeDocker()
    fake.add("tgit-dev-bot-1", service="bot", labels={"tgit.sha": "22d9f0c5abcdef0123"})
    fake.add("unhealthy", service="unhealthy", health="unhealthy")
    fake.add("vobot-api-1", managed=False)
    return fake


@pytest.fixture
async def docker(fake: FakeDocker) -> AsyncIterator[DockerClient]:
    async with fake.client() as client:
        yield client


@pytest.fixture
def tg(
    texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, docker: DockerClient
) -> TgClient:
    checker = Checker(
        docker,
        SnapshotStore(sessions),
        load_checks(ROOT / "config"),
        httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200))),
    )
    return TgClient(build_dispatcher(texts, sessions, access, repo, docker=docker, checker=checker))


def short(fake: FakeDocker, name: str) -> str:
    return next(c.id[:12] for c in fake.containers.values() if c.name == name)


async def test_list_shows_managed_containers_with_lights(tg: TgClient, texts: Texts) -> None:
    await tg.send(ADMIN, texts.menu.status)

    [reply] = tg.session.sent(SendMessage)
    assert "🟢 tgit-dev-bot-1 · running, healthy" in reply.text
    assert "🔴 unhealthy · running, unhealthy" in reply.text
    assert "vobot" not in reply.text
    labels = [b.text for row in reply.reply_markup.inline_keyboard for b in row]
    assert texts.status.check_button in labels and texts.status.refresh in labels


async def test_card(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.send(ADMIN, texts.menu.status)
    await tg.press(ADMIN, StatusCb(a=StatusAct.CARD, id=short(fake, "tgit-dev-bot-1")).pack())

    [card] = tg.session.sent(EditMessageText)
    assert card.text.startswith("🟢 tgit-dev-bot-1")
    assert "Ревизия: 22d9f0c5abcd" in card.text
    assert "CPU: 5.0% · RAM: 64.0 МБ из 256.0 МБ (25%)" in card.text
    assert "рестартов: 0" in card.text


async def test_card_of_stopped_container(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    fake.add("stopped", state="exited", health=None, exit_code=1)
    await tg.press(ADMIN, StatusCb(a=StatusAct.CARD, id=short(fake, "stopped")).pack())

    [card] = tg.session.sent(EditMessageText)
    assert card.text.startswith("🔴 stopped")
    assert "exited (exit 1)" in card.text
    assert f"CPU: {texts.status.none}" in card.text


async def test_run_all_checks(tg: TgClient, texts: Texts) -> None:
    await tg.press(ADMIN, StatusCb(a=StatusAct.CHECK).pack())

    [report] = tg.session.sent(EditMessageText)
    assert report.text.startswith("🩺 Проверки · без ошибок 1 из 2")
    assert "✅ tgit-dev-bot-1" in report.text
    assert "❌ unhealthy" in report.text
    assert f"❌ {texts.status.check_health} · unhealthy" in report.text
    [ack, *_] = tg.session.sent(AnswerCallbackQuery)
    assert ack.text == texts.status.checking


async def test_checks_of_one_container_offer_way_back(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.press(ADMIN, StatusCb(a=StatusAct.CHECK, id=short(fake, "unhealthy")).pack())

    [report] = tg.session.sent(EditMessageText)
    assert "без ошибок 0 из 1" in report.text
    labels = [b.text for row in report.reply_markup.inline_keyboard for b in row]
    assert texts.status.to_card in labels


async def test_foreign_or_vanished_container_is_refused(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.press(ADMIN, StatusCb(a=StatusAct.CARD, id=short(fake, "vobot-api-1")).pack())

    assert not tg.session.sent(EditMessageText)
    [alert] = tg.session.sent(AnswerCallbackQuery)
    assert alert.text == texts.status.unavailable and alert.show_alert


async def test_docker_down(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    fake.down = True
    await tg.send(ADMIN, texts.menu.status)

    assert [r.text for r in tg.session.sent(SendMessage)] == [texts.status.unavailable]


async def test_without_docker_configured(
    texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo
) -> None:
    tg = TgClient(build_dispatcher(texts, sessions, access, repo))
    await tg.send(ADMIN, texts.menu.status)
    await tg.press(ADMIN, StatusCb(a=StatusAct.CHECK).pack())

    assert [r.text for r in tg.session.sent(SendMessage)] == [texts.status.unavailable]
    assert tg.session.sent(AnswerCallbackQuery)[0].text == texts.status.unavailable


async def test_refresh_list(tg: TgClient, texts: Texts) -> None:
    await tg.press(ADMIN, StatusCb(a=StatusAct.LIST).pack())

    [edit] = tg.session.sent(EditMessageText)
    assert edit.text.startswith("🩺 Статус · 2 конт.")


@pytest.mark.parametrize(
    ("seconds", "expected"), [(5, "5 с"), (125, "2 мин"), (3 * 3600 + 120, "3 ч 2 мин"), (2 * 86400 + 3600, "2 д 1 ч")]
)
def test_duration(texts: Texts, seconds: float, expected: str) -> None:
    assert duration(texts.status, seconds) == expected