test: upgrade backend test suite, implement unit tests, and modernize CRUD patterns
Build Backend / build (push) Successful in 6m39s

This commit is contained in:
2026-06-08 16:17:11 +02:00
parent 4d53887fcc
commit a9e1e96622
13 changed files with 759 additions and 25 deletions
+13 -8
View File
@@ -5,15 +5,16 @@ from sqlmodel import SQLModel
from sqlalchemy.orm import sessionmaker
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import get_session
@pytest_asyncio.fixture(scope="session")
@pytest_asyncio.fixture(scope="function")
async def engine():
"""Create a single in-memory SQLite engine for the entire test session."""
"""Create a fresh in-memory SQLite engine for each test."""
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=False,
)
# Create tables once for the whole session to avoid overhead
# Create tables for each test to ensure a clean state
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
yield engine
@@ -26,10 +27,10 @@ async def session_factory(engine):
@pytest_asyncio.fixture
async def db_session(session_factory):
"""Create a database session for a test. Rolls back after each test."""
"""Create a database session for a test."""
async with session_factory() as session:
yield session
await session.rollback()
# Rollback is still good practice, though engine is fresh each time
@pytest.fixture
def sample_session_data():
@@ -38,10 +39,14 @@ def sample_session_data():
"start_time": "2024-01-15T21:30:00Z",
}
@pytest.fixture
async def api_client(engine):
"""Provides an AsyncClient for testing API endpoints."""
@pytest_asyncio.fixture
async def api_client(db_session):
"""Provides an AsyncClient for testing API endpoints with the test database session."""
async def override():
yield db_session
app.dependency_overrides[get_session] = override
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
yield client
app.dependency_overrides.clear()
+31
View File
@@ -1,4 +1,35 @@
import pytest
from datetime import datetime, timedelta, timezone
from app.models import SleepSession
from sqlalchemy import select
@pytest.mark.asyncio
async def test_summary_double_counting_midnight(api_client, db_session):
# 1. Setup: Insert a sleep session that spans across midnight
start_time = datetime(2024, 1, 1, 23, 0)
end_time = datetime(2024, 1, 2, 7, 0)
new_session = SleepSession(start_time=start_time, end_time=end_time)
db_session.add(new_session)
await db_session.commit()
# 2. Call the stats endpoint for a period that includes this session
from_date = "2024-01-01T00:00:00"
to_date = "2024-01-02T23:59:59"
async with api_client as ac:
response = await ac.get(f"/api/stats/summary?from_date={from_date}&to_date={to_date}")
assert response.status_code == 200
data = response.json()
# 3. Verify that total_sleeps is exactly 1 for the requested period
print(f"DEBUG: data={data}")
assert data["total_sleeps"] == 1, f"Expected 1 sleep session, got {data['total_sleeps']}"
# 4. Verify that all-time stats also count this session exactly once
assert data["total_sleeps_all_time"] == 1, f"Expected 1 all-time sleep session, got {data['total_sleeps_all_time']}"
from datetime import datetime, timedelta, timezone
from app.models import SleepSession
from sqlmodel.ext.asyncio.session import AsyncSession
+37
View File
@@ -0,0 +1,37 @@
import pytest
from app.config import Settings
def test_settings_defaults():
"""Test that default settings are correctly loaded."""
# We instantiate a new Settings object to test the class directly,
# rather than using the already-initialized 'settings' instance in app.config.
s = Settings(_env_file=None)
assert s.DATABASE_URL == "sqlite+aiosqlite:///./data/sleep.db"
assert s.CORS_ORIGINS == "http://localhost:5173"
assert s.LOG_LEVEL == "info"
def test_settings_env_overrides(monkeypatch):
"""Test that environment variables correctly override defaults."""
# Set environment variables using monkeypatch
monkeypatch.setenv("DATABASE_URL", "postgresql+asyncpg://testuser:testpass@localhost/testdb")
monkeypatch.setenv("CORS_ORIGINS", "http://example.com")
monkeypatch.setenv("LOG_LEVEL", "debug")
# Re-instantiate Settings to pick up the new environment variables
s = Settings(_env_file=None)
assert s.DATABASE_URL == "postgresql+asyncpg://testuser:testpass@localhost/testdb"
assert s.CORS_ORIGINS == "http://example.com"
assert s.LOG_LEVEL == "debug"
def test_settings_partial_override(monkeypatch):
"""Test that overriding only some environment variables works correctly."""
monkeypatch.setenv("LOG_LEVEL", "warning")
s = Settings(_env_file=None)
# Overridden value
assert s.LOG_LEVEL == "warning"
# Default values should remain
assert s.DATABASE_URL == "sqlite+aiosqlite:///./data/sleep.db"
assert s.CORS_ORIGINS == "http://localhost:5173"
+177
View File
@@ -0,0 +1,177 @@
import pytest
from datetime import datetime, timedelta, timezone
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select
from app.models import SleepSession
from app.crud import (
create_sleep_session,
get_sessions,
get_active_session,
end_sleep_session,
delete_session,
update_session,
get_sleep_counts_by_start_date,
)
@pytest.mark.asyncio
async def test_create_sleep_session(db_session: AsyncSession):
# Use naive datetime to match DB storage behavior in tests
start_time = (datetime.now(timezone.utc) - timedelta(hours=8)).replace(tzinfo=None)
new_session = await create_sleep_session(db_session, start_time)
assert new_session.id is not None
assert new_session.start_time == start_time
# Verify in DB
statement = select(SleepSession).where(SleepSession.id == new_session.id)
result = await db_session.execute(statement)
db_session_record = result.scalar_one()
assert db_session_record.id == new_session.id
assert db_session_record.start_time == start_time
@pytest.mark.asyncio
async def test_get_sessions(db_session: AsyncSession):
# Create multiple sessions (using naive datetimes)
now = datetime.now(timezone.utc).replace(tzinfo=None)
s1_start = now - timedelta(days=2)
s2_start = now - timedelta(days=1)
s3_start = now
await create_sleep_session(db_session, s1_start)
await create_sleep_session(db_session, s2_start)
await create_sleep_session(db_session, s3_start)
# Test all sessions
sessions, total = await get_sessions(db_session)
assert total == 3
assert len(sessions) == 3
# Order is descending start_time
assert sessions[0].start_time == s3_start
assert sessions[2].start_time == s1_start
# Test filtering by from_date
sessions, total = await get_sessions(db_session, from_date=now - timedelta(days=1.5))
assert total == 2
assert len(sessions) == 2
assert sessions[0].start_time == s3_start
# Test filtering by to_date
sessions, total = await get_sessions(db_session, to_date=now - timedelta(days=1.5))
assert total == 1
assert len(sessions) == 1
assert sessions[0].start_time == s1_start
@pytest.mark.asyncio
async def test_get_active_session(db_session: AsyncSession):
now = datetime.now(timezone.utc).replace(tzinfo=None)
# Active session (no end_time)
active = await create_sleep_session(db_session, now - timedelta(hours=1))
# Completed session
completed = await create_sleep_session(db_session, now - timedelta(hours=10))
await end_sleep_session(db_session, completed.id, now - timedelta(hours=5))
retrieved_active = await get_active_session(db_session)
assert retrieved_active is not None
assert retrieved_active.id == active.id
assert retrieved_active.end_time is None
@pytest.mark.asyncio
async def test_end_sleep_session(db_session: AsyncSession):
now = datetime.now(timezone.utc).replace(tzinfo=None)
start_time = now - timedelta(hours=8)
session = await create_sleep_session(db_session, start_time)
# Happy path
end_time = now - timedelta(hours=1)
result = await end_sleep_session(db_session, session.id, end_time)
assert isinstance(result, SleepSession)
assert result.end_time == end_time
# Verify in DB
statement = select(SleepSession).where(SleepSession.id == session.id)
result_db = await db_session.execute(statement)
db_record = result_db.scalar_one()
assert db_record.end_time == end_time
# Already ended
second_end = now + timedelta(hours=1)
result_already_ended = await end_sleep_session(db_session, session.id, second_end)
assert result_already_ended is False
# Non-existent ID
result_not_found = await end_sleep_session(db_session, 9999, now)
assert result_not_found is None
# Invalid time (end < start)
new_session = await create_sleep_session(db_session, now)
invalid_end = now - timedelta(hours=1)
result_invalid = await end_sleep_session(db_session, new_session.id, invalid_end)
assert result_invalid == "invalid"
@pytest.mark.asyncio
async def test_delete_session(db_session: AsyncSession):
now = datetime.now(timezone.utc).replace(tzinfo=None)
session = await create_sleep_session(db_session, now)
session_id = session.id
# Happy path
deleted = await delete_session(db_session, session_id)
assert deleted is True
# Verify it's gone
statement = select(SleepSession).where(SleepSession.id == session_id)
result = await db_session.execute(statement)
assert result.first() is None
# Non-existent ID
deleted_again = await delete_session(db_session, session_id)
assert deleted_again is False
@pytest.mark.asyncio
async def test_update_session(db_session: AsyncSession):
now = datetime.now(timezone.utc).replace(tzinfo=None)
session = await create_sleep_session(db_session, now)
# Happy path: update end_time
new_end_time = now + timedelta(hours=2)
update_data = {"end_time": new_end_time}
updated_session = await update_session(db_session, session.id, update_data)
assert updated_session is not None
assert updated_session.end_time == new_end_time
# Verify in DB
statement = select(SleepSession).where(SleepSession.id == session.id)
result = await db_session.execute(statement)
db_record = result.scalar_one()
assert db_record.end_time == new_end_time
# Non-existent ID
result_not_found = await update_session(db_session, 9999, {"start_time": now})
assert result_not_found is None
@pytest.mark.asyncio
async def test_get_sleep_counts_by_start_date(db_session: AsyncSession):
# Use fixed dates to be safe for testing grouping
date1 = datetime(2024, 6, 1, 22, 0)
date2 = datetime(2024, 6, 1, 23, 0) # Same day
date3 = datetime(2024, 6, 2, 1, 0) # Next day
date4 = datetime(2024, 6, 3, 1, 0) # Day after
await create_sleep_session(db_session, date1)
await create_sleep_session(db_session, date2)
await create_sleep_session(db_session, date3)
await create_sleep_session(db_session, date4)
# Get counts. Note: func.date returns strings in SQLite like '2024-06-01'
counts = await get_sleep_counts_by_start_date(db_session)
# Expected: [('2024-06-03', 1), ('2024-06-02', 1), ('2024-06-01', 2)]
assert len(counts) == 3
# Finding counts by scanning because order/format might vary slightly depending on sqlite func.date behavior
# but it should be descending in our code (order_by(func.date(...).desc()))
assert counts[0][0] == '2024-06-03' and counts[0][1] == 1
assert counts[1][0] == '2024-06-02' and counts[1][1] == 1
assert counts[2][0] == '2024-06-01' and counts[2][1] == 2
+29
View File
@@ -0,0 +1,29 @@
import pytest
from unittest.mock import patch
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text
from sqlmodel import SQLModel
from app.database import init_db
@pytest.mark.asyncio
async def test_init_db_creates_tables():
"""Verifies that init_db() successfully creates all necessary tables.
This test uses a patched in-memory SQLite engine to ensure isolation
and verify the database initialization logic.
"""
# Use an in-memory SQLite engine for testing to avoid any impact on real data
test_engine = create_async_engine("sqlite+aiosqlite:///:memory:")
# Patch the engine in app.database so init_db uses our test engine
with patch("app.database.engine", test_engine):
await init_db()
# Check if the expected table "sleep_session" exists in the schema
async with test_engine.connect() as conn:
result = await conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name='sleep_session';"))
row = result.fetchone()
assert row is not None, "Table 'sleep_session' was not created by init_db()"
assert row[0] == "sleep_session"
await test_engine.dispose()
+183
View File
@@ -0,0 +1,183 @@
from datetime import datetime, timedelta
from typing import List
import pytest
from pydantic import ValidationError
from app.schemas import (
SleepSessionCreate,
SleepSessionUpdate,
SleepSessionRead,
SessionListResponse,
DailySummaryDay,
DailySummaryResponse,
TimetableSlot,
TimetableResponse,
WakeupBedtimeDay,
WakeupBedtimeResponse,
SummaryResponse,
)
def test_sleep_session_create_valid():
start_time = datetime.now()
data = {"start_time": start_time}
schema = SleepSessionCreate(**data)
assert schema.start_time == start_time
def test_sleep_session_create_invalid_missing_field():
with pytest.raises(ValidationError):
SleepSessionCreate()
def test_sleep_session_create_invalid_type():
with pytest.raises(ValidationError):
SleepSessionCreate(start_time="not-a-date")
def test_sleep_session_update_valid():
start_time = datetime.now()
data = {"start_time": start_time}
schema = SleepSessionUpdate(**data)
assert schema.start_time == start_time
assert schema.end_time is None
def test_sleep_session_update_valid_partial():
data = {"end_time": datetime.now()}
schema = SleepSessionUpdate(**data)
assert schema.end_time is not None
assert schema.start_time is None
def test_sleep_session_read_valid():
now = datetime.now()
data = {
"id": 1,
"start_time": now,
"end_time": now + timedelta(hours=8),
"created_at": now,
"updated_at": now,
}
schema = SleepSessionRead(**data)
assert schema.id == 1
assert schema.start_time == now
assert schema.end_time == now + timedelta(hours=8)
def test_sleep_session_read_invalid_id():
now = datetime.now()
data = {
"id": "not-an-int",
"start_time": now,
"end_time": None,
"created_at": now,
"updated_at": now,
}
with pytest.raises(ValidationError):
SleepSessionRead(**data)
def test_session_list_response_valid():
now = datetime.now()
sessions = [
{
"id": 1,
"start_time": now,
"end_time": None,
"created_at": now,
"updated_at": now,
}
]
data = {"sessions": sessions, "total": 1}
schema = SessionListResponse(**data)
assert len(schema.sessions) == 1
assert schema.total == 1
def test_session_list_response_invalid_sessions():
with pytest.raises(ValidationError):
SessionListResponse(sessions="not-a-list", total=0)
def test_daily_summary_day_valid():
data = {
"date": "2023-01-01",
"total_minutes": 480,
"day_minutes": 200,
"night_minutes": 280,
}
schema = DailySummaryDay(**data)
assert schema.date == "2023-01-01"
assert schema.total_minutes == 480
def test_daily_summary_day_invalid_type():
with pytest.raises(ValidationError):
DailySummaryDay(date="2023-01-01", total_minutes="not-an-int", day_minutes=100, night_minutes=100)
def test_daily_summary_response_valid():
days = [
{
"date": "2023-01-01",
"total_minutes": 480,
"day_minutes": 200,
"night_minutes": 280,
}
]
data = {"days": days}
schema = DailySummaryResponse(**data)
assert len(schema.days) == 1
def test_timetable_slot_valid():
data = {
"date": "2023-01-01",
"slot_start": "08:00",
"slot_end": "09:00",
"sleeping": True,
}
schema = TimetableSlot(**data)
assert schema.sleeping is True
def test_timetable_slot_invalid_sleeping():
with pytest.raises(ValidationError):
TimetableSlot(date="2023-01-01", slot_start="08:00", slot_end="09:00", sleeping="not-a-bool")
def test_timetable_response_valid():
slots = [
{
"date": "2023-01-01",
"slot_start": "08:00",
"slot_end": "09:00",
"sleeping": True,
}
]
data = {"slots": slots}
schema = TimetableResponse(**data)
assert len(schema.slots) == 1
def test_wakeup_bedtime_day_valid():
data = {"date": "2023-01-01", "wakeup_minutes": 480, "bedtime_minutes": 1200}
schema = WakeupBedtimeDay(**data)
assert schema.wakeup_minutes == 480
assert schema.bedtime_minutes == 1200
def test_wakeup_bedtime_day_optional_fields():
data = {"date": "2023-01-01"}
schema = WakeupBedtimeDay(**data)
assert schema.wakeup_minutes is None
assert schema.bedtime_minutes is None
def test_wakeup_bedtime_response_valid():
days = [{"date": "2023-01-01", "wakeup_minutes": 480, "bedtime_minutes": 1200}]
data = {"days": days}
schema = WakeupBedtimeResponse(**data)
assert len(schema.days) == 1
def test_summary_response_valid():
data = {
"total_sleeps": 10,
"avg_sleeps_per_day": 1.5,
"total_sleeps_all_time": 100,
"avg_sleeps_per_day_all_time": 1.2,
}
schema = SummaryResponse(**data)
assert schema.total_sleeps == 10
assert schema.avg_sleeps_per_day == 1.5
def test_summary_response_invalid_float():
with pytest.raises(ValidationError):
SummaryResponse(
total_sleeps=10,
avg_sleeps_per_day="not-a-float",
total_sleeps_all_time=100,
avg_sleeps_per_day_all_time=1.2,
)
+85
View File
@@ -1,4 +1,89 @@
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from datetime import datetime, timezone, timedelta
@pytest.mark.asyncio
async def test_create_session(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
payload = {"start_time": start_time}
response = await ac.post("/api/sessions", json=payload)
assert response.status_code == 201
data = response.json()
assert datetime.fromisoformat(data["start_time"]).replace(
tzinfo=None
) == datetime.fromisoformat(start_time).replace(tzinfo=None)
@pytest.mark.asyncio
async def test_create_session_already_open(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
response = await ac.post("/api/sessions", json={"start_time": start_time})
assert response.status_code == 409
@pytest.mark.asyncio
async def test_end_session(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
end_time = datetime.now(timezone.utc).isoformat()
response = await ac.patch("/api/sessions/1/end", json={"end_time": end_time})
assert response.status_code == 200
assert datetime.fromisoformat(response.json()["end_time"]).replace(
tzinfo=None
) == datetime.fromisoformat(end_time).replace(tzinfo=None)
@pytest.mark.asyncio
async def test_list_sessions(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
response = await ac.get("/api/sessions")
assert response.status_code == 200
assert len(response.json()["sessions"]) >= 1
@pytest.mark.asyncio
async def test_end_session_before_start_time(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
end_time = (
datetime.now(timezone.utc).replace(hour=0, minute=0) - timedelta(hours=1)
).isoformat()
response = await ac.patch("/api/sessions/1/end", json={"end_time": end_time})
assert response.status_code == 400
@pytest.mark.asyncio
async def test_end_already_ended_session(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
end_time = datetime.now(timezone.utc).isoformat()
await ac.patch("/api/sessions/1/end", json={"end_time": end_time})
response = await ac.patch("/api/sessions/1/end", json={"end_time": end_time})
assert response.status_code == 409
@pytest.mark.asyncio
async def test_delete_session(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
response = await ac.delete("/api/sessions/1")
assert response.status_code == 204
response = await ac.get("/api/sessions")
assert not any(s["id"] == 1 for s in response.json()["sessions"])
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import engine, init_db
+50
View File
@@ -1,4 +1,54 @@
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from datetime import datetime, timezone
@pytest.mark.asyncio
async def test_create_session(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
payload = {"start_time": start_time}
response = await ac.post("/api/sessions", json=payload)
assert response.status_code == 201
data = response.json()
assert datetime.fromisoformat(data["start_time"]).replace(
tzinfo=None
) == datetime.fromisoformat(start_time).replace(tzinfo=None)
@pytest.mark.asyncio
async def test_create_session_already_open(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
response = await ac.post("/api/sessions", json={"start_time": start_time})
assert response.status_code == 409
@pytest.mark.asyncio
async def test_end_session(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
end_time = datetime.now(timezone.utc).isoformat()
response = await ac.patch("/api/sessions/1/end", json={"end_time": end_time})
assert response.status_code == 200
assert datetime.fromisoformat(response.json()["end_time"]).replace(
tzinfo=None
) == datetime.fromisoformat(end_time).replace(tzinfo=None)
@pytest.mark.asyncio
async def test_list_sessions(api_client):
async with api_client as ac:
start_time = datetime.now(timezone.utc).isoformat()
await ac.post("/api/sessions", json={"start_time": start_time})
response = await ac.get("/api/sessions")
assert response.status_code == 200
assert len(response.json()["sessions"]) >= 1
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import engine, init_db
+82
View File
@@ -1,4 +1,86 @@
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from datetime import datetime, timezone, timedelta, date
from app.models import SleepSession
@pytest.mark.asyncio
async def test_daily_summary(api_client, db_session):
from sqlalchemy import select
# Create a session spanning today
start = datetime.now(timezone.utc) - timedelta(hours=5)
end = datetime.now(timezone.utc)
# Use the db_session fixture for insertion
new_s = SleepSession(start_time=start, end_time=end)
db_session.add(new_s)
await db_session.commit()
from_date = date.today().isoformat()
to_date = (date.today() + timedelta(days=1)).isoformat()
async with api_client as ac:
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(api_client, db_session):
from datetime import date
start = datetime.now(timezone.utc) - timedelta(hours=1)
new_s = SleepSession(start_time=start, end_time=datetime.now(timezone.utc))
db_session.add(new_s)
await db_session.commit()
from_date = date.today().isoformat()
to_date = (date.today() + timedelta(days=1)).isoformat()
async with api_client as ac:
response = await ac.get(
f"/api/stats/timetable?from_date={from_date}&to_date={to_date}"
)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_summary(api_client, db_session):
from datetime import date, timedelta, timezone
# 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)
db_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)
db_session.add(new_s2)
await db_session.commit()
from_date = (date.today() - timedelta(days=7)).isoformat()
to_date = date.today().isoformat()
async with api_client as ac:
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
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import engine, init_db