mirrors / light-heart-labs--ods

项目文件夹

light-heart-labs--ods
文件
main
T
wehub-resource-sync 491752a783 chore: import upstream snapshot with attribution
2026-07-13 12:34:03 +08:00

3499 行
141 KiB
Python
原始文件 永久链接 Blame 文件历史

"""Tests for extensions portal endpoints."""
import contextlib
import json
import os
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
import yaml
from fastapi import HTTPException
from models import ServiceStatus
from routers.extensions import _assert_not_core
# --- Helpers ---
def _make_catalog_ext(ext_id, name="Test", category="optional",
gpu_backends=None, env_vars=None, features=None):
return {
"id": ext_id,
"name": name,
"description": f"Description for {name}",
"category": category,
"gpu_backends": gpu_backends or ["nvidia", "amd", "apple"],
"compose_file": "compose.yaml",
"depends_on": [],
"port": 8080,
"external_port_default": 8080,
"health_endpoint": "/health",
"env_vars": env_vars or [],
"tags": [],
"features": features or [],
}
def _make_service_status(sid, status="healthy"):
return ServiceStatus(
id=sid, name=sid, port=8080, external_port=8080, status=status,
)
def can_create_symlinks(tmp_path: Path) -> bool:
target = tmp_path / "symlink-target"
link = tmp_path / "symlink-probe"
target.write_text("probe", encoding="utf-8")
try:
link.symlink_to(target)
except (OSError, NotImplementedError):
return False
return link.is_symlink()
def _patch_extensions_config(monkeypatch, catalog, services=None,
gpu_backend="nvidia", tmp_path=None):
"""Apply standard patches for extensions router tests."""
monkeypatch.setattr("routers.extensions.EXTENSION_CATALOG", catalog)
monkeypatch.setattr("routers.extensions.SERVICES", services or {})
monkeypatch.setattr("routers.extensions.GPU_BACKEND", gpu_backend)
lib_dir = (tmp_path / "lib") if tmp_path else Path("/tmp/nonexistent-lib")
user_dir = (tmp_path / "user") if tmp_path else Path("/tmp/nonexistent-user")
monkeypatch.setattr("routers.extensions.EXTENSIONS_LIBRARY_DIR", lib_dir)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
monkeypatch.setattr("routers.extensions.DATA_DIR",
str(tmp_path or "/tmp/nonexistent"))
# --- Catalog endpoint ---
class TestExtensionsCatalog:
def test_catalog_returns_enriched_extensions(self, test_client, monkeypatch, tmp_path):
"""Catalog endpoint returns extensions with status enrichment."""
catalog = [_make_catalog_ext("test-svc", "Test Service")]
services = {"test-svc": {"host": "localhost", "port": 8080, "name": "Test"}}
_patch_extensions_config(monkeypatch, catalog, services, tmp_path=tmp_path)
mock_svc = _make_service_status("test-svc", "healthy")
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[mock_svc]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert len(data["extensions"]) == 1
assert data["extensions"][0]["status"] == "enabled"
assert data["extensions"][0]["installable"] is False
assert "summary" in data
assert data["gpu_backend"] == "nvidia"
def test_catalog_category_filter(self, test_client, monkeypatch, tmp_path):
"""Category filter returns only matching extensions."""
catalog = [
_make_catalog_ext("svc-a", "A", category="ai"),
_make_catalog_ext("svc-b", "B", category="tools"),
]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog?category=ai",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert len(data["extensions"]) == 1
assert data["extensions"][0]["id"] == "svc-a"
def test_catalog_gpu_compatible_filter(self, test_client, monkeypatch, tmp_path):
"""gpu_compatible filter excludes incompatible extensions."""
catalog = [
_make_catalog_ext("compat", "Compatible", gpu_backends=["nvidia"]),
_make_catalog_ext("incompat", "Incompatible", gpu_backends=["amd"]),
]
_patch_extensions_config(monkeypatch, catalog, gpu_backend="nvidia",
tmp_path=tmp_path)
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog?gpu_compatible=true",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
ids = [e["id"] for e in data["extensions"]]
assert "compat" in ids
assert "incompat" not in ids
def test_catalog_summary_counts(self, test_client, monkeypatch, tmp_path):
"""Summary counts correctly reflect extension statuses."""
catalog = [
_make_catalog_ext("enabled-svc", "Enabled"),
_make_catalog_ext("disabled-svc", "Disabled"),
_make_catalog_ext("not-installed", "Not Installed"),
_make_catalog_ext("incompat", "Incompatible", gpu_backends=["amd"]),
]
services = {
"enabled-svc": {"host": "localhost", "port": 8080, "name": "Enabled"},
"disabled-svc": {"host": "localhost", "port": 8081, "name": "Disabled"},
}
_patch_extensions_config(monkeypatch, catalog, services,
gpu_backend="nvidia", tmp_path=tmp_path)
mock_svcs = [
_make_service_status("enabled-svc", "healthy"),
_make_service_status("disabled-svc", "down"),
]
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=mock_svcs):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
summary = resp.json()["summary"]
assert summary["total"] == 4
assert summary["enabled"] == 1
assert summary["disabled"] == 1
assert summary["not_installed"] == 1
assert summary["incompatible"] == 1
assert summary["installed"] == 2
def test_catalog_empty_when_no_catalog(self, test_client, monkeypatch, tmp_path):
"""Missing catalog file results in empty extensions list."""
_patch_extensions_config(monkeypatch, [], tmp_path=tmp_path)
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["extensions"] == []
assert data["summary"]["total"] == 0
def test_catalog_requires_auth(self, test_client):
"""GET /api/extensions/catalog without auth → 401."""
resp = test_client.get("/api/extensions/catalog")
assert resp.status_code == 401
# --- Detail endpoint ---
class TestExtensionDetail:
def test_detail_returns_extension(self, test_client, monkeypatch, tmp_path):
"""Detail endpoint returns correct extension with setup instructions."""
catalog = [_make_catalog_ext("test-svc", "Test Service")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/test-svc",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["id"] == "test-svc"
assert data["name"] == "Test Service"
assert data["status"] == "not_installed"
assert "manifest" in data
assert "setup_instructions" in data
assert data["setup_instructions"]["cli_enable"] == "ods enable test-svc"
assert data["setup_instructions"]["cli_disable"] == "ods disable test-svc"
def test_detail_404_for_unknown(self, test_client, monkeypatch, tmp_path):
"""404 for service_id not in catalog."""
_patch_extensions_config(monkeypatch, [], tmp_path=tmp_path)
resp = test_client.get(
"/api/extensions/nonexistent",
headers=test_client.auth_headers,
)
assert resp.status_code == 404
def test_detail_rejects_path_traversal(self, test_client, monkeypatch, tmp_path):
"""Regex validation rejects path traversal and invalid service IDs."""
_patch_extensions_config(monkeypatch, [], tmp_path=tmp_path)
for bad_id in ["..etc", ".hidden", "UPPERCASE", "-starts-dash"]:
resp = test_client.get(
f"/api/extensions/{bad_id}",
headers=test_client.auth_headers,
)
assert resp.status_code == 404, f"Expected 404 for: {bad_id}"
def test_detail_path_traversal_with_slashes(self, test_client):
"""Path traversal with slashes never reaches the handler."""
# Starlette normalizes ../etc/passwd out of the route
resp = test_client.get(
"/api/extensions/../etc/passwd",
headers=test_client.auth_headers,
)
assert resp.status_code == 404
resp = test_client.get(
"/api/extensions/../../",
headers=test_client.auth_headers,
)
assert resp.status_code in (404, 307)
def test_detail_requires_auth(self, test_client):
"""GET /api/extensions/{id} without auth → 401."""
resp = test_client.get("/api/extensions/test-svc")
assert resp.status_code == 401
# --- User-installed extension status ---
class TestUserExtensionStatus:
def test_user_ext_compose_yaml_healthy(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml + healthy service → enabled."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text("version: '3'")
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
mock_svc = _make_service_status("my-ext", "healthy")
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[mock_svc]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["id"] == "my-ext"
assert ext["status"] == "enabled"
def test_user_ext_compose_yaml_no_service(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml but no running container → stopped."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text("version: '3'")
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
# No service in health results — svc is None → stopped
with patch("user_extensions.get_user_services_cached",
return_value={}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["id"] == "my-ext"
assert ext["status"] == "stopped"
def test_user_ext_compose_yaml_disabled(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml.disabled → disabled."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml.disabled").write_text("version: '3'")
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["id"] == "my-ext"
assert ext["status"] == "disabled"
# --- Mutation test helpers ---
_SAFE_COMPOSE = "services:\n svc:\n image: test:latest\n"
def _setup_library_ext(tmp_path, service_id, compose_content=None):
"""Create a library extension directory with compose.yaml and manifest."""
lib_dir = tmp_path / "lib"
lib_dir.mkdir(exist_ok=True)
ext_dir = lib_dir / service_id
ext_dir.mkdir(exist_ok=True)
(ext_dir / "compose.yaml").write_text(compose_content or _SAFE_COMPOSE)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": service_id, "name": service_id},
}))
return lib_dir
def _setup_user_ext(tmp_path, service_id, enabled=True, manifest=None):
"""Create a user-installed extension directory."""
user_dir = tmp_path / "user"
user_dir.mkdir(exist_ok=True)
ext_dir = user_dir / service_id
ext_dir.mkdir(exist_ok=True)
if enabled:
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
else:
(ext_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
if manifest:
(ext_dir / "manifest.yaml").write_text(yaml.dump(manifest))
return user_dir
def _patch_mutation_config(monkeypatch, tmp_path, lib_dir=None, user_dir=None):
"""Patch config values for mutation endpoint tests."""
lib_dir = lib_dir or (tmp_path / "lib")
user_dir = user_dir or (tmp_path / "user")
lib_dir.mkdir(exist_ok=True)
user_dir.mkdir(exist_ok=True)
monkeypatch.setattr("routers.extensions.EXTENSIONS_LIBRARY_DIR", lib_dir)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.EXTENSIONS_DIR",
tmp_path / "builtin")
monkeypatch.setattr("routers.extensions.CORE_SERVICE_IDS",
frozenset({"dashboard-api", "open-webui", "hermes", "hermes-proxy"}))
# --- Install endpoint ---
class TestInstallExtension:
def test_install_copies_and_enables(self, test_client, monkeypatch, tmp_path):
"""Install copies from library and keeps compose.yaml enabled."""
lib_dir = _setup_library_ext(tmp_path, "my-ext")
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["id"] == "my-ext"
assert data["action"] == "installed"
assert "restart_required" in data
user_dir = tmp_path / "user"
assert (user_dir / "my-ext").is_dir()
assert (user_dir / "my-ext" / "compose.yaml").exists()
def test_install_cleans_broken_directory(self, test_client, monkeypatch, tmp_path):
"""Install succeeds when dest dir exists but has no compose files (broken state)."""
lib_dir = _setup_library_ext(tmp_path, "my-ext")
# Create a broken user extension directory (no compose.yaml or compose.yaml.disabled)
user_dir = tmp_path / "user"
user_dir.mkdir(exist_ok=True)
broken_dir = user_dir / "my-ext"
broken_dir.mkdir(exist_ok=True)
(broken_dir / "manifest.yaml").write_text("leftover: true\n")
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir,
user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "installed"
assert (user_dir / "my-ext" / "compose.yaml").exists()
def test_install_stages_tmp_under_user_extensions_dir(
self, test_client, monkeypatch, tmp_path,
):
"""Library installs must not require write access to the /data mount root."""
import tempfile
lib_dir = _setup_library_ext(tmp_path, "my-ext")
user_dir = tmp_path / "user"
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir,
user_dir=user_dir)
captured = {}
real_mkdtemp = tempfile.mkdtemp
def fake_mkdtemp(*args, **kwargs):
captured["dir"] = Path(kwargs["dir"])
return real_mkdtemp(*args, **kwargs)
monkeypatch.setattr("routers.extensions.tempfile.mkdtemp", fake_mkdtemp)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert captured["dir"] == user_dir / ".tmp"
assert (user_dir / "my-ext" / "compose.yaml").exists()
def test_install_already_installed_409(self, test_client, monkeypatch, tmp_path):
"""409 when extension is already installed."""
lib_dir = _setup_library_ext(tmp_path, "my-ext")
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir,
user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 409
def test_install_retries_after_error_progress(self, test_client, monkeypatch, tmp_path):
"""A terminal install error should allow retrying the library install."""
lib_dir = _setup_library_ext(tmp_path, "my-ext")
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
(progress_dir / "my-ext.json").write_text(json.dumps({
"service_id": "my-ext",
"status": "error",
"error": "previous compose resolve failed",
"started_at": "2026-01-01T00:00:00+00:00",
"updated_at": "2026-01-01T00:00:00+00:00",
}))
(user_dir / "my-ext" / "stale.txt").write_text("left over")
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir,
user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert not (user_dir / "my-ext" / "stale.txt").exists()
assert (user_dir / "my-ext" / "compose.yaml").exists()
assert resp.json()["action"] == "installed"
def test_install_unknown_extension_404(self, test_client, monkeypatch, tmp_path):
"""404 when extension is not in the library."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.post(
"/api/extensions/nonexistent/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 404
def test_install_rejects_library_entry_without_compose(
self, test_client, monkeypatch, tmp_path,
):
"""400 when the library entry exists but ships no deployable compose.yaml.
Mirrors the dify/jan/fooocus shape: directory present, manifest
present, but only `compose.yaml.disabled` or `compose.yaml.reference`
on disk. The catalog/UI already hides the Install button for these via
`_is_installable`, but a direct POST must also reject — otherwise the
copytree succeeds but the host agent can't start anything, surfacing
as a cryptic post-install failure instead of a clean 400.
"""
lib_dir = tmp_path / "lib"
lib_dir.mkdir(exist_ok=True)
ext_dir = lib_dir / "reference-only"
ext_dir.mkdir(exist_ok=True)
# Only .disabled — no deployable compose.yaml
(ext_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "reference-only", "name": "reference-only"},
}))
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/reference-only/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
body = resp.json()
assert "compose.yaml" in body["detail"]
# Verify nothing was copied to user-extensions/
assert not (tmp_path / "user" / "reference-only").exists()
def test_install_rejects_library_entry_with_only_reference_compose(
self, test_client, monkeypatch, tmp_path,
):
"""Same shape, .reference suffix variant (mirrors fooocus).
Some library entries ship `compose.yaml.reference` instead of
`.disabled`. Either suffix is reference material — only literal
`compose.yaml` is deployable.
"""
lib_dir = tmp_path / "lib"
lib_dir.mkdir(exist_ok=True)
ext_dir = lib_dir / "reference-only"
ext_dir.mkdir(exist_ok=True)
(ext_dir / "compose.yaml.reference").write_text(_SAFE_COMPOSE)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "reference-only", "name": "reference-only"},
}))
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/reference-only/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert not (tmp_path / "user" / "reference-only").exists()
def test_install_core_service_403(self, test_client, monkeypatch, tmp_path):
"""403 when trying to install a core service."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.post(
"/api/extensions/dashboard-api/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 403
def test_install_rejects_privileged(self, test_client, monkeypatch, tmp_path):
"""400 when compose uses privileged mode."""
bad_compose = "services:\n svc:\n image: test\n privileged: true\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext",
compose_content=bad_compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "privileged" in resp.json()["detail"]
def test_install_rejects_docker_socket(self, test_client, monkeypatch, tmp_path):
"""400 when compose mounts Docker socket."""
bad_compose = (
"services:\n svc:\n image: test\n"
" volumes:\n - /var/run/docker.sock:/var/run/docker.sock\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext",
compose_content=bad_compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "Docker socket mount" in resp.json()["detail"]
def test_install_allows_library_build_context(self, test_client, monkeypatch, tmp_path):
"""Library extensions with build: context are allowed (trusted)."""
bad_compose = "services:\n svc:\n build: .\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext",
compose_content=bad_compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert resp.json()["action"] == "installed"
def test_install_requires_auth(self, test_client):
"""POST install without auth → 401."""
resp = test_client.post("/api/extensions/my-ext/install")
assert resp.status_code == 401
# test_install_writes_pending_change removed — v3 uses host agent, no pending changes file
def test_install_allows_library_host_gateway_extra_host(
self, test_client, monkeypatch, tmp_path,
):
"""Bundled library extensions may use the host-gateway bridge."""
compose = (
"services:\n"
" svc:\n"
" image: test:latest\n"
" extra_hosts:\n"
" - host.docker.internal:host-gateway\n"
)
lib_dir = _setup_library_ext(tmp_path, "host-gateway-ext",
compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/host-gateway-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert resp.json()["action"] == "installed"
def test_install_rejects_untrusted_extra_hosts(
self, test_client, monkeypatch, tmp_path,
):
"""User-installed extension compose files cannot add host aliases."""
compose = (
"services:\n"
" svc:\n"
" image: test:latest\n"
" extra_hosts:\n"
" - host.docker.internal:host-gateway\n"
)
user_dir = tmp_path / "user"
ext_dir = user_dir / "host-gateway-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml.disabled").write_text(compose)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "host-gateway-ext", "name": "host-gateway-ext"},
}))
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/host-gateway-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "extra_hosts" in resp.json()["detail"]
def test_install_rejects_unapproved_library_extra_hosts(
self, test_client, monkeypatch, tmp_path,
):
"""Trusted library status only permits the known host-gateway mapping."""
compose = (
"services:\n"
" svc:\n"
" image: test:latest\n"
" extra_hosts:\n"
" - metadata.google.internal:169.254.169.254\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-host-ext",
compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-host-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "unsupported extra_hosts" in resp.json()["detail"]
# --- Enable endpoint ---
class TestEnableExtension:
def test_enable_renames_to_compose_yaml(self, test_client, monkeypatch, tmp_path):
"""Enable renames compose.yaml.disabled → compose.yaml."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
assert data["restart_required"] is True
assert (user_dir / "my-ext" / "compose.yaml").exists()
assert not (user_dir / "my-ext" / "compose.yaml.disabled").exists()
def test_enable_stopped_starts_without_rename(self, test_client, monkeypatch, tmp_path):
"""Enable when compose.yaml exists (stopped) → starts without rename."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
# compose.yaml still exists (no rename happened)
assert (user_dir / "my-ext" / "compose.yaml").exists()
def test_enable_allows_core_service_dependency(self, test_client, monkeypatch, tmp_path):
"""Enable succeeds when depends_on includes a core service."""
manifest = {"service": {"depends_on": ["open-webui"]}}
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False,
manifest=manifest)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
def test_enable_missing_dependency_400(self, test_client, monkeypatch, tmp_path):
"""400 when a dependency is not enabled."""
manifest = {"service": {"depends_on": ["missing-dep"]}}
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False,
manifest=manifest)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
detail = resp.json()["detail"]
assert "missing-dep" in detail["missing_dependencies"]
assert detail["auto_enable_available"] is True
def test_enable_core_service_403(self, test_client, monkeypatch, tmp_path):
"""403 when trying to enable a core service."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.post(
"/api/extensions/open-webui/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 403
def test_enable_requires_auth(self, test_client):
"""POST enable without auth → 401."""
resp = test_client.post("/api/extensions/my-ext/enable")
assert resp.status_code == 401
def test_enable_rejects_build_context(self, test_client, monkeypatch, tmp_path):
"""400 when user extension compose contains a build context."""
bad_compose = "services:\n svc:\n build: .\n"
user_dir = tmp_path / "user"
user_dir.mkdir(exist_ok=True)
ext_dir = user_dir / "bad-ext"
ext_dir.mkdir(exist_ok=True)
(ext_dir / "compose.yaml.disabled").write_text(bad_compose)
(ext_dir / "manifest.yaml").write_text("schema_version: ods.services.v1\nservice:\n id: bad-ext\n name: bad-ext\n")
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/bad-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "local build" in resp.json()["detail"]
class TestEnableExtensionHookReturnHandling:
"""pre_start failures must block start; post_start failures surface as warnings."""
def test_enable_pre_start_failure_blocks_start(
self, test_client, monkeypatch, tmp_path,
):
"""pre_start False → start NOT called, error progress written, agent_ok=False."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
start_calls = []
def fake_start(action, sid):
start_calls.append((action, sid))
return True
monkeypatch.setattr("routers.extensions._call_agent", fake_start)
# pre_start returns False, post_start would return True (but should not be reached)
monkeypatch.setattr(
"routers.extensions._call_agent_hook",
lambda sid, hook: hook != "pre_start",
)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
assert data["restart_required"] is True
assert "Run 'ods restart'" in data["message"]
assert data["warnings"] == []
# start was never called for the service whose pre_start failed
assert ("start", "my-ext") not in start_calls
# Error progress file should record the pre_start failure
progress_file = tmp_path / "extension-progress" / "my-ext.json"
assert progress_file.exists()
progress = json.loads(progress_file.read_text())
assert progress["status"] == "error"
assert "pre_start hook failed" in progress["error"]
def test_enable_post_start_failure_returns_warning(
self, test_client, monkeypatch, tmp_path,
):
"""post_start False → start IS called, response carries a warning, success path."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
start_calls = []
def fake_start(action, sid):
start_calls.append((action, sid))
return True
monkeypatch.setattr("routers.extensions._call_agent", fake_start)
# pre_start succeeds, post_start fails
monkeypatch.setattr(
"routers.extensions._call_agent_hook",
lambda sid, hook: hook != "post_start",
)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
# post_start failure is non-fatal — service still reports success
assert data["restart_required"] is False
assert data["message"] == "Extension enabled and started."
assert ("start", "my-ext") in start_calls
assert isinstance(data["warnings"], list)
assert len(data["warnings"]) == 1
assert "my-ext" in data["warnings"][0]
assert "post_start hook failed" in data["warnings"][0]
def test_enable_both_hooks_succeed_no_warnings(
self, test_client, monkeypatch, tmp_path,
):
"""Baseline: pre_start + post_start True → no warnings, agent_ok True."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
monkeypatch.setattr(
"routers.extensions._call_agent", lambda action, sid: True,
)
monkeypatch.setattr(
"routers.extensions._call_agent_hook", lambda sid, hook: True,
)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
assert data["restart_required"] is False
assert data["warnings"] == []
assert data["message"] == "Extension enabled and started."
def test_multi_svc_pre_start_failure_on_dep_does_not_warn(
self, test_client, monkeypatch, tmp_path,
):
"""Multi-svc mixed-outcome: dep pre_start fails, main proceeds.
Covers fork issue #494's "pre_start failures must be terminal for
the failing service but must not pollute the warnings array" path.
Dep's pre_start failure writes its own error progress and is
treated as terminal (no start, no post_start) — the warnings
accumulator only collects post_start failures. Main service's
hooks all succeed but agent_ok stays False because one of the
services in enabled_services failed pre_start, so
restart_required is True.
"""
user_dir = tmp_path / "user"
user_dir.mkdir(exist_ok=True)
# Dep ext (no further deps).
dep_dir = user_dir / "dep"
dep_dir.mkdir()
(dep_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
(dep_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "dep", "name": "dep"},
}))
# Main ext, depends_on dep.
main_dir = user_dir / "main-ext"
main_dir.mkdir()
(main_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
(main_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "main-ext", "name": "main-ext",
"depends_on": ["dep"]},
}))
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
monkeypatch.setattr(
"routers.extensions._call_agent", lambda action, sid: True,
)
# pre_start fails for dep; everything else succeeds.
monkeypatch.setattr(
"routers.extensions._call_agent_hook",
lambda sid, hook: not (sid == "dep" and hook == "pre_start"),
)
monkeypatch.setattr(
"routers.extensions._call_agent_invalidate_compose_cache",
lambda: None,
)
resp = test_client.post(
"/api/extensions/main-ext/enable?auto_enable_deps=true",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
# pre_start failure on dep keeps agent_ok False → restart required.
assert data["restart_required"] is True
# warnings collects only post_start failures, so dep's pre_start
# failure must NOT appear here.
assert data["warnings"] == []
# Both services were activated (dep auto-enabled, then main).
assert "dep" in data["enabled_services"]
assert "main-ext" in data["enabled_services"]
# Dep got an error-progress file recording the pre_start failure.
dep_progress = tmp_path / "extension-progress" / "dep.json"
assert dep_progress.exists()
progress = json.loads(dep_progress.read_text())
assert progress["status"] == "error"
assert "pre_start hook failed" in progress["error"]
def test_multi_svc_post_start_failure_on_main_only_warns_main(
self, test_client, monkeypatch, tmp_path,
):
"""Multi-svc mixed-outcome: dep all-pass, main post_start fails.
Covers fork issue #494's "warnings must name the failing service
and not double-count clean dependencies" path. Dep enables clean
with no warning entry; main's post_start failure produces exactly
one warning string identifying main-ext. Post_start is non-fatal,
so agent_ok stays True and restart_required is False.
"""
user_dir = tmp_path / "user"
user_dir.mkdir(exist_ok=True)
dep_dir = user_dir / "dep"
dep_dir.mkdir()
(dep_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
(dep_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "dep", "name": "dep"},
}))
main_dir = user_dir / "main-ext"
main_dir.mkdir()
(main_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
(main_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "main-ext", "name": "main-ext",
"depends_on": ["dep"]},
}))
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
monkeypatch.setattr(
"routers.extensions._call_agent", lambda action, sid: True,
)
# post_start fails ONLY for main-ext.
monkeypatch.setattr(
"routers.extensions._call_agent_hook",
lambda sid, hook: not (sid == "main-ext" and hook == "post_start"),
)
monkeypatch.setattr(
"routers.extensions._call_agent_invalidate_compose_cache",
lambda: None,
)
resp = test_client.post(
"/api/extensions/main-ext/enable?auto_enable_deps=true",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
# post_start is non-fatal → agent_ok stays True.
assert data["restart_required"] is False
# Exactly one warning, naming the main service only.
assert isinstance(data["warnings"], list)
assert len(data["warnings"]) == 1
assert "main-ext" in data["warnings"][0]
assert "post_start hook failed" in data["warnings"][0]
# Dep must NOT show up in warnings (it cleanly enabled).
assert all("dep" not in w.split(":")[0] for w in data["warnings"])
assert "dep" in data["enabled_services"]
assert "main-ext" in data["enabled_services"]
# --- Disable endpoint ---
class TestDisableExtension:
def test_disable_renames_to_disabled(self, test_client, monkeypatch, tmp_path):
"""Disable renames compose.yaml → compose.yaml.disabled."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "disabled"
assert data["restart_required"] is True
assert (user_dir / "my-ext" / "compose.yaml.disabled").exists()
assert not (user_dir / "my-ext" / "compose.yaml").exists()
def test_disable_builtin_delegates_to_host_agent(
self, test_client, monkeypatch, tmp_path,
):
builtin_root = tmp_path / "builtin"
ext_dir = builtin_root / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
_patch_mutation_config(monkeypatch, tmp_path)
monkeypatch.setattr("routers.extensions.EXTENSIONS_DIR", builtin_root)
monkeypatch.setattr("routers.extensions._call_agent", lambda action, sid: True)
calls = []
def _mock_compose_rename(action, service_id):
calls.append((action, service_id))
(ext_dir / "compose.yaml").rename(ext_dir / "compose.yaml.disabled")
return True
monkeypatch.setattr(
"routers.extensions._call_agent_compose_rename",
_mock_compose_rename,
)
resp = test_client.post(
"/api/extensions/my-ext/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert resp.json()["action"] == "disabled"
assert calls == [("deactivate", "my-ext")]
assert (ext_dir / "compose.yaml.disabled").exists()
assert not (ext_dir / "compose.yaml").exists()
def test_disable_unlinks_progress_file(self, test_client, monkeypatch, tmp_path):
"""Disable removes the stale progress file so status reflects reality."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
progress_file = tmp_path / "extension-progress" / "my-ext.json"
progress_file.parent.mkdir(parents=True, exist_ok=True)
progress_file.write_text('{"status": "started", "updated_at": "2026-04-10T00:00:00"}')
resp = test_client.post(
"/api/extensions/my-ext/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert not progress_file.exists()
def test_disable_already_disabled_409(self, test_client, monkeypatch, tmp_path):
"""409 when extension is already disabled."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 409
def test_disable_core_service_403(self, test_client, monkeypatch, tmp_path):
"""403 when trying to disable a core service."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.post(
"/api/extensions/dashboard-api/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 403
def test_disable_warns_about_dependents(self, test_client, monkeypatch, tmp_path):
"""Disable warns about extensions that depend on this one."""
user_dir = tmp_path / "user"
user_dir.mkdir()
# Extension to disable
ext_dir = user_dir / "my-ext"
ext_dir.mkdir()
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
# Dependent extension
dep_dir = user_dir / "dependent-ext"
dep_dir.mkdir()
(dep_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
(dep_dir / "manifest.yaml").write_text(
yaml.dump({"service": {"depends_on": ["my-ext"]}}),
)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert "dependent-ext" in data["dependents_warning"]
def test_disable_requires_auth(self, test_client):
"""POST disable without auth → 401."""
resp = test_client.post("/api/extensions/my-ext/disable")
assert resp.status_code == 401
def test_disable_skips_data_info(self, test_client, monkeypatch, tmp_path):
"""include_data_info=false → data_info is None (skips expensive dir scan)."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/disable?include_data_info=false",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert resp.json()["data_info"] is None
# --- Uninstall endpoint ---
class TestUninstallExtension:
def test_uninstall_removes_dir(self, test_client, monkeypatch, tmp_path):
"""Uninstall removes the extension directory."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.delete(
"/api/extensions/my-ext",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "uninstalled"
assert not (user_dir / "my-ext").exists()
def test_uninstall_unlinks_progress_file(self, test_client, monkeypatch, tmp_path):
"""Uninstall removes the stale progress file so status reflects reality."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
progress_file = tmp_path / "extension-progress" / "my-ext.json"
progress_file.parent.mkdir(parents=True, exist_ok=True)
progress_file.write_text('{"status": "started", "updated_at": "2026-04-10T00:00:00"}')
resp = test_client.delete(
"/api/extensions/my-ext",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert not progress_file.exists()
def test_uninstall_rejects_enabled_400(self, test_client, monkeypatch, tmp_path):
"""400 when extension is still enabled."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.delete(
"/api/extensions/my-ext",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "Disable extension before uninstalling" in resp.json()["detail"]
def test_uninstall_core_service_403(self, test_client, monkeypatch, tmp_path):
"""403 when trying to uninstall a core service."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.delete(
"/api/extensions/open-webui",
headers=test_client.auth_headers,
)
assert resp.status_code == 403
def test_uninstall_requires_auth(self, test_client):
"""DELETE without auth → 401."""
resp = test_client.delete("/api/extensions/my-ext")
assert resp.status_code == 401
# --- Compose-flags cache invalidation ---
class TestComposeCacheInvalidation:
"""Every successful compose mutation must invalidate the host .compose-flags cache."""
def _spy(self, monkeypatch):
calls = []
monkeypatch.setattr(
"routers.extensions._call_agent_invalidate_compose_cache",
lambda: calls.append(1),
)
return calls
def test_install_invalidates_cache(self, test_client, monkeypatch, tmp_path):
lib_dir = _setup_library_ext(tmp_path, "my-ext")
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
calls = self._spy(monkeypatch)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert len(calls) == 1
def test_enable_invalidates_cache(self, test_client, monkeypatch, tmp_path):
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
calls = self._spy(monkeypatch)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert len(calls) == 1
def test_enable_stopped_invalidates_cache(self, test_client, monkeypatch, tmp_path):
"""Stopped-start branch: compose.yaml already exists (library extension
enabled flow). Cache must be invalidated BEFORE the host agent start
call so it sees the new compose set."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
order: list[str] = []
monkeypatch.setattr(
"routers.extensions._call_agent_invalidate_compose_cache",
lambda: order.append("invalidate"),
)
monkeypatch.setattr(
"routers.extensions._call_agent",
lambda action, svc: order.append(f"agent:{action}:{svc}") or True,
)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert order.count("invalidate") == 1
assert order.index("invalidate") < order.index("agent:start:my-ext")
def test_disable_invalidates_cache(self, test_client, monkeypatch, tmp_path):
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
calls = self._spy(monkeypatch)
resp = test_client.post(
"/api/extensions/my-ext/disable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert len(calls) == 1
def test_uninstall_invalidates_cache(self, test_client, monkeypatch, tmp_path):
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=False)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
calls = self._spy(monkeypatch)
resp = test_client.delete(
"/api/extensions/my-ext",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert len(calls) == 1
# --- Path traversal on mutation endpoints ---
class TestMutationPathTraversal:
def test_path_traversal_all_mutations(self, test_client, monkeypatch, tmp_path):
"""Path traversal IDs are rejected on all mutation endpoints."""
_patch_mutation_config(monkeypatch, tmp_path)
bad_ids = ["..etc", ".hidden", "UPPERCASE", "-starts-dash"]
endpoints = [
("POST", "/api/extensions/{}/install"),
("POST", "/api/extensions/{}/enable"),
("POST", "/api/extensions/{}/disable"),
("DELETE", "/api/extensions/{}"),
("DELETE", "/api/extensions/{}/data"),
]
for bad_id in bad_ids:
for method, pattern in endpoints:
url = pattern.format(bad_id)
if method == "POST":
resp = test_client.post(
url, headers=test_client.auth_headers,
)
elif "/data" in pattern:
# Purge endpoint requires a JSON body (PurgeRequest)
resp = test_client.request(
"DELETE", url, headers=test_client.auth_headers,
json={"confirm": False},
)
else:
resp = test_client.delete(
url, headers=test_client.auth_headers,
)
assert resp.status_code == 404, (
f"Expected 404 for {method} {url}, got {resp.status_code}"
)
# --- Compose security scan edge cases ---
class TestComposeScanEdgeCases:
def test_scan_rejects_cap_add_sys_admin(self, test_client, monkeypatch, tmp_path):
"""400 when compose adds SYS_ADMIN capability."""
compose = "services:\n svc:\n image: test\n cap_add:\n - SYS_ADMIN\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "SYS_ADMIN" in resp.json()["detail"]
def test_scan_rejects_pid_host(self, test_client, monkeypatch, tmp_path):
"""400 when compose uses pid: host."""
compose = "services:\n svc:\n image: test\n pid: host\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "host PID" in resp.json()["detail"]
def test_scan_rejects_network_mode_host(self, test_client, monkeypatch, tmp_path):
"""400 when compose uses network_mode: host."""
compose = "services:\n svc:\n image: test\n network_mode: host\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "host network" in resp.json()["detail"]
def test_scan_rejects_user_root(self, test_client, monkeypatch, tmp_path):
"""400 when compose runs as user: root."""
compose = "services:\n svc:\n image: test\n user: root\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "root" in resp.json()["detail"]
def test_scan_rejects_user_0_colon_0(self, test_client, monkeypatch, tmp_path):
"""400 when compose runs as user: '0:0' (root bypass variant)."""
compose = 'services:\n svc:\n image: test\n user: "0:0"\n'
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "root" in resp.json()["detail"]
def test_scan_rejects_absolute_host_path_mount(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose mounts an absolute host path."""
compose = (
"services:\n svc:\n image: test\n"
" volumes:\n - /etc/passwd:/etc/passwd:ro\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "absolute host path" in resp.json()["detail"]
def test_scan_rejects_run_docker_sock(self, test_client, monkeypatch, tmp_path):
"""400 when compose mounts /run/docker.sock (variant path)."""
compose = (
"services:\n svc:\n image: test\n"
" volumes:\n - /run/docker.sock:/var/run/docker.sock\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "Docker socket mount" in resp.json()["detail"]
def test_scan_rejects_bare_port_binding(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose uses bare host:container port binding."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - '8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "127.0.0.1" in resp.json()["detail"]
def test_scan_allows_localhost_port_binding(
self, test_client, monkeypatch, tmp_path,
):
"""Safe compose with 127.0.0.1 port binding passes scan."""
compose = (
"services:\n svc:\n image: test:latest\n"
" ports:\n - '127.0.0.1:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "safe-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/safe-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
def test_scan_rejects_0000_port_binding(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose binds to 0.0.0.0 explicitly."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - '0.0.0.0:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "127.0.0.1" in resp.json()["detail"]
def test_scan_allows_bind_address_var_with_loopback_default(
self, test_client, monkeypatch, tmp_path,
):
"""${BIND_ADDRESS:-127.0.0.1} is the sanctioned LAN-toggle pattern (PR #964)."""
compose = (
"services:\n svc:\n image: test:latest\n"
" ports:\n - '${BIND_ADDRESS:-127.0.0.1}:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "bind-ok", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bind-ok/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
def test_scan_allows_arbitrary_var_name_with_loopback_default(
self, test_client, monkeypatch, tmp_path,
):
"""Any ${VAR:-127.0.0.1} form is accepted, not just BIND_ADDRESS."""
compose = (
"services:\n svc:\n image: test:latest\n"
" ports:\n - '${MY_HOST:-127.0.0.1}:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "bind-var", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bind-var/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
def test_scan_rejects_var_with_non_loopback_default(
self, test_client, monkeypatch, tmp_path,
):
"""A variable defaulting to 0.0.0.0 must NOT be accepted."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - '${BIND_ADDRESS:-0.0.0.0}:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-default", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-default/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "127.0.0.1" in resp.json()["detail"]
def test_scan_rejects_var_without_default(
self, test_client, monkeypatch, tmp_path,
):
"""A bare ${VAR} (no default) is unsafe — it binds 0.0.0.0 when unset."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - '${BIND_ADDRESS}:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "no-default", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/no-default/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "127.0.0.1" in resp.json()["detail"]
def test_scan_allows_dict_port_with_bind_address_default(
self, test_client, monkeypatch, tmp_path,
):
"""Dict-form port with host_ip: ${VAR:-127.0.0.1} is also accepted."""
compose = (
"services:\n svc:\n image: test:latest\n"
" ports:\n - target: 80\n"
" published: 8080\n"
" host_ip: '${BIND_ADDRESS:-127.0.0.1}'\n"
)
lib_dir = _setup_library_ext(tmp_path, "dict-ok", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/dict-ok/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
def test_scan_rejects_core_service_name(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose service name collides with a core service."""
compose = "services:\n open-webui:\n image: test:latest\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "conflicts with core service" in resp.json()["detail"]
@pytest.mark.parametrize("service_name", ["hermes", "hermes-proxy"])
def test_scan_rejects_hermes_core_service_name(
self, test_client, monkeypatch, tmp_path, service_name,
):
"""Hermes built-ins must be protected from user extension shadowing."""
compose = f"services:\n {service_name}:\n image: test:latest\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "conflicts with core service" in resp.json()["detail"]
def test_scan_rejects_cap_add_sys_ptrace(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose adds SYS_PTRACE (expanded blocklist)."""
compose = "services:\n svc:\n image: test\n cap_add:\n - SYS_PTRACE\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "SYS_PTRACE" in resp.json()["detail"]
def test_scan_rejects_lowercase_cap(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose adds lowercase capability (case-insensitive check)."""
compose = "services:\n svc:\n image: test\n cap_add:\n - sys_admin\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "dangerous capability" in resp.json()["detail"]
def test_scan_rejects_ipc_host(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose uses ipc: host."""
compose = "services:\n svc:\n image: test\n ipc: host\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "host IPC" in resp.json()["detail"]
def test_scan_rejects_userns_mode_host(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose uses userns_mode: host."""
compose = "services:\n svc:\n image: test\n userns_mode: host\n"
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "host user namespace" in resp.json()["detail"]
def test_scan_rejects_named_volume_bind_mount(
self, test_client, monkeypatch, tmp_path,
):
"""400 when top-level volume uses driver_opts to bind-mount host path."""
compose = (
"services:\n svc:\n image: test:latest\n"
" volumes:\n - mydata:/data\n"
"volumes:\n mydata:\n driver_opts:\n"
" type: none\n o: bind\n device: /etc\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "bind-mount host path" in resp.json()["detail"]
def test_scan_rejects_dict_port_without_localhost(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose uses dict-form port binding without 127.0.0.1."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - target: 80\n published: 8080\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "127.0.0.1" in resp.json()["detail"]
def test_scan_rejects_bare_port_no_colon(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose uses bare port without colon (e.g. '8080')."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - '8080'\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "bare port" in resp.json()["detail"]
def test_scan_rejects_security_opt_equals_separator(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose uses security_opt with '=' separator."""
compose = (
"services:\n svc:\n image: test\n"
" security_opt:\n - seccomp=unconfined\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "dangerous security_opt" in resp.json()["detail"]
def test_scan_rejects_deploy_resources_devices(
self, test_client, monkeypatch, tmp_path,
):
"""400 when compose requests GPU passthrough via
deploy.resources.reservations.devices (Compose v2 GPU syntax).
Library installs default to skip_gpu_passthrough_check=False."""
compose = (
"services:\n svc:\n image: test\n"
" deploy:\n"
" resources:\n"
" reservations:\n"
" devices:\n"
" - driver: nvidia\n"
" count: 1\n"
" capabilities: [gpu]\n"
)
lib_dir = _setup_library_ext(tmp_path, "bad-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/bad-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "GPU passthrough" in resp.json()["detail"]
# --- Direct unit tests for the port-binding helpers ---
class TestHostPartIsLoopback:
"""Direct unit tests for `_host_part_is_loopback` — pin the regex
behaviour against future refactors. Triggered through the install
endpoint by TestComposeScanEdgeCases above; these tests are the
fast-feedback layer."""
def test_literal_loopback(self):
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("127.0.0.1") is True
def test_var_with_loopback_default(self):
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("${BIND_ADDRESS:-127.0.0.1}") is True
assert _host_part_is_loopback("${MY_HOST:-127.0.0.1}") is True
def test_rejects_var_without_default(self):
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("${BIND_ADDRESS}") is False
def test_rejects_non_loopback_default(self):
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("${BIND_ADDRESS:-0.0.0.0}") is False
assert _host_part_is_loopback("${BIND_ADDRESS:-localhost}") is False
def test_rejects_assignment_default_form(self):
"""Compose's ${VAR:=default} (assignment) is not the same as
${VAR:-default} (substitution); reject defensively."""
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("${BIND_ADDRESS:=127.0.0.1}") is False
def test_rejects_dash_only_default_form(self):
"""${VAR-default} (only-if-unset) differs from ${VAR:-default}
(only-if-unset-or-empty). Strictly require the colon form."""
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("${BIND_ADDRESS-127.0.0.1}") is False
def test_rejects_zero_padded_loopback(self):
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("127.000.000.001") is False
assert _host_part_is_loopback("${BIND_ADDRESS:-127.000.000.001}") is False
def test_rejects_ipv6_loopback(self):
"""IPv6 binds aren't in scope for the LAN toggle."""
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("::1") is False
assert _host_part_is_loopback("[::1]") is False
def test_rejects_trailing_newline(self):
"""fullmatch must defend against `$` matching before \\n."""
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("127.0.0.1\n") is False
assert _host_part_is_loopback("${BIND_ADDRESS:-127.0.0.1}\n") is False
def test_rejects_empty_and_whitespace(self):
from routers.extensions import _host_part_is_loopback
assert _host_part_is_loopback("") is False
assert _host_part_is_loopback(" 127.0.0.1") is False
assert _host_part_is_loopback("127.0.0.1 ") is False
class TestSplitPortHost:
"""Direct unit tests for `_split_port_host` — naive str.split(':') is
wrong on the `:-` default operator inside `${VAR:-127.0.0.1}`. These
tests pin the malformed-input behaviour as fail-closed."""
def test_literal_host_three_parts(self):
from routers.extensions import _split_port_host
assert _split_port_host("127.0.0.1:8080:80") == ("127.0.0.1", "8080:80")
def test_var_with_default(self):
from routers.extensions import _split_port_host
assert _split_port_host("${BIND_ADDRESS:-127.0.0.1}:8080:80") == (
"${BIND_ADDRESS:-127.0.0.1}", "8080:80",
)
def test_var_with_default_and_proto(self):
from routers.extensions import _split_port_host
assert _split_port_host("${BIND_ADDRESS:-127.0.0.1}:8554:8554/udp") == (
"${BIND_ADDRESS:-127.0.0.1}", "8554:8554/udp",
)
def test_var_no_default(self):
"""`${VAR}:8080:80` — no `:-` default, but still has the brace."""
from routers.extensions import _split_port_host
assert _split_port_host("${BIND_ADDRESS}:8080:80") == (
"${BIND_ADDRESS}", "8080:80",
)
def test_var_with_default_alone(self):
"""`${VAR:-127.0.0.1}` with NO host:container suffix — must
return rest='' so the caller's `':' not in core` check kicks in."""
from routers.extensions import _split_port_host
host, rest = _split_port_host("${BIND_ADDRESS:-127.0.0.1}")
assert rest == ""
def test_malformed_no_closing_brace(self):
"""`${VAR:-127.0.0.1` (missing `}`) — fail closed."""
from routers.extensions import _split_port_host
host, rest = _split_port_host("${BIND_ADDRESS:-127.0.0.1")
assert rest == ""
def test_malformed_no_separator_after_brace(self):
"""`${VAR:-127.0.0.1}8080:80` (no `:` between `}` and host port)."""
from routers.extensions import _split_port_host
host, rest = _split_port_host("${BIND_ADDRESS:-127.0.0.1}8080:80")
assert rest == ""
def test_two_part_with_digit_host_returns_no_host(self):
"""`8080:80` — host position is a port number, no host_ip; treat
as no-host so caller rejects (binds 0.0.0.0)."""
from routers.extensions import _split_port_host
assert _split_port_host("8080:80") == (None, "8080:80")
def test_bare_port_returns_no_host(self):
from routers.extensions import _split_port_host
assert _split_port_host("8080") == (None, "8080")
def test_empty_string(self):
from routers.extensions import _split_port_host
assert _split_port_host("") == (None, "")
class TestScanComposePortBindingRegressionLocks:
"""Regression locks for forms that must STAY rejected even though
they vaguely look like loopback bindings."""
def test_ipv6_loopback_bracketed_rejected(
self, test_client, monkeypatch, tmp_path,
):
"""`[::1]:8080:80` is not in the policy; must be rejected."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - '[::1]:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "ipv6-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/ipv6-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "127.0.0.1" in resp.json()["detail"]
def test_var_with_proto_suffix_accepted(
self, test_client, monkeypatch, tmp_path,
):
"""`${VAR:-127.0.0.1}:8554:8554/udp` is the sanctioned pattern
with an explicit /proto suffix (e.g. frigate's WebRTC port)."""
compose = (
"services:\n svc:\n image: test:latest\n"
" ports:\n - '${BIND_ADDRESS:-127.0.0.1}:8554:8554/udp'\n"
)
lib_dir = _setup_library_ext(tmp_path, "udp-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/udp-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
def test_hostname_in_host_position_rejected(
self, test_client, monkeypatch, tmp_path,
):
"""A hostname like `localhost` is not loopback under the regex —
the runtime resolution might not even resolve to 127.0.0.1
(IPv6 ::1, /etc/hosts override, etc.)."""
compose = (
"services:\n svc:\n image: test\n"
" ports:\n - 'localhost:8080:80'\n"
)
lib_dir = _setup_library_ext(tmp_path, "host-ext", compose_content=compose)
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/host-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
# --- skip_name_collision flag isolation ---
class TestScanComposeSkipNameCollision:
"""Direct unit tests for the skip_name_collision parameter added for
built-in activation (fork issue #338)."""
def test_rejects_core_name_by_default(self, tmp_path):
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text("services:\n open-webui:\n image: test\n")
with pytest.raises(HTTPException) as exc:
_scan_compose_content(compose, skip_name_collision=False)
assert exc.value.status_code == 400
assert "conflicts with core service" in exc.value.detail
def test_allows_core_name_when_skipped(self, tmp_path):
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text("services:\n open-webui:\n image: test\n")
_scan_compose_content(compose, skip_name_collision=True)
def test_privileged_still_blocked_when_skipped(self, tmp_path):
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text("services:\n svc:\n image: test\n privileged: true\n")
with pytest.raises(HTTPException) as exc:
_scan_compose_content(compose, skip_name_collision=True)
assert "privileged" in exc.value.detail
def test_docker_socket_still_blocked_when_skipped(self, tmp_path):
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text("services:\n svc:\n image: test\n volumes:\n - /var/run/docker.sock:/var/run/docker.sock\n")
with pytest.raises(HTTPException) as exc:
_scan_compose_content(compose, skip_name_collision=True)
assert "Docker socket" in exc.value.detail
# --- skip_gpu_passthrough_check flag isolation ---
class TestScanComposeSkipGpuPassthroughCheck:
"""Direct unit tests for the skip_gpu_passthrough_check parameter that
permits built-in extensions (e.g. comfyui's nvidia overlay) to declare
deploy.resources.reservations.devices while user extensions cannot."""
_GPU_COMPOSE = (
"services:\n svc:\n image: test\n"
" deploy:\n"
" resources:\n"
" reservations:\n"
" devices:\n"
" - driver: nvidia\n"
" count: 1\n"
" capabilities: [gpu]\n"
)
def test_rejects_deploy_devices_by_default(self, tmp_path):
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(self._GPU_COMPOSE)
with pytest.raises(HTTPException) as exc:
_scan_compose_content(compose)
assert exc.value.status_code == 400
assert "GPU passthrough" in exc.value.detail
def test_allows_deploy_devices_when_skipped(self, tmp_path):
"""Built-in compose paths pass skip_gpu_passthrough_check=True so the
legitimate NVIDIA reservation in docker-compose.nvidia.yml does not
get rejected when the dashboard-api re-scans during activate/enable.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(self._GPU_COMPOSE)
# Should not raise
_scan_compose_content(compose, skip_gpu_passthrough_check=True)
def test_handles_null_resources_without_500(self, tmp_path):
"""Strict regression for the audit-flagged bug: `deploy: { resources: null }`.
Pre-fix code did `deploy.get("resources", {}).get("reservations", {})`.
`dict.get(key, default)` returns the value when the key is present,
NOT the default — so `{"resources": None}.get("resources", {})` yields
None, and the next `.get()` AttributeError'd → 500 to the caller.
The fix's `isinstance(resources, dict)` guard short-circuits cleanly
because no GPU passthrough request can be expressed via null resources.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(
"services:\n svc:\n image: test\n"
" deploy:\n"
" resources: null\n"
)
# Should not raise — no GPU request can be expressed via null resources.
_scan_compose_content(compose)
def test_handles_null_reservations_without_500(self, tmp_path):
"""Defense-in-depth: `resources: { reservations: null }`.
The pre-fix code was already safe at this level — its leaf check
`isinstance(reservations, dict) and reservations.get("devices")`
short-circuited on `None`. This test locks the behavior in so a
future refactor that drops the leaf isinstance check (e.g. relying
only on the new outer guards) cannot reintroduce a 500 here.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(
"services:\n svc:\n image: test\n"
" deploy:\n"
" resources:\n"
" reservations: null\n"
)
# Should not raise.
_scan_compose_content(compose)
def test_handles_null_deploy_without_500(self, tmp_path):
"""Defense-in-depth: `deploy: null`.
The pre-fix code was already safe at this level via
`deploy = svc_def.get("deploy") or {}` — None is falsy and falls
through to `{}`. This test locks the behavior in so a future
refactor that drops the `or {}` short-circuit (e.g. switching to
explicit isinstance gating without the falsy fallback) cannot
reintroduce a 500 here.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(
"services:\n svc:\n image: test\n"
" deploy: null\n"
)
# Should not raise.
_scan_compose_content(compose)
# --- skip_root_user_check flag isolation ---
class TestScanComposeSkipRootUserCheck:
"""Direct unit tests for the skip_root_user_check parameter that permits
built-in extensions (e.g. openclaw, which uses `user: "0:0"` to perform
init-time chown before dropping privileges via setpriv) to declare a
root user, while user/library extensions cannot. Regression guard for
the openclaw init-time chown + setpriv pattern."""
_ROOT_COMPOSE = (
'services:\n svc:\n image: test\n user: "0:0"\n'
)
def test_builtin_with_root_user_accepted(self, tmp_path):
"""A built-in extension with user: 0:0 (init-time chown + setpriv
pattern, e.g. openclaw) must be accepted via
skip_root_user_check=True. Regression guard: built-ins with
`user: '0:0'` must be accepted when skip_root_user_check=True.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(self._ROOT_COMPOSE)
# Should not raise
_scan_compose_content(compose, skip_root_user_check=True)
def test_user_extension_with_root_user_rejected(self, tmp_path):
"""User/library extensions (skip_root_user_check defaulting to False)
still reject user: "0:0". Regression guard to ensure the
new parameter doesn't accidentally weaken security for non-built-ins.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(self._ROOT_COMPOSE)
with pytest.raises(HTTPException) as exc:
_scan_compose_content(compose)
assert exc.value.status_code == 400
assert "runs as root" in exc.value.detail
def test_privileged_still_blocked_when_skipped(self, tmp_path):
"""Other security checks remain active when skip_root_user_check=True;
a built-in cannot smuggle in `privileged: true` under the root-user
exemption.
"""
from routers.extensions import _scan_compose_content
compose = tmp_path / "compose.yaml"
compose.write_text(
'services:\n svc:\n image: test\n user: "0:0"\n'
" privileged: true\n",
)
with pytest.raises(HTTPException) as exc:
_scan_compose_content(compose, skip_root_user_check=True)
assert "privileged" in exc.value.detail
# --- Size quota enforcement ---
class TestInstallSizeQuota:
def test_install_rejects_oversized_extension(
self, test_client, monkeypatch, tmp_path,
):
"""400 when extension exceeds 50MB size limit."""
lib_dir = _setup_library_ext(tmp_path, "huge-ext")
# Write a file that exceeds the limit
big_file = lib_dir / "huge-ext" / "big.bin"
big_file.write_bytes(b"\x00" * (50 * 1024 * 1024 + 1))
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/huge-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "50MB" in resp.json()["detail"]
# --- Extension lifecycle status (stopped / health-based) ---
class TestExtensionLifecycleStatus:
def test_user_extension_enabled_and_healthy(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml + healthy container → enabled."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "my-ext", "name": "My Ext", "port": 8080,
"health": "/health"},
}))
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
mock_svc = _make_service_status("my-ext", "healthy")
with patch("user_extensions.get_user_services_cached",
return_value={"my-ext": {"host": "my-ext", "port": 8080,
"health": "/health", "name": "My Ext"}}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
with patch("helpers.check_service_health", new_callable=AsyncMock,
return_value=mock_svc):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["status"] == "enabled"
def test_user_extension_enabled_but_unhealthy(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml + unhealthy container → stopped."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "my-ext", "name": "My Ext", "port": 8080,
"health": "/health"},
}))
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
mock_svc = _make_service_status("my-ext", "down")
with patch("user_extensions.get_user_services_cached",
return_value={"my-ext": {"host": "my-ext", "port": 8080,
"health": "/health", "name": "My Ext"}}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
with patch("helpers.check_service_health", new_callable=AsyncMock,
return_value=mock_svc):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["status"] == "stopped"
def test_user_extension_http_unhealthy_returns_unhealthy(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml + HTTP 4xx/5xx health → unhealthy."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
(ext_dir / "manifest.yaml").write_text(yaml.dump({
"schema_version": "ods.services.v1",
"service": {"id": "my-ext", "name": "My Ext", "port": 8080,
"health": "/health"},
}))
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
mock_svc = _make_service_status("my-ext", "unhealthy")
with patch("user_extensions.get_user_services_cached",
return_value={"my-ext": {"host": "my-ext", "port": 8080,
"health": "/health", "name": "My Ext"}}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
with patch("helpers.check_service_health", new_callable=AsyncMock,
return_value=mock_svc):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
ext = data["extensions"][0]
assert ext["status"] == "unhealthy"
# Unhealthy counts toward "installed" and has its own summary bucket
assert data["summary"]["unhealthy"] == 1
assert data["summary"]["installed"] == 1
assert data["summary"]["stopped"] == 0
def test_user_extension_disabled_unchanged(self, test_client, monkeypatch, tmp_path):
"""User extension with compose.yaml.disabled → disabled (unchanged)."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
with patch("user_extensions.get_user_services_cached",
return_value={}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["status"] == "disabled"
def test_core_service_status_unchanged(self, test_client, monkeypatch, tmp_path):
"""Core service healthy → enabled, unhealthy → disabled (unchanged)."""
catalog = [_make_catalog_ext("core-svc", "Core Service")]
services = {"core-svc": {"host": "localhost", "port": 8080, "name": "Core"}}
_patch_extensions_config(monkeypatch, catalog, services, tmp_path=tmp_path)
mock_svc = _make_service_status("core-svc", "healthy")
with patch("user_extensions.get_user_services_cached",
return_value={}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[mock_svc]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
ext = resp.json()["extensions"][0]
assert ext["status"] == "enabled"
def test_catalog_includes_user_extension_health(self, test_client, monkeypatch, tmp_path):
"""Catalog response includes 'stopped' in summary counts."""
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
catalog = [_make_catalog_ext("my-ext", "My Extension")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
# No health data → stopped
with patch("user_extensions.get_user_services_cached",
return_value={}):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
summary = resp.json()["summary"]
assert summary["stopped"] == 1
assert summary["installed"] == 1
def test_enable_stopped_extension(self, test_client, monkeypatch, tmp_path):
"""Enable when compose.yaml exists (stopped) → starts without rename."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["action"] == "enabled"
# compose.yaml should still exist (not renamed)
assert (user_dir / "my-ext" / "compose.yaml").exists()
def test_enable_stopped_writes_error_progress_on_agent_failure(
self, test_client, monkeypatch, tmp_path,
):
"""Enable-stopped path writes error progress with restart guidance on agent failure."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
monkeypatch.setattr("routers.extensions._call_agent",
lambda action, sid: False)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert resp.json()["restart_required"] is True
progress_file = Path(tmp_path) / "extension-progress" / "my-ext.json"
assert progress_file.exists(), "enable-stopped path must write progress on agent failure"
data = json.loads(progress_file.read_text())
assert data["status"] == "error"
assert "ods restart" in data["error"]
def test_install_error_progress_includes_restart_guidance(
self, test_client, monkeypatch, tmp_path,
):
"""Install failure-path error message contains 'ods restart' actionable guidance."""
lib_dir = _setup_library_ext(tmp_path, "my-ext")
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
monkeypatch.setattr("routers.extensions._call_agent_install",
lambda sid: False)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
progress_file = Path(tmp_path) / "extension-progress" / "my-ext.json"
assert progress_file.exists()
data = json.loads(progress_file.read_text())
assert data["status"] == "error"
assert "ods restart" in data["error"]
def test_enable_stopped_rejects_malicious_compose(self, test_client, monkeypatch, tmp_path):
"""Enable stopped ext with malicious compose.yaml → 400."""
bad_compose = "services:\n svc:\n image: test\n privileged: true\n"
user_dir = tmp_path / "user"
user_dir.mkdir(exist_ok=True)
ext_dir = user_dir / "bad-ext"
ext_dir.mkdir(exist_ok=True)
(ext_dir / "compose.yaml").write_text(bad_compose)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/bad-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "privileged" in resp.json()["detail"]
def test_stale_started_unhealthy(self, monkeypatch, tmp_path):
"""Stale 'started' progress + container reporting unhealthy → 'unhealthy'.
Covers fork issue #485 (and the L194 branch added by merged PR #1037):
when the installer wrote ``status="started"`` more than 5 min ago
(i.e., past the 300s recency window in _compute_extension_status),
the progress entry must NOT keep the catalog stuck on "installing".
Instead the user-extension health-check branch must run and surface
the container's actual ServiceStatus — here, "unhealthy".
The progress timestamp is computed as ``now - 305s`` rather than a
fixed past date because ``_read_progress`` suppresses any
non-error progress older than 3600s; a fixed date would silently
return None and exercise the wrong path (the "no progress at all"
case rather than the "stale started progress" case).
"""
from datetime import datetime, timedelta, timezone
from routers.extensions import _compute_extension_status
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
(ext_dir / "compose.yaml").write_text(_SAFE_COMPOSE)
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
# 305s old: past the 300s "started"-recency window in
# _compute_extension_status, but well within _read_progress's
# 3600s general-staleness ceiling, so the progress is read but
# ignored and the user-ext health branch runs.
stale_ts = (datetime.now(timezone.utc) - timedelta(seconds=305)).isoformat()
progress_data = {
"service_id": "my-ext",
"status": "started",
"phase_label": "Service started",
"error": None,
"started_at": stale_ts,
"updated_at": stale_ts,
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
ext = _make_catalog_ext("my-ext")
services_by_id = {"my-ext": _make_service_status("my-ext", "unhealthy")}
status = _compute_extension_status(ext, services_by_id)
assert status == "unhealthy"
# --- Symlink handling ---
class TestSymlinkHandling:
def test_copytree_safe_skips_symlinks(self, tmp_path):
"""_copytree_safe skips symlinks in source directory."""
if os.name == "nt" and not can_create_symlinks(tmp_path):
pytest.skip("Windows symlink creation requires Developer Mode or administrator privileges")
from routers.extensions import _copytree_safe
src = tmp_path / "src"
src.mkdir()
(src / "real.txt").write_text("real content")
(src / "link.txt").symlink_to(src / "real.txt")
dst = tmp_path / "dst"
_copytree_safe(src, dst)
assert (dst / "real.txt").exists()
assert not (dst / "link.txt").exists()
def test_enable_stopped_rejects_symlinked_compose(
self, test_client, monkeypatch, tmp_path,
):
"""Enable stopped ext rejects a compose.yaml that is a symlink."""
if os.name == "nt" and not can_create_symlinks(tmp_path):
pytest.skip("Windows symlink creation requires Developer Mode or administrator privileges")
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
# Create a real file and symlink compose.yaml to it
real_compose = tmp_path / "real-compose.yaml"
real_compose.write_text(_SAFE_COMPOSE)
(ext_dir / "compose.yaml").symlink_to(real_compose)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "symlink" in resp.json()["detail"]
def test_enable_rejects_symlinked_compose(
self, test_client, monkeypatch, tmp_path,
):
"""Enable rejects a compose.yaml.disabled that is a symlink."""
if os.name == "nt" and not can_create_symlinks(tmp_path):
pytest.skip("Windows symlink creation requires Developer Mode or administrator privileges")
user_dir = tmp_path / "user"
ext_dir = user_dir / "my-ext"
ext_dir.mkdir(parents=True)
# Create a real file and symlink the .disabled to it
real_compose = tmp_path / "real-compose.yaml"
real_compose.write_text(_SAFE_COMPOSE)
(ext_dir / "compose.yaml.disabled").symlink_to(real_compose)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
resp = test_client.post(
"/api/extensions/my-ext/enable",
headers=test_client.auth_headers,
)
assert resp.status_code == 400
assert "symlink" in resp.json()["detail"]
# --- Purge extension data ---
class TestPurgeExtensionData:
def test_purge_happy_path(self, test_client, monkeypatch, tmp_path):
"""Purge succeeds for disabled extension with existing data dir."""
_patch_mutation_config(monkeypatch, tmp_path)
data_dir = tmp_path / "my-ext"
data_dir.mkdir()
(data_dir / "some-file.db").write_text("data")
with patch("routers.extensions._extensions_lock", return_value=contextlib.nullcontext()), \
patch("helpers.dir_size_gb", return_value=1.5):
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 200
data = resp.json()
assert data["id"] == "my-ext"
assert data["action"] == "purged"
assert data["size_gb_freed"] == 1.5
assert not data_dir.exists()
def test_purge_unlinks_progress_file(self, test_client, monkeypatch, tmp_path):
"""Purge also deletes the per-service install-progress entry so the UI
does not keep showing a stale 'installing' status."""
_patch_mutation_config(monkeypatch, tmp_path)
data_dir = tmp_path / "my-ext"
data_dir.mkdir()
(data_dir / "some-file.db").write_text("data")
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
progress_file = progress_dir / "my-ext.json"
progress_file.write_text(
'{"service_id": "my-ext", "status": "started",'
' "phase_label": "stale", "error": null,'
' "started_at": "2026-04-10T00:00:00+00:00",'
' "updated_at": "2026-04-10T00:00:00+00:00"}'
)
with patch("routers.extensions._extensions_lock", return_value=contextlib.nullcontext()), \
patch("helpers.dir_size_gb", return_value=0.1):
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 200
assert not progress_file.exists(), "purge must unlink the progress file"
def test_purge_400_when_enabled_builtin(self, test_client, monkeypatch, tmp_path):
"""400 when extension is still enabled (compose.yaml in built-in dir)."""
_patch_mutation_config(monkeypatch, tmp_path)
# Create compose.yaml in the built-in extensions dir
builtin_dir = tmp_path / "builtin" / "my-ext"
builtin_dir.mkdir(parents=True)
(builtin_dir / "compose.yaml").write_text("version: '3'")
# Also need a data dir to get past later checks
data_dir = tmp_path / "my-ext"
data_dir.mkdir()
with patch("routers.extensions._extensions_lock", return_value=contextlib.nullcontext()):
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 400
assert "still enabled" in resp.json()["detail"]
def test_purge_400_when_enabled_user(self, test_client, monkeypatch, tmp_path):
"""400 when extension is still enabled (compose.yaml in user dir)."""
user_dir = _setup_user_ext(tmp_path, "my-ext", enabled=True)
_patch_mutation_config(monkeypatch, tmp_path, user_dir=user_dir)
with patch("routers.extensions._extensions_lock", return_value=contextlib.nullcontext()):
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 400
assert "still enabled" in resp.json()["detail"]
def test_purge_403_core_service(self, test_client, monkeypatch, tmp_path):
"""403 when trying to purge a core service."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.request(
"DELETE", "/api/extensions/open-webui/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 403
assert "always-on service" in resp.json()["detail"].lower()
def test_purge_404_invalid_id(self, test_client, monkeypatch, tmp_path):
"""404 for service_id that fails regex validation."""
_patch_mutation_config(monkeypatch, tmp_path)
for bad_id in ["..etc", ".hidden", "UPPERCASE", "-starts-dash"]:
resp = test_client.request(
"DELETE", f"/api/extensions/{bad_id}/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 404, f"Expected 404 for: {bad_id}"
def test_purge_404_no_data_dir(self, test_client, monkeypatch, tmp_path):
"""404 when valid ID but no data directory exists."""
_patch_mutation_config(monkeypatch, tmp_path)
with patch("routers.extensions._extensions_lock", return_value=contextlib.nullcontext()):
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
assert resp.status_code == 404
assert "No data directory" in resp.json()["detail"]
def test_purge_400_confirm_false(self, test_client, monkeypatch, tmp_path):
"""400 when data exists but confirm is false."""
_patch_mutation_config(monkeypatch, tmp_path)
data_dir = tmp_path / "my-ext"
data_dir.mkdir()
with patch("routers.extensions._extensions_lock", return_value=contextlib.nullcontext()):
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
headers=test_client.auth_headers,
json={"confirm": False},
)
assert resp.status_code == 400
assert "Confirmation required" in resp.json()["detail"]
# Data dir should still exist
assert data_dir.exists()
def test_purge_path_traversal(self, test_client, monkeypatch, tmp_path):
"""Path traversal attempts are blocked by regex or path check."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.request(
"DELETE", "/api/extensions/..%2fetc/data",
headers=test_client.auth_headers,
json={"confirm": True},
)
# Should fail at regex or Starlette routing level
assert resp.status_code in (404, 422)
def test_purge_requires_auth(self, test_client):
"""DELETE /api/extensions/{id}/data without auth → 401."""
resp = test_client.request(
"DELETE", "/api/extensions/my-ext/data",
json={"confirm": True},
)
assert resp.status_code == 401
# --- Orphaned storage ---
class TestOrphanedStorage:
def test_orphaned_requires_auth(self, test_client):
"""GET /api/storage/orphaned without auth → 401."""
resp = test_client.get("/api/storage/orphaned")
assert resp.status_code == 401
def test_orphaned_empty_data_dir(self, test_client, monkeypatch, tmp_path):
"""Empty data dir returns empty orphaned list."""
data_dir = tmp_path / "data"
data_dir.mkdir()
monkeypatch.setattr("routers.extensions.DATA_DIR", str(data_dir))
monkeypatch.setattr("routers.extensions.SERVICES", {})
resp = test_client.get(
"/api/storage/orphaned",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["orphaned"] == []
assert data["total_gb"] == 0
def test_orphaned_nonexistent_data_dir(self, test_client, monkeypatch, tmp_path):
"""Non-existent data dir returns empty orphaned list."""
monkeypatch.setattr("routers.extensions.DATA_DIR",
str(tmp_path / "nonexistent"))
monkeypatch.setattr("routers.extensions.SERVICES", {})
resp = test_client.get(
"/api/storage/orphaned",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["orphaned"] == []
assert data["total_gb"] == 0
def test_orphaned_excludes_known_services(self, test_client, monkeypatch, tmp_path):
"""Dirs matching SERVICES keys are not listed as orphaned."""
data_dir = tmp_path / "data"
data_dir.mkdir()
(data_dir / "known-svc").mkdir()
monkeypatch.setattr("routers.extensions.DATA_DIR", str(data_dir))
monkeypatch.setattr("routers.extensions.SERVICES",
{"known-svc": {"host": "localhost", "port": 8080}})
with patch("helpers.dir_size_gb", return_value=2.0):
resp = test_client.get(
"/api/storage/orphaned",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["orphaned"] == []
assert data["total_gb"] == 0
def test_orphaned_excludes_system_dirs(self, test_client, monkeypatch, tmp_path):
"""System dirs (models, config, etc.) are not listed as orphaned."""
data_dir = tmp_path / "data"
data_dir.mkdir()
for name in ("models", "config", "user-extensions", "extensions-library"):
(data_dir / name).mkdir()
monkeypatch.setattr("routers.extensions.DATA_DIR", str(data_dir))
monkeypatch.setattr("routers.extensions.SERVICES", {})
with patch("helpers.dir_size_gb", return_value=1.0):
resp = test_client.get(
"/api/storage/orphaned",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["orphaned"] == []
assert data["total_gb"] == 0
def test_orphaned_includes_unknown_dirs(self, test_client, monkeypatch, tmp_path):
"""Dirs not in SERVICES or system_dirs are listed as orphaned."""
data_dir = tmp_path / "data"
data_dir.mkdir()
(data_dir / "mystery-data").mkdir()
(data_dir / "leftover-ext").mkdir()
monkeypatch.setattr("routers.extensions.DATA_DIR", str(data_dir))
monkeypatch.setattr("routers.extensions.SERVICES", {})
with patch("helpers.dir_size_gb", return_value=3.0):
resp = test_client.get(
"/api/storage/orphaned",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert len(data["orphaned"]) == 2
names = [o["name"] for o in data["orphaned"]]
assert "mystery-data" in names
assert "leftover-ext" in names
assert data["orphaned"][0]["size_gb"] == 3.0
assert data["total_gb"] == 6.0
def test_orphaned_skips_files(self, test_client, monkeypatch, tmp_path):
"""Regular files in data dir are not listed."""
data_dir = tmp_path / "data"
data_dir.mkdir()
(data_dir / "some-file.txt").write_text("not a directory")
(data_dir / "orphan-dir").mkdir()
monkeypatch.setattr("routers.extensions.DATA_DIR", str(data_dir))
monkeypatch.setattr("routers.extensions.SERVICES", {})
with patch("helpers.dir_size_gb", return_value=0.5):
resp = test_client.get(
"/api/storage/orphaned",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert len(data["orphaned"]) == 1
assert data["orphaned"][0]["name"] == "orphan-dir"
assert data["total_gb"] == 0.5
# --- Install progress tracking ---
class TestInstallProgress:
def test_progress_endpoint_no_progress(self, test_client, monkeypatch, tmp_path):
"""GET progress when no file exists → idle."""
_patch_mutation_config(monkeypatch, tmp_path)
resp = test_client.get(
"/api/extensions/my-ext/progress",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["service_id"] == "my-ext"
assert data["status"] == "idle"
def test_progress_endpoint_during_install(self, test_client, monkeypatch, tmp_path):
"""GET progress with active progress file → returns data."""
_patch_mutation_config(monkeypatch, tmp_path)
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
progress_data = {
"service_id": "my-ext",
"status": "pulling",
"phase_label": "Downloading image...",
"error": None,
"started_at": "2026-04-06T10:00:00+00:00",
"updated_at": "2026-04-06T10:00:05+00:00",
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
resp = test_client.get(
"/api/extensions/my-ext/progress",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "pulling"
assert data["phase_label"] == "Downloading image..."
def test_status_installing_when_progress_pulling(self, monkeypatch, tmp_path):
"""Progress file with status 'pulling' → _compute_extension_status returns 'installing'."""
from routers.extensions import _compute_extension_status
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", tmp_path / "user")
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
progress_data = {
"service_id": "my-ext",
"status": "pulling",
"phase_label": "Downloading image...",
"error": None,
"started_at": now,
"updated_at": now,
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
ext = _make_catalog_ext("my-ext")
status = _compute_extension_status(ext, {})
assert status == "installing"
def test_status_installing_when_progress_starting(self, monkeypatch, tmp_path):
"""Progress file with status 'starting' → _compute_extension_status returns 'installing'."""
from routers.extensions import _compute_extension_status
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", tmp_path / "user")
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
progress_data = {
"service_id": "my-ext",
"status": "starting",
"phase_label": "Starting container...",
"error": None,
"started_at": now,
"updated_at": now,
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
ext = _make_catalog_ext("my-ext")
status = _compute_extension_status(ext, {})
assert status == "installing"
def test_status_setting_up_when_progress_setup_hook(self, monkeypatch, tmp_path):
"""Progress file with status 'setup_hook' → returns 'setting_up'."""
from routers.extensions import _compute_extension_status
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", tmp_path / "user")
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
progress_data = {
"service_id": "my-ext",
"status": "setup_hook",
"phase_label": "Running setup...",
"error": None,
"started_at": now,
"updated_at": now,
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
ext = _make_catalog_ext("my-ext")
status = _compute_extension_status(ext, {})
assert status == "setting_up"
def test_status_error_when_progress_error(self, monkeypatch, tmp_path):
"""Progress file with status 'error' → returns 'error'."""
from routers.extensions import _compute_extension_status
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", tmp_path / "user")
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
progress_data = {
"service_id": "my-ext",
"status": "error",
"phase_label": "Installation failed",
"error": "something went wrong",
"started_at": now,
"updated_at": now,
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
ext = _make_catalog_ext("my-ext")
status = _compute_extension_status(ext, {})
assert status == "error"
def test_status_cli_installed_for_oneshot_started_recent(self, monkeypatch, tmp_path):
"""One-shot extension (port=0) with recent 'started' progress →
'cli_installed'. Regression: previously the install toast cycled
through 'installing' / 'stopped' because there is no healthcheck
for a CLI-only container that exits 0 after init."""
from routers.extensions import _compute_extension_status
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", tmp_path / "user")
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
progress_data = {
"service_id": "aider",
"status": "started",
"phase_label": "Service started",
"error": None,
"started_at": now,
"updated_at": now,
}
(progress_dir / "aider.json").write_text(json.dumps(progress_data))
ext = _make_catalog_ext("aider")
ext["port"] = 0 # one-shot CLI extension marker
ext["startup_check"] = False
status = _compute_extension_status(ext, {})
assert status == "cli_installed"
def test_status_cli_installed_for_oneshot_user_dir_compose(self, monkeypatch, tmp_path):
"""Steady-state: a one-shot extension (port=0) installed under
USER_EXTENSIONS_DIR with compose.yaml present should remain
'cli_installed' even when no recent progress file exists."""
from routers.extensions import _compute_extension_status
user_dir = tmp_path / "user"
user_dir.mkdir()
aider_dir = user_dir / "aider"
aider_dir.mkdir()
(aider_dir / "compose.yaml").write_text(
"services:\n aider:\n image: paulgauthier/aider\n"
)
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_dir)
monkeypatch.setattr("routers.extensions.GPU_BACKEND", "nvidia")
monkeypatch.setattr("routers.extensions.SERVICES", {})
ext = _make_catalog_ext("aider")
ext["port"] = 0 # one-shot CLI extension marker
ext["startup_check"] = False
status = _compute_extension_status(ext, {})
assert status == "cli_installed"
def test_stale_progress_ignored(self, monkeypatch, tmp_path):
"""Progress file >1 hour old → _read_progress returns None."""
from routers.extensions import _read_progress
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
# Set updated_at to far in the past (well over 1 hour)
progress_data = {
"service_id": "my-ext",
"status": "pulling",
"phase_label": "Downloading image...",
"error": None,
"started_at": "2020-01-01T00:00:00+00:00",
"updated_at": "2020-01-01T00:00:00+00:00",
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
result = _read_progress("my-ext")
assert result is None
def test_stale_error_progress_preserved(self, monkeypatch, tmp_path):
"""Stale progress file with status 'error' → _read_progress still returns it (not None)."""
from routers.extensions import _read_progress
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
progress_data = {
"service_id": "my-ext",
"status": "error",
"phase_label": "Installation failed",
"error": "something went wrong",
"started_at": "2020-01-01T00:00:00+00:00",
"updated_at": "2020-01-01T00:00:00+00:00",
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
result = _read_progress("my-ext")
assert result is not None
assert result["status"] == "error"
def test_progress_cleanup_removes_old_started(self, monkeypatch, tmp_path):
"""_cleanup_stale_progress() removes 'started' files >15 min old."""
from routers.extensions import _cleanup_stale_progress
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
progress_dir = tmp_path / "extension-progress"
progress_dir.mkdir()
progress_data = {
"service_id": "my-ext",
"status": "started",
"phase_label": "Service started",
"error": None,
"started_at": "2020-01-01T00:00:00+00:00",
"updated_at": "2020-01-01T00:00:00+00:00",
}
(progress_dir / "my-ext.json").write_text(json.dumps(progress_data))
_cleanup_stale_progress()
assert not (progress_dir / "my-ext.json").exists()
def test_install_returns_progress_endpoint(self, test_client, monkeypatch, tmp_path):
"""Install response includes progress_endpoint field."""
lib_dir = _setup_library_ext(tmp_path, "my-ext")
_patch_mutation_config(monkeypatch, tmp_path, lib_dir=lib_dir)
resp = test_client.post(
"/api/extensions/my-ext/install",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
data = resp.json()
assert data["progress_endpoint"] == "/api/extensions/my-ext/progress"
# --- Config sync (delegated to host agent) ---
class TestSyncExtensionConfig:
"""The dashboard-api container has /ods/config bind-mounted
read-only, so _sync_extension_config must NOT touch the filesystem
locally — it forwards to the host agent. End-to-end file-copy behaviour
is covered by the host-agent wire test (TestSyncExtensionConfigWire)."""
def test_delegates_to_host_agent(self, monkeypatch):
"""_sync_extension_config calls _call_agent_sync_config with the service id."""
from routers import extensions as ext_mod
calls = []
def _fake(sid):
calls.append(sid)
return True
monkeypatch.setattr(ext_mod, "_call_agent_sync_config", _fake)
result = ext_mod._sync_extension_config("my-ext")
assert calls == ["my-ext"]
assert result is True
def test_returns_false_on_agent_failure(self, monkeypatch):
"""Agent failure surfaces as a False return; caller decides what to do."""
from routers import extensions as ext_mod
monkeypatch.setattr(
ext_mod, "_call_agent_sync_config", lambda _sid: False,
)
assert ext_mod._sync_extension_config("my-ext") is False
def test_call_agent_sync_config_sends_post(self, monkeypatch):
"""The HTTP helper POSTs to /v1/extension/sync_config with bearer auth."""
from routers import extensions as ext_mod
captured = {}
class _FakeResp:
status = 200
def __enter__(self): return self
def __exit__(self, *a): return False
def _fake_urlopen(req, timeout=None):
captured["url"] = req.full_url
captured["method"] = req.get_method()
captured["auth"] = req.get_header("Authorization")
captured["body"] = req.data
return _FakeResp()
monkeypatch.setattr(ext_mod, "AGENT_URL", "http://agent:7710")
monkeypatch.setattr(ext_mod, "ODS_AGENT_KEY", "secret")
monkeypatch.setattr(ext_mod.urllib.request, "urlopen", _fake_urlopen)
assert ext_mod._call_agent_sync_config("my-ext") is True
assert captured["url"] == "http://agent:7710/v1/extension/sync_config"
assert captured["method"] == "POST"
assert captured["auth"] == "Bearer secret"
assert b'"service_id"' in captured["body"]
assert b'"my-ext"' in captured["body"]
# --- Error progress ---
class TestWriteErrorProgress:
def test_sets_error_status_on_existing_progress(self, monkeypatch, tmp_path):
"""Error progress overwrites status but preserves started_at."""
from routers.extensions import _write_initial_progress, _write_error_progress
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
_write_initial_progress("my-ext")
progress_file = tmp_path / "extension-progress" / "my-ext.json"
initial = json.loads(progress_file.read_text())
assert initial["status"] == "pulling"
_write_error_progress("my-ext", "Host agent failed")
updated = json.loads(progress_file.read_text())
assert updated["status"] == "error"
assert updated["error"] == "Host agent failed"
assert updated["started_at"] == initial["started_at"]
def test_creates_error_file_when_no_prior_progress(self, monkeypatch, tmp_path):
"""Error progress can be written even without prior progress file."""
from routers.extensions import _write_error_progress
monkeypatch.setattr("routers.extensions.DATA_DIR", str(tmp_path))
_write_error_progress("my-ext", "Agent unreachable")
progress_file = tmp_path / "extension-progress" / "my-ext.json"
data = json.loads(progress_file.read_text())
assert data["status"] == "error"
assert data["error"] == "Agent unreachable"
assert "phase_label" in data
# --- _activate_service: built-in (EXTENSIONS_DIR) branch ---
class TestActivateServiceBuiltinBranch:
"""_activate_service must resolve services from EXTENSIONS_DIR (built-in)
when not present under USER_EXTENSIONS_DIR — required so templates can
enable built-in extensions like n8n, tts, etc."""
def test_activate_service_resolves_builtin_with_disabled_compose(
self, monkeypatch, tmp_path,
):
"""Built-in extension with compose.yaml.disabled is renamed to compose.yaml."""
from routers.extensions import _activate_service
builtin_root = tmp_path / "builtin"
user_root = tmp_path / "user"
builtin_root.mkdir()
user_root.mkdir()
ext_dir = builtin_root / "fakesvc"
ext_dir.mkdir()
(ext_dir / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
monkeypatch.setattr("routers.extensions.EXTENSIONS_DIR", builtin_root)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_root)
calls = []
def _mock_compose_rename(action, service_id):
calls.append((action, service_id))
(ext_dir / "compose.yaml.disabled").rename(ext_dir / "compose.yaml")
return True
monkeypatch.setattr(
"routers.extensions._call_agent_compose_rename",
_mock_compose_rename,
)
result = _activate_service("fakesvc")
assert result == {"id": "fakesvc", "action": "enabled"}
assert calls == [("activate", "fakesvc")]
assert (ext_dir / "compose.yaml").exists()
assert not (ext_dir / "compose.yaml.disabled").exists()
def test_activate_service_resolves_builtin_already_enabled(
self, monkeypatch, tmp_path,
):
"""Built-in extension already enabled returns idempotent action without mutation."""
from routers.extensions import _activate_service
builtin_root = tmp_path / "builtin"
user_root = tmp_path / "user"
builtin_root.mkdir()
user_root.mkdir()
ext_dir = builtin_root / "fakesvc"
ext_dir.mkdir()
enabled_compose = ext_dir / "compose.yaml"
enabled_compose.write_text(_SAFE_COMPOSE)
monkeypatch.setattr("routers.extensions.EXTENSIONS_DIR", builtin_root)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_root)
result = _activate_service("fakesvc")
assert result == {"id": "fakesvc", "action": "already_enabled"}
assert enabled_compose.exists()
assert not (ext_dir / "compose.yaml.disabled").exists()
def test_activate_service_user_dir_takes_precedence_over_builtin(
self, monkeypatch, tmp_path,
):
"""When the same id exists in both, the user-installed copy wins."""
from routers.extensions import _activate_service
builtin_root = tmp_path / "builtin"
user_root = tmp_path / "user"
builtin_root.mkdir()
user_root.mkdir()
# User dir: disabled, expected to be activated
user_ext = user_root / "fakesvc"
user_ext.mkdir()
(user_ext / "compose.yaml.disabled").write_text(_SAFE_COMPOSE)
# Built-in: already enabled, must remain untouched
builtin_ext = builtin_root / "fakesvc"
builtin_ext.mkdir()
builtin_compose = builtin_ext / "compose.yaml"
builtin_compose.write_text(_SAFE_COMPOSE)
monkeypatch.setattr("routers.extensions.EXTENSIONS_DIR", builtin_root)
monkeypatch.setattr("routers.extensions.USER_EXTENSIONS_DIR", user_root)
result = _activate_service("fakesvc")
assert result == {"id": "fakesvc", "action": "enabled"}
assert (user_ext / "compose.yaml").exists()
assert not (user_ext / "compose.yaml.disabled").exists()
# Built-in untouched
assert builtin_compose.exists()
class TestAssertNotCoreAllowsBuiltins:
"""_assert_not_core blocks only the 4 always-on base-compose services."""
@pytest.mark.parametrize("service_id", [
"n8n", "tts", "whisper", "comfyui", "litellm", "openclaw",
"perplexica", "searxng", "privacy-shield", "token-spy", "qdrant",
"embeddings", "ape", "langfuse", "opencode", "hermes", "hermes-proxy",
])
def test_assert_not_core_allows_builtin_extension(self, service_id):
"""Built-in extensions are toggleable and must not be blocked."""
_assert_not_core(service_id)
@pytest.mark.parametrize("service_id", [
"llama-server", "open-webui", "dashboard", "dashboard-api",
])
def test_assert_not_core_blocks_always_on(self, service_id):
"""Always-on base-compose services must raise 403."""
with pytest.raises(HTTPException) as exc_info:
_assert_not_core(service_id)
assert exc_info.value.status_code == 403
def test_production_core_service_ids_include_hermes_services():
"""The production anti-shadowing allowlist must cover Hermes built-ins."""
core_ids_path = Path(__file__).resolve().parents[4] / "config" / "core-service-ids.json"
core_ids = set(json.loads(core_ids_path.read_text(encoding="utf-8")))
assert {"hermes", "hermes-proxy"} <= core_ids
class TestCallAgentErrorNarrowing:
"""_call_agent swallows network errors but not programmer errors."""
def test_call_agent_returns_false_on_urlerror(self, monkeypatch, caplog):
"""Network failures produce (False, warning) — callers rely on this."""
import logging
import urllib.error
from routers import extensions as ext_module
def _raise(*_args, **_kwargs):
raise urllib.error.URLError("timeout")
monkeypatch.setattr(ext_module.urllib.request, "urlopen", _raise)
with caplog.at_level(logging.WARNING, logger="routers.extensions"):
result = ext_module._call_agent("start", "svc-x")
assert result is False
assert any("Host agent unreachable" in r.message for r in caplog.records)
def test_call_agent_reraises_non_network_errors(self, monkeypatch):
"""Programmer errors (e.g. AttributeError) must not be swallowed."""
from routers import extensions as ext_module
def _raise(*_args, **_kwargs):
raise AttributeError("boom")
monkeypatch.setattr(ext_module.urllib.request, "urlopen", _raise)
with pytest.raises(AttributeError):
ext_module._call_agent("start", "svc-x")
def test_catalog_logs_when_cleanup_future_fails(
self, test_client, monkeypatch, tmp_path, caplog,
):
"""Stale-progress cleanup failures are logged, not lost to fire-and-forget."""
import logging
catalog = [_make_catalog_ext("test-svc", "Test Service")]
_patch_extensions_config(monkeypatch, catalog, tmp_path=tmp_path)
def _boom():
raise RuntimeError("cleanup exploded")
monkeypatch.setattr(
"routers.extensions._cleanup_stale_progress", _boom,
)
with caplog.at_level(logging.ERROR, logger="routers.extensions"):
with patch("helpers.get_all_services", new_callable=AsyncMock,
return_value=[]):
resp = test_client.get(
"/api/extensions/catalog",
headers=test_client.auth_headers,
)
assert resp.status_code == 200
assert any(
"stale-progress cleanup failed" in r.message for r in caplog.records
)
def test_extensions_lock_falls_back_when_data_root_is_unwritable(
tmp_path, monkeypatch,
):
"""Extension installs should still lock when /data itself is not writable."""
from routers import extensions as ext_module
blocked_parent = tmp_path / "blocked-parent"
blocked_parent.write_text("not a directory", encoding="utf-8")
fallback_lock = tmp_path / "config" / ".extensions-lock"
monkeypatch.setattr(
ext_module,
"_extensions_lock_candidates",
lambda: [blocked_parent / ".extensions-lock", fallback_lock],
)
with ext_module._extensions_lock():
assert fallback_lock.exists()
在新工单中引用 查看 Git Blame 复制永久链接