diff --git a/tests/test_containers.py b/tests/test_containers.py
new file mode 100644
index 0000000..eedadab
--- /dev/null
+++ b/tests/test_containers.py
@@ -0,0 +1,269 @@
+"""Tests for Containers page routes."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+from compose_farm.config import Config, Host
+from compose_farm.glances import ContainerStats
+from compose_farm.web.app import create_app
+from compose_farm.web.routes.containers import (
+ _format_bytes,
+ _infer_stack_service,
+ _parse_image,
+ _parse_uptime_seconds,
+)
+
+# Byte size constants for tests
+KB = 1024
+MB = KB * 1024
+GB = MB * 1024
+
+
+class TestFormatBytes:
+ """Tests for _format_bytes function (uses humanize library)."""
+
+ def test_bytes(self) -> None:
+ assert _format_bytes(500) == "500 Bytes"
+ assert _format_bytes(0) == "0 Bytes"
+
+ def test_kilobytes(self) -> None:
+ assert _format_bytes(KB) == "1.0 KiB"
+ assert _format_bytes(KB * 5) == "5.0 KiB"
+ assert _format_bytes(KB + 512) == "1.5 KiB"
+
+ def test_megabytes(self) -> None:
+ assert _format_bytes(MB) == "1.0 MiB"
+ assert _format_bytes(MB * 100) == "100.0 MiB"
+ assert _format_bytes(MB * 512) == "512.0 MiB"
+
+ def test_gigabytes(self) -> None:
+ assert _format_bytes(GB) == "1.0 GiB"
+ assert _format_bytes(GB * 2) == "2.0 GiB"
+
+
+class TestParseImage:
+ """Tests for _parse_image function."""
+
+ def test_simple_image_with_tag(self) -> None:
+ assert _parse_image("nginx:latest") == ("nginx", "latest")
+ assert _parse_image("redis:7") == ("redis", "7")
+
+ def test_image_without_tag(self) -> None:
+ assert _parse_image("nginx") == ("nginx", "latest")
+
+ def test_registry_image(self) -> None:
+ assert _parse_image("ghcr.io/user/repo:v1.0") == ("ghcr.io/user/repo", "v1.0")
+ assert _parse_image("docker.io/library/nginx:alpine") == (
+ "docker.io/library/nginx",
+ "alpine",
+ )
+
+ def test_image_with_port_in_registry(self) -> None:
+ # Registry with port should not be confused with tag
+ assert _parse_image("localhost:5000/myimage") == ("localhost:5000/myimage", "latest")
+
+
+class TestParseUptimeSeconds:
+ """Tests for _parse_uptime_seconds function."""
+
+ def test_seconds(self) -> None:
+ assert _parse_uptime_seconds("17 seconds") == 17
+ assert _parse_uptime_seconds("1 second") == 1
+
+ def test_minutes(self) -> None:
+ assert _parse_uptime_seconds("5 minutes") == 300
+ assert _parse_uptime_seconds("1 minute") == 60
+
+ def test_hours(self) -> None:
+ assert _parse_uptime_seconds("2 hours") == 7200
+ assert _parse_uptime_seconds("an hour") == 3600
+ assert _parse_uptime_seconds("1 hour") == 3600
+
+ def test_days(self) -> None:
+ assert _parse_uptime_seconds("3 days") == 259200
+ assert _parse_uptime_seconds("a day") == 86400
+
+ def test_empty(self) -> None:
+ assert _parse_uptime_seconds("") == 0
+ assert _parse_uptime_seconds("-") == 0
+
+
+class TestInferStackService:
+ """Tests for _infer_stack_service function."""
+
+ def test_underscore_separator(self) -> None:
+ assert _infer_stack_service("mystack_web_1") == ("mystack", "web")
+ assert _infer_stack_service("app_db_1") == ("app", "db")
+
+ def test_hyphen_separator(self) -> None:
+ assert _infer_stack_service("mystack-web-1") == ("mystack", "web")
+ assert _infer_stack_service("compose-farm-api-1") == ("compose", "farm-api")
+
+ def test_simple_name(self) -> None:
+ # No separator - use name for both
+ assert _infer_stack_service("nginx") == ("nginx", "nginx")
+ assert _infer_stack_service("traefik") == ("traefik", "traefik")
+
+ def test_single_part_with_separator(self) -> None:
+ # Edge case: separator with empty second part
+ assert _infer_stack_service("single_") == ("single", "")
+
+
+class TestContainersPage:
+ """Tests for containers page endpoint."""
+
+ @pytest.fixture
+ def client(self) -> TestClient:
+ app = create_app()
+ return TestClient(app)
+
+ @pytest.fixture
+ def mock_config(self) -> Config:
+ return Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={
+ "nas": Host(address="192.168.1.6"),
+ "nuc": Host(address="192.168.1.2"),
+ },
+ stacks={"test": "nas"},
+ glances_stack="glances",
+ )
+
+ def test_containers_page_without_glances(self, client: TestClient) -> None:
+ """Test containers page shows warning when Glances not configured."""
+ with patch("compose_farm.web.routes.containers.get_config") as mock:
+ mock.return_value = Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={"nas": Host(address="192.168.1.6")},
+ stacks={"test": "nas"},
+ glances_stack=None,
+ )
+ response = client.get("/live-stats")
+
+ assert response.status_code == 200
+ assert "Glances not configured" in response.text
+
+ def test_containers_page_with_glances(self, client: TestClient, mock_config: Config) -> None:
+ """Test containers page loads when Glances is configured."""
+ with patch("compose_farm.web.routes.containers.get_config") as mock:
+ mock.return_value = mock_config
+ response = client.get("/live-stats")
+
+ assert response.status_code == 200
+ assert "Live Stats" in response.text
+ assert "container-rows" in response.text
+
+
+class TestContainersRowsAPI:
+ """Tests for containers rows HTML endpoint."""
+
+ @pytest.fixture
+ def client(self) -> TestClient:
+ app = create_app()
+ return TestClient(app)
+
+ def test_rows_without_glances(self, client: TestClient) -> None:
+ """Test rows endpoint returns error when Glances not configured."""
+ with patch("compose_farm.web.routes.containers.get_config") as mock:
+ mock.return_value = Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={"nas": Host(address="192.168.1.6")},
+ stacks={"test": "nas"},
+ glances_stack=None,
+ )
+ response = client.get("/api/containers/rows")
+
+ assert response.status_code == 200
+ assert "Glances not configured" in response.text
+
+ def test_rows_returns_html(self, client: TestClient) -> None:
+ """Test rows endpoint returns HTML table rows."""
+ mock_containers = [
+ ContainerStats(
+ name="nginx",
+ host="nas",
+ status="running",
+ image="nginx:latest",
+ cpu_percent=5.5,
+ memory_usage=104857600,
+ memory_limit=1073741824,
+ memory_percent=9.77,
+ network_rx=1000,
+ network_tx=500,
+ uptime="2 hours",
+ ports="80->80/tcp",
+ engine="docker",
+ stack="web",
+ service="nginx",
+ ),
+ ]
+
+ with (
+ patch("compose_farm.web.routes.containers.get_config") as mock_config,
+ patch(
+ "compose_farm.web.routes.containers.fetch_all_container_stats",
+ new_callable=AsyncMock,
+ ) as mock_fetch,
+ ):
+ mock_config.return_value = Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={"nas": Host(address="192.168.1.6")},
+ stacks={"test": "nas"},
+ glances_stack="glances",
+ )
+ mock_fetch.return_value = mock_containers
+
+ response = client.get("/api/containers/rows")
+
+ assert response.status_code == 200
+ assert "
has attributes
+ assert "nginx" in response.text
+ assert "running" in response.text
+
+ def test_rows_have_data_sort_attributes(self, client: TestClient) -> None:
+ """Test rows have data-sort attributes for client-side sorting."""
+ mock_containers = [
+ ContainerStats(
+ name="alpha",
+ host="nas",
+ status="running",
+ image="nginx:latest",
+ cpu_percent=10.0,
+ memory_usage=100,
+ memory_limit=1000,
+ memory_percent=10.0,
+ network_rx=100,
+ network_tx=100,
+ uptime="1 hour",
+ ports="",
+ engine="docker",
+ stack="alpha",
+ service="web",
+ ),
+ ]
+
+ with (
+ patch("compose_farm.web.routes.containers.get_config") as mock_config,
+ patch(
+ "compose_farm.web.routes.containers.fetch_all_container_stats",
+ new_callable=AsyncMock,
+ ) as mock_fetch,
+ ):
+ mock_config.return_value = Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={"nas": Host(address="192.168.1.6")},
+ stacks={"test": "nas"},
+ glances_stack="glances",
+ )
+ mock_fetch.return_value = mock_containers
+
+ response = client.get("/api/containers/rows")
+ assert response.status_code == 200
+ # Check that cells have data-sort attributes
+ assert 'data-sort="alpha"' in response.text # stack
+ assert 'data-sort="web"' in response.text # service
+ assert 'data-sort="3600"' in response.text # uptime (1 hour = 3600s)
+ assert 'data-sort="10' in response.text # cpu
diff --git a/tests/test_glances.py b/tests/test_glances.py
new file mode 100644
index 0000000..cfc92c5
--- /dev/null
+++ b/tests/test_glances.py
@@ -0,0 +1,349 @@
+"""Tests for Glances integration."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import httpx
+import pytest
+
+from compose_farm.config import Config, Host
+from compose_farm.glances import (
+ DEFAULT_GLANCES_PORT,
+ ContainerStats,
+ HostStats,
+ fetch_all_container_stats,
+ fetch_all_host_stats,
+ fetch_container_stats,
+ fetch_host_stats,
+)
+
+
+class TestHostStats:
+ """Tests for HostStats dataclass."""
+
+ def test_host_stats_creation(self) -> None:
+ stats = HostStats(
+ host="nas",
+ cpu_percent=25.5,
+ mem_percent=50.0,
+ swap_percent=10.0,
+ load=2.5,
+ disk_percent=75.0,
+ )
+ assert stats.host == "nas"
+ assert stats.cpu_percent == 25.5
+ assert stats.mem_percent == 50.0
+ assert stats.disk_percent == 75.0
+ assert stats.error is None
+
+ def test_host_stats_from_error(self) -> None:
+ stats = HostStats.from_error("nas", "Connection refused")
+ assert stats.host == "nas"
+ assert stats.cpu_percent == 0
+ assert stats.mem_percent == 0
+ assert stats.error == "Connection refused"
+
+
+class TestFetchHostStats:
+ """Tests for fetch_host_stats function."""
+
+ @pytest.mark.asyncio
+ async def test_fetch_host_stats_success(self) -> None:
+ quicklook_response = httpx.Response(
+ 200,
+ json={
+ "cpu": 25.5,
+ "mem": 50.0,
+ "swap": 5.0,
+ "load": 2.5,
+ },
+ )
+ fs_response = httpx.Response(
+ 200,
+ json=[
+ {"mnt_point": "/", "percent": 65.0},
+ {"mnt_point": "/mnt/data", "percent": 80.0},
+ ],
+ )
+
+ async def mock_get(url: str) -> httpx.Response:
+ if "quicklook" in url:
+ return quicklook_response
+ return fs_response
+
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(side_effect=mock_get)
+
+ stats = await fetch_host_stats("nas", "192.168.1.6")
+
+ assert stats.host == "nas"
+ assert stats.cpu_percent == 25.5
+ assert stats.mem_percent == 50.0
+ assert stats.swap_percent == 5.0
+ assert stats.load == 2.5
+ assert stats.disk_percent == 65.0 # Root filesystem
+ assert stats.error is None
+
+ @pytest.mark.asyncio
+ async def test_fetch_host_stats_http_error(self) -> None:
+ mock_response = httpx.Response(500)
+
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(return_value=mock_response)
+
+ stats = await fetch_host_stats("nas", "192.168.1.6")
+
+ assert stats.host == "nas"
+ assert stats.error == "HTTP 500"
+ assert stats.cpu_percent == 0
+
+ @pytest.mark.asyncio
+ async def test_fetch_host_stats_timeout(self) -> None:
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(side_effect=httpx.TimeoutException("timeout"))
+
+ stats = await fetch_host_stats("nas", "192.168.1.6")
+
+ assert stats.host == "nas"
+ assert stats.error == "timeout"
+
+ @pytest.mark.asyncio
+ async def test_fetch_host_stats_connection_error(self) -> None:
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(
+ side_effect=httpx.ConnectError("Connection refused")
+ )
+
+ stats = await fetch_host_stats("nas", "192.168.1.6")
+
+ assert stats.host == "nas"
+ assert stats.error is not None
+ assert "Connection refused" in stats.error
+
+
+class TestFetchAllHostStats:
+ """Tests for fetch_all_host_stats function."""
+
+ @pytest.mark.asyncio
+ async def test_fetch_all_host_stats(self) -> None:
+ config = Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={
+ "nas": Host(address="192.168.1.6"),
+ "nuc": Host(address="192.168.1.2"),
+ },
+ stacks={"test": "nas"},
+ )
+
+ quicklook_response = httpx.Response(
+ 200,
+ json={
+ "cpu": 25.5,
+ "mem": 50.0,
+ "swap": 5.0,
+ "load": 2.5,
+ },
+ )
+ fs_response = httpx.Response(
+ 200,
+ json=[{"mnt_point": "/", "percent": 70.0}],
+ )
+
+ async def mock_get(url: str) -> httpx.Response:
+ if "quicklook" in url:
+ return quicklook_response
+ return fs_response
+
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(side_effect=mock_get)
+
+ stats = await fetch_all_host_stats(config)
+
+ assert "nas" in stats
+ assert "nuc" in stats
+ assert stats["nas"].cpu_percent == 25.5
+ assert stats["nuc"].cpu_percent == 25.5
+ assert stats["nas"].disk_percent == 70.0
+
+
+class TestDefaultPort:
+ """Tests for default Glances port constant."""
+
+ def test_default_port(self) -> None:
+ assert DEFAULT_GLANCES_PORT == 61208
+
+
+class TestContainerStats:
+ """Tests for ContainerStats dataclass."""
+
+ def test_container_stats_creation(self) -> None:
+ stats = ContainerStats(
+ name="nginx",
+ host="nas",
+ status="running",
+ image="nginx:latest",
+ cpu_percent=5.5,
+ memory_usage=104857600, # 100MB
+ memory_limit=1073741824, # 1GB
+ memory_percent=9.77,
+ network_rx=1000000,
+ network_tx=500000,
+ uptime="2 hours",
+ ports="80->80/tcp",
+ engine="docker",
+ )
+ assert stats.name == "nginx"
+ assert stats.host == "nas"
+ assert stats.cpu_percent == 5.5
+
+
+class TestFetchContainerStats:
+ """Tests for fetch_container_stats function."""
+
+ @pytest.mark.asyncio
+ async def test_fetch_container_stats_success(self) -> None:
+ mock_response = httpx.Response(
+ 200,
+ json=[
+ {
+ "name": "nginx",
+ "status": "running",
+ "image": ["nginx:latest"],
+ "cpu_percent": 5.5,
+ "memory_usage": 104857600,
+ "memory_limit": 1073741824,
+ "network": {"cumulative_rx": 1000, "cumulative_tx": 500},
+ "uptime": "2 hours",
+ "ports": "80->80/tcp",
+ "engine": "docker",
+ },
+ {
+ "name": "redis",
+ "status": "running",
+ "image": ["redis:7"],
+ "cpu_percent": 1.2,
+ "memory_usage": 52428800,
+ "memory_limit": 1073741824,
+ "network": {},
+ "uptime": "3 hours",
+ "ports": "",
+ "engine": "docker",
+ },
+ ],
+ )
+
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(return_value=mock_response)
+
+ containers, error = await fetch_container_stats("nas", "192.168.1.6")
+
+ assert error is None
+ assert containers is not None
+ assert len(containers) == 2
+ assert containers[0].name == "nginx"
+ assert containers[0].host == "nas"
+ assert containers[0].cpu_percent == 5.5
+ assert containers[1].name == "redis"
+
+ @pytest.mark.asyncio
+ async def test_fetch_container_stats_empty_on_error(self) -> None:
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(side_effect=httpx.TimeoutException("timeout"))
+
+ containers, error = await fetch_container_stats("nas", "192.168.1.6")
+
+ assert containers is None
+ assert error == "Connection timed out"
+
+ @pytest.mark.asyncio
+ async def test_fetch_container_stats_handles_string_image(self) -> None:
+ """Test that image field works as string (not just list)."""
+ mock_response = httpx.Response(
+ 200,
+ json=[
+ {
+ "name": "test",
+ "status": "running",
+ "image": "myimage:v1", # String instead of list
+ "cpu_percent": 0,
+ "memory_usage": 0,
+ "memory_limit": 1,
+ "network": {},
+ "uptime": "",
+ "ports": "",
+ "engine": "docker",
+ },
+ ],
+ )
+
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(return_value=mock_response)
+
+ containers, error = await fetch_container_stats("nas", "192.168.1.6")
+
+ assert error is None
+ assert containers is not None
+ assert len(containers) == 1
+ assert containers[0].image == "myimage:v1"
+
+
+class TestFetchAllContainerStats:
+ """Tests for fetch_all_container_stats function."""
+
+ @pytest.mark.asyncio
+ async def test_fetch_all_container_stats(self) -> None:
+ config = Config(
+ compose_dir=Path("/opt/compose"),
+ hosts={
+ "nas": Host(address="192.168.1.6"),
+ "nuc": Host(address="192.168.1.2"),
+ },
+ stacks={"test": "nas"},
+ )
+
+ mock_response = httpx.Response(
+ 200,
+ json=[
+ {
+ "name": "nginx",
+ "status": "running",
+ "image": ["nginx:latest"],
+ "cpu_percent": 5.5,
+ "memory_usage": 104857600,
+ "memory_limit": 1073741824,
+ "network": {},
+ "uptime": "2 hours",
+ "ports": "",
+ "engine": "docker",
+ },
+ ],
+ )
+
+ with patch("httpx.AsyncClient") as mock_client:
+ mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_client.return_value)
+ mock_client.return_value.__aexit__ = AsyncMock(return_value=None)
+ mock_client.return_value.get = AsyncMock(return_value=mock_response)
+
+ containers = await fetch_all_container_stats(config)
+
+ # 2 hosts x 1 container each = 2 containers
+ assert len(containers) == 2
+ hosts = {c.host for c in containers}
+ assert "nas" in hosts
+ assert "nuc" in hosts
diff --git a/tests/test_registry.py b/tests/test_registry.py
new file mode 100644
index 0000000..bb52c0d
--- /dev/null
+++ b/tests/test_registry.py
@@ -0,0 +1,182 @@
+"""Tests for registry module."""
+
+from compose_farm.registry import (
+ DOCKER_HUB_ALIASES,
+ ImageRef,
+ RegistryClient,
+ TagCheckResult,
+ _find_updates,
+ _parse_version,
+)
+
+
+class TestImageRef:
+ """Tests for ImageRef parsing."""
+
+ def test_parse_simple_image(self) -> None:
+ """Test parsing simple image name."""
+ ref = ImageRef.parse("nginx")
+ assert ref.registry == "docker.io"
+ assert ref.namespace == "library"
+ assert ref.name == "nginx"
+ assert ref.tag == "latest"
+
+ def test_parse_image_with_tag(self) -> None:
+ """Test parsing image with tag."""
+ ref = ImageRef.parse("nginx:1.25")
+ assert ref.registry == "docker.io"
+ assert ref.namespace == "library"
+ assert ref.name == "nginx"
+ assert ref.tag == "1.25"
+
+ def test_parse_image_with_namespace(self) -> None:
+ """Test parsing image with namespace."""
+ ref = ImageRef.parse("linuxserver/jellyfin:latest")
+ assert ref.registry == "docker.io"
+ assert ref.namespace == "linuxserver"
+ assert ref.name == "jellyfin"
+ assert ref.tag == "latest"
+
+ def test_parse_ghcr_image(self) -> None:
+ """Test parsing GitHub Container Registry image."""
+ ref = ImageRef.parse("ghcr.io/user/repo:v1.0.0")
+ assert ref.registry == "ghcr.io"
+ assert ref.namespace == "user"
+ assert ref.name == "repo"
+ assert ref.tag == "v1.0.0"
+
+ def test_parse_image_with_digest(self) -> None:
+ """Test parsing image with digest."""
+ ref = ImageRef.parse("nginx:latest@sha256:abc123")
+ assert ref.registry == "docker.io"
+ assert ref.name == "nginx"
+ assert ref.tag == "latest"
+ assert ref.digest == "sha256:abc123"
+
+ def test_full_name_with_namespace(self) -> None:
+ """Test full_name property with namespace."""
+ ref = ImageRef.parse("linuxserver/jellyfin")
+ assert ref.full_name == "linuxserver/jellyfin"
+
+ def test_full_name_without_namespace(self) -> None:
+ """Test full_name property for official images."""
+ ref = ImageRef.parse("nginx")
+ assert ref.full_name == "library/nginx"
+
+ def test_display_name_official_image(self) -> None:
+ """Test display_name for official Docker Hub images."""
+ ref = ImageRef.parse("nginx:latest")
+ assert ref.display_name == "nginx"
+
+ def test_display_name_hub_with_namespace(self) -> None:
+ """Test display_name for Docker Hub images with namespace."""
+ ref = ImageRef.parse("linuxserver/jellyfin")
+ assert ref.display_name == "linuxserver/jellyfin"
+
+ def test_display_name_other_registry(self) -> None:
+ """Test display_name for other registries."""
+ ref = ImageRef.parse("ghcr.io/user/repo")
+ assert ref.display_name == "ghcr.io/user/repo"
+
+
+class TestParseVersion:
+ """Tests for version parsing."""
+
+ def test_parse_semver(self) -> None:
+ """Test parsing semantic version."""
+ assert _parse_version("1.2.3") == (1, 2, 3)
+
+ def test_parse_version_with_v_prefix(self) -> None:
+ """Test parsing version with v prefix."""
+ assert _parse_version("v1.2.3") == (1, 2, 3)
+ assert _parse_version("V1.2.3") == (1, 2, 3)
+
+ def test_parse_two_part_version(self) -> None:
+ """Test parsing two-part version."""
+ assert _parse_version("1.25") == (1, 25)
+
+ def test_parse_single_number(self) -> None:
+ """Test parsing single number version."""
+ assert _parse_version("7") == (7,)
+
+ def test_parse_invalid_version(self) -> None:
+ """Test parsing non-version tags."""
+ assert _parse_version("latest") is None
+ assert _parse_version("stable") is None
+ assert _parse_version("alpine") is None
+
+
+class TestFindUpdates:
+ """Tests for finding available updates."""
+
+ def test_find_updates_with_newer_versions(self) -> None:
+ """Test finding newer versions."""
+ current = "1.0.0"
+ tags = ["0.9.0", "1.0.0", "1.1.0", "2.0.0"]
+ updates = _find_updates(current, tags)
+ assert updates == ["2.0.0", "1.1.0"]
+
+ def test_find_updates_no_newer(self) -> None:
+ """Test when already on latest."""
+ current = "2.0.0"
+ tags = ["1.0.0", "1.5.0", "2.0.0"]
+ updates = _find_updates(current, tags)
+ assert updates == []
+
+ def test_find_updates_non_version_tag(self) -> None:
+ """Test with non-version current tag."""
+ current = "latest"
+ tags = ["1.0.0", "2.0.0"]
+ updates = _find_updates(current, tags)
+ # Can't determine updates for non-version tags
+ assert updates == []
+
+
+class TestRegistryClient:
+ """Tests for unified registry client."""
+
+ def test_docker_hub_normalization(self) -> None:
+ """Test Docker Hub aliases are normalized."""
+ for alias in DOCKER_HUB_ALIASES:
+ client = RegistryClient(alias)
+ assert client.registry == "docker.io"
+ assert client.registry_url == "https://registry-1.docker.io"
+
+ def test_ghcr_client(self) -> None:
+ """Test GitHub Container Registry client."""
+ client = RegistryClient("ghcr.io")
+ assert client.registry == "ghcr.io"
+ assert client.registry_url == "https://ghcr.io"
+
+ def test_generic_registry(self) -> None:
+ """Test generic registry client."""
+ client = RegistryClient("quay.io")
+ assert client.registry == "quay.io"
+ assert client.registry_url == "https://quay.io"
+
+
+class TestTagCheckResult:
+ """Tests for TagCheckResult."""
+
+ def test_create_result(self) -> None:
+ """Test creating a result."""
+ ref = ImageRef.parse("nginx:1.25")
+ result = TagCheckResult(
+ image=ref,
+ current_digest="sha256:abc",
+ available_updates=["1.26", "1.27"],
+ )
+ assert result.image.name == "nginx"
+ assert result.available_updates == ["1.26", "1.27"]
+ assert result.error is None
+
+ def test_result_with_error(self) -> None:
+ """Test result with error."""
+ ref = ImageRef.parse("nginx")
+ result = TagCheckResult(
+ image=ref,
+ current_digest="",
+ error="Connection refused",
+ )
+ assert result.error == "Connection refused"
+ assert result.available_updates == []
diff --git a/tests/web/test_htmx_browser.py b/tests/web/test_htmx_browser.py
index 3396653..390c779 100644
--- a/tests/web/test_htmx_browser.py
+++ b/tests/web/test_htmx_browser.py
@@ -134,6 +134,13 @@ def test_config(tmp_path_factory: pytest.TempPathFactory) -> Path:
else:
(svc / "compose.yaml").write_text(f"services:\n {name}:\n image: test/{name}\n")
+ # Create glances stack (required for containers page)
+ glances_dir = compose_dir / "glances"
+ glances_dir.mkdir()
+ (glances_dir / "compose.yaml").write_text(
+ "services:\n glances:\n image: nicolargo/glances\n"
+ )
+
# Create config with multiple hosts
config = tmp / "compose-farm.yaml"
config.write_text(f"""
@@ -151,6 +158,8 @@ stacks:
nextcloud: server-2
jellyfin: server-2
redis: server-1
+ glances: all
+glances_stack: glances
""")
# Create state (plex and nextcloud running, grafana and jellyfin not started)
@@ -245,7 +254,7 @@ class TestHTMXSidebarLoading:
# Verify actual stacks from test config appear
stacks = page.locator("#sidebar-stacks li")
- assert stacks.count() == 5 # plex, grafana, nextcloud, jellyfin, redis
+ assert stacks.count() == 6 # plex, grafana, nextcloud, jellyfin, redis, glances
# Check specific stacks are present
content = page.locator("#sidebar-stacks").inner_text()
@@ -348,7 +357,7 @@ class TestDashboardContent:
# From test config: 2 hosts, 5 stacks, 2 running (plex, nextcloud)
assert "2" in stats # hosts count
- assert "5" in stats # stacks count
+ assert "6" in stats # stacks count
def test_pending_shows_not_started_stacks(self, page: Page, server_url: str) -> None:
"""Pending operations shows grafana and jellyfin as not started."""
@@ -476,9 +485,9 @@ class TestSidebarFilter:
page.goto(server_url)
page.wait_for_selector("#sidebar-stacks", timeout=TIMEOUT)
- # Initially all 4 stacks visible
+ # Initially all 6 stacks visible
visible_items = page.locator("#sidebar-stacks li:not([hidden])")
- assert visible_items.count() == 5
+ assert visible_items.count() == 6
# Type in filter to match only "plex"
self._filter_sidebar(page, "plex")
@@ -493,9 +502,9 @@ class TestSidebarFilter:
page.goto(server_url)
page.wait_for_selector("#sidebar-stacks", timeout=TIMEOUT)
- # Initial count should be (5)
+ # Initial count should be (6)
count_badge = page.locator("#sidebar-count")
- assert "(5)" in count_badge.inner_text()
+ assert "(6)" in count_badge.inner_text()
# Filter to show only stacks containing "x" (plex, nextcloud)
self._filter_sidebar(page, "x")
@@ -524,13 +533,14 @@ class TestSidebarFilter:
# Select server-1 from dropdown
page.locator("#sidebar-host-select").select_option("server-1")
- # Only plex, grafana, and redis (server-1 stacks) should be visible
+ # plex, grafana, redis (server-1), and glances (all) should be visible
visible = page.locator("#sidebar-stacks li:not([hidden])")
- assert visible.count() == 3
+ assert visible.count() == 4
content = visible.all_inner_texts()
assert any("plex" in s for s in content)
assert any("grafana" in s for s in content)
+ assert any("glances" in s for s in content)
assert not any("nextcloud" in s for s in content)
assert not any("jellyfin" in s for s in content)
@@ -562,7 +572,7 @@ class TestSidebarFilter:
self._filter_sidebar(page, "")
# All stacks visible again
- assert page.locator("#sidebar-stacks li:not([hidden])").count() == 5
+ assert page.locator("#sidebar-stacks li:not([hidden])").count() == 6
class TestCommandPalette:
@@ -884,7 +894,7 @@ class TestContentStability:
# Remember sidebar state
initial_count = page.locator("#sidebar-stacks li").count()
- assert initial_count == 5
+ assert initial_count == 6
# Navigate away
page.locator("#sidebar-stacks a", has_text="plex").click()
@@ -2329,3 +2339,227 @@ class TestTerminalNavigationIsolation:
# Terminal should still be collapsed (no task to reconnect to)
terminal_toggle = page.locator("#terminal-toggle")
assert not terminal_toggle.is_checked(), "Terminal should remain collapsed after navigation"
+
+
+class TestContainersPagePause:
+ """Test containers page auto-refresh pause mechanism.
+
+ The containers page auto-refreshes every 3 seconds. When a user opens
+ an action dropdown, refresh should pause to prevent the dropdown from
+ closing unexpectedly.
+ """
+
+ # Mock HTML for container rows with action dropdowns
+ MOCK_ROWS_HTML = """
+
+| 1 |
+plex |
+server |
+ |
+nas |
+nginx:latest |
+running |
+1 hour |
+5% |
+100MB |
+↓1KB ↑1KB |
+
+
+| 2 |
+redis |
+redis |
+ |
+nas |
+redis:7 |
+running |
+2 hours |
+1% |
+50MB |
+↓500B ↑500B |
+
+"""
+
+ def test_dropdown_pauses_refresh(self, page: Page, server_url: str) -> None:
+ """Opening action dropdown pauses auto-refresh.
+
+ Bug: focusin event triggers pause, but focusout fires shortly after
+ when focus moves within the dropdown, causing refresh to resume
+ while dropdown is still visually open.
+ """
+ # Mock container rows and update checks
+ page.route(
+ "**/api/containers/rows/*",
+ lambda route: route.fulfill(
+ status=200,
+ content_type="text/html",
+ body=self.MOCK_ROWS_HTML,
+ ),
+ )
+ page.route(
+ "**/api/containers/check-updates",
+ lambda route: route.fulfill(
+ status=200,
+ content_type="application/json",
+ body='{"results": []}',
+ ),
+ )
+
+ page.goto(f"{server_url}/live-stats")
+
+ # Wait for container rows to load
+ page.wait_for_function(
+ "document.querySelectorAll('#container-rows tr:not(.loading-row)').length > 0",
+ timeout=TIMEOUT,
+ )
+
+ # Wait for timer to start
+ page.wait_for_function(
+ "document.getElementById('refresh-timer')?.textContent?.includes('↻')",
+ timeout=TIMEOUT,
+ )
+
+ # Click on a dropdown to open it
+ dropdown_label = page.locator(".dropdown label").first
+ dropdown_label.click()
+
+ # Wait a moment for focusin to trigger
+ page.wait_for_timeout(200)
+
+ # Verify pause is engaged
+ timer_text = page.locator("#refresh-timer").inner_text()
+
+ assert timer_text == "❚❚", (
+ f"Refresh should be paused after clicking dropdown. timer='{timer_text}'"
+ )
+ assert "❚❚" in timer_text, f"Timer should show pause icon, got '{timer_text}'"
+
+ def test_refresh_stays_paused_while_dropdown_open(self, page: Page, server_url: str) -> None:
+ """Refresh remains paused for duration dropdown is open (>5s refresh interval).
+
+ This is the critical test for the pause bug: refresh should stay paused
+ for longer than the 3-second refresh interval while dropdown is open.
+ """
+ # Mock container rows and update checks
+ page.route(
+ "**/api/containers/rows/*",
+ lambda route: route.fulfill(
+ status=200,
+ content_type="text/html",
+ body=self.MOCK_ROWS_HTML,
+ ),
+ )
+ page.route(
+ "**/api/containers/check-updates",
+ lambda route: route.fulfill(
+ status=200,
+ content_type="application/json",
+ body='{"results": []}',
+ ),
+ )
+
+ page.goto(f"{server_url}/live-stats")
+
+ # Wait for container rows to load
+ page.wait_for_function(
+ "document.querySelectorAll('#container-rows tr:not(.loading-row)').length > 0",
+ timeout=TIMEOUT,
+ )
+
+ # Wait for timer to start
+ page.wait_for_function(
+ "document.getElementById('refresh-timer')?.textContent?.includes('↻')",
+ timeout=TIMEOUT,
+ )
+
+ # Record a marker in the first row to detect if refresh happened
+ page.evaluate("""
+ const firstRow = document.querySelector('#container-rows tr');
+ if (firstRow) firstRow.dataset.testMarker = 'original';
+ """)
+
+ # Click dropdown to pause
+ dropdown_label = page.locator(".dropdown label").first
+ dropdown_label.click()
+ page.wait_for_timeout(200)
+
+ # Confirm paused
+ assert page.locator("#refresh-timer").inner_text() == "❚❚"
+
+ # Wait longer than the 5-second refresh interval
+ page.wait_for_timeout(6000)
+
+ # Check if still paused
+ timer_text = page.locator("#refresh-timer").inner_text()
+
+ # Check if the row was replaced (marker would be gone)
+ marker = page.evaluate("""
+ document.querySelector('#container-rows tr')?.dataset?.testMarker
+ """)
+
+ assert timer_text == "❚❚", f"Refresh should still be paused after 6s. timer='{timer_text}'"
+ assert marker == "original", (
+ "Table was refreshed while dropdown was open - pause mechanism failed"
+ )
+
+ def test_refresh_resumes_after_dropdown_closes(self, page: Page, server_url: str) -> None:
+ """Refresh resumes after dropdown is closed."""
+ # Mock container rows and update checks
+ page.route(
+ "**/api/containers/rows/*",
+ lambda route: route.fulfill(
+ status=200,
+ content_type="text/html",
+ body=self.MOCK_ROWS_HTML,
+ ),
+ )
+ page.route(
+ "**/api/containers/check-updates",
+ lambda route: route.fulfill(
+ status=200,
+ content_type="application/json",
+ body='{"results": []}',
+ ),
+ )
+
+ page.goto(f"{server_url}/live-stats")
+
+ # Wait for container rows to load
+ page.wait_for_function(
+ "document.querySelectorAll('#container-rows tr:not(.loading-row)').length > 0",
+ timeout=TIMEOUT,
+ )
+
+ # Wait for timer to start
+ page.wait_for_function(
+ "document.getElementById('refresh-timer')?.textContent?.includes('↻')",
+ timeout=TIMEOUT,
+ )
+
+ # Click dropdown to pause
+ dropdown_label = page.locator(".dropdown label").first
+ dropdown_label.click()
+ page.wait_for_timeout(200)
+
+ assert page.locator("#refresh-timer").inner_text() == "❚❚"
+
+ # Close dropdown by pressing Escape or clicking elsewhere
+ page.keyboard.press("Escape")
+ page.wait_for_timeout(300) # Wait for focusout timeout (150ms) + buffer
+
+ # Verify refresh resumed
+ timer_text = page.locator("#refresh-timer").inner_text()
+
+ assert timer_text != "❚❚", (
+ f"Refresh should resume after closing dropdown. timer='{timer_text}'"
+ )
+ assert "↻" in timer_text, f"Timer should show countdown, got '{timer_text}'"
diff --git a/uv.lock b/uv.lock
index 0fffc67..52533d3 100644
--- a/uv.lock
+++ b/uv.lock
@@ -242,6 +242,7 @@ dependencies = [
[package.optional-dependencies]
web = [
{ name = "fastapi", extra = ["standard"] },
+ { name = "humanize" },
{ name = "jinja2" },
{ name = "websockets" },
]
@@ -270,6 +271,7 @@ dev = [
requires-dist = [
{ name = "asyncssh", specifier = ">=2.14.0" },
{ name = "fastapi", extras = ["standard"], marker = "extra == 'web'", specifier = ">=0.109.0" },
+ { name = "humanize", marker = "extra == 'web'", specifier = ">=4.0.0" },
{ name = "jinja2", marker = "extra == 'web'", specifier = ">=3.1.0" },
{ name = "pydantic", specifier = ">=2.0.0" },
{ name = "pyyaml", specifier = ">=6.0" },
@@ -781,6 +783,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
]
+[[package]]
+name = "humanize"
+version = "4.15.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/ba/66/a3921783d54be8a6870ac4ccffcd15c4dc0dd7fcce51c6d63b8c63935276/humanize-4.15.0.tar.gz", hash = "sha256:1dd098483eb1c7ee8e32eb2e99ad1910baefa4b75c3aff3a82f4d78688993b10", size = 83599, upload-time = "2025-12-20T20:16:13.19Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/c5/7b/bca5613a0c3b542420cf92bd5e5fb8ebd5435ce1011a091f66bb7693285e/humanize-4.15.0-py3-none-any.whl", hash = "sha256:b1186eb9f5a9749cd9cb8565aee77919dd7c8d076161cf44d70e59e3301e1769", size = 132203, upload-time = "2025-12-20T20:16:11.67Z" },
+]
+
[[package]]
name = "identify"
version = "2.6.15"