+
+
Docker Topology
+ {{ topo.containers|length }} containers · {{ topo.networks|length }} networks
+
+ Live read-only view of this host's Docker daemon, routed through {{ topo.proxy }}. The proxy exposes only GET on containers and networks — psyc cannot start, stop, exec into, or modify anything from here.
+ ← back to admin
+
+ {% if topo.error %}
+ ✗ Socket-proxy unreachable: {{ topo.error }}
+ {% endif %}
+
+ {% if topo.containers %}
+
+
+ re-settle
+ drag any node · scroll to zoom
+
+
+
+
+
Networks
+
+ {% for n in topo.networks %}
+
+
+
+
{{ n.name }}
+
{{ n.driver }} · {{ n.scope }}{% if n.internal %} · internal{% endif %}
+
+
{{ n.containers|length }}
+
+ {% if n.containers %}
+
+ {% for c in n.containers %}
+ {{ c.name }} {{ c.ip or '—' }}
+ {% else %}
+
(no attached containers)
+ {% endif %}
+
+ {% endfor %}
+
Containers
+
+ Name Image State Networks Ports
+ {% for c in topo.containers %}
+
+ {{ c.name }} {{ c.id }}
{{ c.image }}
+ {{ c.state }}
+ {% for net in c.networks %}{{ net.name }} {{ net.ip or '—' }}
+ {% for p in c.ports %}{{ p }}{% if not loop.last %}
+
+ {% endfor %}
+
+
+
+{% endblock %}
diff --git a/tests/test_docker_view.py b/tests/test_docker_view.py
new file mode 100644
index 0000000..05f3c86
--- /dev/null
+++ b/tests/test_docker_view.py
@@ -0,0 +1,124 @@
+"""Docker topology — normalization + error handling against the socket-proxy."""
+
+from __future__ import annotations
+
+import pytest
+
+from psyc.cockpit import docker_view
+
+
+_CONTAINERS_FIXTURE = [
+ {
+ "Id": "abcdef1234567890",
+ "Names": ["/psyc-cockpit-1"],
+ "Image": "psyc:latest",
+ "State": "running",
+ "Status": "Up 5 minutes (healthy)",
+ "NetworkSettings": {"Networks": {"backend": {"IPAddress": "172.20.0.5"}}},
+ "Ports": [{"IP": "0.0.0.0", "PrivatePort": 8767, "PublicPort": 8767, "Type": "tcp"}],
+ },
+ {
+ "Id": "fedcba0987654321",
+ "Names": ["/some-stopped"],
+ "Image": "alpine",
+ "State": "exited",
+ "Status": "Exited (0) 2 hours ago",
+ "NetworkSettings": {"Networks": {}},
+ "Ports": [],
+ },
+]
+
+_NETWORKS_FIXTURE = [
+ {
+ "Id": "n1", "Name": "backend", "Driver": "bridge", "Scope": "local", "Internal": False,
+ "IPAM": {"Config": [{"Subnet": "172.20.0.0/16", "Gateway": "172.20.0.1"}]},
+ "Containers": {
+ "abcdef1234567890": {
+ "Name": "psyc-cockpit-1", "IPv4Address": "172.20.0.5/16",
+ "MacAddress": "02:42:ac:14:00:05",
+ },
+ },
+ },
+ {"Id": "n2", "Name": "bridge", "Driver": "bridge", "Scope": "local", "Internal": False, "Containers": {}},
+]
+
+
+def _fake_get_factory(monkeypatch, payloads: dict):
+ """Patch docker_view._get to return canned payloads by path.
+
+ Unknown paths raise DockerProxyError (mimicking a proxy that blocks the
+ endpoint), so callers like host_info() exercise their fallback paths.
+ """
+ def fake_get(path: str):
+ for prefix, body in payloads.items():
+ if path.startswith(prefix):
+ return body
+ raise docker_view.DockerProxyError(f"blocked: {path}")
+ monkeypatch.setattr(docker_view, "_get", fake_get)
+
+
+def test_list_containers_normalizes_fields(monkeypatch):
+ _fake_get_factory(monkeypatch, {"/containers/json": _CONTAINERS_FIXTURE})
+ out = docker_view.list_containers()
+ by_name = {c["name"]: c for c in out}
+ # running comes before exited
+ assert out[0]["state"] == "running"
+ assert out[-1]["state"] == "exited"
+ cockpit = by_name["psyc-cockpit-1"]
+ assert cockpit["id"] == "abcdef123456"
+ assert cockpit["image"] == "psyc:latest"
+ assert cockpit["networks"][0]["name"] == "backend"
+ assert cockpit["networks"][0]["ip"] == "172.20.0.5"
+ assert "0.0.0.0:8767->8767/tcp" in cockpit["ports"]
+
+
+def test_list_networks_attaches_containers_with_ip(monkeypatch):
+ _fake_get_factory(monkeypatch, {"/networks": _NETWORKS_FIXTURE})
+ out = docker_view.list_networks()
+ backend = next(n for n in out if n["name"] == "backend")
+ assert backend["driver"] == "bridge"
+ assert backend["subnet"] == "172.20.0.0/16"
+ assert backend["gateway"] == "172.20.0.1"
+ assert backend["containers"][0]["name"] == "psyc-cockpit-1"
+ assert backend["containers"][0]["ip"] == "172.20.0.5"
+ assert backend["containers"][0]["mac"] == "02:42:ac:14:00:05"
+ # default networks pushed to bottom
+ assert out[-1]["name"] == "bridge"
+
+
+def test_list_containers_extracts_published_ports(monkeypatch):
+ _fake_get_factory(monkeypatch, {"/containers/json": _CONTAINERS_FIXTURE})
+ out = docker_view.list_containers()
+ cockpit = next(c for c in out if c["name"] == "psyc-cockpit-1")
+ assert cockpit["published_ports"] == ["8767/tcp"]
+ # stopped container with no ports → empty
+ stopped = next(c for c in out if c["name"] == "some-stopped")
+ assert stopped["published_ports"] == []
+
+
+def test_host_info_falls_back_when_proxy_blocks_info(monkeypatch):
+ def boom(path):
+ raise docker_view.DockerProxyError("info forbidden")
+ monkeypatch.setattr(docker_view, "_get", boom)
+ info = docker_view.host_info()
+ assert info["name"] == "docker host"
+
+
+def test_topology_returns_both_with_no_error(monkeypatch):
+ _fake_get_factory(monkeypatch, {
+ "/containers/json": _CONTAINERS_FIXTURE,
+ "/networks": _NETWORKS_FIXTURE,
+ })
+ snap = docker_view.topology()
+ assert snap["error"] is None
+ assert len(snap["containers"]) == 2
+ assert len(snap["networks"]) == 2
+
+
+def test_topology_surfaces_proxy_failure(monkeypatch):
+ def boom(path):
+ raise docker_view.DockerProxyError("connection refused")
+ monkeypatch.setattr(docker_view, "_get", boom)
+ snap = docker_view.topology()
+ assert snap["error"] is not None and "connection refused" in snap["error"]
+ assert snap["containers"] == [] and snap["networks"] == []