Initial commit
This commit is contained in:
@@ -0,0 +1,47 @@
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||||
from sqlmodel import SQLModel
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from app.main import app
|
||||
|
||||
@pytest_asyncio.fixture(scope="session")
|
||||
async def engine():
|
||||
"""Create a single in-memory SQLite engine for the entire test session."""
|
||||
engine = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
echo=False,
|
||||
)
|
||||
# Create tables once for the whole session to avoid overhead
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(SQLModel.metadata.create_all)
|
||||
yield engine
|
||||
await engine.dispose()
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def session_factory(engine):
|
||||
"""Create async session factory."""
|
||||
return sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session(session_factory):
|
||||
"""Create a database session for a test. Rolls back after each test."""
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
await session.rollback()
|
||||
|
||||
@pytest.fixture
|
||||
def sample_session_data():
|
||||
"""Sample sleep session data for testing."""
|
||||
return {
|
||||
"start_time": "2024-01-15T21:30:00Z",
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
async def api_client(engine):
|
||||
"""Provides an AsyncClient for testing API endpoints."""
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) as client:
|
||||
yield client
|
||||
@@ -0,0 +1,69 @@
|
||||
import pytest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from app.main import app
|
||||
from app.database import engine
|
||||
from app.models import SQLModel
|
||||
|
||||
@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_list_sessions_date_only_filter(setup_db):
|
||||
# Test that passing a date (midnight) also works and includes the whole day
|
||||
target_date = datetime(2023, 10, 1, tzinfo=timezone.utc)
|
||||
other_date = datetime(2023, 10, 2, tzinfo=timezone.utc)
|
||||
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) as ac:
|
||||
# 1. Create a session that starts LATER in the day on OTHER date (should be excluded)
|
||||
# We end it immediately to avoid 409s in the test
|
||||
start_time_other = other_date.replace(hour=20, minute=0)
|
||||
resp_other = await ac.post("/api/sessions", json={"start_time": start_time_other.isoformat()})
|
||||
assert resp_other.status_code == 201
|
||||
session_id_other = resp_other.json()["id"]
|
||||
await ac.patch(f"/api/sessions/{session_id_other}/end", json={"end_time": (start_time_other + timedelta(minutes=30)).isoformat()})
|
||||
|
||||
# 2. Create a session that starts ON the target_date (should be included)
|
||||
start_time_target = target_date.replace(hour=20, minute=0)
|
||||
resp_target = await ac.post("/api/sessions", json={"start_time": start_time_target.isoformat()})
|
||||
assert resp_target.status_code == 201
|
||||
session_id_target = resp_target.json()["id"]
|
||||
await ac.patch(f"/api/sessions/{session_id_target}/end", json={"end_time": (start_time_target + timedelta(minutes=30)).isoformat()})
|
||||
|
||||
# 3. Create a session that starts EARLIER on the target_date (should be included)
|
||||
start_time_early = target_date.replace(hour=2, minute=0)
|
||||
resp_early = await ac.post("/api/sessions", json={"start_time": start_time_early.isoformat()})
|
||||
assert resp_early.status_code == 201
|
||||
session_id_early = resp_early.json()["id"]
|
||||
await ac.patch(f"/api/sessions/{session_id_early}/end", json={"end_time": (start_time_early + timedelta(minutes=30)).isoformat()})
|
||||
|
||||
# 4. Create a session that is on the OTHER date (should be excluded)
|
||||
start_time_other2 = other_date.replace(hour=2, minute=0)
|
||||
resp_other2 = await ac.post("/api/sessions", json={"start_time": start_time_other2.isoformat()})
|
||||
assert resp_other2.status_code == 201
|
||||
session_id_other2 = resp_other2.json()["id"]
|
||||
await ac.patch(f"/api/sessions/{session_id_other2}/end", json={"end_time": (start_time_other2 + timedelta(minutes=30)).isoformat()})
|
||||
|
||||
# Query with to_date = target_date (midnight)
|
||||
# This should be expanded to 23:59:59 on target_date.
|
||||
# Should include the 2 sessions on target_date (target and early).
|
||||
# Should exclude the 2 sessions on other_date.
|
||||
to_date_str = target_date.strftime("%Y-%m-%d") # "2023-10-01"
|
||||
print(f"\nGET: to_date={to_date_str}")
|
||||
|
||||
response = await ac.get(f"/api/sessions?to_date={to_date_str}")
|
||||
print(f"GET RESPONSE: {response.status_code}, body={response.json()}")
|
||||
|
||||
assert response.status_code == 200
|
||||
sessions = response.json()["sessions"]
|
||||
|
||||
# We expect exactly 2 sessions (the ones on target_date)
|
||||
assert len(sessions) == 2, f"Expected 2 sessions (the ones on {to_date_str}), but got {len(sessions)}. (Body: {response.json()})"
|
||||
for s in sessions:
|
||||
# Verify that the sessions returned are indeed from the target_date
|
||||
assert target_date.strftime("%Y-%m-%d") in s["start_time"]
|
||||
@@ -0,0 +1,111 @@
|
||||
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
|
||||
|
||||
|
||||
@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_create_session(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) as ac2:
|
||||
list_resp = await ac2.get("/api/sessions")
|
||||
assert not any(s["id"] == 1 for s in list_resp.json()["sessions"])
|
||||
@@ -0,0 +1,68 @@
|
||||
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
|
||||
|
||||
|
||||
@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_create_session(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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(setup_db):
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) 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
|
||||
@@ -0,0 +1,73 @@
|
||||
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}"
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()["slots"]) > 0
|
||||
@@ -0,0 +1,121 @@
|
||||
import pytest
|
||||
from datetime import datetime, time, timedelta
|
||||
from app.utils.time_utils import get_day_night_splits, minutes_after, day_start_pattern
|
||||
|
||||
class TestGetDayNightSplits:
|
||||
def test_zero_length_session(self):
|
||||
start = datetime(2024, 1, 15, 10, 0)
|
||||
end = datetime(2024, 1, 15, 10, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
assert len(segments) == 0
|
||||
|
||||
def test_session_entirely_in_day(self):
|
||||
start = datetime(2024, 1, 15, 10, 0)
|
||||
end = datetime(2024, 1, 15, 14, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
assert len(segments) == 1
|
||||
assert segments[0]["type"] == "day"
|
||||
|
||||
def test_session_entirely_in_night(self):
|
||||
start = datetime(2024, 1, 15, 22, 0)
|
||||
end = datetime(2024, 1, 16, 2, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
assert len(segments) == 1
|
||||
assert segments[0]["type"] == "night"
|
||||
|
||||
def test_session_spans_day_night_boundary(self):
|
||||
start = datetime(2024, 1, 15, 18, 0)
|
||||
end = datetime(2024, 1, 15, 22, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
assert len(segments) == 2
|
||||
assert segments[0]["type"] == "day"
|
||||
assert segments[1]["type"] == "night"
|
||||
|
||||
def test_session_spans_midnight(self):
|
||||
start = datetime(2024, 1, 15, 23, 0)
|
||||
end = datetime(2024, 1, 16, 2, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
assert len(segments) == 1
|
||||
assert segments[0]["type"] == "night"
|
||||
|
||||
def test_session_spans_two_days(self):
|
||||
# Start Day 1 at 23:00, End Day 2 at 01:00 (Night)
|
||||
# It should catch the transition if day_start/end logic is correct.
|
||||
# But here, no boundary (07:00 or 20:00) is in the session.
|
||||
# So it should just be one segment of type 'night'.
|
||||
start = datetime(2024, 1, 15, 23, 0)
|
||||
end = datetime(2024, 1, 16, 1, 0)
|
||||
day_start = time(7, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, time(20, 0))
|
||||
assert len(segments) == 1
|
||||
assert segments[0]["type"] == "night"
|
||||
|
||||
def test_session_boundary_at_start(self):
|
||||
# Session starts exactly at day_start
|
||||
start = datetime(2024, 1, 15, 7, 0)
|
||||
end = datetime(2024, 1, 15, 10, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
# The loop uses: if session_start < b1 < session_end
|
||||
# Since 7:00 is not > 7:00, b1 won't be added.
|
||||
# So it should result in 1 segment [start, end]
|
||||
assert len(segments) == 1
|
||||
assert segments[0]["type"] == "day"
|
||||
|
||||
def test_session_boundary_at_end(self):
|
||||
# Session ends exactly at day_end
|
||||
start = datetime(2024, 1, 15, 18, 0)
|
||||
end = datetime(2024, 1, 15, 20, 0)
|
||||
day_start = time(7, 0)
|
||||
day_end = time(20, 0)
|
||||
segments = get_day_night_splits(start, end, day_start, day_end)
|
||||
assert len(segments) == 1
|
||||
assert segments[0]["type"] == "day"
|
||||
|
||||
class TestMinutesAfter:
|
||||
def test_minutes_at_reference(self):
|
||||
dt = datetime(2024, 1, 15, 4, 0)
|
||||
ref = time(4, 0)
|
||||
result = minutes_after(dt, ref)
|
||||
assert result == 0
|
||||
|
||||
def test_minutes_after_reference(self):
|
||||
dt = datetime(2024, 1, 15, 7, 30)
|
||||
ref = time(4, 0)
|
||||
result = minutes_after(dt, ref)
|
||||
assert result == 210
|
||||
|
||||
def test_before_reference(self):
|
||||
dt = datetime(2024, 1, 15, 3, 30)
|
||||
ref = time(4, 0)
|
||||
result = minutes_after(dt, ref)
|
||||
assert result == -30
|
||||
|
||||
class TestDayStartPattern:
|
||||
def test_within_day_range(self):
|
||||
assert day_start_pattern(time(12, 0), time(7, 0), time(20, 0)) is True
|
||||
|
||||
def test_outside_day_range(self):
|
||||
assert day_start_pattern(time(3, 0), time(7, 0), time(20, 0)) is False
|
||||
|
||||
def test_at_boundary(self):
|
||||
assert day_start_pattern(time(7, 0), time(7, 0), time(20, 0)) is True
|
||||
|
||||
def test_just_after_boundary(self):
|
||||
# Test that the end boundary is exclusive
|
||||
assert day_start_pattern(time(20, 0), time(7, 0), time(20, 0)) is False
|
||||
|
||||
def test_spans_midnight(self):
|
||||
assert day_start_pattern(time(22, 0), time(20, 0), time(7, 0)) is True
|
||||
assert day_start_pattern(time(3, 0), time(20, 0), time(7, 0)) is True
|
||||
Reference in New Issue
Block a user