Support containers with either webterm-command or webterm-theme label

Previously, only containers with the webterm-command label were detected
by the Docker watcher. Now containers with webterm-theme label (but not
webterm-command) are also picked up and use the default auto command.

Changes:
- Add _has_webterm_label() helper to check for any webterm label
- Update event handler to use the new helper
- Update _get_labeled_containers() to query for both labels
- Add tests for theme-only label detection
This commit is contained in:
GitHub Copilot
2026-01-29 19:18:55 +00:00
parent 30e9bbed4d
commit 3ee35b54f5
2 changed files with 95 additions and 9 deletions
+30 -9
View File
@@ -22,11 +22,18 @@ log = logging.getLogger("webterm")
LABEL_NAME = "webterm-command" LABEL_NAME = "webterm-command"
THEME_LABEL = "webterm-theme" THEME_LABEL = "webterm-theme"
# All labels that trigger container inclusion
WEBTERM_LABELS = (LABEL_NAME, THEME_LABEL)
AUTO_COMMAND_ENV = "WEBTERM_DOCKER_AUTO_COMMAND" AUTO_COMMAND_ENV = "WEBTERM_DOCKER_AUTO_COMMAND"
DEFAULT_COMMAND = "/bin/bash" DEFAULT_COMMAND = "/bin/bash"
AUTO_COMMAND_SENTINEL = "__docker_exec__" AUTO_COMMAND_SENTINEL = "__docker_exec__"
def _has_webterm_label(attributes: dict) -> bool:
"""Check if a container has any webterm label."""
return any(label in attributes for label in WEBTERM_LABELS)
def _get_auto_command() -> str: def _get_auto_command() -> str:
return os.environ.get(AUTO_COMMAND_ENV, DEFAULT_COMMAND) return os.environ.get(AUTO_COMMAND_ENV, DEFAULT_COMMAND)
@@ -117,13 +124,27 @@ class DockerWatcher:
await writer.wait_closed() await writer.wait_closed()
async def _get_labeled_containers(self) -> list[dict]: async def _get_labeled_containers(self) -> list[dict]:
"""Get all running containers with webterm-command label.""" """Get all running containers with any webterm label.
path = f'/containers/json?filters={{"label":["{LABEL_NAME}"]}}'
status, body = await self._docker_request("GET", path) Queries for both webterm-command and webterm-theme labels,
if status != 200: merging results and deduplicating by container ID.
log.error("Failed to list containers: %s", body) """
return [] seen_ids: set[str] = set()
return json.loads(body) result: list[dict] = []
for label in WEBTERM_LABELS:
path = f'/containers/json?filters={{"label":["{label}"]}}'
status, body = await self._docker_request("GET", path)
if status != 200:
log.error("Failed to list containers for label %s: %s", label, body)
continue
for container in json.loads(body):
container_id = container.get("Id", "")
if container_id and container_id not in seen_ids:
seen_ids.add(container_id)
result.append(container)
return result
def _get_container_command(self, container: dict) -> str: def _get_container_command(self, container: dict) -> str:
"""Get command for container from label. """Get command for container from label.
@@ -260,8 +281,8 @@ class DockerWatcher:
container_id = actor.get("ID", "") container_id = actor.get("ID", "")
attributes = actor.get("Attributes", {}) attributes = actor.get("Attributes", {})
# Only handle containers with our label # Only handle containers with any webterm label
if LABEL_NAME not in attributes: if not _has_webterm_label(attributes):
return return
if action == "start": if action == "start":
+65
View File
@@ -11,6 +11,7 @@ from webterm.docker_watcher import (
LABEL_NAME, LABEL_NAME,
THEME_LABEL, THEME_LABEL,
DockerWatcher, DockerWatcher,
_has_webterm_label,
) )
@@ -289,3 +290,67 @@ async def test_watch_events_recovers_from_errors(docker_watcher, monkeypatch):
monkeypatch.setattr("webterm.docker_watcher.asyncio.open_unix_connection", fail_once) monkeypatch.setattr("webterm.docker_watcher.asyncio.open_unix_connection", fail_once)
monkeypatch.setattr("webterm.docker_watcher.asyncio.sleep", fake_sleep) monkeypatch.setattr("webterm.docker_watcher.asyncio.sleep", fake_sleep)
await docker_watcher._watch_events() await docker_watcher._watch_events()
class TestHasWebtermLabel:
"""Tests for _has_webterm_label helper."""
def test_has_command_label(self):
"""Container with webterm-command label is detected."""
assert _has_webterm_label({LABEL_NAME: "auto"}) is True
def test_has_theme_label(self):
"""Container with webterm-theme label is detected."""
assert _has_webterm_label({THEME_LABEL: "dracula"}) is True
def test_has_both_labels(self):
"""Container with both labels is detected."""
assert _has_webterm_label({LABEL_NAME: "bash", THEME_LABEL: "dark"}) is True
def test_no_webterm_labels(self):
"""Container without webterm labels is not detected."""
assert _has_webterm_label({"other-label": "value"}) is False
def test_empty_attributes(self):
"""Empty attributes means no labels."""
assert _has_webterm_label({}) is False
class TestHandleEventWithThemeLabel:
"""Tests for event handling with theme-only labels."""
@pytest.mark.asyncio
async def test_handle_start_event_theme_only(self):
"""Container with only theme label is picked up."""
manager = MagicMock()
manager.apps_by_slug = {}
manager.apps = []
watcher = DockerWatcher(manager)
async def mock_request(method, path):
if "/containers/" in path and "/json" in path:
import json
return 200, json.dumps({
"Id": "abc123",
"Name": "/themed-container",
"Config": {"Labels": {THEME_LABEL: "monokai"}},
})
return 404, ""
watcher._docker_request = mock_request
event = {
"Action": "start",
"Actor": {
"ID": "abc123",
"Attributes": {THEME_LABEL: "monokai"}, # Only theme label
},
}
await watcher._handle_event(event)
# Should add container with auto command
manager.add_app.assert_called_once()
call_args = manager.add_app.call_args
assert call_args[0][1] == AUTO_COMMAND_SENTINEL # command arg
assert call_args[1]["theme"] == "monokai"