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

# #region Test.Api.AgentConversations [C:3] [TYPE Module] [SEMANTICS test,agent,conversations,routes]
# @BRIEF Tests for agent conversation API routes — list, save, history, delete, active, llm-config.
# @RELATION BINDS_TO -> [AgentChat.Api.Conversations]

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 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)


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

    app = FastAPI()
    app.include_router(router)
    app.include_router(agent_router)

    _NOW = __import__("datetime").datetime.now()
    mock_user = User(
        id="user-1", username="testuser", email="test@x.com",
        auth_source="LOCAL",
        created_at=_NOW,
        roles=[RoleSchema(id="r1", name="User", description="", permissions=[])],
    )

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


# ── list_conversations ──

class TestListConversations:
    """GET /api/assistant/conversations"""

    def test_success(self):
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_db.query.return_value = mock_query
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 2
        _NOW = __import__("datetime").datetime.now()
        mock_msg = MagicMock()
        mock_msg.role = "user"
        mock_msg.tool_calls = []
        mock_msg.state = None
        mock_msg.text = "Hello"
        mock_conv = MagicMock()
        mock_conv.id = "conv-1"
        mock_conv.title = "Test Conversation"
        mock_conv.updated_at = _NOW
        mock_conv.messages = [mock_msg]
        mock_query.order_by.return_value.offset.return_value.limit.return_value.all.return_value = [mock_conv, mock_conv]

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.get("/api/assistant/conversations")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data["items"]) == 2
        assert data["items"][0]["id"] == "conv-1"

    def test_with_search(self):
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_db.query.return_value = mock_query
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 0
        mock_query.order_by.return_value.offset.return_value.limit.return_value.all.return_value = []

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.get("/api/assistant/conversations?search=test&include_archived=true")
        assert resp.status_code == 200

    def test_empty(self):
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_db.query.return_value = mock_query
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 0
        mock_query.order_by.return_value.offset.return_value.limit.return_value.all.return_value = []

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.get("/api/assistant/conversations")
        assert resp.status_code == 200
        assert resp.json()["items"] == []


# ── save_conversation ──

class TestSaveConversation:
    """POST /api/agent/conversations/save"""

    SAVE_PAYLOAD = {
        "conversation_id": "conv-1",
        "title": "New Chat",
        "messages": [
            {"id": "msg-1", "role": "user", "text": "Hello"},
            {"id": "msg-2", "role": "assistant", "text": "Hi there", "tool_calls": [{"name": "search"}]},
        ],
    }

    def test_create_new(self):
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.first.return_value = None

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.post("/api/agent/conversations/save", json=self.SAVE_PAYLOAD)
        assert resp.status_code == 200
        data = resp.json()
        assert data["saved"] is True
        assert data["conversation_id"] == "conv-1"

    def test_update_existing(self):
        existing_conv = MagicMock()
        existing_conv.updated_at = __import__("datetime").datetime.now()
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.first.side_effect = [existing_conv, None, None]

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.post("/api/agent/conversations/save", json=self.SAVE_PAYLOAD)
        assert resp.status_code == 200

    def test_skip_empty_message_id(self):
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.first.return_value = None

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.post("/api/agent/conversations/save", json={
            "conversation_id": "conv-1",
            "messages": [{"id": "", "role": "user", "text": "skip me"}],
        })
        assert resp.status_code == 200

    def test_skip_duplicate_message(self):
        existing_msg = MagicMock()
        existing_msg.id = "msg-2"
        mock_db = MagicMock()
        # 3 .first() calls: 1 for conv lookup → None, msg-1 lookup → None, msg-2 lookup → existing_msg
        mock_db.query.return_value.filter.return_value.first.side_effect = [None, None, existing_msg]

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.post("/api/agent/conversations/save", json=self.SAVE_PAYLOAD)
        assert resp.status_code == 200


# ── get_history ──

class TestGetHistory:
    """GET /api/assistant/history"""

    def test_success(self):
        mock_db = MagicMock()
        _NOW = __import__("datetime").datetime.now()
        mock_conv = MagicMock()
        mock_conv.id = "conv-1"
        mock_msg = MagicMock()
        mock_msg.id = "msg-1"
        mock_msg.conversation_id = "conv-1"
        mock_msg.role = "user"
        mock_msg.text = "Hello"
        mock_msg.tool_calls = None
        mock_msg.attachments = None
        mock_msg.created_at = _NOW
        mock_conv.messages = [mock_msg]
        mock_db.query.return_value.filter.return_value.first.return_value = mock_conv

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.get("/api/assistant/history?conversation_id=conv-1")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data["items"]) == 1
        assert data["items"][0]["text"] == "Hello"

    def test_not_found(self):
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.first.return_value = None

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.get("/api/assistant/history?conversation_id=unknown")
        assert resp.status_code == 404


# ── delete_conversation ──

class TestDeleteConversation:
    """DELETE /api/assistant/conversations/{conversation_id}"""

    def test_success(self):
        mock_db = MagicMock()
        mock_conv = MagicMock()
        mock_conv.is_archived = False
        mock_db.query.return_value.filter.return_value.first.return_value = mock_conv

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.delete("/api/assistant/conversations/conv-1")
        assert resp.status_code == 200
        assert resp.json()["deleted"] is True
        assert mock_conv.is_archived is True

    def test_not_found(self):
        mock_db = MagicMock()
        mock_db.query.return_value.filter.return_value.first.return_value = None

        from src.core.database import get_db
        client = _make_client({get_db: lambda: mock_db})
        resp = client.delete("/api/assistant/conversations/unknown")
        assert resp.status_code == 404


# ── check_active_session ──

class TestCheckActiveSession:
    """GET /api/agent/conversations/active"""

    def test_returns_false(self):
        client = _make_client()
        resp = client.get("/api/agent/conversations/active")
        assert resp.status_code == 200
        assert resp.json()["active"] is False


# ── get_agent_llm_config ──

class TestGetAgentLlmConfig:
    """GET /api/agent/llm-config"""

    def test_success_with_preferred(self):
        mock_db = MagicMock()
        mock_config_mgr = MagicMock()
        mock_config = MagicMock()
        mock_config.settings.llm = {"assistant_planner_provider": "prov-1"}
        mock_config_mgr.get_config.return_value = mock_config

        mock_svc = MagicMock()
        mock_provider = MagicMock()
        mock_provider.id = "prov-1"
        mock_provider.provider_type = "openai"
        mock_provider.base_url = "https://api.openai.com"
        mock_provider.default_model = "gpt-4"
        mock_provider.name = "GPT-4 Provider"
        mock_provider.is_active = True
        mock_svc.get_all_providers.return_value = [mock_provider]
        mock_svc.get_decrypted_api_key.return_value = "sk-real-key"

        from src.core.database import get_db
        from src.dependencies import get_config_manager
        with patch("src.api.routes.agent_conversations.LLMProviderService", return_value=mock_svc):
            client = _make_client({
                get_db: lambda: mock_db,
                get_config_manager: lambda: mock_config_mgr,
            })
            resp = client.get("/api/agent/llm-config")
        assert resp.status_code == 200
        data = resp.json()
        assert data["configured"] is True
        assert data["provider_type"] == "openai"

    def test_fallback_to_active(self):
        mock_db = MagicMock()
        mock_config_mgr = MagicMock()
        mock_config = MagicMock()
        mock_config.settings.llm = {}
        mock_config_mgr.get_config.return_value = mock_config

        mock_svc = MagicMock()
        mock_provider = MagicMock()
        mock_provider.id = "prov-2"
        mock_provider.provider_type = "anthropic"
        mock_provider.base_url = ""
        mock_provider.default_model = "claude-3"
        mock_provider.name = "Claude"
        mock_provider.is_active = True
        mock_svc.get_all_providers.return_value = [mock_provider]
        mock_svc.get_decrypted_api_key.return_value = "sk-anthropic"

        from src.core.database import get_db
        from src.dependencies import get_config_manager
        with patch("src.api.routes.agent_conversations.LLMProviderService", return_value=mock_svc):
            client = _make_client({
                get_db: lambda: mock_db,
                get_config_manager: lambda: mock_config_mgr,
            })
            resp = client.get("/api/agent/llm-config")
        assert resp.status_code == 200
        assert resp.json()["configured"] is True

    def test_no_active_provider(self):
        mock_db = MagicMock()
        mock_config_mgr = MagicMock()
        mock_config = MagicMock()
        mock_config.settings.llm = {}
        mock_config_mgr.get_config.return_value = mock_config

        mock_svc = MagicMock()
        mock_svc.get_all_providers.return_value = []

        from src.core.database import get_db
        from src.dependencies import get_config_manager
        with patch("src.api.routes.agent_conversations.LLMProviderService", return_value=mock_svc):
            client = _make_client({
                get_db: lambda: mock_db,
                get_config_manager: lambda: mock_config_mgr,
            })
            resp = client.get("/api/agent/llm-config")
        assert resp.status_code == 200
        assert resp.json()["configured"] is False
        assert resp.json()["reason"] == "no_active_provider"

    def test_invalid_api_key(self):
        mock_db = MagicMock()
        mock_config_mgr = MagicMock()
        mock_config = MagicMock()
        mock_config.settings.llm = {}
        mock_config_mgr.get_config.return_value = mock_config

        mock_svc = MagicMock()
        mock_provider = MagicMock()
        mock_provider.id = "prov-1"
        mock_provider.is_active = True
        mock_svc.get_all_providers.return_value = [mock_provider]
        mock_svc.get_decrypted_api_key.return_value = ""

        from src.core.database import get_db
        from src.dependencies import get_config_manager
        with patch("src.api.routes.agent_conversations.LLMProviderService", return_value=mock_svc):
            client = _make_client({
                get_db: lambda: mock_db,
                get_config_manager: lambda: mock_config_mgr,
            })
            resp = client.get("/api/agent/llm-config")
        assert resp.status_code == 200
        assert resp.json()["configured"] is False
        assert resp.json()["reason"] == "invalid_api_key"
# #endregion Test.Api.AgentConversations