Files
ElebobaandClaude Opus 5.5 95a72437ec
CI / ruff · mypy · pytest (pull_request) Failing after 3m40s
feat(pipeline): уведомления CI/CD — PipelineWatcher, CI загрузок на Gitea Actions
- app/pipeline: протокол CIProvider, клиент Gitea Actions (прогон по head_sha, упавший
  шаг, лог джоба, ретраи), PipelineWatcher — опрос только при незавершённых push,
  сообщение о push обновляется CI ⏳ → ✅/❌ → CD 🚀/⛔, автору — отдельные уведомления
- деплой сервисов — только после зелёного CI; старый push не откатывает сервис,
  который развернёт более новый; после рестарта бот продолжает с того же места
- CI_PROVIDER=none — прежнее поведение (деплой сразу после push)
- шаблон tgit-uploads: .gitea/workflows/check.yml; .gitea/ защищён от правок из бота
- миграция 0002: старые push помечены skipped, чтобы не слать им «таймаут CI»
- ADR-016, ARCHITECTURE §2.5/§5.2, README, DEPLOY, .env.example

Refs #11

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-19 00:24:08 +03:00

248 lines
10 KiB
Python

"""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 AsyncIterator, 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.pipeline.watcher import PipelineWatcher
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
async def tg(
texts: Texts, sessions: SessionMaker, access: AccessService, repo: UploadsRepo, fake_host: Path
) -> AsyncIterator[TgClient]:
loops: list[asyncio.Task[None]] = []
def dispatcher(bot: Bot) -> Dispatcher:
alerts = AlertService(
sessions, TelegramNotifier(bot, access, sessions, texts), load_alerts_config(ROOT / "config")
)
# no CI configured: the watcher deploys right after push
watcher = PipelineWatcher(sessions, bot, texts, None, Deployer(repo, client(), alerts), poll_interval=0.05)
loops.append(asyncio.create_task(watcher.run()))
return build_dispatcher(texts, sessions, access, repo, alerts=alerts, runner=client(), pipeline=watcher)
yield TgClient(dispatcher)
for loop in loops:
loop.cancel()
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.text for e in tg.session.sent(EditMessageText) if "Запушено" in (e.text or "")]
assert "CI" not in pushed[0] and pushed[0].endswith(texts.pipeline.cd_running.format(services="hello"))
await until(
lambda: tg.session.sent(EditMessageText)[-1].text.endswith(texts.pipeline.cd_success.format(services="hello"))
)
[done] = [m for m in tg.session.sent(SendMessage) if "развёрнуты" in (m.text or "")]
assert done.text.startswith("🚀 CD: сервисы hello развёрнуты, ревизия ")
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