tgit/tests/bot/test_logs.py

from collections.abc import AsyncIterator

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

from app.access.service import AccessService
from app.bot.routers.logs import LogsAct, LogsCb
from app.db.models import LogMode, LogQuery
from app.db.session import SessionMaker
from app.main import build_dispatcher
from app.repo.git import UploadsRepo
from app.runtime.docker import DockerClient
from app.texts import Texts
from tests.conftest import ADMIN
from tests.fakes import TgClient
from tests.runtime.fake_docker import FakeDocker, now_minus


@pytest.fixture
def fake() -> FakeDocker:
    fake = FakeDocker()
    fake.add(
        "crash",
        logs=[
            (now_minus(120), "boot"),
            (now_minus(30), "connect db"),
            (now_minus(29), "ERROR connection refused"),
            (now_minus(28), "retry"),
            (now_minus(2), "<b>html-ish</b> & stuff"),
        ],
    )
    fake.add("vobot-api-1", managed=False, logs=[(now_minus(1), "SECRET")])
    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:
    return TgClient(build_dispatcher(texts, sessions, access, repo, docker=docker))


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


async def journal(sessions: SessionMaker) -> list[LogQuery]:
    async with sessions() as s:
        return list(await s.scalars(select(LogQuery).order_by(LogQuery.id)))


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

    [reply] = tg.session.sent(SendMessage)
    assert reply.text == texts.logs.title
    labels = [b.text for row in reply.reply_markup.inline_keyboard for b in row]
    assert "🟢 crash" in labels and not any("vobot" in label for label in labels)


async def test_tail_is_escaped_and_journaled(tg: TgClient, fake: FakeDocker, sessions: SessionMaker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.TAIL, id=cid(fake, "crash"), n=50).pack())

    [reply] = tg.session.sent(SendMessage)
    assert "строк: 5" in reply.text
    assert "&lt;b&gt;html-ish&lt;/b&gt; &amp; stuff" in reply.text
    [row] = await journal(sessions)
    assert (row.mode, row.container, row.query, row.lines) == (LogMode.TAIL, "crash", {"tail": 50}, 5)


async def test_since_period(tg: TgClient, fake: FakeDocker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.SINCE, id=cid(fake, "crash"), n=5).pack())

    [reply] = tg.session.sent(SendMessage)
    assert "за 5 мин · строк: 1" in reply.text


async def test_grep_flow(tg: TgClient, texts: Texts, fake: FakeDocker, sessions: SessionMaker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.GREP, id=cid(fake, "crash")).pack())
    await tg.send(ADMIN, "-i -C 1 error")

    ask, result = tg.session.sent(SendMessage)
    assert ask.text.startswith("🔎 Выражение grep")
    assert "<code>grep -i -C 1 error</code>" in result.text
    assert "совпадений: 1" in result.text
    assert "connect db\nERROR connection refused\nretry" in result.text
    [row] = await journal(sessions)
    assert row.mode is LogMode.GREP and row.query["pattern"] == "error" and row.lines == 1

    tg.session.requests.clear()
    await tg.send(ADMIN, "error")  # state is cleared after one search
    assert not any("совпадений" in (m.text or "") for m in tg.session.sent(SendMessage))


async def test_grep_error_keeps_waiting(tg: TgClient, fake: FakeDocker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.GREP, id=cid(fake, "crash")).pack())
    await tg.send(ADMIN, "-r error")
    await tg.send(ADMIN, "-c retry")

    _, err, result = tg.session.sent(SendMessage)
    assert "Флаг -r не поддерживается" in err.text
    assert "совпадений: 1" in result.text and "<pre>" not in result.text


async def test_grep_nothing_found(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.GREP, id=cid(fake, "crash")).pack())
    await tg.send(ADMIN, "nope")

    assert tg.session.sent(SendMessage)[-1].text.endswith(texts.logs.nothing)


async def test_menu_button_escapes_grep_state(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.GREP, id=cid(fake, "crash")).pack())
    await tg.send(ADMIN, texts.menu.logs)

    assert tg.session.sent(SendMessage)[-1].text == texts.logs.title


async def test_long_result_goes_as_file(tg: TgClient, fake: FakeDocker) -> None:
    crash = next(c for c in fake.containers.values() if c.name == "crash")
    crash.logs = [(now_minus(1), f"line {i} " + "x" * 50) for i in range(200)]

    await tg.press(ADMIN, LogsCb(a=LogsAct.TAIL, id=cid(fake, "crash"), n=200).pack())

    [doc] = tg.session.sent(SendDocument)
    assert doc.document.filename == "crash.log"
    assert doc.document.data.decode().count("\n") == 199


async def test_foreign_container_refused(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.TAIL, id=cid(fake, "vobot-api-1"), n=50).pack())

    assert not tg.session.sent(SendMessage)
    assert tg.session.sent(AnswerCallbackQuery)[0].text == texts.logs.unavailable


async def test_open_container_and_back_to_list(tg: TgClient, texts: Texts, fake: FakeDocker) -> None:
    await tg.press(ADMIN, LogsCb(a=LogsAct.OPEN, id=cid(fake, "crash")).pack())
    await tg.press(ADMIN, LogsCb(a=LogsAct.LIST).pack())
    await tg.press(ADMIN, LogsCb(a=LogsAct.AGENT, id=cid(fake, "crash")).pack())

    card, listing = tg.session.sent(EditMessageText)
    assert card.text.startswith("📜 Логи · crash")
    assert listing.text == texts.logs.title
    assert tg.session.sent(AnswerCallbackQuery)[-1].text == texts.logs.agent_soon


async def test_without_docker(texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo) -> None:
    tg = TgClient(build_dispatcher(texts, sessions, access, repo))
    await tg.send(ADMIN, texts.menu.logs)

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