"""Tests for extensions portal endpoints."""

import contextlib
import json
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 _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"] == "dream enable test-svc"
        assert data["setup_instructions"]["cli_disable"] == "dream 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": "dream.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"}))


# --- 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_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_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": "dream.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": "dream.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


# --- 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: dream.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 'dream 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": "dream.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": "dream.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": "dream.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": "dream.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"]

    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": "dream.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": "dream.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": "dream.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 "dream restart" in data["error"]

    def test_install_error_progress_includes_restart_guidance(
        self, test_client, monkeypatch, tmp_path,
    ):
        """Install failure-path error message contains 'dream 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 "dream 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."""
        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."""
        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."""
        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 /dream-server/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, "DREAM_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", "dreamforge", "langfuse", "opencode",
    ])
    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


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
        )
