feat(status): раздел «Статус» — контейнеры, карточка, проверки (#14)

- 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>

tests/bot/test_common.py

@@ -46,7 +46,7 @@ async def test_outsider_stale_button_is_swallowed(tg: TgClient) -> None:
     assert len(tg.session.sent(AnswerCallbackQuery)) == 1
 
 
-@pytest.mark.parametrize("section", ["status", "logs", "alerts"])
+@pytest.mark.parametrize("section", ["logs", "alerts"])
 async def test_menu_button_opens_section_with_nav(tg: TgClient, texts: Texts, section: str) -> None:
     await tg.send(ADMIN, getattr(texts.menu, section))
 

tests/bot/test_status.py

@@ -0,0 +1,143 @@
+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

tests/runtime/fake_docker.py

@@ -0,0 +1,160 @@
+"""An in-memory Docker Engine API behind httpx.MockTransport, GET-only like docker-socket-proxy."""
+
+import hashlib
+import json
+import struct
+from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from urllib.parse import parse_qs
+
+import httpx
+
+from app.config import DockerSettings
+from app.runtime.docker import DockerClient
+
+STARTED = datetime(2026, 10, 17, 8, 0, tzinfo=UTC)
+MiB = 1024 * 1024
+
+
+def frame(text: str, stream: int = 1) -> bytes:
+    data = text.encode()
+    return struct.pack(">BxxxI", stream, len(data)) + data
+
+
+@dataclass
+class FakeContainer:
+    id: str
+    name: str
+    service: str = ""
+    state: str = "running"
+    health: str | None = "healthy"
+    exit_code: int = 0
+    restart_count: int = 0
+    oom: bool = False
+    managed: bool = True
+    labels: dict[str, str] = field(default_factory=dict)
+    logs: list[tuple[datetime, str]] = field(default_factory=list)  # (when, line)
+    cpu_pct: float = 5.0
+    mem_bytes: int = 64 * MiB
+    mem_limit: int = 256 * MiB
+    stats_status: int = 200
+
+    def all_labels(self) -> dict[str, str]:
+        base = {"com.docker.compose.service": self.service} if self.service else {}
+        return {**base, **({"tgit.managed": "true"} if self.managed else {}), **self.labels}
+
+    def status_line(self) -> str:
+        if self.state == "running":
+            return "Up 2 hours" + {
+                "healthy": " (healthy)",
+                "unhealthy": " (unhealthy)",
+                "starting": " (health: starting)",
+            }.get(self.health or "", "")
+        if self.state == "exited":
+            return f"Exited ({self.exit_code}) 5 minutes ago"
+        return self.state.capitalize()
+
+
+class FakeDocker:
+    def __init__(self) -> None:
+        self.containers: dict[str, FakeContainer] = {}
+        self.requests: list[httpx.Request] = []
+        self.down = False
+
+    def add(self, name: str, **kw: Any) -> FakeContainer:
+        cid = hashlib.sha256(name.encode()).hexdigest()
+        c = FakeContainer(id=cid, name=name, **kw)
+        self.containers[cid] = c
+        return c
+
+    def client(self, **settings: Any) -> DockerClient:
+        return DockerClient(
+            DockerSettings(url="http://docker-proxy:2375", **settings), transport=httpx.MockTransport(self)
+        )
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        self.requests.append(request)
+        if self.down:
+            raise httpx.ConnectError("connection refused", request=request)
+        if request.method != "GET":
+            return httpx.Response(403, text="forbidden by proxy")
+        path = request.url.path
+        q = {k: v[0] for k, v in parse_qs(request.url.query.decode()).items()}
+        if path == "/_ping":
+            return httpx.Response(200, text="OK")
+        if path == "/containers/json":
+            return httpx.Response(200, json=self._list(q))
+        parts = path.strip("/").split("/")
+        if len(parts) == 3 and parts[0] == "containers" and parts[1] in self.containers:
+            c = self.containers[parts[1]]
+            match parts[2]:
+                case "json":
+                    return httpx.Response(200, json=self._inspect(c))
+                case "stats":
+                    if c.stats_status != 200:
+                        return httpx.Response(c.stats_status, text="stats broken")
+                    return httpx.Response(200, json=self._stats(c))
+                case "logs":
+                    since = datetime.fromtimestamp(int(q.get("since", "0")), UTC)
+                    body = b"".join(frame(line + "\n", 2 if "ERR" in line else 1) for at, line in c.logs if at >= since)
+                    return httpx.Response(200, content=body)
+        return httpx.Response(404, json={"message": "No such container"})
+
+    def _list(self, q: dict[str, str]) -> list[dict[str, Any]]:
+        wanted = json.loads(q.get("filters", "{}")).get("label", [])
+        out = []
+        for c in self.containers.values():
+            labels = c.all_labels()
+            if all(labels.get(k) == v for k, _, v in (w.partition("=") for w in wanted)):
+                out.append(
+                    {
+                        "Id": c.id,
+                        "Names": [f"/{c.name}"],
+                        "Image": f"{c.name}:latest",
+                        "State": c.state,
+                        "Status": c.status_line(),
+                        "Labels": labels,
+                    }
+                )
+        return out
+
+    def _inspect(self, c: FakeContainer) -> dict[str, Any]:
+        state: dict[str, Any] = {
+            "Status": c.state,
+            "ExitCode": c.exit_code,
+            "OOMKilled": c.oom,
+            "StartedAt": STARTED.strftime("%Y-%m-%dT%H:%M:%S.123456789Z"),
+            "FinishedAt": "0001-01-01T00:00:00Z",
+        }
+        if c.health is not None:
+            state["Health"] = {"Status": c.health}
+        return {
+            "Id": c.id,
+            "Name": f"/{c.name}",
+            "RestartCount": c.restart_count,
+            "State": state,
+            "Config": {"Image": f"{c.name}:latest", "Labels": c.all_labels()},
+        }
+
+    @staticmethod
+    def _stats(c: FakeContainer) -> dict[str, Any]:
+        # 2 CPUs, system delta 1e9 → cpu_pct = delta/1e9*2*100
+        delta = int(c.cpu_pct / 200 * 1e9)
+        return {
+            "cpu_stats": {
+                "cpu_usage": {"total_usage": 10_000 + delta},
+                "system_cpu_usage": 2_000_000_000,
+                "online_cpus": 2,
+            },
+            "precpu_stats": {"cpu_usage": {"total_usage": 10_000}, "system_cpu_usage": 1_000_000_000},
+            "memory_stats": {
+                "usage": c.mem_bytes + 10 * MiB,
+                "limit": c.mem_limit,
+                "stats": {"inactive_file": 10 * MiB},
+            },
+        }
+
+
+def now_minus(minutes: float) -> datetime:
+    return datetime.now(UTC) - timedelta(minutes=minutes)

tests/runtime/test_checks.py

@@ -0,0 +1,203 @@
+import asyncio
+import contextlib
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime, timedelta
+
+import httpx
+import pytest
+
+from app.db.models import ContainerSnapshot
+from app.db.session import SessionMaker
+from app.runtime import checks
+from app.runtime.checks import Checker, CheckName, ChecksConfig, Verdict, count_errors, load_checks
+from app.runtime.docker import DockerClient, Light
+from app.runtime.snapshots import SnapshotStore, collect_once, snapshot_loop
+from tests.conftest import ROOT
+from tests.runtime.fake_docker import FakeDocker, now_minus
+
+CFG = load_checks(ROOT / "config")
+
+
+def probe(request: httpx.Request) -> httpx.Response:
+    if request.url.host == "broken":
+        raise httpx.ConnectError("refused", request=request)
+    return httpx.Response(200 if request.url.path == "/health" else 503)
+
+
+@pytest.fixture
+def fake() -> FakeDocker:
+    fake = FakeDocker()
+    fake.add("hello", service="hello", labels={"tgit.health_url": "http://hello:8000/health"})
+    fake.add(
+        "unhealthy", service="unhealthy", health="unhealthy", labels={"tgit.health_url": "http://unhealthy:8000/x"}
+    )
+    fake.add("crashed", state="exited", health=None, exit_code=137, oom=True)
+    return fake
+
+
+@pytest.fixture
+async def docker(fake: FakeDocker) -> AsyncIterator[DockerClient]:
+    async with fake.client() as client:
+        await client.containers()  # the bot always lists before it inspects
+        yield client
+
+
+@pytest.fixture
+def store(sessions: SessionMaker) -> SnapshotStore:
+    return SnapshotStore(sessions)
+
+
+@pytest.fixture
+def checker(docker: DockerClient, store: SnapshotStore) -> Checker:
+    return Checker(docker, store, CFG, httpx.AsyncClient(transport=httpx.MockTransport(probe)))
+
+
+def verdicts(report: checks.ContainerReport) -> dict[CheckName, Verdict]:
+    return {r.name: r.verdict for r in report.results}
+
+
+async def test_config_loads_and_bad_pattern_rejected() -> None:
+    assert CFG.window == 300
+    with pytest.raises(ValueError):
+        ChecksConfig(error_pattern="(unclosed")
+
+
+async def test_healthy_vs_unhealthy_vs_crashed(checker: Checker) -> None:
+    reports = {r.details.name: r for r in await checker.check_all()}
+
+    hello = verdicts(reports["hello"])
+    assert hello[CheckName.RUNNING] is Verdict.OK
+    assert hello[CheckName.HEALTH] is Verdict.OK
+    assert hello[CheckName.HTTP] is Verdict.OK
+    assert hello[CheckName.RESTARTS] is Verdict.SKIP  # no history yet
+    assert reports["hello"].ok
+
+    bad = verdicts(reports["unhealthy"])
+    assert bad[CheckName.HEALTH] is Verdict.FAIL
+    assert bad[CheckName.HTTP] is Verdict.FAIL
+    assert not reports["unhealthy"].ok
+
+    crashed = reports["crashed"]
+    assert verdicts(crashed)[CheckName.RUNNING] is Verdict.FAIL
+    assert "exit 137, OOM" in crashed.results[0].detail
+    assert verdicts(crashed)[CheckName.CPU] is Verdict.SKIP
+    assert crashed.details.light is Light.RED
+
+
+async def test_restarts_compared_with_history(
+    checker: Checker, fake: FakeDocker, docker: DockerClient, store: SnapshotStore
+) -> None:
+    hello = next(c for c in fake.containers.values() if c.name == "hello")
+    assert await collect_once(docker, store, now=datetime.now(UTC) - timedelta(minutes=3)) == 2  # running ones only
+    hello.restart_count = 2
+
+    report = await checker.check(hello.id)
+
+    result = next(r for r in report.results if r.name is CheckName.RESTARTS)
+    assert result.verdict is Verdict.FAIL
+    assert result.detail == "0 → 2"
+
+
+async def test_old_history_outside_window_is_ignored(
+    checker: Checker, fake: FakeDocker, sessions: SessionMaker
+) -> None:
+    hello = next(c for c in fake.containers.values() if c.name == "hello")
+    async with sessions.begin() as s:
+        s.add(
+            ContainerSnapshot(container="hello", at=now_minus(30), restart_count=0, cpu_pct=1, mem_pct=1, health=None)
+        )
+    hello.restart_count = 5
+
+    report = await checker.check(hello.id)
+
+    assert verdicts(report)[CheckName.RESTARTS] is Verdict.SKIP
+
+
+async def test_thresholds_and_log_errors(checker: Checker, fake: FakeDocker) -> None:
+    hello = next(c for c in fake.containers.values() if c.name == "hello")
+    hello.cpu_pct, hello.mem_bytes = 150.0, 250 * 1024 * 1024
+    hello.logs = [
+        (now_minus(20), "Traceback from long ago"),
+        (now_minus(1), "all good"),
+        (now_minus(1), '{"level": "ERROR", "msg": "boom"}'),
+    ]
+
+    report = await checker.check(hello.id)
+
+    v = verdicts(report)
+    assert v[CheckName.CPU] is Verdict.FAIL
+    assert v[CheckName.MEM] is Verdict.FAIL
+    log_result = next(r for r in report.results if r.name is CheckName.LOG_ERRORS)
+    assert log_result.verdict is Verdict.FAIL
+    assert log_result.detail.startswith("1: ")  # the old Traceback is outside the window
+
+
+async def test_broken_or_slow_check_does_not_spoil_report(
+    checker: Checker, fake: FakeDocker, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    async def boom(ctx: checks.Context) -> checks.CheckResult:
+        raise RuntimeError("kaputt")
+
+    async def slow(ctx: checks.Context) -> checks.CheckResult:
+        await asyncio.sleep(10)
+        raise AssertionError
+
+    monkeypatch.setattr(checks, "CHECKS", [(CheckName.HEALTH, boom), (CheckName.HTTP, slow), *checks.CHECKS[:1]])
+    fast = Checker(checker._docker, checker._store, CFG.model_copy(update={"check_timeout": 0.05}), checker._http)
+    hello = next(c for c in fake.containers.values() if c.name == "hello")
+
+    report = await fast.check(hello.id)
+
+    by_name = {r.name: r for r in report.results}
+    assert by_name[CheckName.HEALTH].detail == "RuntimeError: kaputt"
+    assert by_name[CheckName.HTTP].detail == "timeout"
+    assert by_name[CheckName.RUNNING].verdict is Verdict.OK
+
+
+async def test_stats_failure_skips_cpu_and_mem(checker: Checker, fake: FakeDocker) -> None:
+    hello = next(c for c in fake.containers.values() if c.name == "hello")
+    hello.stats_status = 500
+
+    report = await checker.check(hello.id)
+
+    assert verdicts(report)[CheckName.CPU] is Verdict.SKIP
+    assert verdicts(report)[CheckName.RUNNING] is Verdict.OK
+
+
+async def test_http_probe_connection_error(fake: FakeDocker, checker: Checker) -> None:
+    hello = next(c for c in fake.containers.values() if c.name == "hello")
+    hello.labels["tgit.health_url"] = "http://broken/health"
+
+    report = await checker.check(hello.id)
+
+    result = next(r for r in report.results if r.name is CheckName.HTTP)
+    assert (result.verdict, result.detail) == (Verdict.FAIL, "ConnectError")
+
+
+async def test_snapshot_store_latest(docker: DockerClient, store: SnapshotStore) -> None:
+    await collect_once(docker, store, now=now_minus(1))
+    await collect_once(docker, store)
+
+    latest = await store.latest("hello")
+
+    assert latest is not None and latest.cpu_pct == 5.0 and latest.health == "healthy"
+
+
+def test_count_errors_reports_first_hit() -> None:
+    hits, first = count_errors("ok\nERROR one\nERROR two\n", r"ERROR")
+    assert (hits, first) == (2, "ERROR one")
+
+
+async def test_snapshot_loop_survives_docker_outage(
+    fake: FakeDocker, docker: DockerClient, store: SnapshotStore
+) -> None:
+    fake.down = True
+    task = asyncio.create_task(snapshot_loop(docker, store, interval=0.01))
+    await asyncio.sleep(0.05)
+    fake.down = False
+    await asyncio.sleep(0.05)
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+    assert await store.latest("hello") is not None

tests/runtime/test_docker.py

@@ -0,0 +1,110 @@
+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