"""Regression tests for issues found during code review."""

import importlib
import json
import sys
import types
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

from src.preset_manager import PresetManager


class _FakeColumn:
    def __init__(self, name):
        self.name = name

    def __eq__(self, value):
        return ("eq", self.name, value)


class _FakeModelEndpoint:
    id = _FakeColumn("id")
    is_enabled = _FakeColumn("is_enabled")
    owner = _FakeColumn("owner")


class _FakeDbSession:
    id = _FakeColumn("id")
    endpoint_url = _FakeColumn("endpoint_url")


class _FakeQuery:
    def __init__(self, rows):
        self.rows = list(rows)

    def filter(self, *conditions):
        for condition in conditions:
            if isinstance(condition, tuple) and condition[0] == "eq":
                _, field, value = condition
                self.rows = [row for row in self.rows if getattr(row, field) == value]
        return self

    def first(self):
        return self.rows[0] if self.rows else None

    def all(self):
        return list(self.rows)


class _FakeDb:
    def __init__(self, rows):
        self.rows = rows

    def query(self, model):
        return _FakeQuery(self.rows)

    def close(self):
        pass


def _default_chat_endpoint():
    from routes.model_routes import setup_model_routes

    router = setup_model_routes(model_discovery=None)
    for route in router.routes:
        if getattr(route, "path", "") == "/api/default-chat":
            return route.endpoint
    raise AssertionError("/api/default-chat route not found")


def _install_model_route_import_stubs(monkeypatch):
    core_mod = types.ModuleType("core")
    core_mod.__path__ = []
    db_mod = types.ModuleType("core.database")
    db_mod.SessionLocal = lambda: _FakeDb([])
    db_mod.ModelEndpoint = _FakeModelEndpoint
    db_mod.Session = _FakeDbSession
    db_mod.Document = MagicMock()
    db_mod.DocumentVersion = MagicMock()
    db_mod.GalleryImage = MagicMock()
    middleware_mod = types.ModuleType("core.middleware")
    middleware_mod.require_admin = lambda request: None
    multipart_mod = types.ModuleType("python_multipart")
    multipart_mod.__version__ = "0.0.13"
    models_mod = types.ModuleType("core.models")
    models_mod.ChatMessage = MagicMock()
    exceptions_mod = types.ModuleType("core.exceptions")
    exceptions_mod.SessionNotFoundError = type("SessionNotFoundError", (Exception,), {})
    session_mgr_mod = types.ModuleType("core.session_manager")
    session_mgr_mod.SessionManager = MagicMock()

    monkeypatch.delitem(sys.modules, "routes.model_routes", raising=False)
    monkeypatch.delitem(sys.modules, "routes.chat_routes", raising=False)
    monkeypatch.delitem(sys.modules, "routes.session_routes", raising=False)
    monkeypatch.setitem(sys.modules, "core", core_mod)
    monkeypatch.setitem(sys.modules, "core.database", db_mod)
    monkeypatch.setitem(sys.modules, "core.middleware", middleware_mod)
    monkeypatch.setitem(sys.modules, "python_multipart", multipart_mod)
    monkeypatch.setitem(sys.modules, "core.models", models_mod)
    monkeypatch.setitem(sys.modules, "core.exceptions", exceptions_mod)
    monkeypatch.setitem(sys.modules, "core.session_manager", session_mgr_mod)


def _install_core_auth_stub(monkeypatch):
    """Install the narrow auth surface needed by tool-policy tests."""
    core_mod = types.ModuleType("core")
    core_mod.__path__ = []
    auth_mod = types.ModuleType("core.auth")
    auth_mod.AuthManager = MagicMock()
    core_mod.auth = auth_mod
    monkeypatch.setitem(sys.modules, "core", core_mod)
    monkeypatch.setitem(sys.modules, "core.auth", auth_mod)
    return auth_mod


def _install_core_middleware_stub(monkeypatch):
    """Install the narrow middleware surface needed by loopback tool tests."""
    core_mod = types.ModuleType("core")
    core_mod.__path__ = []
    middleware_mod = types.ModuleType("core.middleware")
    middleware_mod.INTERNAL_TOOL_HEADER = "X-Internal-Tool"
    middleware_mod.INTERNAL_TOOL_TOKEN = "test-token"
    core_mod.middleware = middleware_mod
    monkeypatch.setitem(sys.modules, "core", core_mod)
    monkeypatch.setitem(sys.modules, "core.middleware", middleware_mod)
    return middleware_mod


def test_providers_requires_admin_before_discovery_and_cache(monkeypatch):
    _install_model_route_import_stubs(monkeypatch)
    import routes.model_routes as model_routes

    class _Discovery:
        def __init__(self):
            self.calls = 0

        def get_providers(self):
            self.calls += 1
            return {"providers": [{"host": "internal.example"}]}

    discovery = _Discovery()
    router = model_routes.setup_model_routes(discovery)
    endpoint = next(
        route.endpoint
        for route in router.routes
        if getattr(route, "path", "") == "/api/providers"
    )
    request = SimpleNamespace()

    assert endpoint(request, refresh=True) == {"providers": [{"host": "internal.example"}]}
    assert discovery.calls == 1

    def deny_admin(_request):
        raise PermissionError("admin required")

    monkeypatch.setattr(model_routes, "require_admin", deny_admin)

    with pytest.raises(PermissionError):
        endpoint(request, refresh=True)
    with pytest.raises(PermissionError):
        endpoint(request, refresh=False)
    assert discovery.calls == 1


def test_default_chat_does_not_auto_pick_shared_endpoint_for_fresh_user(monkeypatch):
    _install_model_route_import_stubs(monkeypatch)
    import routes.model_routes as model_routes
    import routes.prefs_routes as prefs_routes

    shared_ep = SimpleNamespace(
        id="shared",
        base_url="http://localhost:11434",
        is_enabled=True,
        owner=None,
        cached_models='["shared-model"]',
    )

    def scoped_owner_filter(query, model_cls, user, *, include_shared=True):
        query.rows = [
            row for row in query.rows
            if row.owner == user or (include_shared and row.owner is None)
        ]
        return query

    monkeypatch.setattr(model_routes, "ModelEndpoint", _FakeModelEndpoint)
    monkeypatch.setattr(model_routes, "SessionLocal", lambda: _FakeDb([shared_ep]))
    monkeypatch.setattr(model_routes, "_load_settings", lambda: {})
    monkeypatch.setattr(model_routes, "owner_filter", scoped_owner_filter)
    monkeypatch.setattr(model_routes, "_normalize_base", lambda base: base.rstrip("/"))
    monkeypatch.setattr(model_routes, "build_chat_url", lambda base: f"{base}/chat/completions")
    monkeypatch.setattr(prefs_routes, "_load_for_user", lambda user: {})

    request = SimpleNamespace(
        state=SimpleNamespace(current_user="fresh"),
        app=SimpleNamespace(state=SimpleNamespace(
            auth_manager=SimpleNamespace(is_admin=lambda user: False)
        )),
    )

    assert _default_chat_endpoint()(request) == {
        "endpoint_id": "",
        "endpoint_url": "",
        "model": "",
    }


def test_default_chat_uses_owned_endpoint_as_regular_user_last_resort(monkeypatch):
    _install_model_route_import_stubs(monkeypatch)
    import routes.model_routes as model_routes
    import routes.prefs_routes as prefs_routes

    owned_ep = SimpleNamespace(
        id="owned",
        base_url="http://localhost:11434",
        is_enabled=True,
        owner="fresh",
        cached_models='["owned-model"]',
    )

    def scoped_owner_filter(query, model_cls, user, *, include_shared=True):
        query.rows = [
            row for row in query.rows
            if row.owner == user or (include_shared and row.owner is None)
        ]
        return query

    monkeypatch.setattr(model_routes, "ModelEndpoint", _FakeModelEndpoint)
    monkeypatch.setattr(model_routes, "SessionLocal", lambda: _FakeDb([owned_ep]))
    monkeypatch.setattr(model_routes, "_load_settings", lambda: {})
    monkeypatch.setattr(model_routes, "owner_filter", scoped_owner_filter)
    monkeypatch.setattr(model_routes, "_normalize_base", lambda base: base.rstrip("/"))
    monkeypatch.setattr(model_routes, "build_chat_url", lambda base: f"{base}/chat/completions")
    monkeypatch.setattr(prefs_routes, "_load_for_user", lambda user: {})

    request = SimpleNamespace(
        state=SimpleNamespace(current_user="fresh"),
        app=SimpleNamespace(state=SimpleNamespace(
            auth_manager=SimpleNamespace(is_admin=lambda user: False)
        )),
    )

    assert _default_chat_endpoint()(request) == {
        "endpoint_id": "owned",
        "endpoint_url": "http://localhost:11434/chat/completions",
        "model": "owned-model",
    }


def test_preset_manager_persists_inject_fields(tmp_path):
    manager = PresetManager(str(tmp_path))

    ok = manager.update_custom(
        temperature=0.7,
        max_tokens=2048,
        system_prompt="Be useful.",
        name="Custom",
        enabled=True,
        inject_prefix="PREFIX",
        inject_suffix="SUFFIX",
    )

    assert ok is True
    assert manager.presets["custom"]["inject_prefix"] == "PREFIX"
    assert manager.presets["custom"]["inject_suffix"] == "SUFFIX"

    reloaded = PresetManager(str(tmp_path))
    assert reloaded.presets["custom"]["inject_prefix"] == "PREFIX"
    assert reloaded.presets["custom"]["inject_suffix"] == "SUFFIX"


def test_preset_manager_default_custom_preset_starts_disabled(tmp_path):
    manager = PresetManager(str(tmp_path))

    custom = manager.presets["custom"]

    assert custom["enabled"] is False
    assert custom["system_prompt"] == ""
    assert custom["temperature"] == 1.0
    assert custom["max_tokens"] == 0


def test_preset_manager_migrates_legacy_default_custom_preset_disabled(tmp_path):
    presets_file = tmp_path / "presets.json"
    presets_file.write_text(
        json.dumps({
            "custom": {
                "name": "Custom",
                "temperature": 0.7,
                "max_tokens": 4096,
                "system_prompt": "You are a helpful, balanced assistant. Match your response style to the user's needs.",
            }
        }),
        encoding="utf-8",
    )

    manager = PresetManager(str(tmp_path))
    custom = manager.presets["custom"]

    assert custom["enabled"] is False
    assert custom["system_prompt"] == ""
    assert custom["temperature"] == 1.0
    assert custom["max_tokens"] == 0


def test_normalize_thinking_handles_lowercase_thinking_process(monkeypatch):
    for mod_name in [
        "starlette.middleware",
        "starlette.middleware.base",
        "core.models",
        "core.database",
        "routes.prefs_routes",
        "routes.research_routes",
        "src.llm_core",
        "src.context_compactor",
        "src.model_context",
        "src.auth_helpers",
    ]:
        if mod_name not in sys.modules:
            monkeypatch.setitem(sys.modules, mod_name, MagicMock())

    chat_helpers = importlib.import_module("routes.chat_helpers")

    text = (
        "Thinking process:\n"
        "Analyze the Request: The user is explicitly instructing me to use the tag.\n\n"
        "hi"
    )

    normalized = chat_helpers._normalize_thinking(text)

    assert normalized == (
        "<think>Analyze the Request: The user is explicitly instructing me to use the tag.</think>\n\n"
        "hi"
    )


@pytest.mark.asyncio
async def test_build_chat_context_incognito_does_not_duplicate_current_user_message(monkeypatch):
    for mod_name in [
        "starlette.middleware",
        "starlette.middleware.base",
        "core.models",
        "core.database",
        "routes.prefs_routes",
        "routes.research_routes",
        "src.llm_core",
        "src.context_compactor",
        "src.model_context",
        "src.auth_helpers",
    ]:
        if mod_name not in sys.modules:
            monkeypatch.setitem(sys.modules, mod_name, MagicMock())

    chat_helpers = importlib.import_module("routes.chat_helpers")

    async def fake_preprocess(chat_handler, message, att_ids, sess, **kwargs):
        # **kwargs absorbs auto_opened_docs (added when PDF imports auto-create
        # docs) and any other future preprocess kwargs without the test fixture
        # having to be updated each time.
        return chat_helpers.PreprocessedMessage(
            enhanced_message=message,
            user_content=message,
            text_for_context=message,
            youtube_transcripts=[],
            attachment_meta=[],
        )

    def fake_extract_preset(chat_handler, preset_id):
        return chat_helpers.PresetInfo(
            temperature=0.7,
            max_tokens=1024,
            system_prompt=None,
            character_name=None,
        )

    def fake_add_user_message(sess, chat_handler, preprocessed, incognito=False):
        sess.messages.append({"role": "user", "content": preprocessed.user_content})

    async def fake_maybe_compact(sess, endpoint_url, model, messages, headers, owner=None):
        return messages, 123, False

    monkeypatch.setattr(chat_helpers, "preprocess", fake_preprocess)
    monkeypatch.setattr(chat_helpers, "extract_preset", fake_extract_preset)
    monkeypatch.setattr(chat_helpers, "add_user_message", fake_add_user_message)
    monkeypatch.setattr(chat_helpers, "load_prefs_for_user", lambda user: {})
    monkeypatch.setattr(chat_helpers, "effective_user", lambda request: "tester")
    monkeypatch.setattr(chat_helpers, "normalize_model_id", lambda endpoint_url, model, **kwargs: None)
    monkeypatch.setattr(chat_helpers, "maybe_compact", fake_maybe_compact)
    monkeypatch.setattr(chat_helpers, "trim_for_context", lambda messages, context_length: messages)

    sess = SimpleNamespace(
        endpoint_url="http://localhost:8000/v1",
        model="test-model",
        headers={},
        messages=[],
        get_context_messages=lambda: list(sess.messages),
    )
    request = SimpleNamespace()
    chat_handler = SimpleNamespace()
    chat_processor = SimpleNamespace(
        build_context_preface=lambda **kwargs: ([], [], []),
    )

    ctx = await chat_helpers.build_chat_context(
        sess=sess,
        request=request,
        chat_handler=chat_handler,
        chat_processor=chat_processor,
        message="hello",
        session_id="s1",
        incognito=True,
    )

    user_messages = [m for m in ctx.messages if m.get("role") == "user" and m.get("content") == "hello"]
    assert len(user_messages) == 1


@pytest.mark.asyncio
async def test_admin_agent_tools_require_admin(monkeypatch):
    auth_mod = _install_core_auth_stub(monkeypatch)
    from src.tool_execution import execute_tool_block

    class FakeAuth:
        is_configured = True

        def is_admin(self, username):
            return False

    monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())

    for tool_name in ("manage_tokens", "app_api", "serve_preset"):
        desc, result = await execute_tool_block(
            SimpleNamespace(tool_type=tool_name, content='{"action":"create","name":"bad"}'),
            owner="regular-user",
        )

        assert desc == f"{tool_name}: BLOCKED"
        assert result["exit_code"] == 1
        assert "requires an admin" in result["error"]


@pytest.mark.asyncio
async def test_app_api_blocks_shell_routes_before_loopback(monkeypatch):
    import httpx
    from src.tool_implementations import do_app_api

    class UnexpectedAsyncClient:
        def __init__(self, *args, **kwargs):
            raise AssertionError("app_api should block shell routes before loopback")

    monkeypatch.setattr(httpx, "AsyncClient", UnexpectedAsyncClient)

    for path in ("/api/shell/exec", "api/shell/stream"):
        result = await do_app_api(
            json.dumps(
                {
                    "action": "call",
                    "method": "POST",
                    "path": path,
                    "body": {"command": "echo should-not-run"},
                }
            ),
            owner="admin",
        )

        assert result["exit_code"] == 1
        assert "Path blocked for safety" in result["error"]
        assert "Sensitive endpoints" in result["error"]


@pytest.mark.asyncio
async def test_app_api_blocks_cookbook_host_control_routes_before_loopback(monkeypatch):
    import httpx
    from src.tool_implementations import do_app_api

    class UnexpectedAsyncClient:
        def __init__(self, *args, **kwargs):
            raise AssertionError("app_api should block host-control routes before loopback")

    monkeypatch.setattr(httpx, "AsyncClient", UnexpectedAsyncClient)

    blocked_calls = (
        (
            "api/cookbook/packages/install",
            {"pip": "hf_transfer"},
            "package installation is host code execution",
        ),
        (
            "/api/cookbook/rebuild-engine",
            {"engine": "llamacpp"},
            "engine rebuild mutates local or remote host state",
        ),
        (
            "/api/cookbook/kill-pid",
            {"pid": 12345, "signal": "TERM"},
            "process signalling is host control",
        ),
    )

    for path, body, error_text in blocked_calls:
        result = await do_app_api(
            json.dumps(
                {
                    "action": "call",
                    "method": "POST",
                    "path": path,
                    "body": body,
                }
            ),
            owner="admin",
        )

        assert result["exit_code"] == 1
        assert error_text in result["error"]


@pytest.mark.asyncio
async def test_app_api_endpoint_discovery_hides_shell_routes(monkeypatch):
    _install_core_middleware_stub(monkeypatch)
    import httpx
    from src.tool_implementations import do_app_api

    class FakeResponse:
        def json(self):
            return {
                "paths": {
                    "/api/shell/exec": {"post": {"summary": "Execute Shell Command"}},
                    "/api/shell/stream": {"post": {"summary": "Stream Shell Command"}},
                    "/api/auth/settings": {"get": {"summary": "Auth Settings"}},
                    "/api/cookbook/gpus": {"get": {"summary": "List GPUs"}},
                }
            }

    class FakeAsyncClient:
        def __init__(self, *args, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, exc_type, exc, tb):
            return False

        async def get(self, *args, **kwargs):
            return FakeResponse()

    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)

    result = await do_app_api(json.dumps({"action": "endpoints"}), owner="admin")

    assert result["exit_code"] == 0
    paths = {(endpoint["method"], endpoint["path"]) for endpoint in result["endpoints"]}
    assert ("GET", "/api/cookbook/gpus") in paths
    assert ("POST", "/api/shell/exec") not in paths
    assert ("POST", "/api/shell/stream") not in paths
    assert ("GET", "/api/auth/settings") not in paths
    assert all(not endpoint["path"].startswith("/api/shell") for endpoint in result["endpoints"])


@pytest.mark.asyncio
async def test_app_api_endpoint_discovery_hides_cookbook_host_control_routes(monkeypatch):
    _install_core_middleware_stub(monkeypatch)
    import httpx
    from src.tool_implementations import do_app_api

    class FakeResponse:
        def json(self):
            return {
                "paths": {
                    "/api/cookbook/packages": {"get": {"summary": "List Cookbook Packages"}},
                    "/api/cookbook/packages/install": {"post": {"summary": "Install Package"}},
                    "/api/cookbook/rebuild-engine": {"post": {"summary": "Rebuild Engine"}},
                    "/api/cookbook/kill-pid": {"post": {"summary": "Kill Process"}},
                    "/api/cookbook/gpus": {"get": {"summary": "List GPUs"}},
                }
            }

    class FakeAsyncClient:
        def __init__(self, *args, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, exc_type, exc, tb):
            return False

        async def get(self, *args, **kwargs):
            return FakeResponse()

    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)

    result = await do_app_api(json.dumps({"action": "endpoints", "filter": "cookbook"}), owner="admin")

    assert result["exit_code"] == 0
    paths = {(endpoint["method"], endpoint["path"]) for endpoint in result["endpoints"]}
    assert ("GET", "/api/cookbook/packages") in paths
    assert ("GET", "/api/cookbook/gpus") in paths
    assert ("POST", "/api/cookbook/packages/install") not in paths
    assert ("POST", "/api/cookbook/rebuild-engine") not in paths
    assert ("POST", "/api/cookbook/kill-pid") not in paths


@pytest.mark.asyncio
async def test_public_agent_policy_blocks_sensitive_tools(monkeypatch):
    auth_mod = _install_core_auth_stub(monkeypatch)
    from src.tool_execution import execute_tool_block

    class FakeAuth:
        is_configured = True

        def is_admin(self, username):
            return False

    monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())

    for tool_name in ("send_email", "read_file", "mcp__email__send_email"):
        desc, result = await execute_tool_block(
            SimpleNamespace(tool_type=tool_name, content="{}"),
            owner="regular-user",
        )
        assert desc == f"{tool_name}: BLOCKED"
        assert result["exit_code"] == 1
        assert "restricted to admin users" in result["error"]


@pytest.mark.asyncio
async def test_email_mcp_non_object_args_fail_before_dispatch(monkeypatch):
    import src.tool_execution as tool_execution
    from src.tool_execution import execute_tool_block

    class FakeMcp:
        def __init__(self):
            self.calls = []

        async def call_tool(self, name, args):
            self.calls.append((name, args))
            return {"output": "called", "exit_code": 0}

    fake = FakeMcp()
    monkeypatch.setattr(tool_execution, "_owner_is_admin", lambda owner: True)
    monkeypatch.setattr(tool_execution, "get_mcp_manager", lambda: fake)

    desc, result = await execute_tool_block(
        SimpleNamespace(tool_type="mcp__email__list_emails", content='["INBOX"]'),
        owner="alice",
    )

    assert desc == "mcp: mcp__email__list_emails"
    assert result["exit_code"] == 1
    assert "JSON object" in result["error"]
    assert fake.calls == []


@pytest.mark.asyncio
async def test_email_mcp_dispatch_includes_hidden_owner(monkeypatch):
    import src.tool_execution as tool_execution
    from src.tool_execution import execute_tool_block

    class FakeMcp:
        def __init__(self):
            self.calls = []

        async def call_tool(self, name, args):
            self.calls.append((name, args))
            return {"output": "called", "exit_code": 0}

    fake = FakeMcp()
    monkeypatch.setattr(tool_execution, "_owner_is_admin", lambda owner: True)
    monkeypatch.setattr(tool_execution, "get_mcp_manager", lambda: fake)

    desc, result = await execute_tool_block(
        SimpleNamespace(tool_type="mcp__email__list_emails", content='{"folder":"INBOX"}'),
        owner="alice",
    )

    assert desc == "mcp: mcp__email__list_emails"
    assert result["exit_code"] == 0
    assert fake.calls == [
        ("mcp__email__list_emails", {"folder": "INBOX", "_odysseus_owner": "alice"}),
    ]


def test_public_agent_policy_hides_sensitive_tools(monkeypatch):
    auth_mod = _install_core_auth_stub(monkeypatch)
    from src.tool_security import blocked_tools_for_owner

    class FakeAuth:
        is_configured = True

        def is_admin(self, username):
            return False

    monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())

    blocked = blocked_tools_for_owner("regular-user")

    assert "send_email" in blocked
    assert "read_file" in blocked
    assert "app_api" in blocked
    assert "serve_preset" in blocked
    assert "manage_tasks" in blocked


def test_presetup_does_not_grant_admin_tools_when_auth_enabled(monkeypatch):
    """Pre-setup window: auth is enabled but no admin user exists yet.

    This must NOT be treated as single-user/admin at the tool layer — the
    server-execution tools (bash/python) stay blocked as defense-in-depth so
    an unauthenticated caller that slips past the auth middleware (e.g. via a
    loopback bypass) can't reach an RCE before setup completes.
    """
    monkeypatch.delenv("AUTH_ENABLED", raising=False)  # default: enabled
    auth_mod = _install_core_auth_stub(monkeypatch)

    class FakeAuth:
        is_configured = False

        def is_admin(self, username):
            return False

    monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())

    from src.tool_security import (
        blocked_tools_for_owner,
        owner_is_admin_or_single_user,
    )

    assert owner_is_admin_or_single_user(None) is False
    blocked = blocked_tools_for_owner(None)
    assert "bash" in blocked
    assert "python" in blocked


def test_single_user_mode_keeps_full_tool_access_when_auth_disabled(monkeypatch):
    """Intentional single-user mode (AUTH_ENABLED=false) keeps full tool
    access even with no admin user — this is the default local/self-host UX
    and must not regress."""
    monkeypatch.setenv("AUTH_ENABLED", "false")
    auth_mod = _install_core_auth_stub(monkeypatch)

    class FakeAuth:
        is_configured = False

        def is_admin(self, username):
            return False

    monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())

    from src.tool_security import (
        blocked_tools_for_owner,
        owner_is_admin_or_single_user,
    )

    assert owner_is_admin_or_single_user(None) is True
    assert blocked_tools_for_owner(None) == set()


def test_auth_disabled_configured_mode_keeps_full_tool_access(monkeypatch):
    """AUTH_ENABLED=false is still intentional single-user mode after setup.

    Once an admin account exists, AuthManager.is_configured becomes true. The
    tool gate must still honor explicit auth-disabled mode before requiring an
    owner/admin match, otherwise agent mode hides email/MCP/local tools from the
    operator.
    """
    monkeypatch.setenv("AUTH_ENABLED", "false")
    auth_mod = _install_core_auth_stub(monkeypatch)

    class FakeAuth:
        is_configured = True

        def is_admin(self, username):
            return False

    monkeypatch.setattr(auth_mod, "AuthManager", lambda: FakeAuth())

    from src.tool_security import (
        blocked_tools_for_owner,
        owner_is_admin_or_single_user,
    )

    assert owner_is_admin_or_single_user(None) is True
    assert blocked_tools_for_owner(None) == set()


@pytest.mark.asyncio
async def test_webhook_tool_reuses_private_url_validation():
    class FakeDb:
        def close(self):
            pass

    fake_core_db = types.ModuleType("core.database")
    fake_core_db.SessionLocal = lambda: FakeDb()
    fake_core_db.Webhook = object
    fake_src_db = types.ModuleType("src.database")
    fake_src_db.SessionLocal = fake_core_db.SessionLocal
    fake_src_db.Webhook = object
    # Importing do_manage_webhooks below re-executes src.webhook_manager bound to
    # the faked src.database, whose Webhook is plain `object`. Save BOTH the
    # sys.modules entry AND the parent-package attribute (src.webhook_manager) so
    # the real module can be restored afterwards. Without this the polluted
    # module leaks into the cache and breaks sibling tests that call
    # WebhookManager._deliver (which evaluates `Webhook.id == webhook_id`).
    _ABSENT = object()
    _wm_saved_module = sys.modules.get("src.webhook_manager", _ABSENT)
    _src_pkg = sys.modules.get("src")
    _wm_saved_attr = (
        getattr(_src_pkg, "webhook_manager", _ABSENT) if _src_pkg is not None else _ABSENT
    )

    # Drop both bindings so the import re-executes against the fake src.database,
    # still exercising the intended import path.
    sys.modules.pop("src.webhook_manager", None)
    if _src_pkg is not None and hasattr(_src_pkg, "webhook_manager"):
        delattr(_src_pkg, "webhook_manager")

    monkeypatch = pytest.MonkeyPatch()
    monkeypatch.setitem(sys.modules, "core.database", fake_core_db)
    monkeypatch.setitem(sys.modules, "src.database", fake_src_db)

    from src.tool_implementations import do_manage_webhooks

    try:
        result = await do_manage_webhooks(
            '{"action":"add","url":"http://127.0.0.1:8000/hook","events":"chat.completed"}',
            owner="admin",
        )
    finally:
        monkeypatch.undo()
        # Restore src.webhook_manager to its exact pre-test state at BOTH the
        # sys.modules and parent-package attribute level.
        if _wm_saved_module is _ABSENT:
            sys.modules.pop("src.webhook_manager", None)
        else:
            sys.modules["src.webhook_manager"] = _wm_saved_module
        if _src_pkg is not None:
            if _wm_saved_attr is _ABSENT:
                if hasattr(_src_pkg, "webhook_manager"):
                    delattr(_src_pkg, "webhook_manager")
            else:
                setattr(_src_pkg, "webhook_manager", _wm_saved_attr)

    assert result["exit_code"] == 1
    assert "private/internal" in result["error"]


def test_default_chat_skips_hidden_first_model(monkeypatch):
    """get_default_chat picks first visible model when default_model is empty
    and the first cached model is hidden."""
    _install_model_route_import_stubs(monkeypatch)
    import routes.model_routes as model_routes
    import routes.prefs_routes as prefs_routes

    ep = SimpleNamespace(
        id="ep1",
        base_url="http://localhost:11434",
        is_enabled=True,
        owner="fresh",
        cached_models='["hidden-model", "visible-model"]',
        hidden_models='["hidden-model"]',
    )

    monkeypatch.setattr(model_routes, "ModelEndpoint", _FakeModelEndpoint)
    monkeypatch.setattr(model_routes, "SessionLocal", lambda: _FakeDb([ep]))
    monkeypatch.setattr(model_routes, "_load_settings", lambda: {})
    monkeypatch.setattr(model_routes, "owner_filter", lambda q, m, u, **kw: q)
    monkeypatch.setattr(model_routes, "_normalize_base", lambda base: base.rstrip("/"))
    monkeypatch.setattr(model_routes, "build_chat_url", lambda base: f"{base}/chat/completions")
    monkeypatch.setattr(prefs_routes, "_load_for_user", lambda user: {})

    request = SimpleNamespace(
        state=SimpleNamespace(current_user="fresh"),
        app=SimpleNamespace(state=SimpleNamespace(
            auth_manager=SimpleNamespace(is_admin=lambda user: False)
        )),
    )

    result = _default_chat_endpoint()(request)
    assert result["model"] == "visible-model", f"Expected visible-model, got {result['model']!r}"


def test_default_chat_admin_skips_hidden_first_model(monkeypatch):
    """Admin user with global defaults also skips hidden models in fallback."""
    _install_model_route_import_stubs(monkeypatch)
    import routes.model_routes as model_routes

    ep = SimpleNamespace(
        id="ep1",
        base_url="http://localhost:11434",
        is_enabled=True,
        owner=None,
        cached_models='["hidden-model", "visible-model"]',
        hidden_models='["hidden-model"]',
    )

    monkeypatch.setattr(model_routes, "ModelEndpoint", _FakeModelEndpoint)
    monkeypatch.setattr(model_routes, "SessionLocal", lambda: _FakeDb([ep]))
    monkeypatch.setattr(model_routes, "_load_settings", lambda: {})
    monkeypatch.setattr(model_routes, "owner_filter", lambda q, m, u, **kw: q)
    monkeypatch.setattr(model_routes, "_normalize_base", lambda base: base.rstrip("/"))
    monkeypatch.setattr(model_routes, "build_chat_url", lambda base: f"{base}/chat/completions")

    request = SimpleNamespace(
        state=SimpleNamespace(current_user="admin"),
        app=SimpleNamespace(state=SimpleNamespace(
            auth_manager=SimpleNamespace(is_admin=lambda user: True)
        )),
    )

    result = _default_chat_endpoint()(request)
    assert result["model"] == "visible-model"


def test_default_chat_all_models_hidden_returns_empty_model(monkeypatch):
    """When all cached models are hidden, get_default_chat returns model: ''."""
    _install_model_route_import_stubs(monkeypatch)
    import routes.model_routes as model_routes

    ep = SimpleNamespace(
        id="ep1",
        base_url="http://localhost:11434",
        is_enabled=True,
        owner=None,
        cached_models='["hidden-a", "hidden-b"]',
        hidden_models='["hidden-a", "hidden-b"]',
    )

    monkeypatch.setattr(model_routes, "ModelEndpoint", _FakeModelEndpoint)
    monkeypatch.setattr(model_routes, "SessionLocal", lambda: _FakeDb([ep]))
    monkeypatch.setattr(model_routes, "_load_settings", lambda: {})
    monkeypatch.setattr(model_routes, "owner_filter", lambda q, m, u, **kw: q)
    monkeypatch.setattr(model_routes, "_normalize_base", lambda base: base.rstrip("/"))
    monkeypatch.setattr(model_routes, "build_chat_url", lambda base: f"{base}/chat/completions")

    request = SimpleNamespace(
        state=SimpleNamespace(current_user="admin"),
        app=SimpleNamespace(state=SimpleNamespace(
            auth_manager=SimpleNamespace(is_admin=lambda user: True)
        )),
    )

    result = _default_chat_endpoint()(request)
    assert result["model"] == "", f"Expected empty model, got {result['model']!r}"


def test_visible_models_filters_hidden_first(monkeypatch):
    """_visible_models removes hidden models from the list."""
    from routes.model_routes import _visible_models

    result = _visible_models(
        '["hidden-model", "visible-model"]',
        '["hidden-model"]',
    )
    assert result == ["visible-model"]


def test_visible_models_all_hidden_returns_empty(monkeypatch):
    """_visible_models returns [] when all models are hidden."""
    from routes.model_routes import _visible_models

    result = _visible_models(
        '["hidden-a", "hidden-b"]',
        '["hidden-a", "hidden-b"]',
    )
    assert result == []


def test_visible_models_no_hidden_returns_all(monkeypatch):
    """_visible_models returns full list when no hidden_models."""
    from routes.model_routes import _visible_models

    result = _visible_models(
        '["model-a", "model-b"]',
        None,
    )
    assert result == ["model-a", "model-b"]


def test_visible_models_empty_cached_returns_empty(monkeypatch):
    """_visible_models returns [] for empty cached list."""
    from routes.model_routes import _visible_models

    result = _visible_models([], None)
    assert result == []