178 lines
6.7 KiB
Python
178 lines
6.7 KiB
Python
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
|
|
|