AgentEvalTool/tests/unit/test_intelligent_eval_model.py

"""Smoke tests for intelligent eval data model (ticket 01)."""

import pytest
from agenteval.intelligent_eval.models import (
    IntelligentEval,
    IntelligentEvalMessage,
    IntelligentEvalSession,
    IntelligentEvalSessionStatus,
    IntelligentEvalStatus,
)
from agenteval.intelligent_eval.repository import (
    IntelligentEvalMessageRepository,
    IntelligentEvalRepository,
    IntelligentEvalSessionRepository,
)
from agenteval.storage.db import EvalTargetDB
from sqlalchemy.pool import StaticPool
from sqlmodel import Session, SQLModel, create_engine


@pytest.fixture
def db_session():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    SQLModel.metadata.create_all(engine)
    session = Session(engine)
    target = EvalTargetDB(id="t1", name="test-target")
    session.add(target)
    session.commit()
    yield session
    session.close()


class TestIntelligentEvalRepository:
    def test_create_and_get(self, db_session):
        repo = IntelligentEvalRepository(db_session)
        ev = repo.create(IntelligentEval(
            name="test-eval",
            target_id="t1",
            goal="evaluate customer service",
            seeds={"personas": [], "goals": []},
            intent="test intent",
            role_description="impatient customer",
            time_window_hours=24,
        ))
        assert ev.id is not None
        assert ev.status == IntelligentEvalStatus.DRAFT
        assert ev.name == "test-eval"
        assert ev.time_window_hours == 24

        fetched = repo.get(ev.id)
        assert fetched is not None
        assert fetched.goal == "evaluate customer service"
        assert fetched.seeds == {"personas": [], "goals": []}

    def test_list_all(self, db_session):
        repo = IntelligentEvalRepository(db_session)
        repo.create(IntelligentEval(name="eval-1", target_id="t1"))
        repo.create(IntelligentEval(name="eval-2", target_id="t1"))
        all_evals = repo.list_all()
        assert len(all_evals) == 2

    def test_status_transitions(self, db_session):
        repo = IntelligentEvalRepository(db_session)
        ev = repo.create(IntelligentEval(name="eval", target_id="t1"))
        assert ev.status == IntelligentEvalStatus.DRAFT

        ev = repo.transition_status(ev.id, IntelligentEvalStatus.PLANNING)
        assert ev.status == IntelligentEvalStatus.PLANNING

        ev = repo.transition_status(ev.id, IntelligentEvalStatus.PENDING_APPROVAL)
        assert ev.status == IntelligentEvalStatus.PENDING_APPROVAL

        ev = repo.transition_status(ev.id, IntelligentEvalStatus.EXECUTING)
        assert ev.status == IntelligentEvalStatus.EXECUTING
        assert ev.started_at is not None

        ev = repo.transition_status(ev.id, IntelligentEvalStatus.COMPLETED)
        assert ev.status == IntelligentEvalStatus.COMPLETED
        assert ev.completed_at is not None

    def test_plan_and_report_json(self, db_session):
        repo = IntelligentEvalRepository(db_session)
        ev = repo.create(IntelligentEval(name="eval", target_id="t1"))

        plan = {"dimensions": ["退货"], "virtual_users": [], "estimated_sessions": 3}
        ev.plan = plan
        ev.status = IntelligentEvalStatus.PENDING_APPROVAL
        ev = repo.update(ev)
        assert ev.plan == plan

        report = {"summary": "good", "findings": []}
        ev.report = report
        ev = repo.update(ev)
        assert ev.report == report


class TestIntelligentEvalSessionRepository:
    def test_create_and_list(self, db_session):
        eval_repo = IntelligentEvalRepository(db_session)
        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))

        sess_repo = IntelligentEvalSessionRepository(db_session)
        sess = sess_repo.create(IntelligentEvalSession(
            eval_id=ev.id,
            target_id="t1",
            persona={"name": "user1", "patience": "low"},
            goal="complete return",
            dimension="退货流程",
        ))
        assert sess.id is not None
        assert sess.status == IntelligentEvalSessionStatus.RUNNING
        assert sess.persona == {"name": "user1", "patience": "low"}

        sessions = sess_repo.list_by_eval(ev.id)
        assert len(sessions) == 1

    def test_close_session(self, db_session):
        eval_repo = IntelligentEvalRepository(db_session)
        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))

        sess_repo = IntelligentEvalSessionRepository(db_session)
        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))

        verdict = {"goal_achieved": True, "issues": []}
        closed = sess_repo.close(sess.id, verdict)
        assert closed.status == IntelligentEvalSessionStatus.COMPLETED
        assert closed.verdict == verdict
        assert closed.closed_at is not None

    def test_increment_turns(self, db_session):
        eval_repo = IntelligentEvalRepository(db_session)
        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))

        sess_repo = IntelligentEvalSessionRepository(db_session)
        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))
        assert sess.turn_count == 0

        sess_repo.increment_turns(sess.id)
        sess_repo.increment_turns(sess.id)
        updated = sess_repo.get(sess.id)
        assert updated.turn_count == 2


class TestIntelligentEvalMessageRepository:
    def test_create_and_list(self, db_session):
        eval_repo = IntelligentEvalRepository(db_session)
        ev = eval_repo.create(IntelligentEval(name="eval", target_id="t1"))

        sess_repo = IntelligentEvalSessionRepository(db_session)
        sess = sess_repo.create(IntelligentEvalSession(eval_id=ev.id, target_id="t1"))

        msg_repo = IntelligentEvalMessageRepository(db_session)
        msg_repo.create(IntelligentEvalMessage(session_id=sess.id, role="user", content="hello"))
        msg_repo.create(IntelligentEvalMessage(session_id=sess.id, role="assistant", content="hi", latency_ms=120))

        messages = msg_repo.list_by_session(sess.id)
        assert len(messages) == 2
        assert messages[0].role == "user"
        assert messages[1].role == "assistant"
        assert messages[1].latency_ms == 120