42668eeeb1
Create/update/response schemas for Project, Experiment, Run, Endpoint, Webhook, Score, Auth (setup/login/token), Export, and Health. All use Pydantic v2 ConfigDict(from_attributes=True) for ORM compatibility. RunDetailResponse nests StageResults and Scores. ExportRunRow provides flat scorer_name→value dict for CSV/JSON export. 30 tests added.
339 lines
9.8 KiB
Python
339 lines
9.8 KiB
Python
"""Tests for backend/schemas.py."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from models import ExperimentStatus, RunStatus
|
|
from schemas import (
|
|
EndpointCreate,
|
|
EndpointResponse,
|
|
EndpointUpdate,
|
|
ExperimentCreate,
|
|
ExperimentResponse,
|
|
ExperimentUpdate,
|
|
ExportResponse,
|
|
ExportRunRow,
|
|
HealthResponse,
|
|
LoginRequest,
|
|
ProjectCreate,
|
|
ProjectResponse,
|
|
ProjectUpdate,
|
|
RunDetailResponse,
|
|
RunResponse,
|
|
ScoreInput,
|
|
ScoreResponse,
|
|
SetupRequest,
|
|
StageResultResponse,
|
|
TokenResponse,
|
|
UserResponse,
|
|
WebhookCreate,
|
|
WebhookResponse,
|
|
WebhookUpdate,
|
|
)
|
|
|
|
|
|
NOW = datetime.now(timezone.utc)
|
|
UUID1 = uuid.uuid4()
|
|
UUID2 = uuid.uuid4()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Project schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProjectSchemas:
|
|
def test_create_valid(self) -> None:
|
|
p = ProjectCreate(name="My Project", description="desc")
|
|
assert p.name == "My Project"
|
|
assert p.description == "desc"
|
|
|
|
def test_create_name_required(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
ProjectCreate() # type: ignore[call-arg]
|
|
|
|
def test_create_empty_name_rejected(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
ProjectCreate(name="")
|
|
|
|
def test_update_partial(self) -> None:
|
|
p = ProjectUpdate(name="New Name")
|
|
assert p.name == "New Name"
|
|
assert p.description is None
|
|
|
|
def test_response_from_attributes(self) -> None:
|
|
class Fake:
|
|
id = UUID1
|
|
name = "Proj"
|
|
description = None
|
|
owner_id = UUID2
|
|
created_at = NOW
|
|
updated_at = NOW
|
|
|
|
r = ProjectResponse.model_validate(Fake())
|
|
assert r.id == UUID1
|
|
assert r.name == "Proj"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Experiment schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExperimentSchemas:
|
|
def test_create_minimal(self) -> None:
|
|
e = ExperimentCreate(name="Exp 1")
|
|
assert e.name == "Exp 1"
|
|
assert e.sample_data is None
|
|
|
|
def test_create_with_all_fields(self) -> None:
|
|
e = ExperimentCreate(
|
|
name="Full",
|
|
description="desc",
|
|
sample_data={"key": "value"},
|
|
pipeline_stages={"stages": []},
|
|
scoring_config={"scorer": "exact"},
|
|
parameter_space={"temp": [0.5, 1.0]},
|
|
)
|
|
assert e.parameter_space == {"temp": [0.5, 1.0]}
|
|
|
|
def test_update_status(self) -> None:
|
|
e = ExperimentUpdate(status=ExperimentStatus.running)
|
|
assert e.status == ExperimentStatus.running
|
|
|
|
def test_response_from_attributes(self) -> None:
|
|
class Fake:
|
|
id = UUID1
|
|
project_id = UUID2
|
|
name = "Exp"
|
|
description = None
|
|
sample_data = None
|
|
pipeline_stages = None
|
|
scoring_config = None
|
|
parameter_space = None
|
|
status = ExperimentStatus.draft
|
|
created_at = NOW
|
|
updated_at = NOW
|
|
|
|
r = ExperimentResponse.model_validate(Fake())
|
|
assert r.status == ExperimentStatus.draft
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRunSchemas:
|
|
def test_response_from_attributes(self) -> None:
|
|
class Fake:
|
|
id = UUID1
|
|
experiment_id = UUID2
|
|
config_hash = "abc123"
|
|
config = {"model": "gpt-4"}
|
|
status = RunStatus.completed
|
|
started_at = NOW
|
|
completed_at = NOW
|
|
duration_ms = 1234
|
|
tokens_in = 100
|
|
tokens_out = 200
|
|
cost_estimate = 0.003
|
|
|
|
r = RunResponse.model_validate(Fake())
|
|
assert r.duration_ms == 1234
|
|
assert r.cost_estimate == 0.003
|
|
|
|
def test_detail_response_nested(self) -> None:
|
|
data = {
|
|
"id": UUID1,
|
|
"experiment_id": UUID2,
|
|
"config_hash": "abc",
|
|
"config": {},
|
|
"status": RunStatus.pending,
|
|
"started_at": None,
|
|
"completed_at": None,
|
|
"duration_ms": None,
|
|
"tokens_in": None,
|
|
"tokens_out": None,
|
|
"cost_estimate": None,
|
|
"stage_results": [],
|
|
"scores": [],
|
|
}
|
|
r = RunDetailResponse(**data)
|
|
assert r.stage_results == []
|
|
assert r.scores == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Score schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestScoreSchemas:
|
|
def test_input_valid(self) -> None:
|
|
s = ScoreInput(scorer_name="exact_match", value=0.95, metadata={"note": "ok"})
|
|
assert s.value == 0.95
|
|
assert s.metadata == {"note": "ok"}
|
|
|
|
def test_input_missing_name(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
ScoreInput(value=0.5) # type: ignore[call-arg]
|
|
|
|
def test_response_from_attributes(self) -> None:
|
|
class Fake:
|
|
id = UUID1
|
|
run_id = UUID2
|
|
scorer_name = "bleu"
|
|
value = 0.8
|
|
scorer_metadata = {"n": 4}
|
|
created_at = NOW
|
|
|
|
r = ScoreResponse.model_validate(Fake())
|
|
assert r.scorer_metadata == {"n": 4}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Endpoint schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEndpointSchemas:
|
|
def test_create_valid(self) -> None:
|
|
e = EndpointCreate(name="OpenAI", url="https://api.openai.com/v1")
|
|
assert e.api_key is None
|
|
|
|
def test_create_empty_name_rejected(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
EndpointCreate(name="", url="https://example.com")
|
|
|
|
def test_update_partial(self) -> None:
|
|
e = EndpointUpdate(url="https://new-url.com")
|
|
assert e.name is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Webhook schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWebhookSchemas:
|
|
def test_create_valid(self) -> None:
|
|
w = WebhookCreate(
|
|
event_type="run.completed",
|
|
url="https://hooks.example.com/promptlooper",
|
|
headers={"Authorization": "Bearer xyz"},
|
|
)
|
|
assert w.is_active is True
|
|
|
|
def test_create_inactive(self) -> None:
|
|
w = WebhookCreate(
|
|
event_type="run.failed",
|
|
url="https://example.com",
|
|
is_active=False,
|
|
)
|
|
assert w.is_active is False
|
|
|
|
def test_update_partial(self) -> None:
|
|
w = WebhookUpdate(is_active=False)
|
|
assert w.event_type is None
|
|
assert w.is_active is False
|
|
|
|
def test_response_from_attributes(self) -> None:
|
|
class Fake:
|
|
id = UUID1
|
|
event_type = "run.completed"
|
|
url = "https://example.com"
|
|
headers = None
|
|
is_active = True
|
|
|
|
r = WebhookResponse.model_validate(Fake())
|
|
assert r.event_type == "run.completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAuthSchemas:
|
|
def test_setup_password_min_length(self) -> None:
|
|
with pytest.raises(ValidationError):
|
|
SetupRequest(username="admin", password="short")
|
|
|
|
def test_setup_valid(self) -> None:
|
|
s = SetupRequest(username="admin", password="securepass123")
|
|
assert s.username == "admin"
|
|
|
|
def test_login_valid(self) -> None:
|
|
l = LoginRequest(username="user", password="pass")
|
|
assert l.username == "user"
|
|
|
|
def test_token_response(self) -> None:
|
|
t = TokenResponse(access_token="jwt.token.here")
|
|
assert t.token_type == "bearer"
|
|
|
|
def test_user_response_from_attributes(self) -> None:
|
|
class Fake:
|
|
id = UUID1
|
|
username = "admin"
|
|
is_admin = True
|
|
created_at = NOW
|
|
|
|
r = UserResponse.model_validate(Fake())
|
|
assert r.is_admin is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Export schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExportSchemas:
|
|
def test_export_run_row(self) -> None:
|
|
row = ExportRunRow(
|
|
run_id=UUID1,
|
|
experiment_id=UUID2,
|
|
config_hash="abc",
|
|
config={"model": "gpt-4"},
|
|
status=RunStatus.completed,
|
|
duration_ms=500,
|
|
tokens_in=10,
|
|
tokens_out=20,
|
|
cost_estimate=0.001,
|
|
scores={"exact_match": 1.0, "bleu": 0.85},
|
|
)
|
|
assert row.scores["bleu"] == 0.85
|
|
|
|
def test_export_run_row_default_scores(self) -> None:
|
|
row = ExportRunRow(
|
|
run_id=UUID1,
|
|
experiment_id=UUID2,
|
|
config_hash="abc",
|
|
config={},
|
|
status=RunStatus.pending,
|
|
)
|
|
assert row.scores == {}
|
|
|
|
def test_export_response(self) -> None:
|
|
r = ExportResponse(
|
|
experiment_id=UUID1,
|
|
experiment_name="Test Exp",
|
|
rows=[],
|
|
)
|
|
assert r.rows == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Health schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHealthSchema:
|
|
def test_health_response(self) -> None:
|
|
h = HealthResponse(database=True, redis=False)
|
|
assert h.status == "ok"
|
|
assert h.database is True
|
|
assert h.redis is False
|