ElebobaandClaude Opus 5.5
55286495cb
CI / ruff · mypy · pytest (pull_request) Successful in 1m11s
- docker-socket-proxy в Compose (только GET /containers/*), сокет только у него - app/runtime/docker.py: read-only клиент; видны только контейнеры с меткой tgit.managed=true, чужие id отклоняются до запроса - app/runtime/checks.py: running, health, рестарты за окно, CPU/RAM, ошибки в логе, HTTP-проба; у каждой проверки свой таймаут, отчёт всегда полный - app/runtime/snapshots.py: снапшоты раз в 15 с в container_snapshots - бот: список с 🟢🟡🔴⚪, карточка, «🩺 Запустить проверки», «🔄 Обновить» - деплой ставит метки tgit.sha / tgit.deployed_at на контейнер бота - имя джоба CD без неотрисованного выражения Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
import pytest
|
|
|
|
from app.runtime.docker import DockerError, Light, demux_logs, light, parse_stats, parse_time
|
|
from tests.runtime.fake_docker import FakeDocker, MiB, frame, now_minus
|
|
|
|
|
|
@pytest.fixture
|
|
def fake() -> FakeDocker:
|
|
fake = FakeDocker()
|
|
fake.add(
|
|
"tgit-dev-bot-1", service="bot", labels={"tgit.sha": "526eb56aa", "tgit.deployed_at": "2026-10-17T07:52:00Z"}
|
|
)
|
|
fake.add("tgit-dev-postgres-1", service="postgres")
|
|
fake.add("vobot-api-1", managed=False) # someone else's project on the same host
|
|
return fake
|
|
|
|
|
|
async def test_lists_only_managed_containers(fake: FakeDocker) -> None:
|
|
async with fake.client() as docker:
|
|
names = [c.name for c in await docker.containers()]
|
|
|
|
assert names == ["tgit-dev-bot-1", "tgit-dev-postgres-1"]
|
|
assert all(r.method == "GET" for r in fake.requests)
|
|
|
|
|
|
async def test_foreign_container_is_refused_without_a_request(fake: FakeDocker) -> None:
|
|
foreign = next(c for c in fake.containers.values() if not c.managed)
|
|
async with fake.client() as docker:
|
|
await docker.containers()
|
|
fake.requests.clear()
|
|
with pytest.raises(DockerError):
|
|
await docker.inspect(foreign.id)
|
|
with pytest.raises(DockerError):
|
|
await docker.logs(foreign.id)
|
|
with pytest.raises(DockerError):
|
|
await docker.resolve(foreign.id[:12])
|
|
|
|
assert all("/containers/json" in str(r.url) for r in fake.requests) # only re-listings, never the foreign id
|
|
|
|
|
|
async def test_inspect_and_stats(fake: FakeDocker) -> None:
|
|
async with fake.client() as docker:
|
|
bot = next(c for c in await docker.containers() if c.service == "bot")
|
|
details = await docker.inspect(await docker.resolve(bot.short_id))
|
|
stats = await docker.stats(bot.id)
|
|
|
|
assert details.light is Light.GREEN
|
|
assert details.sha == "526eb56aa"
|
|
assert details.deployed_at is not None and details.deployed_at.hour == 7
|
|
assert details.started_at is not None and details.started_at.microsecond == 123456 # nanoseconds trimmed
|
|
assert stats.cpu_pct == 5.0
|
|
assert stats.mem_bytes == 64 * MiB # page cache subtracted
|
|
assert stats.mem_pct == 25.0
|
|
|
|
|
|
async def test_logs_are_demultiplexed_and_since_filtered(fake: FakeDocker) -> None:
|
|
bot = next(c for c in fake.containers.values() if c.service == "bot")
|
|
bot.logs = [(now_minus(10), "old line"), (now_minus(1), "fresh line"), (now_minus(0.5), "ERR on stderr")]
|
|
async with fake.client() as docker:
|
|
await docker.containers()
|
|
text = await docker.logs(bot.id, since=now_minus(5))
|
|
|
|
assert text == "fresh line\nERR on stderr\n"
|
|
|
|
|
|
async def test_unreachable_docker_is_a_docker_error(fake: FakeDocker) -> None:
|
|
fake.down = True
|
|
async with fake.client() as docker:
|
|
assert not await docker.ping()
|
|
with pytest.raises(DockerError, match="unreachable"):
|
|
await docker.containers()
|
|
|
|
|
|
async def test_ping(fake: FakeDocker) -> None:
|
|
async with fake.client() as docker:
|
|
assert await docker.ping()
|
|
|
|
|
|
def test_demux_plain_tty_stream_passes_through() -> None:
|
|
assert demux_logs(b"plain text\n") == "plain text\n"
|
|
assert demux_logs(frame("a\n") + frame("b\n", 2)) == "a\nb\n"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("state", "health", "code", "expected"),
|
|
[
|
|
("running", "healthy", 0, Light.GREEN),
|
|
("running", None, 0, Light.GREEN),
|
|
("running", "starting", 0, Light.YELLOW),
|
|
("running", "unhealthy", 0, Light.RED),
|
|
("restarting", None, 0, Light.RED),
|
|
("exited", None, 137, Light.RED),
|
|
("exited", None, 0, Light.WHITE),
|
|
("created", None, 0, Light.YELLOW),
|
|
],
|
|
)
|
|
def test_light(state: str, health: str | None, code: int, expected: Light) -> None:
|
|
assert light(state, health, code) is expected
|
|
|
|
|
|
def test_parse_time_edge_cases() -> None:
|
|
assert parse_time(None) is None
|
|
assert parse_time("0001-01-01T00:00:00Z") is None
|
|
assert parse_time("garbage") is None
|
|
|
|
|
|
def test_parse_stats_without_previous_sample() -> None:
|
|
stats = parse_stats({"cpu_stats": {"cpu_usage": {"total_usage": 5}}, "memory_stats": {}})
|
|
assert stats.cpu_pct == 0.0
|
|
assert stats.mem_pct == 0.0
|