109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
import pytest
|
|
from httpx import AsyncClient, ASGITransport
|
|
from app.main import app
|
|
from app.database import engine, init_db
|
|
from app.models import SQLModel
|
|
from datetime import datetime, timezone, timedelta
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
|
|
@pytest.fixture
|
|
async def setup_db():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.drop_all)
|
|
await conn.run_sync(SQLModel.metadata.create_all)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_daily_summary(setup_db):
|
|
from datetime import date
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
# Create a session spanning today
|
|
start = datetime.now(timezone.utc) - timedelta(hours=5)
|
|
end = datetime.now(timezone.utc)
|
|
# We need to manually insert since the API only allows POST for start_time
|
|
from app.database import get_session
|
|
from app.models import SleepSession
|
|
from sqlalchemy import select
|
|
|
|
async with AsyncSession(engine) as session:
|
|
new_s = SleepSession(start_time=start, end_time=end)
|
|
session.add(new_s)
|
|
await session.commit()
|
|
|
|
from_date = date.today().isoformat()
|
|
# Use the next day as to_date to cover the whole current day
|
|
to_date = (date.today() + timedelta(days=1)).isoformat()
|
|
response = await ac.get(
|
|
f"/api/stats/daily-summary?from_date={from_date}&to_date={to_date}"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["days"]) >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timetable(setup_db):
|
|
from datetime import date
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
from app.database import get_session
|
|
from app.models import SleepSession
|
|
from sqlalchemy import select
|
|
|
|
async with AsyncSession(engine) as session:
|
|
start = datetime.now(timezone.utc) - timedelta(hours=1)
|
|
new_s = SleepSession(start_time=start, end_time=datetime.now(timezone.utc))
|
|
session.add(new_s)
|
|
await session.commit()
|
|
|
|
from_date = date.today().isoformat()
|
|
to_date = (date.today() + timedelta(days=1)).isoformat()
|
|
response = await ac.get(
|
|
f"/api/stats/timetable?from_date={from_date}&to_date={to_date}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_summary(setup_db):
|
|
from datetime import date, timedelta, timezone
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app), base_url="http://test"
|
|
) as ac:
|
|
from app.database import get_session
|
|
from app.models import SleepSession
|
|
from sqlalchemy import select
|
|
|
|
async with AsyncSession(engine) as session:
|
|
# Create two sessions, one a few days ago and one today
|
|
start1 = datetime.now(timezone.utc) - timedelta(days=5)
|
|
end1 = start1 + timedelta(hours=8)
|
|
new_s1 = SleepSession(start_time=start1, end_time=end1)
|
|
session.add(new_s1)
|
|
|
|
start2 = datetime.now(timezone.utc) - timedelta(hours=2)
|
|
end2 = datetime.now(timezone.utc)
|
|
new_s2 = SleepSession(start_time=start2, end_time=end2)
|
|
session.add(new_s2)
|
|
|
|
await session.commit()
|
|
|
|
from_date = (date.today() - timedelta(days=7)).isoformat()
|
|
to_date = date.today().isoformat()
|
|
response = await ac.get(
|
|
f"/api/stats/summary?from_date={from_date}&to_date={to_date}"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total_sleeps"] == 2
|
|
assert data["total_sleeps_all_time"] == 2
|
|
assert data["avg_sleeps_per_day_all_time"] > 0
|
|
|