File size: 1,705 Bytes
856334f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | import pytest
from sqlalchemy import select
from app.db.models import AgentRun, User, BillingEvent
@pytest.mark.asyncio
async def test_create_agent_run(db_session):
run = AgentRun(
agent_id="web_research",
agent_name="Web Research Agent",
query="test query",
status="success",
tokens_used=150,
cost=0.15,
)
db_session.add(run)
await db_session.commit()
stmt = select(AgentRun).where(AgentRun.agent_id == "web_research")
result = await db_session.execute(stmt)
db_run = result.scalar_one()
assert db_run.agent_name == "Web Research Agent"
assert db_run.status == "success"
assert db_run.tokens_used == 150
@pytest.mark.asyncio
async def test_create_user(db_session):
user = User(
email="test@example.com",
name="Test User",
role="admin",
)
db_session.add(user)
await db_session.commit()
stmt = select(User).where(User.email == "test@example.com")
result = await db_session.execute(stmt)
db_user = result.scalar_one()
assert db_user.name == "Test User"
assert db_user.role == "admin"
assert db_user.is_active is True
@pytest.mark.asyncio
async def test_create_billing_event(db_session):
event = BillingEvent(
agent_id="web_research",
amount=150.0,
currency="usd",
event_type="usage",
status="completed",
)
db_session.add(event)
await db_session.commit()
stmt = select(BillingEvent).where(BillingEvent.agent_id == "web_research")
result = await db_session.execute(stmt)
db_event = result.scalar_one()
assert db_event.amount == 150.0
assert db_event.status == "completed"
|