CI / ruff · mypy · pytest (pull_request) Successful in 9m40s
- app/logs/grep.py: подмножество grep (шаблон, -i -v -E -c -C N), разбор shlex без исполнения, BRE → regex, таймаут на строку и бюджет на поиск, лимиты 200 символов / 500 строк вывода - app/logs/journal.py: каждый поиск в log_queries (L-6) - бот: выбор контейнера, хвост 50/200 строк или 5 мин/1 ч/24 ч, свой grep по 24 ч, длинный результат — файлом .log (L-5) - логи Docker: не больше 50 000 строк и 5 МБ, сохраняются новейшие - fix: кнопки меню и /menu, /help больше не проглатываются шагами ввода (имя файла, содержимое, сообщение коммита, grep) — фильтр free_input Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
162 lines
5.9 KiB
Python
162 lines
5.9 KiB
Python
"""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)
|
|
picked = [line for at, line in c.logs if at >= since][-int(q.get("tail", "1000000")) :]
|
|
body = b"".join(frame(line + "\n", 2 if "ERR" in line else 1) for line in picked)
|
|
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)
|