233 lines
9.2 KiB
Python
233 lines
9.2 KiB
Python
# #region Test.Assistant.AdminRoutes [C:2] [TYPE Module] [SEMANTICS test,assistant,admin,route,audit,conversation]
|
|
# @BRIEF Tests for _admin_routes.py — admin route handlers for assistant.
|
|
# @RELATION BINDS_TO -> [AssistantAdminRoutes]
|
|
# @TEST_EDGE: list_conversations -> Returns empty (deprecated)
|
|
# @TEST_EDGE: delete_conversation -> Deletes from cache and DB
|
|
# @TEST_EDGE: delete_conversation_not_found -> 404 on missing
|
|
# @TEST_EDGE: get_history -> Returns paginated messages
|
|
# @TEST_EDGE: get_assistant_audit -> Returns audit records
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_global_state():
|
|
"""Clean global in-memory stores before each test."""
|
|
from src.api.routes.assistant._schemas import ASSISTANT_AUDIT, CONVERSATIONS, USER_ACTIVE_CONVERSATION
|
|
ASSISTANT_AUDIT.clear()
|
|
CONVERSATIONS.clear()
|
|
USER_ACTIVE_CONVERSATION.clear()
|
|
|
|
|
|
class TestListConversations:
|
|
"""list_conversations — deprecated endpoint."""
|
|
|
|
def test_returns_empty(self):
|
|
from src.api.routes.assistant._admin_routes import list_conversations
|
|
|
|
result = __import__("asyncio").run(list_conversations(
|
|
page=1, page_size=20, include_archived=False, archived_only=False,
|
|
search=None, current_user=None, db=MagicMock(),
|
|
))
|
|
assert result["items"] == []
|
|
assert result["total"] == 0
|
|
assert result["page"] == 1
|
|
|
|
|
|
class TestDeleteConversation:
|
|
pytestmark = pytest.mark.skip(reason="delete_conversation removed during 035 refactoring — needs test rewrite for new API")
|
|
"""delete_conversation — conversation deletion."""
|
|
|
|
def test_delete_success(self):
|
|
from src.api.routes.assistant._schemas import CONVERSATIONS
|
|
from src.api.routes.assistant._admin_routes import delete_conversation
|
|
from src.schemas.auth import User
|
|
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True,
|
|
auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
db = MagicMock()
|
|
|
|
# Add to in-memory cache
|
|
CONVERSATIONS[("user-1", "conv-1")] = [{"role": "user", "text": "hello"}]
|
|
|
|
# Build query chain where .delete() returns int
|
|
q = MagicMock()
|
|
q.filter.return_value = q
|
|
q.delete.return_value = 5
|
|
db.query.return_value = q
|
|
|
|
result = __import__("asyncio").run(delete_conversation(
|
|
"conv-1", current_user=user, db=db,
|
|
))
|
|
assert result["status"] == "success"
|
|
assert result["deleted"] == 5
|
|
assert ("user-1", "conv-1") not in CONVERSATIONS
|
|
|
|
def test_delete_not_found_raises_404(self):
|
|
from src.api.routes.assistant._admin_routes import delete_conversation
|
|
from src.schemas.auth import User
|
|
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True,
|
|
auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
db = MagicMock()
|
|
|
|
q = MagicMock()
|
|
q.filter.return_value = q
|
|
q.delete.return_value = 0
|
|
db.query.return_value = q
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(delete_conversation(
|
|
"conv-nonexistent", current_user=user, db=db,
|
|
))
|
|
assert exc.value.status_code == 404
|
|
|
|
def test_delete_unknown_user_no_cache_entry(self):
|
|
from src.api.routes.assistant._admin_routes import delete_conversation
|
|
from src.schemas.auth import User
|
|
|
|
user = User(id="user-other", username="test", email="test@x.com", is_active=True,
|
|
auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
db = MagicMock()
|
|
|
|
q = MagicMock()
|
|
q.filter.return_value = q
|
|
q.delete.return_value = 0
|
|
db.query.return_value = q
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
__import__("asyncio").run(delete_conversation(
|
|
"conv-1", current_user=user, db=db,
|
|
))
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
class TestGetHistory:
|
|
pytestmark = pytest.mark.skip(reason="get_history removed during 035 refactoring — needs test rewrite for new list_conversations API")
|
|
"""get_history — conversation history retrieval."""
|
|
|
|
def _make_mock_row(self, msg_id="msg-1", conv_id="conv-1", role="user", text="Hello",
|
|
state="sent", task_id=None, confirmation_id=None, payload=None):
|
|
from src.models.assistant import AssistantMessageRecord
|
|
row = MagicMock(spec=AssistantMessageRecord)
|
|
row.id = msg_id
|
|
row.user_id = "user-1"
|
|
row.conversation_id = conv_id
|
|
row.role = role
|
|
row.text = text
|
|
row.state = state
|
|
row.task_id = task_id
|
|
row.confirmation_id = confirmation_id
|
|
row.payload = payload
|
|
row.created_at = datetime.now(UTC)
|
|
return row
|
|
|
|
def test_get_history_basic(self):
|
|
from src.api.routes.assistant._admin_routes import get_history
|
|
from src.schemas.auth import User
|
|
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True,
|
|
auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
db = MagicMock()
|
|
|
|
# Build query mock chain that returns real values
|
|
q = MagicMock()
|
|
# .filter().filter() -> q
|
|
q.filter.return_value = q
|
|
q.count.return_value = 1
|
|
q.order_by.return_value = q
|
|
q.offset.return_value = q
|
|
q.limit.return_value = q
|
|
q.all.return_value = [self._make_mock_row()]
|
|
db.query.return_value = q
|
|
|
|
with patch("src.api.routes.assistant._admin_routes._resolve_or_create_conversation") as mock_resolve, \
|
|
patch("src.api.routes.assistant._admin_routes._cleanup_history_ttl") as mock_cleanup:
|
|
mock_resolve.return_value = "conv-1"
|
|
|
|
result = __import__("asyncio").run(get_history(
|
|
page=1, page_size=20, conversation_id="conv-1",
|
|
from_latest=False, current_user=user, db=db,
|
|
))
|
|
assert result["total"] == 1
|
|
assert len(result["items"]) == 1
|
|
assert result["items"][0]["message_id"] == "msg-1"
|
|
assert result["conversation_id"] == "conv-1"
|
|
|
|
def test_get_history_from_latest(self):
|
|
from src.api.routes.assistant._admin_routes import get_history
|
|
from src.schemas.auth import User
|
|
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True,
|
|
auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
db = MagicMock()
|
|
|
|
q = MagicMock()
|
|
q.filter.return_value = q
|
|
q.count.return_value = 1
|
|
q.order_by.return_value = q
|
|
q.offset.return_value = q
|
|
q.limit.return_value = q
|
|
q.all.return_value = [self._make_mock_row(role="assistant", text="Response", state="success", task_id="task-1")]
|
|
db.query.return_value = q
|
|
|
|
with patch("src.api.routes.assistant._admin_routes._resolve_or_create_conversation") as mock_resolve, \
|
|
patch("src.api.routes.assistant._admin_routes._cleanup_history_ttl") as mock_cleanup:
|
|
mock_resolve.return_value = "conv-1"
|
|
|
|
result = __import__("asyncio").run(get_history(
|
|
page=1, page_size=20, conversation_id="conv-1",
|
|
from_latest=True, current_user=user, db=db,
|
|
))
|
|
assert result["total"] == 1
|
|
assert result["from_latest"] is True
|
|
|
|
|
|
class TestGetAssistantAudit:
|
|
"""get_assistant_audit — audit record retrieval."""
|
|
|
|
def _make_mock_row(self):
|
|
from src.models.assistant import AssistantAuditRecord
|
|
row = MagicMock(spec=AssistantAuditRecord)
|
|
row.id = "audit-1"
|
|
row.user_id = "user-1"
|
|
row.conversation_id = "conv-1"
|
|
row.decision = "executed"
|
|
row.task_id = "task-1"
|
|
row.message = "test command"
|
|
row.payload = {"key": "val"}
|
|
row.created_at = datetime.now(UTC)
|
|
return row
|
|
|
|
def test_get_audit(self):
|
|
from src.api.routes.assistant._admin_routes import get_assistant_audit
|
|
from src.schemas.auth import User
|
|
|
|
user = User(id="user-1", username="test", email="test@x.com", is_active=True,
|
|
auth_source="internal", created_at=datetime.now(UTC), last_login=datetime.now(UTC), roles=[])
|
|
db = MagicMock()
|
|
|
|
q = MagicMock()
|
|
q.filter.return_value = q
|
|
q.order_by.return_value = q
|
|
q.limit.return_value = q
|
|
q.all.return_value = [self._make_mock_row()]
|
|
db.query.return_value = q
|
|
|
|
result = __import__("asyncio").run(get_assistant_audit(
|
|
limit=50, current_user=user, db=db, _=None,
|
|
))
|
|
assert result["total"] == 1
|
|
assert len(result["items"]) == 1
|
|
assert result["items"][0]["id"] == "audit-1"
|
|
assert result["memory_total"] == 0
|
|
# #endregion Test.Assistant.AdminRoutes
|