308 lines
10 KiB
Python
308 lines
10 KiB
Python
"""Tests for Pydantic models validation."""
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
from app.models.auth import LoginRequest, UserResponse, LoginResponse
|
||
from app.models.query import QuestionRequest, BenchQueryRequest, BackendQueryRequest, QueryResponse
|
||
from app.models.settings import EnvironmentSettings, EnvironmentSettingsUpdate, UserSettingsUpdate
|
||
from app.models.analysis import SessionCreate, SessionUpdate
|
||
|
||
|
||
class TestAuthModels:
|
||
"""Tests for authentication models."""
|
||
|
||
def test_login_request_valid(self):
|
||
"""Test valid LoginRequest."""
|
||
request = LoginRequest(
|
||
login="12345678",
|
||
client_ip="192.168.1.1"
|
||
)
|
||
|
||
assert request.login == "12345678"
|
||
assert request.client_ip == "192.168.1.1"
|
||
|
||
def test_login_request_invalid_format(self):
|
||
"""Test LoginRequest with invalid login format."""
|
||
|
||
with pytest.raises(ValidationError):
|
||
LoginRequest(login="1234567", client_ip="192.168.1.1")
|
||
|
||
|
||
with pytest.raises(ValidationError):
|
||
LoginRequest(login="abcd1234", client_ip="192.168.1.1")
|
||
|
||
def test_user_response(self):
|
||
"""Test UserResponse model."""
|
||
user = UserResponse(
|
||
user_id="user-123",
|
||
login="12345678",
|
||
last_login_at="2024-01-01T00:00:00Z",
|
||
created_at="2024-01-01T00:00:00Z"
|
||
)
|
||
|
||
assert user.user_id == "user-123"
|
||
assert user.login == "12345678"
|
||
|
||
def test_login_response(self):
|
||
"""Test LoginResponse model."""
|
||
user = UserResponse(
|
||
user_id="user-123",
|
||
login="12345678",
|
||
last_login_at="2024-01-01T00:00:00Z",
|
||
created_at="2024-01-01T00:00:00Z"
|
||
)
|
||
|
||
response = LoginResponse(
|
||
access_token="token123",
|
||
token_type="bearer",
|
||
user=user
|
||
)
|
||
|
||
assert response.access_token == "token123"
|
||
assert response.token_type == "bearer"
|
||
assert response.user.user_id == "user-123"
|
||
|
||
|
||
class TestQueryModels:
|
||
"""Tests for query models."""
|
||
|
||
def test_question_request_valid(self):
|
||
"""Test valid QuestionRequest."""
|
||
question = QuestionRequest(
|
||
body="What is the weather?",
|
||
with_docs=True
|
||
)
|
||
|
||
assert question.body == "What is the weather?"
|
||
assert question.with_docs is True
|
||
|
||
def test_question_request_default_with_docs(self):
|
||
"""Test QuestionRequest with default with_docs."""
|
||
question = QuestionRequest(body="Test question")
|
||
|
||
assert question.with_docs is True
|
||
|
||
def test_bench_query_request_valid(self):
|
||
"""Test valid BenchQueryRequest."""
|
||
request = BenchQueryRequest(
|
||
environment="ift",
|
||
questions=[
|
||
QuestionRequest(body="Q1", with_docs=True),
|
||
QuestionRequest(body="Q2", with_docs=False)
|
||
]
|
||
)
|
||
|
||
assert request.environment == "ift"
|
||
assert len(request.questions) == 2
|
||
assert request.questions[0].body == "Q1"
|
||
|
||
def test_backend_query_request_valid(self):
|
||
"""Test valid BackendQueryRequest."""
|
||
request = BackendQueryRequest(
|
||
environment="psi",
|
||
questions=[
|
||
QuestionRequest(body="Q1", with_docs=True)
|
||
],
|
||
reset_session=True
|
||
)
|
||
|
||
assert request.environment == "psi"
|
||
assert len(request.questions) == 1
|
||
assert request.reset_session is True
|
||
|
||
def test_backend_query_request_default_reset(self):
|
||
"""Test BackendQueryRequest with default reset_session."""
|
||
request = BackendQueryRequest(
|
||
environment="prod",
|
||
questions=[QuestionRequest(body="Q1")]
|
||
)
|
||
|
||
assert request.reset_session is True
|
||
|
||
def test_query_response(self):
|
||
"""Test QueryResponse model."""
|
||
# Test with RagResponseBenchList (parsed from dict)
|
||
response = QueryResponse(
|
||
request_id="req-123",
|
||
timestamp="2024-01-01T00:00:00Z",
|
||
environment="ift",
|
||
response={"answers": []} # Auto-parsed to RagResponseBenchList
|
||
)
|
||
|
||
assert response.request_id == "req-123"
|
||
assert response.environment == "ift"
|
||
# When dict with "answers" is provided, it's parsed as RagResponseBenchList
|
||
from app.models.query import RagResponseBenchList
|
||
assert isinstance(response.response, RagResponseBenchList)
|
||
|
||
def test_query_response_with_dict(self):
|
||
"""Test QueryResponse model with plain dict (backend mode)."""
|
||
# Use dict that doesn't match RagResponseBenchList schema
|
||
response = QueryResponse(
|
||
request_id="req-456",
|
||
timestamp="2024-01-01T00:00:00Z",
|
||
environment="psi",
|
||
response={"result": "some data", "status": "ok"}
|
||
)
|
||
|
||
assert response.request_id == "req-456"
|
||
assert response.environment == "psi"
|
||
assert isinstance(response.response, dict)
|
||
assert response.response["status"] == "ok"
|
||
|
||
|
||
class TestSettingsModels:
|
||
"""Tests for settings models."""
|
||
|
||
def test_environment_settings_valid(self):
|
||
"""Test valid EnvironmentSettings."""
|
||
settings = EnvironmentSettings(
|
||
apiMode="bench",
|
||
bearerToken="token123",
|
||
systemPlatform="platform",
|
||
systemPlatformUser="user",
|
||
platformUserId="user-123",
|
||
platformId="platform-123",
|
||
withClassify=True,
|
||
resetSessionMode=False
|
||
)
|
||
|
||
assert settings.apiMode == "bench"
|
||
assert settings.bearerToken == "token123"
|
||
assert settings.withClassify is True
|
||
assert settings.resetSessionMode is False
|
||
|
||
def test_environment_settings_defaults(self):
|
||
"""Test EnvironmentSettings with default values (nullable fields)."""
|
||
settings = EnvironmentSettings()
|
||
|
||
assert settings.apiMode == "bench"
|
||
assert settings.bearerToken is None
|
||
assert settings.systemPlatform is None
|
||
assert settings.systemPlatformUser is None
|
||
assert settings.platformUserId is None
|
||
assert settings.platformId is None
|
||
assert settings.withClassify is False
|
||
assert settings.resetSessionMode is True
|
||
|
||
def test_environment_settings_nullable_fields(self):
|
||
"""Test EnvironmentSettings with explicit None values."""
|
||
settings = EnvironmentSettings(
|
||
apiMode="backend",
|
||
bearerToken=None,
|
||
systemPlatform=None,
|
||
withClassify=True
|
||
)
|
||
|
||
assert settings.apiMode == "backend"
|
||
assert settings.bearerToken is None
|
||
assert settings.systemPlatform is None
|
||
assert settings.withClassify is True
|
||
|
||
def test_environment_settings_update_partial(self):
|
||
"""Test EnvironmentSettingsUpdate for partial updates."""
|
||
update = EnvironmentSettingsUpdate(
|
||
bearerToken="new-token",
|
||
withClassify=True
|
||
)
|
||
|
||
assert update.bearerToken == "new-token"
|
||
assert update.withClassify is True
|
||
assert update.apiMode is None # Not provided, should be None
|
||
assert update.systemPlatform is None
|
||
|
||
def test_environment_settings_update_all_none(self):
|
||
"""Test EnvironmentSettingsUpdate with all fields as None."""
|
||
update = EnvironmentSettingsUpdate()
|
||
|
||
assert update.apiMode is None
|
||
assert update.bearerToken is None
|
||
assert update.systemPlatform is None
|
||
assert update.withClassify is None
|
||
|
||
def test_user_settings_update(self):
|
||
"""Test UserSettingsUpdate model with partial updates."""
|
||
update = UserSettingsUpdate(
|
||
settings={
|
||
"ift": EnvironmentSettingsUpdate(apiMode="bench", withClassify=True),
|
||
"psi": EnvironmentSettingsUpdate(bearerToken="token123")
|
||
}
|
||
)
|
||
|
||
assert "ift" in update.settings
|
||
assert "psi" in update.settings
|
||
assert update.settings["ift"].apiMode == "bench"
|
||
assert update.settings["ift"].withClassify is True
|
||
assert update.settings["psi"].bearerToken == "token123"
|
||
assert update.settings["psi"].apiMode is None # Not provided
|
||
|
||
def test_user_settings_update_empty(self):
|
||
"""Test UserSettingsUpdate with empty settings."""
|
||
update = UserSettingsUpdate(settings={})
|
||
|
||
assert update.settings == {}
|
||
|
||
|
||
class TestAnalysisModels:
|
||
"""Tests for analysis session models."""
|
||
|
||
def test_session_create_valid(self):
|
||
"""Test valid SessionCreate."""
|
||
session = SessionCreate(
|
||
environment="ift",
|
||
api_mode="bench",
|
||
request=[{"body": "question 1", "with_docs": True}],
|
||
response={"answers": ["answer 1"]},
|
||
annotations={"0": {"rating": "correct"}}
|
||
)
|
||
|
||
assert session.environment == "ift"
|
||
assert session.api_mode == "bench"
|
||
assert len(session.request) == 1
|
||
assert session.annotations == {"0": {"rating": "correct"}}
|
||
|
||
def test_session_create_default_annotations(self):
|
||
"""Test SessionCreate with default empty annotations."""
|
||
session = SessionCreate(
|
||
environment="psi",
|
||
api_mode="backend",
|
||
request=[],
|
||
response={}
|
||
)
|
||
|
||
assert session.annotations == {}
|
||
|
||
def test_session_update_valid(self):
|
||
"""Test valid SessionUpdate."""
|
||
update = SessionUpdate(
|
||
annotations={
|
||
"0": {
|
||
"overall": {
|
||
"rating": "incorrect",
|
||
"comment": "Ответ неполный"
|
||
},
|
||
"body_research": {
|
||
"issues": ["missing_info"],
|
||
"comment": "Не указаны процентные ставки"
|
||
}
|
||
},
|
||
"1": {
|
||
"overall": {
|
||
"rating": "correct",
|
||
"comment": "Ответ корректный"
|
||
}
|
||
}
|
||
}
|
||
)
|
||
|
||
assert "0" in update.annotations
|
||
assert "1" in update.annotations
|
||
assert update.annotations["0"]["overall"]["rating"] == "incorrect"
|
||
assert update.annotations["1"]["overall"]["rating"] == "correct"
|
||
|
||
def test_session_update_empty_annotations(self):
|
||
"""Test SessionUpdate with empty annotations."""
|
||
update = SessionUpdate(annotations={})
|
||
|
||
assert update.annotations == {}
|