health_sleep API add tests and route for create and delete

Added comprehensive unit tests for health_sleep CRUD, models, and router. Added route for creating and deleting sleep entries
This commit is contained in:
João Vitória Silva
2025-12-05 14:19:29 +00:00
parent 73f2bc3588
commit a4d52a8e03
6 changed files with 1500 additions and 2 deletions

View File

@@ -1,6 +1,6 @@
from typing import Annotated, Callable
from fastapi import APIRouter, Depends, Security
from fastapi import APIRouter, Depends, Security, HTTPException
from sqlalchemy.orm import Session
import health_sleep.schema as health_sleep_schema
@@ -112,6 +112,105 @@ async def read_health_sleep_all_pagination(
return health_sleep_schema.HealthSleepListResponse(total=total, records=records)
@router.post("", status_code=201)
async def create_health_sleep(
health_sleep: health_sleep_schema.HealthSleep,
_check_scopes: Annotated[
Callable, Security(auth_security.check_scopes, scopes=["health:write"])
],
token_user_id: Annotated[
int,
Depends(auth_security.get_sub_from_access_token),
],
db: Annotated[
Session,
Depends(core_database.get_db),
],
) -> health_sleep_schema.HealthSleep:
"""
Create or update health sleep data for a user.
This endpoint creates new health sleep data or updates existing data if an entry
for the specified date already exists. The operation is determined automatically
based on whether sleep data exists for the given date.
Args:
health_sleep (health_sleep_schema.HealthSleep): The health sleep data to create
or update, including the date and sleep duration.
_check_scopes (Callable): Security dependency that verifies the user has
'health:write' scope.
token_user_id (int): The ID of the authenticated user extracted from the
access token.
db (Session): Database session dependency for database operations.
Returns:
health_sleep_schema.HealthSleep: The created or updated health sleep data.
Raises:
HTTPException: 400 error if the date field is not provided in the request.
"""
if not health_sleep.date:
raise HTTPException(status_code=400, detail="Date field is required.")
# Convert date to string format for CRUD function
date_str = health_sleep.date.isoformat()
# Check if health_sleep for this date already exists
sleep_for_date = health_sleep_crud.get_health_sleep_by_date(
token_user_id, date_str, db
)
if sleep_for_date:
health_sleep.id = sleep_for_date.id
# Updates the health_sleep in the database and returns it
return health_sleep_crud.edit_health_sleep(token_user_id, health_sleep, db)
else:
# Creates the health_sleep in the database and returns it
return health_sleep_crud.create_health_sleep(token_user_id, health_sleep, db)
@router.put("")
async def edit_health_sleep(
health_sleep: health_sleep_schema.HealthSleep,
_check_scopes: Annotated[
Callable, Security(auth_security.check_scopes, scopes=["health:write"])
],
token_user_id: Annotated[
int,
Depends(auth_security.get_sub_from_access_token),
],
db: Annotated[
Session,
Depends(core_database.get_db),
],
) -> health_sleep_schema.HealthSleep:
"""
Edit health sleep data for a user.
This endpoint updates existing health sleep records in the database for the authenticated user.
Requires 'health:write' scope for authorization.
Args:
health_sleep (health_sleep_schema.HealthSleep): The health sleep data to be updated,
containing the new values for the health sleep record.
_check_scopes (Callable): Security dependency that verifies the user has 'health:write'
scope permission.
token_user_id (int): The user ID extracted from the JWT access token, used to identify
the user making the request.
db (Session): Database session dependency for performing database operations.
Returns:
health_sleep_schema.HealthSleep: The updated health sleep record with the new values
as stored in the database.
Raises:
HTTPException: May raise various HTTP exceptions if authorization fails, user is not
found, or database operations fail.
"""
# Updates the health_sleep in the database and returns it
return health_sleep_crud.edit_health_sleep(token_user_id, health_sleep, db)
@router.delete("/{health_sleep_id}", status_code=204)
async def delete_health_sleep(
health_sleep_id: int,
@@ -146,7 +245,7 @@ async def delete_health_sleep(
HTTPException: May be raised by dependencies if:
- The access token is invalid or expired
- The user lacks required 'health:write' scope
- The health steps record doesn't exist or doesn't belong to the user
- The health sleep record doesn't exist or doesn't belong to the user
"""
# Deletes entry from database
health_sleep_crud.delete_health_sleep(token_user_id, health_sleep_id, db)

View File

@@ -0,0 +1 @@
# tests/health_sleep/__init__.py

View File

@@ -0,0 +1,494 @@
import pytest
from datetime import datetime, date as datetime_date
from decimal import Decimal
from unittest.mock import MagicMock, patch
from fastapi import HTTPException, status
from sqlalchemy.exc import IntegrityError
import health_sleep.crud as health_sleep_crud
import health_sleep.schema as health_sleep_schema
import health_sleep.models as health_sleep_models
class TestGetHealthSleepNumber:
"""
Test suite for get_health_sleep_number function.
"""
def test_get_health_sleep_number_success(self, mock_db):
"""
Test successful count of health sleep records for a user.
"""
# Arrange
user_id = 1
expected_count = 5
mock_query = mock_db.query.return_value
mock_query.filter.return_value.count.return_value = expected_count
# Act
result = health_sleep_crud.get_health_sleep_number(user_id, mock_db)
# Assert
assert result == expected_count
mock_db.query.assert_called_once_with(health_sleep_models.HealthSleep)
def test_get_health_sleep_number_zero(self, mock_db):
"""
Test count when user has no health sleep records.
"""
# Arrange
user_id = 1
mock_query = mock_db.query.return_value
mock_query.filter.return_value.count.return_value = 0
# Act
result = health_sleep_crud.get_health_sleep_number(user_id, mock_db)
# Assert
assert result == 0
def test_get_health_sleep_number_exception(self, mock_db):
"""
Test exception handling in get_health_sleep_number.
"""
# Arrange
user_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.get_health_sleep_number(user_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert exc_info.value.detail == "Internal Server Error"
class TestGetAllHealthSleepByUserId:
"""
Test suite for get_all_health_sleep_by_user_id function.
"""
def test_get_all_health_sleep_by_user_id_success(self, mock_db):
"""
Test successful retrieval of all health sleep records for user.
"""
# Arrange
user_id = 1
mock_sleep1 = MagicMock(spec=health_sleep_models.HealthSleep)
mock_sleep2 = MagicMock(spec=health_sleep_models.HealthSleep)
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_filter.order_by.return_value.all.return_value = [
mock_sleep1,
mock_sleep2,
]
# Act
result = health_sleep_crud.get_all_health_sleep_by_user_id(user_id, mock_db)
# Assert
assert result == [mock_sleep1, mock_sleep2]
mock_db.query.assert_called_once_with(health_sleep_models.HealthSleep)
def test_get_all_health_sleep_by_user_id_empty(self, mock_db):
"""
Test retrieval when user has no health sleep records.
"""
# Arrange
user_id = 1
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_filter.order_by.return_value.all.return_value = []
# Act
result = health_sleep_crud.get_all_health_sleep_by_user_id(user_id, mock_db)
# Assert
assert result == []
def test_get_all_health_sleep_by_user_id_exception(self, mock_db):
"""
Test exception handling in get_all_health_sleep_by_user_id.
"""
# Arrange
user_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.get_all_health_sleep_by_user_id(user_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestGetHealthSleepWithPagination:
"""
Test suite for get_health_sleep_with_pagination function.
"""
def test_get_health_sleep_with_pagination_success(self, mock_db):
"""
Test successful retrieval of paginated health sleep records.
"""
# Arrange
user_id = 1
page_number = 2
num_records = 5
mock_sleep1 = MagicMock(spec=health_sleep_models.HealthSleep)
mock_sleep2 = MagicMock(spec=health_sleep_models.HealthSleep)
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_order = mock_filter.order_by.return_value
mock_offset = mock_order.offset.return_value
mock_offset.limit.return_value.all.return_value = [
mock_sleep1,
mock_sleep2,
]
# Act
result = health_sleep_crud.get_health_sleep_with_pagination(
user_id, mock_db, page_number, num_records
)
# Assert
assert result == [mock_sleep1, mock_sleep2]
mock_order.offset.assert_called_once_with(5)
mock_offset.limit.assert_called_once_with(5)
def test_get_health_sleep_with_pagination_defaults(self, mock_db):
"""
Test pagination with default values.
"""
# Arrange
user_id = 1
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_order = mock_filter.order_by.return_value
mock_offset = mock_order.offset.return_value
mock_offset.limit.return_value.all.return_value = []
# Act
result = health_sleep_crud.get_health_sleep_with_pagination(user_id, mock_db)
# Assert
mock_order.offset.assert_called_once_with(0)
mock_offset.limit.assert_called_once_with(5)
def test_get_health_sleep_with_pagination_exception(self, mock_db):
"""
Test exception handling in get_health_sleep_with_pagination.
"""
# Arrange
user_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.get_health_sleep_with_pagination(user_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestGetHealthSleepByDate:
"""
Test suite for get_health_sleep_by_date function.
"""
def test_get_health_sleep_by_date_success(self, mock_db):
"""
Test successful retrieval of health sleep by date.
"""
# Arrange
user_id = 1
test_date = "2024-01-15"
mock_sleep = MagicMock(spec=health_sleep_models.HealthSleep)
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = mock_sleep
# Act
result = health_sleep_crud.get_health_sleep_by_date(user_id, test_date, mock_db)
# Assert
assert result == mock_sleep
mock_db.query.assert_called_once_with(health_sleep_models.HealthSleep)
def test_get_health_sleep_by_date_not_found(self, mock_db):
"""
Test retrieval when no record exists for date.
"""
# Arrange
user_id = 1
test_date = "2024-01-15"
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = None
# Act
result = health_sleep_crud.get_health_sleep_by_date(user_id, test_date, mock_db)
# Assert
assert result is None
def test_get_health_sleep_by_date_exception(self, mock_db):
"""
Test exception handling in get_health_sleep_by_date.
"""
# Arrange
user_id = 1
test_date = "2024-01-15"
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.get_health_sleep_by_date(user_id, test_date, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestCreateHealthSleep:
"""
Test suite for create_health_sleep function.
"""
def test_create_health_sleep_success(self, mock_db):
"""
Test successful creation of health sleep entry.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(
date=datetime_date(2024, 1, 15),
total_sleep_seconds=28800,
sleep_score_overall=85,
)
mock_db_sleep = MagicMock()
mock_db_sleep.id = 1
mock_db.add.return_value = None
mock_db.commit.return_value = None
mock_db.refresh.return_value = None
with patch.object(
health_sleep_models,
"HealthSleep",
return_value=mock_db_sleep,
):
# Act
result = health_sleep_crud.create_health_sleep(
user_id, health_sleep, mock_db
)
# Assert
assert result.id == 1
assert result.total_sleep_seconds == 28800
mock_db.add.assert_called_once()
mock_db.commit.assert_called_once()
mock_db.refresh.assert_called_once()
@patch("health_sleep.crud.func")
def test_create_health_sleep_with_none_date(self, mock_func, mock_db):
"""
Test creation with None date sets current date.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(
date=None, total_sleep_seconds=28800
)
# Mock func.now() to return a proper date object
mock_func.now.return_value = datetime_date(2024, 1, 15)
mock_db_sleep = MagicMock()
mock_db_sleep.id = 1
mock_db.add.return_value = None
mock_db.commit.return_value = None
mock_db.refresh.return_value = None
with patch.object(
health_sleep_models,
"HealthSleep",
return_value=mock_db_sleep,
):
# Act
result = health_sleep_crud.create_health_sleep(
user_id, health_sleep, mock_db
)
# Assert
mock_func.now.assert_called_once()
assert result.id == 1
assert result.date == datetime_date(2024, 1, 15)
def test_create_health_sleep_duplicate_entry(self, mock_db):
"""
Test creation with duplicate entry raises conflict error.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(
date=datetime_date(2024, 1, 15), total_sleep_seconds=28800
)
mock_db_sleep = MagicMock()
mock_db.add.return_value = None
mock_db.commit.side_effect = IntegrityError("Duplicate entry", None, None)
with patch.object(
health_sleep_models,
"HealthSleep",
return_value=mock_db_sleep,
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.create_health_sleep(user_id, health_sleep, mock_db)
assert exc_info.value.status_code == status.HTTP_409_CONFLICT
assert "Duplicate entry error" in exc_info.value.detail
mock_db.rollback.assert_called_once()
def test_create_health_sleep_exception(self, mock_db):
"""
Test exception handling in create_health_sleep.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(
date=datetime_date(2024, 1, 15), total_sleep_seconds=28800
)
mock_db.add.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.create_health_sleep(user_id, health_sleep, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
mock_db.rollback.assert_called_once()
class TestEditHealthSleep:
"""
Test suite for edit_health_sleep function.
"""
def test_edit_health_sleep_success(self, mock_db):
"""
Test successful edit of health sleep entry.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(
id=1,
date=datetime_date(2024, 1, 15),
total_sleep_seconds=32400,
sleep_score_overall=90,
)
mock_db_sleep = MagicMock(spec=health_sleep_models.HealthSleep)
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = mock_db_sleep
# Act
result = health_sleep_crud.edit_health_sleep(user_id, health_sleep, mock_db)
# Assert
assert result.total_sleep_seconds == 32400
mock_db.commit.assert_called_once()
def test_edit_health_sleep_not_found(self, mock_db):
"""
Test edit when health sleep record not found.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(
id=999, date=datetime_date(2024, 1, 15), total_sleep_seconds=32400
)
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = None
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.edit_health_sleep(user_id, health_sleep, mock_db)
assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
assert exc_info.value.detail == "Health sleep not found"
def test_edit_health_sleep_exception(self, mock_db):
"""
Test exception handling in edit_health_sleep.
"""
# Arrange
user_id = 1
health_sleep = health_sleep_schema.HealthSleep(id=1, total_sleep_seconds=32400)
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.edit_health_sleep(user_id, health_sleep, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
mock_db.rollback.assert_called_once()
class TestDeleteHealthSleep:
"""
Test suite for delete_health_sleep function.
"""
def test_delete_health_sleep_success(self, mock_db):
"""
Test successful deletion of health sleep entry.
"""
# Arrange
user_id = 1
health_sleep_id = 1
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_filter.delete.return_value = 1
# Act
health_sleep_crud.delete_health_sleep(user_id, health_sleep_id, mock_db)
# Assert
mock_db.commit.assert_called_once()
mock_db.query.assert_called_once_with(health_sleep_models.HealthSleep)
def test_delete_health_sleep_not_found(self, mock_db):
"""
Test deletion when health sleep record not found.
"""
# Arrange
user_id = 1
health_sleep_id = 999
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_filter.delete.return_value = 0
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.delete_health_sleep(user_id, health_sleep_id, mock_db)
assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
assert f"Health sleep with id {health_sleep_id}" in exc_info.value.detail
def test_delete_health_sleep_exception(self, mock_db):
"""
Test exception handling in delete_health_sleep.
"""
# Arrange
user_id = 1
health_sleep_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_sleep_crud.delete_health_sleep(user_id, health_sleep_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
mock_db.rollback.assert_called_once()

View File

@@ -0,0 +1,131 @@
import pytest
from datetime import date as datetime_date
import health_sleep.models as health_sleep_models
class TestHealthSleepModel:
"""
Test suite for HealthSleep SQLAlchemy model.
"""
def test_health_sleep_model_table_name(self):
"""
Test HealthSleep model has correct table name.
"""
# Assert
assert health_sleep_models.HealthSleep.__tablename__ == "health_sleep"
def test_health_sleep_model_columns_exist(self):
"""
Test HealthSleep model has all expected columns.
"""
# Assert
assert hasattr(health_sleep_models.HealthSleep, "id")
assert hasattr(health_sleep_models.HealthSleep, "user_id")
assert hasattr(health_sleep_models.HealthSleep, "date")
assert hasattr(health_sleep_models.HealthSleep, "sleep_start_time_gmt")
assert hasattr(health_sleep_models.HealthSleep, "sleep_end_time_gmt")
assert hasattr(health_sleep_models.HealthSleep, "total_sleep_seconds")
assert hasattr(health_sleep_models.HealthSleep, "deep_sleep_seconds")
assert hasattr(health_sleep_models.HealthSleep, "light_sleep_seconds")
assert hasattr(health_sleep_models.HealthSleep, "rem_sleep_seconds")
assert hasattr(health_sleep_models.HealthSleep, "avg_heart_rate")
assert hasattr(health_sleep_models.HealthSleep, "sleep_score_overall")
assert hasattr(health_sleep_models.HealthSleep, "source")
def test_health_sleep_model_primary_key(self):
"""
Test HealthSleep model has correct primary key.
"""
# Arrange
id_column = health_sleep_models.HealthSleep.id
# Assert
assert id_column.primary_key is True
assert id_column.autoincrement is True
def test_health_sleep_model_foreign_key(self):
"""
Test HealthSleep model has correct foreign key.
"""
# Arrange
user_id_column = health_sleep_models.HealthSleep.user_id
# Assert
assert user_id_column.nullable is False
assert user_id_column.index is True
def test_health_sleep_model_nullable_fields(self):
"""
Test HealthSleep model nullable fields.
"""
# Assert
assert health_sleep_models.HealthSleep.sleep_start_time_gmt.nullable is True
assert health_sleep_models.HealthSleep.total_sleep_seconds.nullable is True
assert health_sleep_models.HealthSleep.deep_sleep_seconds.nullable is True
assert health_sleep_models.HealthSleep.avg_heart_rate.nullable is True
assert health_sleep_models.HealthSleep.sleep_score_overall.nullable is True
assert health_sleep_models.HealthSleep.source.nullable is True
def test_health_sleep_model_required_fields(self):
"""
Test HealthSleep model required fields.
"""
# Assert
assert health_sleep_models.HealthSleep.user_id.nullable is False
assert health_sleep_models.HealthSleep.date.nullable is False
def test_health_sleep_model_column_types(self):
"""
Test HealthSleep model column types.
"""
# Assert
assert health_sleep_models.HealthSleep.id.type.python_type == int
assert health_sleep_models.HealthSleep.user_id.type.python_type == int
assert health_sleep_models.HealthSleep.date.type.python_type == datetime_date
assert (
health_sleep_models.HealthSleep.total_sleep_seconds.type.python_type == int
)
assert health_sleep_models.HealthSleep.min_heart_rate.type.python_type == int
assert health_sleep_models.HealthSleep.awake_count.type.python_type == int
def test_health_sleep_model_relationship(self):
"""
Test HealthSleep model has relationship to User.
"""
# Assert
assert hasattr(health_sleep_models.HealthSleep, "user")
def test_health_sleep_model_decimal_precision(self):
"""
Test HealthSleep model decimal fields precision.
"""
# Arrange
avg_heart_rate_column = health_sleep_models.HealthSleep.avg_heart_rate
avg_spo2_column = health_sleep_models.HealthSleep.avg_spo2
# Assert
assert avg_heart_rate_column.type.precision == 10
assert avg_heart_rate_column.type.scale == 2
assert avg_spo2_column.type.precision == 10
assert avg_spo2_column.type.scale == 2
def test_health_sleep_model_string_field_lengths(self):
"""
Test HealthSleep model string field max lengths.
"""
# Assert
assert health_sleep_models.HealthSleep.source.type.length == 250
assert health_sleep_models.HealthSleep.garminconnect_sleep_id.type.length == 250
assert health_sleep_models.HealthSleep.sleep_score_duration.type.length == 50
def test_health_sleep_model_date_indexed(self):
"""
Test HealthSleep model date field is indexed.
"""
# Arrange
date_column = health_sleep_models.HealthSleep.date
# Assert
assert date_column.index is True

View File

@@ -0,0 +1,409 @@
import pytest
from datetime import datetime, date as datetime_date
from decimal import Decimal
from unittest.mock import MagicMock, patch, ANY
from fastapi import HTTPException, status
import health_sleep.schema as health_sleep_schema
import health_sleep.models as health_sleep_models
class TestReadHealthSleepAll:
"""
Test suite for read_health_sleep_all endpoint.
"""
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_number")
@patch("health_sleep.router.health_sleep_crud.get_all_health_sleep_by_user_id")
def test_read_health_sleep_all_success(
self, mock_get_all, mock_get_number, fast_api_client, fast_api_app
):
"""
Test successful retrieval of all health sleep records with total count.
"""
# Arrange
mock_sleep1 = MagicMock(spec=health_sleep_models.HealthSleep)
mock_sleep1.id = 1
mock_sleep1.user_id = 1
mock_sleep1.date = datetime_date(2024, 1, 15)
mock_sleep1.sleep_start_time_gmt = datetime(2024, 1, 14, 22, 0, 0)
mock_sleep1.sleep_end_time_gmt = datetime(2024, 1, 15, 6, 0, 0)
mock_sleep1.sleep_start_time_local = None
mock_sleep1.sleep_end_time_local = None
mock_sleep1.total_sleep_seconds = 28800
mock_sleep1.nap_time_seconds = None
mock_sleep1.unmeasurable_sleep_seconds = None
mock_sleep1.deep_sleep_seconds = 7200
mock_sleep1.light_sleep_seconds = 14400
mock_sleep1.rem_sleep_seconds = 7200
mock_sleep1.awake_sleep_seconds = 0
mock_sleep1.avg_heart_rate = Decimal("55.5")
mock_sleep1.min_heart_rate = 45
mock_sleep1.max_heart_rate = 75
mock_sleep1.avg_spo2 = Decimal("97.5")
mock_sleep1.lowest_spo2 = 95
mock_sleep1.highest_spo2 = 99
mock_sleep1.avg_respiration = None
mock_sleep1.lowest_respiration = None
mock_sleep1.highest_respiration = None
mock_sleep1.avg_stress_level = None
mock_sleep1.awake_count = 2
mock_sleep1.restless_moments_count = 5
mock_sleep1.sleep_score_overall = 85
mock_sleep1.sleep_score_duration = "GOOD"
mock_sleep1.sleep_score_quality = "GOOD"
mock_sleep1.garminconnect_sleep_id = None
mock_sleep1.sleep_stages = None
mock_sleep1.source = None
mock_sleep1.hrv_status = None
mock_sleep1.resting_heart_rate = 50
mock_sleep1.avg_skin_temp_deviation = None
mock_sleep1.awake_count_score = None
mock_sleep1.rem_percentage_score = None
mock_sleep1.deep_percentage_score = None
mock_sleep1.light_percentage_score = None
mock_sleep1.avg_sleep_stress = None
mock_sleep1.sleep_stress_score = None
mock_get_all.return_value = [mock_sleep1]
mock_get_number.return_value = 1
# Act
response = fast_api_client.get(
"/health_sleep",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert len(data["records"]) == 1
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_number")
@patch("health_sleep.router.health_sleep_crud.get_all_health_sleep_by_user_id")
def test_read_health_sleep_all_empty(
self, mock_get_all, mock_get_number, fast_api_client, fast_api_app
):
"""
Test retrieval when user has no health sleep records.
"""
# Arrange
mock_get_all.return_value = []
mock_get_number.return_value = 0
# Act
response = fast_api_client.get(
"/health_sleep",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["records"] == []
class TestReadHealthSleepAllPagination:
"""
Test suite for read_health_sleep_all_pagination endpoint.
"""
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_number")
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_with_pagination")
def test_read_health_sleep_all_pagination_success(
self, mock_get_paginated, mock_get_number, fast_api_client, fast_api_app
):
"""
Test successful retrieval of paginated health sleep records with total count.
"""
# Arrange
mock_sleep1 = MagicMock(spec=health_sleep_models.HealthSleep)
mock_sleep1.id = 1
mock_sleep1.user_id = 1
mock_sleep1.date = datetime_date(2024, 1, 15)
mock_sleep1.sleep_start_time_gmt = None
mock_sleep1.sleep_end_time_gmt = None
mock_sleep1.sleep_start_time_local = None
mock_sleep1.sleep_end_time_local = None
mock_sleep1.total_sleep_seconds = 28800
mock_sleep1.nap_time_seconds = None
mock_sleep1.unmeasurable_sleep_seconds = None
mock_sleep1.deep_sleep_seconds = None
mock_sleep1.light_sleep_seconds = None
mock_sleep1.rem_sleep_seconds = None
mock_sleep1.awake_sleep_seconds = None
mock_sleep1.avg_heart_rate = None
mock_sleep1.min_heart_rate = None
mock_sleep1.max_heart_rate = None
mock_sleep1.avg_spo2 = None
mock_sleep1.lowest_spo2 = None
mock_sleep1.highest_spo2 = None
mock_sleep1.avg_respiration = None
mock_sleep1.lowest_respiration = None
mock_sleep1.highest_respiration = None
mock_sleep1.avg_stress_level = None
mock_sleep1.awake_count = None
mock_sleep1.restless_moments_count = None
mock_sleep1.sleep_score_overall = None
mock_sleep1.sleep_score_duration = None
mock_sleep1.sleep_score_quality = None
mock_sleep1.garminconnect_sleep_id = None
mock_sleep1.sleep_stages = None
mock_sleep1.source = None
mock_sleep1.hrv_status = None
mock_sleep1.resting_heart_rate = None
mock_sleep1.avg_skin_temp_deviation = None
mock_sleep1.awake_count_score = None
mock_sleep1.rem_percentage_score = None
mock_sleep1.deep_percentage_score = None
mock_sleep1.light_percentage_score = None
mock_sleep1.avg_sleep_stress = None
mock_sleep1.sleep_stress_score = None
mock_get_paginated.return_value = [mock_sleep1]
mock_get_number.return_value = 10
# Act
response = fast_api_client.get(
"/health_sleep/page_number/1/num_records/5",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total"] == 10
assert len(data["records"]) == 1
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_number")
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_with_pagination")
def test_read_health_sleep_all_pagination_different_page(
self, mock_get_paginated, mock_get_number, fast_api_client, fast_api_app
):
"""
Test paginated retrieval with different page numbers.
"""
# Arrange
mock_get_paginated.return_value = []
mock_get_number.return_value = 20
# Act
response = fast_api_client.get(
"/health_sleep/page_number/2/num_records/10",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total"] == 20
assert data["records"] == []
mock_get_paginated.assert_called_once_with(1, ANY, 2, 10)
class TestCreateHealthSleep:
"""
Test suite for create_health_sleep endpoint.
"""
@patch("health_sleep.router.health_sleep_crud.create_health_sleep")
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_by_date")
def test_create_health_sleep_success(
self,
mock_get_by_date,
mock_create,
fast_api_client,
fast_api_app,
):
"""
Test successful creation of health sleep entry.
"""
# Arrange
mock_get_by_date.return_value = None
created_sleep = health_sleep_schema.HealthSleep(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
total_sleep_seconds=28800,
)
mock_create.return_value = created_sleep
# Act
response = fast_api_client.post(
"/health_sleep",
json={
"date": "2024-01-15",
"total_sleep_seconds": 28800,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 201
data = response.json()
assert data["total_sleep_seconds"] == 28800
@patch("health_sleep.router.health_sleep_crud.edit_health_sleep")
@patch("health_sleep.router.health_sleep_crud.get_health_sleep_by_date")
def test_create_health_sleep_updates_existing(
self, mock_get_by_date, mock_edit, fast_api_client, fast_api_app
):
"""
Test creating health sleep when entry exists updates it.
"""
# Arrange
existing_sleep = MagicMock()
existing_sleep.id = 1
mock_get_by_date.return_value = existing_sleep
updated_sleep = health_sleep_schema.HealthSleep(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
total_sleep_seconds=32400,
)
mock_edit.return_value = updated_sleep
# Act
response = fast_api_client.post(
"/health_sleep",
json={
"date": "2024-01-15",
"total_sleep_seconds": 32400,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 201
mock_edit.assert_called_once()
def test_create_health_sleep_missing_date(self, fast_api_client, fast_api_app):
"""
Test creating health sleep without date field raises error.
"""
# Act
response = fast_api_client.post(
"/health_sleep",
json={
"total_sleep_seconds": 28800,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 400
assert "Date field is required" in response.json()["detail"]
class TestEditHealthSleep:
"""
Test suite for edit_health_sleep endpoint.
"""
@patch("health_sleep.router.health_sleep_crud.edit_health_sleep")
def test_edit_health_sleep_success(self, mock_edit, fast_api_client, fast_api_app):
"""
Test successful edit of health sleep entry.
"""
# Arrange
updated_sleep = health_sleep_schema.HealthSleep(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
total_sleep_seconds=32400,
)
mock_edit.return_value = updated_sleep
# Act
response = fast_api_client.put(
"/health_sleep",
json={
"id": 1,
"date": "2024-01-15",
"total_sleep_seconds": 32400,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total_sleep_seconds"] == 32400
@patch("health_sleep.router.health_sleep_crud.edit_health_sleep")
def test_edit_health_sleep_not_found(
self, mock_edit, fast_api_client, fast_api_app
):
"""
Test edit when health sleep not found.
"""
# Arrange
mock_edit.side_effect = HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Health sleep not found",
)
# Act
response = fast_api_client.put(
"/health_sleep",
json={
"id": 999,
"date": "2024-01-15",
"total_sleep_seconds": 32400,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 404
class TestDeleteHealthSleep:
"""
Test suite for delete_health_sleep endpoint.
"""
@patch("health_sleep.router.health_sleep_crud.delete_health_sleep")
def test_delete_health_sleep_success(
self, mock_delete, fast_api_client, fast_api_app
):
"""
Test successful deletion of health sleep entry.
"""
# Arrange
mock_delete.return_value = None
# Act
response = fast_api_client.delete(
"/health_sleep/1",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 204
mock_delete.assert_called_once_with(1, 1, ANY)
@patch("health_sleep.router.health_sleep_crud.delete_health_sleep")
def test_delete_health_sleep_not_found(
self, mock_delete, fast_api_client, fast_api_app
):
"""
Test deletion when health sleep not found.
"""
# Arrange
mock_delete.side_effect = HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Health sleep with id 999 for user 1 not found",
)
# Act
response = fast_api_client.delete(
"/health_sleep/999",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 404

View File

@@ -0,0 +1,364 @@
import pytest
from datetime import datetime, date as datetime_date
from decimal import Decimal
from pydantic import ValidationError
import health_sleep.schema as health_sleep_schema
class TestHealthSleepSchema:
"""
Test suite for HealthSleep Pydantic schema.
"""
def test_health_sleep_valid_full_data(self):
"""
Test HealthSleep schema with all valid fields.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
sleep_start_time_gmt=datetime(2024, 1, 14, 22, 0, 0),
sleep_end_time_gmt=datetime(2024, 1, 15, 6, 0, 0),
total_sleep_seconds=28800,
deep_sleep_seconds=7200,
light_sleep_seconds=14400,
rem_sleep_seconds=7200,
avg_heart_rate=Decimal("55.5"),
min_heart_rate=45,
max_heart_rate=75,
sleep_score_overall=85,
source=health_sleep_schema.Source.GARMIN,
)
# Assert
assert health_sleep.id == 1
assert health_sleep.user_id == 1
assert health_sleep.date == datetime_date(2024, 1, 15)
assert health_sleep.total_sleep_seconds == 28800
assert health_sleep.sleep_score_overall == 85
assert health_sleep.source == "garmin"
def test_health_sleep_minimal_data(self):
"""
Test HealthSleep schema with minimal required fields.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep()
# Assert
assert health_sleep.id is None
assert health_sleep.user_id is None
assert health_sleep.date is None
assert health_sleep.total_sleep_seconds is None
def test_health_sleep_with_none_values(self):
"""
Test HealthSleep schema allows None for optional fields.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
total_sleep_seconds=28800,
deep_sleep_seconds=None,
avg_heart_rate=None,
)
# Assert
assert health_sleep.id == 1
assert health_sleep.total_sleep_seconds == 28800
assert health_sleep.deep_sleep_seconds is None
assert health_sleep.avg_heart_rate is None
def test_health_sleep_forbid_extra_fields(self):
"""
Test that HealthSleep schema forbids extra fields.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_sleep_schema.HealthSleep(
total_sleep_seconds=28800, extra_field="not allowed"
)
assert "extra_field" in str(exc_info.value)
def test_health_sleep_from_attributes(self):
"""
Test HealthSleep schema can be created from ORM model.
"""
# Arrange
class MockORMModel:
"""Mock ORM model for testing."""
id = 1
user_id = 1
date = datetime_date(2024, 1, 15)
sleep_start_time_gmt = datetime(2024, 1, 14, 22, 0, 0)
sleep_end_time_gmt = datetime(2024, 1, 15, 6, 0, 0)
sleep_start_time_local = None
sleep_end_time_local = None
total_sleep_seconds = 28800
nap_time_seconds = None
unmeasurable_sleep_seconds = None
deep_sleep_seconds = 7200
light_sleep_seconds = 14400
rem_sleep_seconds = 7200
awake_sleep_seconds = 0
avg_heart_rate = Decimal("55.5")
min_heart_rate = 45
max_heart_rate = 75
avg_spo2 = Decimal("97.5")
lowest_spo2 = 95
highest_spo2 = 99
avg_respiration = None
lowest_respiration = None
highest_respiration = None
avg_stress_level = None
awake_count = 2
restless_moments_count = 5
sleep_score_overall = 85
sleep_score_duration = "GOOD"
sleep_score_quality = "GOOD"
garminconnect_sleep_id = None
sleep_stages = None
source = "garmin"
hrv_status = None
resting_heart_rate = 50
avg_skin_temp_deviation = None
awake_count_score = None
rem_percentage_score = None
deep_percentage_score = None
light_percentage_score = None
avg_sleep_stress = None
sleep_stress_score = None
# Act
health_sleep = health_sleep_schema.HealthSleep.model_validate(MockORMModel())
# Assert
assert health_sleep.id == 1
assert health_sleep.total_sleep_seconds == 28800
assert health_sleep.source == "garmin"
def test_health_sleep_validate_assignment(self):
"""
Test that validate_assignment works correctly.
"""
# Arrange
health_sleep = health_sleep_schema.HealthSleep(total_sleep_seconds=28800)
# Act
health_sleep.total_sleep_seconds = 32400
health_sleep.sleep_score_overall = 90
# Assert
assert health_sleep.total_sleep_seconds == 32400
assert health_sleep.sleep_score_overall == 90
def test_health_sleep_heart_rate_validation_valid(self):
"""
Test heart rate validation with valid values.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep(
avg_heart_rate=Decimal("60.5"),
min_heart_rate=45,
max_heart_rate=85,
)
# Assert
assert health_sleep.avg_heart_rate == Decimal("60.5")
assert health_sleep.min_heart_rate == 45
assert health_sleep.max_heart_rate == 85
def test_health_sleep_heart_rate_validation_invalid_low(self):
"""
Test heart rate validation rejects values below 20 bpm.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_sleep_schema.HealthSleep(min_heart_rate=15)
assert "between 20 and 220" in str(exc_info.value)
def test_health_sleep_heart_rate_validation_invalid_high(self):
"""
Test heart rate validation rejects values above 220 bpm.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_sleep_schema.HealthSleep(max_heart_rate=250)
assert "between 20 and 220" in str(exc_info.value)
def test_health_sleep_spo2_validation_valid(self):
"""
Test SpO2 validation with valid values.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep(
avg_spo2=Decimal("97.5"),
lowest_spo2=95,
highest_spo2=99,
)
# Assert
assert health_sleep.avg_spo2 == Decimal("97.5")
assert health_sleep.lowest_spo2 == 95
assert health_sleep.highest_spo2 == 99
def test_health_sleep_spo2_validation_invalid_low(self):
"""
Test SpO2 validation rejects values below 70%.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_sleep_schema.HealthSleep(lowest_spo2=65)
assert "between 70 and 100" in str(exc_info.value)
def test_health_sleep_spo2_validation_invalid_high(self):
"""
Test SpO2 validation rejects values above 100%.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_sleep_schema.HealthSleep(highest_spo2=105)
assert "between 70 and 100" in str(exc_info.value)
def test_health_sleep_time_validation_valid(self):
"""
Test sleep time validation with valid start < end times.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep(
sleep_start_time_gmt=datetime(2024, 1, 14, 22, 0, 0),
sleep_end_time_gmt=datetime(2024, 1, 15, 6, 0, 0),
)
# Assert
assert health_sleep.sleep_start_time_gmt < health_sleep.sleep_end_time_gmt
def test_health_sleep_time_validation_invalid(self):
"""
Test sleep time validation rejects start >= end times.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_sleep_schema.HealthSleep(
sleep_start_time_gmt=datetime(2024, 1, 15, 6, 0, 0),
sleep_end_time_gmt=datetime(2024, 1, 14, 22, 0, 0),
)
assert "before" in str(exc_info.value)
class TestSourceEnum:
"""
Test suite for Source enum.
"""
def test_source_enum_garmin(self):
"""
Test Source enum has GARMIN value.
"""
# Arrange & Act
source = health_sleep_schema.Source.GARMIN
# Assert
assert source.value == "garmin"
def test_source_enum_use_in_schema(self):
"""
Test Source enum can be used in HealthSleep schema.
"""
# Arrange & Act
health_sleep = health_sleep_schema.HealthSleep(
source=health_sleep_schema.Source.GARMIN
)
# Assert
assert health_sleep.source == "garmin"
class TestSleepScoreEnum:
"""
Test suite for SleepScore enum.
"""
def test_sleep_score_enum_values(self):
"""
Test SleepScore enum has all expected values.
"""
# Assert
assert health_sleep_schema.SleepScore.EXCELLENT.value == "EXCELLENT"
assert health_sleep_schema.SleepScore.GOOD.value == "GOOD"
assert health_sleep_schema.SleepScore.FAIR.value == "FAIR"
assert health_sleep_schema.SleepScore.POOR.value == "POOR"
class TestHRVStatusEnum:
"""
Test suite for HRVStatus enum.
"""
def test_hrv_status_enum_values(self):
"""
Test HRVStatus enum has all expected values.
"""
# Assert
assert health_sleep_schema.HRVStatus.BALANCED.value == "BALANCED"
assert health_sleep_schema.HRVStatus.UNBALANCED.value == "UNBALANCED"
assert health_sleep_schema.HRVStatus.LOW.value == "LOW"
assert health_sleep_schema.HRVStatus.POOR.value == "POOR"
class TestHealthSleepListResponse:
"""
Test suite for HealthSleepListResponse schema.
"""
def test_health_sleep_list_response_valid(self):
"""
Test HealthSleepListResponse with valid data.
"""
# Arrange & Act
health_sleep1 = health_sleep_schema.HealthSleep(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
total_sleep_seconds=28800,
)
health_sleep2 = health_sleep_schema.HealthSleep(
id=2,
user_id=1,
date=datetime_date(2024, 1, 16),
total_sleep_seconds=32400,
)
response = health_sleep_schema.HealthSleepListResponse(
total=2, records=[health_sleep1, health_sleep2]
)
# Assert
assert response.total == 2
assert len(response.records) == 2
assert response.records[0].total_sleep_seconds == 28800
assert response.records[1].total_sleep_seconds == 32400
def test_health_sleep_list_response_empty(self):
"""
Test HealthSleepListResponse with empty records.
"""
# Arrange & Act
response = health_sleep_schema.HealthSleepListResponse(total=0, records=[])
# Assert
assert response.total == 0
assert response.records == []