Add CPU sparkline to dashboard in compose mode
- New docker_stats.py module reads container stats from Docker socket using only asyncio + stdlib (no new dependencies) - Calculates CPU % from delta of cpu_usage and system_cpu_usage - Maintains ring buffer of last 30 CPU readings per container - render_sparkline_svg() generates mini SVG chart from history - DockerStatsCollector polls containers every 2 seconds - New /cpu-sparkline.svg endpoint serves sparkline for a container - Dashboard shows sparkline in tile header next to container name - Only active in compose mode (--compose-manifest flag) - Graceful degradation if Docker socket unavailable Bump version to 0.1.17
This commit is contained in:
GitHub Copilot
+1
-1
@@ -1,6 +1,6 @@
|
||||
[tool.poetry]
|
||||
name = "textual-webterm"
|
||||
version = "0.1.16"
|
||||
version = "0.1.17"
|
||||
description = "Serve terminal sessions over the web"
|
||||
authors = ["Will McGugan <will@textualize.io>"]
|
||||
license = "MIT"
|
||||
|
||||
@@ -149,10 +149,12 @@ def app(
|
||||
_config = default_config()
|
||||
|
||||
landing_apps: list = []
|
||||
is_compose_mode = False
|
||||
if landing_manifest:
|
||||
landing_apps = load_landing_yaml(landing_manifest)
|
||||
elif compose_manifest:
|
||||
landing_apps = load_compose_manifest(compose_manifest)
|
||||
is_compose_mode = True
|
||||
|
||||
server = LocalServer(
|
||||
"./",
|
||||
@@ -160,6 +162,7 @@ def app(
|
||||
host=host,
|
||||
port=port,
|
||||
landing_apps=landing_apps,
|
||||
compose_mode=is_compose_mode,
|
||||
)
|
||||
for app_entry in landing_apps:
|
||||
server.add_terminal(app_entry.name, app_entry.command, slug=app_entry.slug)
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Docker container CPU stats via Unix socket.
|
||||
|
||||
Reads container stats from Docker socket using only asyncio and stdlib.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import json
|
||||
import logging
|
||||
import socket
|
||||
from collections import deque
|
||||
from pathlib import Path
|
||||
|
||||
log = logging.getLogger("textual-webterm")
|
||||
|
||||
DOCKER_SOCKET = "/var/run/docker.sock"
|
||||
STATS_HISTORY_SIZE = 30 # Number of CPU readings to keep
|
||||
POLL_INTERVAL = 2.0 # Seconds between polls
|
||||
|
||||
|
||||
class DockerStatsCollector:
|
||||
"""Collects CPU stats from Docker containers via the Docker socket."""
|
||||
|
||||
def __init__(self, socket_path: str = DOCKER_SOCKET) -> None:
|
||||
self._socket_path = socket_path
|
||||
# container_name -> deque of CPU % values (0-100)
|
||||
self._cpu_history: dict[str, deque[float]] = {}
|
||||
self._running = False
|
||||
self._task: asyncio.Task | None = None
|
||||
# Track previous CPU values for delta calculation
|
||||
self._prev_cpu: dict[str, tuple[int, int]] = {}
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
"""Check if Docker socket is available."""
|
||||
return Path(self._socket_path).exists()
|
||||
|
||||
def get_cpu_history(self, container_name: str) -> list[float]:
|
||||
"""Get CPU history for a container."""
|
||||
if container_name not in self._cpu_history:
|
||||
return []
|
||||
return list(self._cpu_history[container_name])
|
||||
|
||||
async def _make_request(self, path: str) -> dict | None:
|
||||
"""Make HTTP request to Docker socket."""
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
def _sync_request() -> bytes | None:
|
||||
try:
|
||||
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
sock.settimeout(5.0)
|
||||
sock.connect(self._socket_path)
|
||||
|
||||
request = f"GET {path} HTTP/1.1\r\nHost: localhost\r\n\r\n"
|
||||
sock.sendall(request.encode())
|
||||
|
||||
# Read response
|
||||
chunks = []
|
||||
while True:
|
||||
chunk = sock.recv(4096)
|
||||
if not chunk:
|
||||
break
|
||||
chunks.append(chunk)
|
||||
sock.close()
|
||||
return b"".join(chunks)
|
||||
except (OSError, TimeoutError):
|
||||
return None
|
||||
|
||||
response = await loop.run_in_executor(None, _sync_request)
|
||||
if response is None:
|
||||
return None
|
||||
|
||||
# Parse HTTP response - find JSON body after headers
|
||||
try:
|
||||
response_str = response.decode("utf-8", errors="replace")
|
||||
# Split headers and body
|
||||
if "\r\n\r\n" in response_str:
|
||||
_, body = response_str.split("\r\n\r\n", 1)
|
||||
else:
|
||||
body = response_str
|
||||
|
||||
# Handle chunked encoding - find the JSON object
|
||||
if body.startswith("{"):
|
||||
json_str = body
|
||||
else:
|
||||
# Skip chunk size line in chunked encoding
|
||||
lines = body.split("\r\n")
|
||||
for line in lines:
|
||||
if line.startswith("{"):
|
||||
json_str = line
|
||||
break
|
||||
else:
|
||||
return None
|
||||
|
||||
return json.loads(json_str)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return None
|
||||
|
||||
def _calculate_cpu_percent(
|
||||
self, container: str, cpu_stats: dict, precpu_stats: dict
|
||||
) -> float | None:
|
||||
"""Calculate CPU percentage from stats.
|
||||
|
||||
Formula: (cpu_delta / system_delta) * num_cpus * 100
|
||||
"""
|
||||
try:
|
||||
cpu_usage = cpu_stats.get("cpu_usage", {})
|
||||
precpu_usage = precpu_stats.get("cpu_usage", {})
|
||||
|
||||
cpu_total = cpu_usage.get("total_usage", 0)
|
||||
precpu_total = precpu_usage.get("total_usage", 0)
|
||||
system_cpu = cpu_stats.get("system_cpu_usage", 0)
|
||||
presystem_cpu = precpu_stats.get("system_cpu_usage", 0)
|
||||
|
||||
# Use previous values if precpu_stats is empty (first read)
|
||||
if precpu_total == 0 and container in self._prev_cpu:
|
||||
precpu_total, presystem_cpu = self._prev_cpu[container]
|
||||
|
||||
# Store current values for next calculation
|
||||
self._prev_cpu[container] = (cpu_total, system_cpu)
|
||||
|
||||
cpu_delta = cpu_total - precpu_total
|
||||
system_delta = system_cpu - presystem_cpu
|
||||
|
||||
if system_delta <= 0 or cpu_delta < 0:
|
||||
return None
|
||||
|
||||
# Get number of CPUs
|
||||
online_cpus = cpu_stats.get("online_cpus")
|
||||
if online_cpus is None:
|
||||
percpu = cpu_usage.get("percpu_usage", [])
|
||||
online_cpus = len(percpu) if percpu else 1
|
||||
|
||||
cpu_percent = (cpu_delta / system_delta) * online_cpus * 100.0
|
||||
return min(cpu_percent, 100.0 * online_cpus) # Cap at max possible
|
||||
|
||||
except (KeyError, TypeError, ZeroDivisionError):
|
||||
return None
|
||||
|
||||
async def _poll_container(self, container_name: str) -> None:
|
||||
"""Poll stats for a single container."""
|
||||
# Docker API uses container name without leading slash
|
||||
path = f"/containers/{container_name}/stats?stream=false"
|
||||
stats = await self._make_request(path)
|
||||
|
||||
if stats is None:
|
||||
return
|
||||
|
||||
cpu_stats = stats.get("cpu_stats", {})
|
||||
precpu_stats = stats.get("precpu_stats", {})
|
||||
|
||||
cpu_percent = self._calculate_cpu_percent(container_name, cpu_stats, precpu_stats)
|
||||
if cpu_percent is not None:
|
||||
if container_name not in self._cpu_history:
|
||||
self._cpu_history[container_name] = deque(maxlen=STATS_HISTORY_SIZE)
|
||||
self._cpu_history[container_name].append(cpu_percent)
|
||||
|
||||
async def _poll_loop(self, containers: list[str]) -> None:
|
||||
"""Background polling loop."""
|
||||
while self._running:
|
||||
for container in containers:
|
||||
if not self._running:
|
||||
break
|
||||
try:
|
||||
await self._poll_container(container)
|
||||
except Exception:
|
||||
log.debug("Error polling stats for %s", container)
|
||||
await asyncio.sleep(POLL_INTERVAL)
|
||||
|
||||
def start(self, containers: list[str]) -> None:
|
||||
"""Start collecting stats for given containers."""
|
||||
if not self.available:
|
||||
log.debug("Docker socket not available at %s", self._socket_path)
|
||||
return
|
||||
|
||||
if self._running:
|
||||
return
|
||||
|
||||
self._running = True
|
||||
self._task = asyncio.create_task(self._poll_loop(containers))
|
||||
log.info("Started Docker stats collection for %d containers", len(containers))
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Stop collecting stats."""
|
||||
self._running = False
|
||||
if self._task:
|
||||
self._task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await self._task
|
||||
self._task = None
|
||||
|
||||
|
||||
def render_sparkline_svg(
|
||||
values: list[float],
|
||||
width: int = 100,
|
||||
height: int = 20,
|
||||
stroke_color: str = "#4ade80",
|
||||
fill_color: str = "rgba(74, 222, 128, 0.2)",
|
||||
) -> str:
|
||||
"""Render a list of values as an SVG sparkline.
|
||||
|
||||
Args:
|
||||
values: List of values to plot (0-100 range expected for CPU %)
|
||||
width: SVG width in pixels
|
||||
height: SVG height in pixels
|
||||
stroke_color: Line color
|
||||
fill_color: Fill color under the line
|
||||
|
||||
Returns:
|
||||
SVG string
|
||||
"""
|
||||
if not values:
|
||||
# Empty placeholder
|
||||
return f'<svg width="{width}" height="{height}" xmlns="http://www.w3.org/2000/svg"></svg>'
|
||||
|
||||
# Normalize values to 0-1 range
|
||||
max_val = max(values) if max(values) > 0 else 1
|
||||
normalized = [v / max_val for v in values]
|
||||
|
||||
# Calculate points
|
||||
points = []
|
||||
x_step = width / max(len(values) - 1, 1)
|
||||
for i, v in enumerate(normalized):
|
||||
x = i * x_step
|
||||
y = height - (v * (height - 2)) - 1 # Leave 1px margin
|
||||
points.append(f"{x:.1f},{y:.1f}")
|
||||
|
||||
path_line = " ".join(points)
|
||||
|
||||
# Create filled area path (line + close to bottom)
|
||||
fill_points = [*points, f"{width},{height}", f"0,{height}"]
|
||||
path_fill = " ".join(fill_points)
|
||||
|
||||
svg = f'''<svg width="{width}" height="{height}" xmlns="http://www.w3.org/2000/svg">
|
||||
<polygon points="{path_fill}" fill="{fill_color}" />
|
||||
<polyline points="{path_line}" fill="none" stroke="{stroke_color}" stroke-width="1.5" />
|
||||
</svg>'''
|
||||
return svg
|
||||
@@ -20,6 +20,7 @@ from rich.style import Style
|
||||
from rich.text import Text
|
||||
|
||||
from . import constants
|
||||
from .docker_stats import DockerStatsCollector, render_sparkline_svg
|
||||
from .exit_poller import ExitPoller
|
||||
from .identity import generate
|
||||
from .poller import Poller
|
||||
@@ -187,6 +188,7 @@ class LocalServer:
|
||||
port: int = 8080,
|
||||
exit_on_idle: int = 0,
|
||||
landing_apps: list | None = None,
|
||||
compose_mode: bool = False,
|
||||
) -> None:
|
||||
self.host = host
|
||||
self.port = port
|
||||
@@ -206,12 +208,16 @@ class LocalServer:
|
||||
|
||||
self._websocket_connections: dict[RouteKey, web.WebSocketResponse] = {}
|
||||
self._landing_apps = landing_apps or []
|
||||
self._compose_mode = compose_mode
|
||||
|
||||
self._screenshot_cache: dict[str, tuple[float, str]] = {}
|
||||
self._screenshot_cache_etag: dict[str, str] = {}
|
||||
self._screenshot_locks: dict[str, asyncio.Lock] = {}
|
||||
self._route_last_activity: dict[str, float] = {}
|
||||
|
||||
# Docker stats collector (only used in compose mode)
|
||||
self._docker_stats: DockerStatsCollector | None = None
|
||||
|
||||
@property
|
||||
def app_count(self) -> int:
|
||||
return len(self.session_manager.apps)
|
||||
@@ -291,6 +297,7 @@ class LocalServer:
|
||||
routes: list[web.AbstractRouteDef] = [
|
||||
web.get("/ws/{route_key}", self._handle_websocket),
|
||||
web.get("/screenshot.svg", self._handle_screenshot),
|
||||
web.get("/cpu-sparkline.svg", self._handle_cpu_sparkline),
|
||||
web.get("/health", self._handle_health_check),
|
||||
web.get("/", self._handle_root),
|
||||
]
|
||||
@@ -324,6 +331,14 @@ class LocalServer:
|
||||
await runner.setup()
|
||||
stack.push_async_callback(runner.cleanup)
|
||||
|
||||
# Start Docker stats collector in compose mode
|
||||
if self._compose_mode and self._landing_apps:
|
||||
self._docker_stats = DockerStatsCollector()
|
||||
if self._docker_stats.available:
|
||||
containers = [app.slug for app in self._landing_apps]
|
||||
self._docker_stats.start(containers)
|
||||
stack.push_async_callback(self._docker_stats.stop)
|
||||
|
||||
site = web.TCPSite(runner, self.host, self.port)
|
||||
await site.start()
|
||||
|
||||
@@ -613,6 +628,34 @@ class LocalServer:
|
||||
headers = {"Cache-Control": "no-cache", "ETag": etag}
|
||||
return web.Response(text=svg, content_type="image/svg+xml", headers=headers)
|
||||
|
||||
async def _handle_cpu_sparkline(self, request: web.Request) -> web.Response:
|
||||
"""Return CPU sparkline SVG for a container."""
|
||||
container = request.query.get("container", "")
|
||||
if not container:
|
||||
raise web.HTTPBadRequest(text="Missing container parameter")
|
||||
|
||||
# Get dimensions from query params
|
||||
try:
|
||||
width = int(request.query.get("width", "100"))
|
||||
except ValueError:
|
||||
width = 100
|
||||
width = max(50, min(300, width))
|
||||
|
||||
try:
|
||||
height = int(request.query.get("height", "20"))
|
||||
except ValueError:
|
||||
height = 20
|
||||
height = max(10, min(100, height))
|
||||
|
||||
# Get CPU history
|
||||
values: list[float] = []
|
||||
if self._docker_stats:
|
||||
values = self._docker_stats.get_cpu_history(container)
|
||||
|
||||
svg = render_sparkline_svg(values, width=width, height=height)
|
||||
headers = {"Cache-Control": "no-cache, max-age=0"}
|
||||
return web.Response(text=svg, content_type="image/svg+xml", headers=headers)
|
||||
|
||||
async def _handle_health_check(self, _request: web.Request) -> web.Response:
|
||||
return web.Response(text="Local server is running")
|
||||
|
||||
@@ -665,6 +708,7 @@ class LocalServer:
|
||||
for app in self._landing_apps
|
||||
]
|
||||
tiles_json = json.dumps(tiles)
|
||||
compose_mode_js = "true" if self._compose_mode else "false"
|
||||
html_content = f"""<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
@@ -674,7 +718,9 @@ class LocalServer:
|
||||
h1 {{ margin-bottom: 8px; }}
|
||||
.grid {{ display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); gap: 12px; }}
|
||||
.tile {{ background: #1e293b; border: 1px solid #334155; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 6px rgba(0,0,0,0.4); }}
|
||||
.tile-header {{ padding: 10px 12px; font-weight: bold; border-bottom: 1px solid #334155; display: flex; align-items: center; gap: 8px; }}
|
||||
.tile-header {{ padding: 10px 12px; font-weight: bold; border-bottom: 1px solid #334155; display: flex; align-items: center; justify-content: space-between; }}
|
||||
.tile-title {{ display: flex; align-items: center; gap: 8px; }}
|
||||
.sparkline {{ opacity: 0.9; }}
|
||||
.tile-body {{ padding: 0; }}
|
||||
.thumb {{ width: 100%; height: 180px; object-fit: contain; background: #0b1220; display: block; }}
|
||||
.meta {{ padding: 8px 12px; color: #94a3b8; font-size: 12px; }}
|
||||
@@ -686,12 +732,25 @@ class LocalServer:
|
||||
<div class=\"grid\" id=\"grid\"></div>
|
||||
<script>
|
||||
const tiles = {tiles_json};
|
||||
const composeMode = {compose_mode_js};
|
||||
function makeTile(tile) {{
|
||||
const card = document.createElement('div');
|
||||
card.className = 'tile';
|
||||
const header = document.createElement('div');
|
||||
header.className = 'tile-header';
|
||||
header.innerHTML = `<span>${{tile.name}}</span>`;
|
||||
const titleSpan = document.createElement('div');
|
||||
titleSpan.className = 'tile-title';
|
||||
titleSpan.innerHTML = `<span>${{tile.name}}</span>`;
|
||||
header.appendChild(titleSpan);
|
||||
if (composeMode) {{
|
||||
const sparkline = document.createElement('img');
|
||||
sparkline.className = 'sparkline';
|
||||
sparkline.width = 80;
|
||||
sparkline.height = 16;
|
||||
sparkline.alt = 'CPU';
|
||||
header.appendChild(sparkline);
|
||||
card.sparkline = sparkline;
|
||||
}}
|
||||
const body = document.createElement('div');
|
||||
body.className = 'tile-body';
|
||||
const img = document.createElement('img');
|
||||
@@ -719,6 +778,9 @@ class LocalServer:
|
||||
const tile = tiles[cards.indexOf(card)];
|
||||
const url = `/screenshot.svg?route_key=${{encodeURIComponent(tile.slug)}}`;
|
||||
card.img.src = url;
|
||||
if (composeMode && card.sparkline) {{
|
||||
card.sparkline.src = `/cpu-sparkline.svg?container=${{encodeURIComponent(tile.slug)}}&width=80&height=16&_t=${{Date.now()}}`;
|
||||
}}
|
||||
}}
|
||||
}}
|
||||
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
"""Tests for docker_stats module."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch, AsyncMock
|
||||
from textual_webterm.docker_stats import (
|
||||
DockerStatsCollector,
|
||||
render_sparkline_svg,
|
||||
STATS_HISTORY_SIZE,
|
||||
)
|
||||
|
||||
|
||||
class TestRenderSparklineSvg:
|
||||
"""Tests for SVG sparkline rendering."""
|
||||
|
||||
def test_empty_values(self):
|
||||
"""Empty values produce empty SVG."""
|
||||
svg = render_sparkline_svg([])
|
||||
assert "<svg" in svg
|
||||
assert "width=" in svg
|
||||
assert "polygon" not in svg # No data to draw
|
||||
|
||||
def test_single_value(self):
|
||||
"""Single value renders correctly."""
|
||||
svg = render_sparkline_svg([50.0])
|
||||
assert "<svg" in svg
|
||||
assert "polygon" in svg
|
||||
assert "polyline" in svg
|
||||
|
||||
def test_multiple_values(self):
|
||||
"""Multiple values render as sparkline."""
|
||||
values = [10.0, 50.0, 30.0, 80.0, 20.0]
|
||||
svg = render_sparkline_svg(values)
|
||||
assert "<svg" in svg
|
||||
assert "polygon" in svg
|
||||
assert "polyline" in svg
|
||||
|
||||
def test_custom_dimensions(self):
|
||||
"""Custom width/height are applied."""
|
||||
svg = render_sparkline_svg([50.0], width=200, height=40)
|
||||
assert 'width="200"' in svg
|
||||
assert 'height="40"' in svg
|
||||
|
||||
def test_custom_colors(self):
|
||||
"""Custom colors are applied."""
|
||||
svg = render_sparkline_svg(
|
||||
[50.0],
|
||||
stroke_color="#ff0000",
|
||||
fill_color="rgba(255, 0, 0, 0.3)",
|
||||
)
|
||||
assert "#ff0000" in svg
|
||||
assert "rgba(255, 0, 0, 0.3)" in svg
|
||||
|
||||
def test_zero_values(self):
|
||||
"""All zero values don't cause division errors."""
|
||||
svg = render_sparkline_svg([0.0, 0.0, 0.0])
|
||||
assert "<svg" in svg
|
||||
|
||||
def test_high_values(self):
|
||||
"""High CPU values (100%+) render correctly."""
|
||||
svg = render_sparkline_svg([100.0, 150.0, 200.0])
|
||||
assert "<svg" in svg
|
||||
|
||||
|
||||
class TestDockerStatsCollector:
|
||||
"""Tests for Docker stats collector."""
|
||||
|
||||
def test_available_checks_socket(self, tmp_path):
|
||||
"""available property checks socket existence."""
|
||||
socket_path = tmp_path / "docker.sock"
|
||||
collector = DockerStatsCollector(str(socket_path))
|
||||
assert collector.available is False
|
||||
|
||||
socket_path.touch()
|
||||
assert collector.available is True
|
||||
|
||||
def test_get_cpu_history_empty(self):
|
||||
"""Empty history returns empty list."""
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
assert collector.get_cpu_history("container1") == []
|
||||
|
||||
def test_get_cpu_history_with_data(self):
|
||||
"""CPU history returns stored values."""
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
collector._cpu_history["test"] = [10.0, 20.0, 30.0]
|
||||
# get_cpu_history converts deque to list
|
||||
collector._cpu_history["test"] = list.__new__(list)
|
||||
collector._cpu_history["test"].extend([10.0, 20.0, 30.0])
|
||||
|
||||
history = collector.get_cpu_history("test")
|
||||
assert history == [10.0, 20.0, 30.0]
|
||||
|
||||
def test_calculate_cpu_percent(self):
|
||||
"""CPU percentage calculation."""
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
|
||||
cpu_stats = {
|
||||
"cpu_usage": {"total_usage": 1000000000},
|
||||
"system_cpu_usage": 10000000000,
|
||||
"online_cpus": 4,
|
||||
}
|
||||
precpu_stats = {
|
||||
"cpu_usage": {"total_usage": 500000000},
|
||||
"system_cpu_usage": 5000000000,
|
||||
}
|
||||
|
||||
result = collector._calculate_cpu_percent("test", cpu_stats, precpu_stats)
|
||||
assert result is not None
|
||||
assert 0 <= result <= 400 # 4 CPUs max
|
||||
|
||||
def test_calculate_cpu_percent_zero_delta(self):
|
||||
"""Zero system delta returns None."""
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
|
||||
cpu_stats = {
|
||||
"cpu_usage": {"total_usage": 1000},
|
||||
"system_cpu_usage": 1000,
|
||||
"online_cpus": 1,
|
||||
}
|
||||
precpu_stats = {
|
||||
"cpu_usage": {"total_usage": 1000},
|
||||
"system_cpu_usage": 1000,
|
||||
}
|
||||
|
||||
result = collector._calculate_cpu_percent("test", cpu_stats, precpu_stats)
|
||||
assert result is None
|
||||
|
||||
def test_start_without_socket(self, tmp_path):
|
||||
"""Start does nothing if socket not available."""
|
||||
collector = DockerStatsCollector(str(tmp_path / "nonexistent.sock"))
|
||||
collector.start(["container1"])
|
||||
assert collector._running is False
|
||||
assert collector._task is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_without_start(self):
|
||||
"""Stop is safe to call without start."""
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
await collector.stop() # Should not raise
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_make_request_no_socket(self):
|
||||
"""Request returns None if socket unavailable."""
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
result = await collector._make_request("/test")
|
||||
assert result is None
|
||||
|
||||
def test_cpu_history_max_size(self):
|
||||
"""CPU history respects max size."""
|
||||
from collections import deque
|
||||
|
||||
collector = DockerStatsCollector("/nonexistent")
|
||||
collector._cpu_history["test"] = deque(maxlen=STATS_HISTORY_SIZE)
|
||||
|
||||
# Add more than max entries
|
||||
for i in range(STATS_HISTORY_SIZE + 10):
|
||||
collector._cpu_history["test"].append(float(i))
|
||||
|
||||
assert len(collector._cpu_history["test"]) == STATS_HISTORY_SIZE
|
||||
|
||||
|
||||
class TestLocalServerSparklineEndpoint:
|
||||
"""Tests for the CPU sparkline endpoint in LocalServer."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sparkline_endpoint_missing_container(self):
|
||||
"""Missing container param returns 400."""
|
||||
from aiohttp.web import HTTPBadRequest
|
||||
from textual_webterm.local_server import LocalServer
|
||||
from textual_webterm.config import Config
|
||||
|
||||
server = LocalServer("./", Config(), compose_mode=True)
|
||||
|
||||
request = MagicMock()
|
||||
request.query = {}
|
||||
|
||||
with pytest.raises(HTTPBadRequest):
|
||||
await server._handle_cpu_sparkline(request)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sparkline_endpoint_returns_svg(self):
|
||||
"""Sparkline endpoint returns SVG."""
|
||||
from textual_webterm.local_server import LocalServer
|
||||
from textual_webterm.config import Config
|
||||
|
||||
server = LocalServer("./", Config(), compose_mode=True)
|
||||
|
||||
request = MagicMock()
|
||||
request.query = {"container": "test", "width": "80", "height": "20"}
|
||||
|
||||
response = await server._handle_cpu_sparkline(request)
|
||||
assert response.content_type == "image/svg+xml"
|
||||
assert "<svg" in response.text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sparkline_with_stats_collector(self):
|
||||
"""Sparkline uses stats collector data when available."""
|
||||
from textual_webterm.local_server import LocalServer
|
||||
from textual_webterm.config import Config
|
||||
|
||||
server = LocalServer("./", Config(), compose_mode=True)
|
||||
server._docker_stats = MagicMock()
|
||||
server._docker_stats.get_cpu_history.return_value = [10.0, 20.0, 30.0]
|
||||
|
||||
request = MagicMock()
|
||||
request.query = {"container": "test"}
|
||||
|
||||
response = await server._handle_cpu_sparkline(request)
|
||||
server._docker_stats.get_cpu_history.assert_called_once_with("test")
|
||||
assert "<svg" in response.text
|
||||
Reference in New Issue
Block a user