busya
ce0369ae5c
Subagents delivered tests across all uncovered backend modules: Schemas (100%): agent, auth, health, profile, settings, validation Services (98-100%): auth, profile, health, llm, mapping, resource, security, git, superset_lookup, sql_table_extractor, rbac API routes (new): auth, admin, health, environments, plugins, dashboards (helpers, projection, actions, listing), git (config, deps, env, helpers) Clean Release (100%): DTO, facade, policy_engine, stages, repos, preparation, source_isolation, compliance Git services: base, remote_providers Agent module: app, run, middleware, langgraph_setup Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching Reports: normalizer, report_service, type_profiles Notifications: service, providers Also: - .gitignore: add .coverage, *.cover, coverage-* dirs - src/schemas/auth.py: fix AD group DN regex (comma in CN=...) - Remove co-located src/services/__tests__/ (caused pytest module collision)
267 lines
9.3 KiB
Python
267 lines
9.3 KiB
Python
# #region Test.Schemas.Agent [C:2] [TYPE Module] [SEMANTICS test,schema,agent]
|
|
# @BRIEF Tests for schemas/agent.py — ConversationItem, ConversationListResponse, ToolCall,
|
|
# AttachmentMeta, MessageItem, HistoryResponse, DeleteResponse, ServiceTokenRequest,
|
|
# ServiceTokenResponse, SaveConversationRequest.
|
|
# @RELATION BINDS_TO -> [Schemas.Agent]
|
|
# @TEST_EDGE: missing_field -> optional fields default correctly
|
|
# @TEST_EDGE: invalid_type -> Pydantic validation rejects bad types
|
|
# @TEST_EDGE: serialization -> round-trip via model_dump/model_validate
|
|
|
|
from datetime import datetime, timezone
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from src.schemas.agent import (
|
|
AttachmentMeta,
|
|
ConversationItem,
|
|
ConversationListResponse,
|
|
DeleteResponse,
|
|
HistoryResponse,
|
|
MessageItem,
|
|
SaveConversationRequest,
|
|
ServiceTokenRequest,
|
|
ServiceTokenResponse,
|
|
ToolCall,
|
|
)
|
|
|
|
|
|
class TestConversationItem:
|
|
"""ConversationItem — required fields and defaults."""
|
|
|
|
def test_minimal(self):
|
|
ts = datetime.now(timezone.utc)
|
|
item = ConversationItem(id="c1", title="Chat", updated_at=ts, message_count=5)
|
|
assert item.id == "c1"
|
|
assert item.title == "Chat"
|
|
assert item.message_count == 5
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
item = ConversationItem(id="c1", title="Chat", updated_at=ts, message_count=5)
|
|
data = item.model_dump()
|
|
restored = ConversationItem.model_validate(data)
|
|
assert restored.id == item.id
|
|
assert restored.title == item.title
|
|
|
|
def test_missing_required_raises(self):
|
|
with pytest.raises(ValidationError):
|
|
ConversationItem()
|
|
|
|
|
|
class TestConversationListResponse:
|
|
"""ConversationListResponse — items list and optional counters."""
|
|
|
|
def test_defaults(self):
|
|
resp = ConversationListResponse(items=[])
|
|
assert resp.items == []
|
|
assert resp.has_next is False
|
|
assert resp.active_total == 0
|
|
assert resp.archived_total == 0
|
|
|
|
def test_with_items(self):
|
|
ts = datetime.now(timezone.utc)
|
|
items = [ConversationItem(id="c1", title="A", updated_at=ts, message_count=1)]
|
|
resp = ConversationListResponse(items=items, has_next=True, active_total=10, archived_total=2)
|
|
assert len(resp.items) == 1
|
|
assert resp.has_next is True
|
|
assert resp.active_total == 10
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
items = [ConversationItem(id="c1", title="A", updated_at=ts, message_count=1)]
|
|
resp = ConversationListResponse(items=items, has_next=True)
|
|
data = resp.model_dump()
|
|
restored = ConversationListResponse.model_validate(data)
|
|
assert restored.has_next is True
|
|
assert len(restored.items) == 1
|
|
|
|
|
|
class TestToolCall:
|
|
"""ToolCall — tool name, input/output/error, status defaults."""
|
|
|
|
def test_defaults(self):
|
|
tc = ToolCall(tool="test_tool")
|
|
assert tc.tool == "test_tool"
|
|
assert tc.input == {}
|
|
assert tc.output is None
|
|
assert tc.error is None
|
|
assert tc.status == "executing"
|
|
|
|
def test_with_all_fields(self):
|
|
tc = ToolCall(
|
|
tool="query", input={"q": "1"}, output={"result": "ok"},
|
|
error=None, status="completed",
|
|
)
|
|
assert tc.tool == "query"
|
|
assert tc.output == {"result": "ok"}
|
|
|
|
def test_serialize_roundtrip(self):
|
|
tc = ToolCall(tool="calc", input={"expr": "2+2"}, status="completed")
|
|
data = tc.model_dump()
|
|
restored = ToolCall.model_validate(data)
|
|
assert restored.tool == "calc"
|
|
assert restored.input == {"expr": "2+2"}
|
|
|
|
|
|
class TestAttachmentMeta:
|
|
"""AttachmentMeta — name, type, size, optional preview_url."""
|
|
|
|
def test_minimal(self):
|
|
am = AttachmentMeta(name="file.pdf", type="pdf", size=1024)
|
|
assert am.name == "file.pdf"
|
|
assert am.type == "pdf"
|
|
assert am.size == 1024
|
|
assert am.preview_url is None
|
|
|
|
def test_with_preview(self):
|
|
am = AttachmentMeta(name="img.png", type="png", size=2048, preview_url="/preview/img.png")
|
|
assert am.preview_url == "/preview/img.png"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
am = AttachmentMeta(name="data.csv", type="csv", size=512)
|
|
data = am.model_dump()
|
|
restored = AttachmentMeta.model_validate(data)
|
|
assert restored.name == "data.csv"
|
|
|
|
|
|
class TestMessageItem:
|
|
"""MessageItem — all fields including nested ToolCall and AttachmentMeta."""
|
|
|
|
def test_minimal(self):
|
|
ts = datetime.now(timezone.utc)
|
|
msg = MessageItem(id="m1", conversation_id="c1", role="user", created_at=ts)
|
|
assert msg.text is None
|
|
assert msg.state is None
|
|
assert msg.tool_calls is None
|
|
assert msg.attachments is None
|
|
|
|
def test_with_nested_objects(self):
|
|
ts = datetime.now(timezone.utc)
|
|
tc = ToolCall(tool="search", status="completed")
|
|
am = AttachmentMeta(name="doc.txt", type="txt", size=100)
|
|
msg = MessageItem(
|
|
id="m1", conversation_id="c1", role="assistant",
|
|
text="Here are results", state="done",
|
|
tool_calls=[tc], attachments=[am], created_at=ts,
|
|
)
|
|
assert len(msg.tool_calls) == 1
|
|
assert len(msg.attachments) == 1
|
|
assert msg.tool_calls[0].tool == "search"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
msg = MessageItem(
|
|
id="m1", conversation_id="c1", role="user",
|
|
text="hello", created_at=ts,
|
|
)
|
|
data = msg.model_dump()
|
|
restored = MessageItem.model_validate(data)
|
|
assert restored.text == "hello"
|
|
|
|
|
|
class TestHistoryResponse:
|
|
"""HistoryResponse — items list and pagination fields."""
|
|
|
|
def test_defaults(self):
|
|
resp = HistoryResponse(items=[])
|
|
assert resp.items == []
|
|
assert resp.has_next is False
|
|
assert resp.conversation_id is None
|
|
|
|
def test_with_conversation_id(self):
|
|
resp = HistoryResponse(items=[], conversation_id="c1")
|
|
assert resp.conversation_id == "c1"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
items = [MessageItem(id="m1", conversation_id="c1", role="user", created_at=ts)]
|
|
resp = HistoryResponse(items=items, has_next=True)
|
|
data = resp.model_dump()
|
|
restored = HistoryResponse.model_validate(data)
|
|
assert restored.has_next is True
|
|
|
|
|
|
class TestDeleteResponse:
|
|
"""DeleteResponse — just a deleted flag defaulting True."""
|
|
|
|
def test_default(self):
|
|
resp = DeleteResponse()
|
|
assert resp.deleted is True
|
|
|
|
def test_false(self):
|
|
resp = DeleteResponse(deleted=False)
|
|
assert resp.deleted is False
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = DeleteResponse()
|
|
data = resp.model_dump()
|
|
restored = DeleteResponse.model_validate(data)
|
|
assert restored.deleted is True
|
|
|
|
|
|
class TestServiceTokenRequest:
|
|
"""ServiceTokenRequest — just service_secret."""
|
|
|
|
def test_required(self):
|
|
req = ServiceTokenRequest(service_secret="s3cr3t")
|
|
assert req.service_secret == "s3cr3t"
|
|
|
|
def test_missing_raises(self):
|
|
with pytest.raises(ValidationError):
|
|
ServiceTokenRequest()
|
|
|
|
def test_serialize_roundtrip(self):
|
|
req = ServiceTokenRequest(service_secret="key")
|
|
data = req.model_dump()
|
|
restored = ServiceTokenRequest.model_validate(data)
|
|
assert restored.service_secret == "key"
|
|
|
|
|
|
class TestServiceTokenResponse:
|
|
"""ServiceTokenResponse — token response with defaults."""
|
|
|
|
def test_defaults(self):
|
|
resp = ServiceTokenResponse(access_token="jwt")
|
|
assert resp.access_token == "jwt"
|
|
assert resp.token_type == "bearer"
|
|
assert resp.expires_in == 86400
|
|
assert resp.role == "agent"
|
|
|
|
def test_custom_values(self):
|
|
resp = ServiceTokenResponse(access_token="tok", token_type="Bearer", expires_in=3600, role="admin")
|
|
assert resp.expires_in == 3600
|
|
assert resp.role == "admin"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
resp = ServiceTokenResponse(access_token="jwt-token")
|
|
data = resp.model_dump()
|
|
restored = ServiceTokenResponse.model_validate(data)
|
|
assert restored.access_token == "jwt-token"
|
|
|
|
|
|
class TestSaveConversationRequest:
|
|
"""SaveConversationRequest — conversation save payload."""
|
|
|
|
def test_defaults(self):
|
|
req = SaveConversationRequest(conversation_id="c1")
|
|
assert req.title == ""
|
|
assert req.user_id == "admin"
|
|
assert req.messages == []
|
|
|
|
def test_with_messages(self):
|
|
req = SaveConversationRequest(conversation_id="c1", title="Chat", user_id="u1", messages=[{"role": "user", "text": "hi"}])
|
|
assert req.title == "Chat"
|
|
assert len(req.messages) == 1
|
|
|
|
def test_missing_required_raises(self):
|
|
with pytest.raises(ValidationError):
|
|
SaveConversationRequest()
|
|
|
|
def test_serialize_roundtrip(self):
|
|
req = SaveConversationRequest(conversation_id="c1", title="Test", messages=[{"role": "user", "text": "hello"}])
|
|
data = req.model_dump()
|
|
restored = SaveConversationRequest.model_validate(data)
|
|
assert restored.title == "Test"
|
|
assert restored.messages == [{"role": "user", "text": "hello"}]
|
|
# #endregion Test.Schemas.Agent
|