ss-tools/backend/tests/api/test_llm.py

# #region Test.Api.Llm [C:3] [TYPE Module] [SEMANTICS test,llm,routes]
# @BRIEF Tests for LLM API routes — providers, fetch-models, status, test, probe.
# @RELATION BINDS_TO -> [Api.Llm.LlmRoutes]

import os

os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-tests")
os.environ.setdefault("DEV_MODE", "true")

import sys
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

_src = str(Path(__file__).resolve().parent.parent.parent / "src")
if _src not in sys.path:
    sys.path.insert(0, _src)

P = "/api/llm"


def _make_provider(**kw):
    p = MagicMock()
    p.id = kw.get("id", "prov-1")
    p.provider_type = kw.get("provider_type", "openai")
    p.name = kw.get("name", "Test Provider")
    p.base_url = kw.get("base_url", "https://api.openai.com")
    p.api_key = kw.get("api_key", "sk-test")
    p.default_model = kw.get("default_model", "gpt-4")
    p.is_active = kw.get("is_active", True)
    p.is_multimodal = kw.get("is_multimodal", True)
    p.max_images = kw.get("max_images", 10)
    p.context_window = kw.get("context_window", 128000)
    p.max_output_tokens = kw.get("max_output_tokens", 4096)
    # New provider capability fields (NULL = derive/legacy defaults)
    p.throughput_class = kw.get("throughput_class")
    p.reasoning_control = kw.get("reasoning_control")
    p.supports_json_object = kw.get("supports_json_object")
    p.default_llm_concurrency = kw.get("default_llm_concurrency")
    p.max_llm_concurrency = kw.get("max_llm_concurrency")
    return p


def _make_client(overrides: dict | None = None) -> TestClient:
    from src.api.routes.llm import router
    from src.core.database import get_db
    from src.dependencies import get_current_user
    from src.schemas.auth import User, RoleSchema

    app = FastAPI()
    app.include_router(router, prefix="/api/llm")

    mock_user = User(
        id="admin-1", username="admin", email="admin@x.com",
        auth_source="LOCAL",
        created_at=__import__("datetime").datetime.now(),
        roles=[RoleSchema(id="r1", name="Admin", description="", is_admin=True, permissions=[])],
    )

    app.dependency_overrides[get_db] = lambda: MagicMock()
    app.dependency_overrides[get_current_user] = lambda: mock_user
    if overrides:
        for dep, fn in overrides.items():
            app.dependency_overrides[dep] = fn
    return TestClient(app)


class TestGetProviders:
    """GET /api/llm/providers"""

    def test_success(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = [_make_provider()]
        mock_svc.get_decrypted_api_key.return_value = "sk-real-key"

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.get(f"{P}/providers")
        assert resp.status_code == 200

    def test_empty(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = []

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.get(f"{P}/providers")
        assert resp.status_code == 200
        assert resp.json() == []


class TestFetchModels:
    """POST /api/llm/providers/fetch-models"""

    def test_success_with_api_key(self):
        mock_db = MagicMock()
        mock_client = MagicMock()
        mock_client.fetch_models = AsyncMock(return_value=["gpt-4"])

        from src.core.database import get_db
        with patch("src.plugins.llm_analysis.service.LLMClient", return_value=mock_client):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/fetch-models", json={
                "base_url": "https://api.openai.com", "provider_type": "openai", "api_key": "sk-test",
            })
        assert resp.status_code == 200
        assert "gpt-4" in resp.json()["models"]

    def test_success_with_provider_id(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = "sk-decrypted"
        mock_client = MagicMock()
        mock_client.fetch_models = AsyncMock(return_value=["gpt-4"])

        from src.core.database import get_db
        with (
            patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc),
            patch("src.plugins.llm_analysis.service.LLMClient", return_value=mock_client),
        ):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/fetch-models", json={"provider_id": "prov-1"})
        assert resp.status_code == 200

    def test_missing_credentials(self):
        client = _make_client()
        resp = client.post(f"{P}/providers/fetch-models", json={
            "base_url": "https://api.openai.com", "provider_type": "openai",
        })
        assert resp.status_code == 400

    def test_provider_not_found(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = None

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/fetch-models", json={"provider_id": "unknown"})
        assert resp.status_code == 404

    def test_missing_base_url(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider(base_url="")
        mock_svc.get_decrypted_api_key.return_value = "sk-decrypted"

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/fetch-models", json={"provider_id": "prov-1"})
        assert resp.status_code == 400

    def test_api_failure(self):
        mock_db = MagicMock()
        mock_client = MagicMock()
        mock_client.fetch_models = AsyncMock(side_effect=RuntimeError("API error"))

        from src.core.database import get_db
        with patch("src.plugins.llm_analysis.service.LLMClient", return_value=mock_client):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/fetch-models", json={
                "base_url": "https://api.openai.com", "provider_type": "openai", "api_key": "sk-test",
            })
        assert resp.status_code == 502


class TestGetLlmStatus:
    """GET /api/llm/status"""

    def test_configured(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = [_make_provider()]
        mock_svc.get_decrypted_api_key.return_value = "sk-real-valid-key-16chars-min"

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.get(f"{P}/status")
        assert resp.status_code == 200
        assert resp.json()["configured"] is True

    def test_no_providers(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = []

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.get(f"{P}/status")
        assert resp.json()["configured"] is False

    def test_no_active_provider(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = [_make_provider(is_active=False)]

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.get(f"{P}/status")
        assert resp.json()["reason"] == "no_active_provider"

    def test_invalid_api_key(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = [_make_provider()]
        mock_svc.get_decrypted_api_key.return_value = ""

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.get(f"{P}/status")
        assert resp.json()["reason"] == "invalid_api_key"


class TestCreateProvider:
    """POST /api/llm/providers"""

    def test_success(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.create_provider.return_value = _make_provider(id="new-prov")

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers", json={
                "provider_type": "openai", "name": "New", "base_url": "https://x.com",
                "api_key": "sk-new", "default_model": "gpt-4",
            })
        assert resp.status_code == 201


class TestUpdateProvider:
    """PUT /api/llm/providers/{provider_id}"""

    def test_success(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.update_provider.return_value = _make_provider(id="prov-1")
        mock_svc.get_decrypted_api_key.return_value = "sk-decrypted"

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.put(f"{P}/providers/prov-1", json={
                "provider_type": "openai", "name": "Updated", "base_url": "https://x.com",
                "api_key": "sk-updated", "default_model": "gpt-4",
            })
        assert resp.status_code == 200

    def test_not_found(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.update_provider.return_value = None

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.put(f"{P}/providers/unknown", json={
                "provider_type": "openai", "name": "X", "base_url": "https://x.com",
                "api_key": "sk-x", "default_model": "gpt-4",
            })
        assert resp.status_code == 404


class TestDeleteProvider:
    """DELETE /api/llm/providers/{provider_id}"""

    def test_success(self):
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.all.return_value = []
        mock_svc = MagicMock()
        mock_svc.delete_provider.return_value = True

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.delete(f"{P}/providers/prov-1")
        assert resp.status_code == 204

    def test_not_found(self):
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.all.return_value = []
        mock_svc = MagicMock()
        mock_svc.delete_provider.return_value = False

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.delete(f"{P}/providers/unknown")
        assert resp.status_code == 404


class TestLlmTestConnection:
    """POST /api/llm/providers/{provider_id}/test"""

    def test_success(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = "sk-real-key"
        mock_client = MagicMock()
        mock_client.test_runtime_connection = AsyncMock()

        from src.core.database import get_db
        with (
            patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc),
            patch("src.plugins.llm_analysis.service.LLMClient", return_value=mock_client),
        ):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/test")
        assert resp.status_code == 200
        assert resp.json()["success"] is True

    def test_not_found(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = None

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/unknown/test")
        assert resp.status_code == 404

    def test_no_api_key(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = ""

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/test")
        assert resp.status_code == 500

    def test_connection_failure(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = "sk-real-key"
        mock_client = MagicMock()
        mock_client.test_runtime_connection = AsyncMock(side_effect=RuntimeError("fail"))

        from src.core.database import get_db
        with (
            patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc),
            patch("src.plugins.llm_analysis.service.LLMClient", return_value=mock_client),
        ):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/test")
        assert resp.status_code == 200
        assert resp.json()["success"] is False


class TestTestProviderConfig:
    """POST /api/llm/providers/test"""

    def test_success(self):
        mock_client = MagicMock()
        mock_client.test_runtime_connection = AsyncMock()

        with patch("src.plugins.llm_analysis.service.LLMClient", return_value=mock_client):
            client = _make_client()
            resp = client.post(f"{P}/providers/test", json={
                "provider_type": "openai", "name": "T", "base_url": "https://x.com",
                "api_key": "sk-test", "default_model": "gpt-4",
            })
        assert resp.status_code == 200
        assert resp.json()["success"] is True

    def test_masked_api_key_rejected(self):
        client = _make_client()
        resp = client.post(f"{P}/providers/test", json={
            "provider_type": "openai", "name": "T", "base_url": "https://x.com",
            "api_key": "********", "default_model": "gpt-4",
        })
        assert resp.status_code == 400


class TestProbeMaxImages:
    """POST /api/llm/providers/{provider_id}/probe-max-images"""

    def test_no_limit_detected(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = "sk-real"
        mock_openai = MagicMock()
        mock_openai.chat.completions = MagicMock()
        mock_openai.chat.completions.create = AsyncMock()

        from src.core.database import get_db
        with (
            patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc),
            patch("openai.AsyncOpenAI", return_value=mock_openai),
        ):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/probe-max-images")
        assert resp.status_code == 200
        assert resp.json()["method"] == "no_limit_detected"

    def test_provider_not_found(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = None

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/unknown/probe-max-images")
        assert resp.status_code == 404

    def test_no_api_key(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = ""

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/probe-max-images")
        assert resp.status_code == 400

    def test_no_default_model(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider(default_model="")
        mock_svc.get_decrypted_api_key.return_value = "sk-real"

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/probe-max-images")
        assert resp.status_code == 400

    def test_limit_from_error_parse(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = "sk-real"
        mock_openai = MagicMock()
        mock_openai.chat.completions = MagicMock()
        mock_openai.chat.completions.create = AsyncMock(side_effect=RuntimeError("At most 5 image(s) per request"))

        from src.core.database import get_db
        with (
            patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc),
            patch("openai.AsyncOpenAI", return_value=mock_openai),
        ):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/probe-max-images")
        assert resp.status_code == 200
        assert resp.json()["max_images"] == 5
        assert resp.json()["method"] == "parse_error"

    def test_first_image_fails(self):
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_provider.return_value = _make_provider()
        mock_svc.get_decrypted_api_key.return_value = "sk-real"
        mock_openai = MagicMock()
        mock_openai.chat.completions = MagicMock()
        mock_openai.chat.completions.create = AsyncMock(side_effect=RuntimeError("unsupported"))

        from src.core.database import get_db
        with (
            patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc),
            patch("openai.AsyncOpenAI", return_value=mock_openai),
        ):
            client = _make_client({get_db: lambda: mock_db})
            resp = client.post(f"{P}/providers/prov-1/probe-max-images")
        assert resp.status_code == 200
        assert resp.json()["max_images"] == 0


# #region Test.Api.LlmPermissionGate [C:3] [TYPE Test] [SEMANTICS test,llm,rbac,denied]
class TestLlmPermissionGate:
    """LLM provider surface is gated on admin:settings — authentication alone is not enough (RBAC-01)."""

    def _non_admin_client(self) -> TestClient:
        from datetime import datetime

        from src.dependencies import get_current_user
        from src.schemas.auth import User

        non_admin = User(
            id="user-1", username="regular", email="u@x.com", auth_source="LOCAL",
            created_at=datetime.now(), roles=[],
        )
        return _make_client({get_current_user: lambda: non_admin})

    def test_get_providers_denied_without_read_permission(self):
        client = self._non_admin_client()
        resp = client.get(f"{P}/providers")
        assert resp.status_code == 403

    def test_get_status_denied_without_read_permission(self):
        client = self._non_admin_client()
        resp = client.get(f"{P}/status")
        assert resp.status_code == 403

    def test_delete_provider_denied_without_write_permission(self):
        client = self._non_admin_client()
        resp = client.delete(f"{P}/providers/prov-1")
        assert resp.status_code == 403

    def test_test_connection_denied_without_write_permission(self):
        client = self._non_admin_client()
        resp = client.post(f"{P}/providers/prov-1/test")
        assert resp.status_code == 403

    # ── Orthogonal edge cases ──

    def _user_client(self, roles) -> TestClient:
        from datetime import datetime

        from src.dependencies import get_current_user
        from src.schemas.auth import User

        user = User(
            id="user-rbac", username="rbac", email="rbac@x.com", auth_source="LOCAL",
            created_at=datetime.now(), roles=roles,
        )
        return _make_client({get_current_user: lambda: user})

    def test_admin_flag_bypass_without_explicit_permission(self):
        """Boundary: is_admin=true bypasses the explicit admin:settings permission requirement."""
        from src.schemas.auth import RoleSchema

        client = self._user_client([
            RoleSchema(id="r", name="Admin", description="", is_admin=True, permissions=[]),
        ])
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = []

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            resp = client.get(f"{P}/providers", )
        assert resp.status_code == 200

    def test_read_only_permission_allows_get_but_not_delete(self):
        """Boundary: admin:settings READ grants reads but the same user cannot mutate."""
        from src.schemas.auth import PermissionSchema, RoleSchema

        read_only = RoleSchema(
            id="r", name="SettingsReader", description="", is_admin=False,
            permissions=[PermissionSchema(id="p", resource="admin:settings", action="READ")],
        )
        client = self._user_client([read_only])
        mock_db = MagicMock()
        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = []

        from src.core.database import get_db
        with patch("src.api.routes.llm.LLMProviderService", return_value=mock_svc):
            get_resp = client.get(f"{P}/providers")
        assert get_resp.status_code == 200
        assert client.delete(f"{P}/providers/prov-1").status_code == 403

    def test_unrelated_permission_denied(self):
        """Boundary: holding an unrelated plugin permission does not open the LLM surface."""
        from src.schemas.auth import PermissionSchema, RoleSchema

        other = RoleSchema(
            id="r", name="StorageWriter", description="", is_admin=False,
            permissions=[PermissionSchema(id="p", resource="plugin:storage", action="WRITE")],
        )
        client = self._user_client([other])
        assert client.get(f"{P}/providers").status_code == 403
# #endregion Test.Api.LlmPermissionGate
# #endregion Test.Api.Llm