github/endurain
1
1
Fork 0
mirror of https://github.com/joaovitoriasilva/endurain.git synced 2026-01-09 15:57:59 -05:00
Code Issues 45 Packages Projects Releases Wiki Activity

Refactor health_steps API and add tests

Browse Source 
Refactored backend health_steps CRUD, models, router, schema, and utils for improved type safety, docstrings, and response structure. Added HealthStepsListResponse for paginated and total-count responses. Updated router endpoints to return total counts and records, and improved error handling. Added comprehensive unit tests for health_steps CRUD, models, and router. Updated frontend service and view to match new API response format.
This commit is contained in:
João Vitória Silva
2025-12-05 12:58:05 +00:00
parent 99396433f5
commit 086efd85bb
13 changed files with 1581 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
backend/app/health_steps/crud.py
View File

@@ -9,7 +9,21 @@ import health_steps.models as health_steps_models
import core.logger as core_logger
def get_health_steps_number(user_id: int, db: Session):
def get_health_steps_number(user_id: int, db: Session) -> int:
"""
Retrieves the total count of health steps records for a specific user.
Args:
user_id (int): The unique identifier of the user whose health steps count is to be retrieved.
db (Session): The database session object used to query the database.
Returns:
int: The total number of health steps records associated with the specified user.
Raises:
HTTPException: If a database error or any other exception occurs during the query execution,
an HTTPException with status code 500 (Internal Server Error) is raised.
"""
try:
# Get the number of health_steps from the database
return (
@@ -29,22 +43,38 @@ def get_health_steps_number(user_id: int, db: Session):
) from err
def get_all_health_steps_by_user_id(user_id: int, db: Session):
def get_all_health_steps_by_user_id(
user_id: int, db: Session
) -> list[health_steps_models.HealthSteps]:
"""
Retrieve all health steps records for a specific user from the database.
This function queries the database to fetch all health steps entries associated
with the given user ID, ordered by date in descending order (most recent first).
Args:
user_id (int): The unique identifier of the user whose health steps records
are to be retrieved.
db (Session): The SQLAlchemy database session used to execute the query.
Returns:
list[HealthSteps]: A list of HealthSteps model instances representing all
health steps records for the specified user, ordered by
date in descending order.
Raises:
HTTPException: A 500 Internal Server Error exception is raised if any
database or unexpected error occurs during the query execution.
The original exception is logged before being re-raised.
"""
try:
# Get the health_steps from the database
health_steps = (
return (
db.query(health_steps_models.HealthSteps)
.filter(health_steps_models.HealthSteps.user_id == user_id)
.order_by(desc(health_steps_models.HealthSteps.date))
.all()
)
# Check if there are health_steps if not return None
if not health_steps:
return None
# Return the health_steps
return health_steps
except Exception as err:
# Log the exception
core_logger.print_to_log(
@@ -59,10 +89,30 @@ def get_all_health_steps_by_user_id(user_id: int, db: Session):
def get_health_steps_with_pagination(
user_id: int, db: Session, page_number: int = 1, num_records: int = 5
):
) -> list[health_steps_models.HealthSteps]:
"""
Retrieve paginated health steps records for a specific user.
This function queries the database to fetch health steps records for a given user,
with support for pagination. Results are ordered by date in descending order.
Args:
user_id (int): The ID of the user whose health steps records are to be retrieved.
db (Session): The database session object used for querying.
page_number (int, optional): The page number to retrieve. Defaults to 1.
num_records (int, optional): The number of records per page. Defaults to 5.
Returns:
list[health_steps_models.HealthSteps]: A list of HealthSteps model instances
representing the user's health steps records for the specified page.
Raises:
HTTPException: A 500 Internal Server Error exception is raised if any
database operation fails or an unexpected error occurs during execution.
"""
try:
# Get the health_steps from the database
health_steps = (
return (
db.query(health_steps_models.HealthSteps)
.filter(health_steps_models.HealthSteps.user_id == user_id)
.order_by(desc(health_steps_models.HealthSteps.date))
@@ -70,13 +120,6 @@ def get_health_steps_with_pagination(
.limit(num_records)
.all()
)
# Check if there are health_steps if not return None
if not health_steps:
return None
# Return the health_steps
return health_steps
except Exception as err:
# Log the exception
core_logger.print_to_log(
@@ -89,10 +132,31 @@ def get_health_steps_with_pagination(
) from err
def get_health_steps_by_date(user_id: int, date: str, db: Session):
def get_health_steps_by_date(
user_id: int, date: str, db: Session
) -> health_steps_models.HealthSteps | None:
"""
Retrieve health steps data for a specific user and date.
This function queries the database to find health steps records matching
the specified user ID and date.
Args:
user_id (int): The unique identifier of the user.
date (str): The date for which to retrieve health steps data (format: YYYY-MM-DD).
db (Session): The database session object for executing queries.
Returns:
health_steps_models.HealthSteps | None: The health steps record if found,
otherwise None.
Raises:
HTTPException: A 500 Internal Server Error if an exception occurs during
the database query operation.
"""
try:
# Get the health_steps from the database
health_steps = (
return (
db.query(health_steps_models.HealthSteps)
.filter(
health_steps_models.HealthSteps.date == date,
@@ -100,13 +164,6 @@ def get_health_steps_by_date(user_id: int, date: str, db: Session):
)
.first()
)
# Check if there are health_steps if not return None
if not health_steps:
return None
# Return the health_steps
return health_steps
except Exception as err:
# Log the exception
core_logger.print_to_log(
@@ -121,7 +178,27 @@ def get_health_steps_by_date(user_id: int, date: str, db: Session):
def create_health_steps(
user_id: int, health_steps: health_steps_schema.HealthSteps, db: Session
):
) -> health_steps_schema.HealthSteps:
"""
Create a new health steps record for a user.
This function creates a new health steps entry in the database for the specified user.
If no date is provided, it defaults to the current date/time.
Args:
user_id (int): The ID of the user for whom the health steps record is being created.
health_steps (health_steps_schema.HealthSteps): The health steps data to be created.
The 'id' and 'user_id' fields are excluded from the input as they are set internally.
db (Session): The database session for executing the database operations.
Returns:
health_steps_schema.HealthSteps: The created health steps record with the assigned ID.
Raises:
HTTPException:
- 409 Conflict: If a health steps entry already exists for the given date.
- 500 Internal Server Error: If any other unexpected error occurs during creation.
"""
try:
# Check if date is None
if health_steps.date is None:
@@ -170,7 +247,30 @@ def create_health_steps(
def edit_health_steps(
user_id, health_steps: health_steps_schema.HealthSteps, db: Session
):
) -> health_steps_schema.HealthSteps:
"""
Edit health steps record for a specific user.
This function updates an existing health steps record in the database for a given user.
It performs validation to ensure the record exists and belongs to the specified user
before applying updates.
Args:
user_id: The ID of the user who owns the health steps record.
health_steps (health_steps_schema.HealthSteps): The health steps object containing
the ID of the record to update and the fields to be modified.
db (Session): The database session used for querying and committing changes.
Returns:
health_steps_schema.HealthSteps: The updated health steps object.
Raises:
HTTPException:
- 404 NOT_FOUND: If the health steps record is not found or doesn't belong
to the specified user.
- 500 INTERNAL_SERVER_ERROR: If an unexpected error occurs during the update
process.
"""
try:
# Get the health_steps from the database
db_health_steps = (
@@ -215,7 +315,29 @@ def edit_health_steps(
) from err
def delete_health_steps(user_id: int, health_steps_id: int, db: Session):
def delete_health_steps(user_id: int, health_steps_id: int, db: Session) -> None:
"""
Delete a health steps record for a specific user.
This function deletes a health steps entry from the database based on the provided
health_steps_id and user_id. It ensures that the record belongs to the specified user
before deletion.
Args:
user_id (int): The ID of the user who owns the health steps record.
health_steps_id (int): The ID of the health steps record to delete.
db (Session): The database session object for executing queries.
Returns:
None
Raises:
HTTPException:
- 404 NOT FOUND if the health steps record with the given ID
for the specified user is not found.
- 500 INTERNAL SERVER ERROR if any other exception occurs during
the deletion process.
"""
try:
# Delete the health_steps
num_deleted = (

26
backend/app/health_steps/models.py
View File

@@ -4,13 +4,37 @@ from sqlalchemy import (
String,
Date,
ForeignKey,
DECIMAL,
)
from sqlalchemy.orm import relationship
from core.database import Base
class HealthSteps(Base):
"""
SQLAlchemy model representing daily step count data for users.
This model stores health and fitness tracking data related to the number of steps
taken by a user on a specific date. It includes information about the data source
and maintains a relationship with the User model.
Attributes:
id (int): Primary key, auto-incremented unique identifier for each record.
user_id (int): Foreign key referencing users.id. Identifies the user who owns
this health data. Cascades on delete and is indexed for performance.
date (Date): The date for which the step count is recorded.
steps (int): The total number of steps taken on the specified date.
source (str, optional): The source or origin of the step data (e.g., fitness
device, mobile app). Maximum length of 250 characters.
user (relationship): SQLAlchemy relationship to the User model, enabling
bidirectional access between users and their health step records.
Table:
health_steps
Relationships:
- Many-to-One with User model through user_id
"""
__tablename__ = "health_steps"
id = Column(Integer, primary_key=True, autoincrement=True)

175
backend/app/health_steps/router.py
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_steps.schema as health_steps_schema
@@ -15,30 +15,9 @@ import core.dependencies as core_dependencies
router = APIRouter()
@router.get(
"/number",
response_model=int,
)
async def read_health_steps_number(
_check_scopes: Annotated[
Callable, Security(auth_security.check_scopes, scopes=["health:read"])
],
token_user_id: Annotated[
int,
Depends(auth_security.get_sub_from_access_token),
],
db: Annotated[
Session,
Depends(core_database.get_db),
],
):
# Get the health_steps number from the database
return health_steps_crud.get_health_steps_number(token_user_id, db)
@router.get(
"",
response_model=list[health_steps_schema.HealthSteps] | None,
response_model=health_steps_schema.HealthStepsListResponse,
)
async def read_health_steps_all(
_check_scopes: Annotated[
@@ -52,14 +31,37 @@ async def read_health_steps_all(
Session,
Depends(core_database.get_db),
],
):
# Get the health_steps from the database
return health_steps_crud.get_all_health_steps_by_user_id(token_user_id, db)
) -> health_steps_schema.HealthStepsListResponse:
"""
Retrieve all health steps records for the authenticated user.
This endpoint fetches all health steps entries associated with the authenticated user's ID.
It requires the 'health:read' scope for authorization.
Args:
_check_scopes (Callable): Security dependency that validates the required scopes.
token_user_id (int): The user ID extracted from the access token.
db (Session): Database session dependency for querying the database.
Returns:
HealthStepsListResponse: A response object containing:
- total (int): The total number of health steps records for the user.
- records (List): A list of all health steps records for the user.
Raises:
HTTPException: May raise authentication or authorization related exceptions
if the token is invalid or the user lacks required permissions.
"""
# Get the total count and records from the database
total = health_steps_crud.get_health_steps_number(token_user_id, db)
records = health_steps_crud.get_all_health_steps_by_user_id(token_user_id, db)
return health_steps_schema.HealthStepsListResponse(total=total, records=records)
@router.get(
"/page_number/{page_number}/num_records/{num_records}",
response_model=list[health_steps_schema.HealthSteps] | None,
response_model=health_steps_schema.HealthStepsListResponse,
)
async def read_health_steps_all_pagination(
page_number: int,
@@ -67,7 +69,7 @@ async def read_health_steps_all_pagination(
_check_scopes: Annotated[
Callable, Security(auth_security.check_scopes, scopes=["health:read"])
],
validate_pagination_values: Annotated[
_validate_pagination_values: Annotated[
Callable, Depends(core_dependencies.validate_pagination_values)
],
token_user_id: Annotated[
@@ -78,12 +80,39 @@ async def read_health_steps_all_pagination(
Session,
Depends(core_database.get_db),
],
):
# Get the health_steps from the database with pagination
return health_steps_crud.get_health_steps_with_pagination(
) -> health_steps_schema.HealthStepsListResponse:
"""
Retrieve paginated health steps records for the authenticated user.
This endpoint returns a paginated list of health steps data for the user identified
by the access token. It enforces proper authentication, authorization (health:read scope),
and pagination parameter validation.
Args:
page_number (int): The page number to retrieve (1-indexed).
num_records (int): The number of records per page.
_check_scopes (Callable): Dependency that validates the user has 'health:read' scope.
validate_pagination_values (Callable): Dependency that validates pagination parameters.
token_user_id (int): The user ID extracted from the access token.
db (Session): Database session dependency.
Returns:
HealthStepsListResponse: A response object containing:
- total (int): The total number of health steps records for the user.
- records (list): A list of paginated health steps records.
Raises:
HTTPException: If authentication fails, authorization is denied, or pagination
parameters are invalid.
"""
# Get the total count and paginated records from the database
total = health_steps_crud.get_health_steps_number(token_user_id, db)
records = health_steps_crud.get_health_steps_with_pagination(
token_user_id, db, page_number, num_records
)
return health_steps_schema.HealthStepsListResponse(total=total, records=records)
@router.post("", status_code=201)
async def create_health_steps(
@@ -99,10 +128,38 @@ async def create_health_steps(
Session,
Depends(core_database.get_db),
],
):
) -> health_steps_schema.HealthSteps:
"""
Create or update health steps data for a user.
This endpoint creates new health steps data or updates existing data if an entry
for the specified date already exists. The operation is determined automatically
based on whether steps data exists for the given date.
Args:
health_steps (health_steps_schema.HealthSteps): The health steps data to create
or update, including the date and step count.
_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_steps_schema.HealthSteps: The created or updated health steps data.
Raises:
HTTPException: 400 error if the date field is not provided in the request.
"""
if not health_steps.date:
raise HTTPException(status_code=400, detail="Date field is required.")
# Convert date to string format for CRUD function
date_str = health_steps.date.isoformat()
# Check if health_steps for this date already exists
steps_for_date = health_steps_crud.get_health_steps_by_date(
token_user_id, health_steps.date, db
token_user_id, date_str, db
)
if steps_for_date:
@@ -128,12 +185,35 @@ async def edit_health_steps(
Session,
Depends(core_database.get_db),
],
):
) -> health_steps_schema.HealthSteps:
"""
Edit health steps data for a user.
This endpoint updates existing health steps records in the database for the authenticated user.
Requires 'health:write' scope for authorization.
Args:
health_steps (health_steps_schema.HealthSteps): The health steps data to be updated,
containing the new values for the health steps 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_steps_schema.HealthSteps: The updated health steps 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_steps in the database and returns it
return health_steps_crud.edit_health_steps(token_user_id, health_steps, db)
@router.delete("/{health_steps_id}")
@router.delete("/{health_steps_id}", status_code=204)
async def delete_health_steps(
health_steps_id: int,
_check_scopes: Annotated[
@@ -147,6 +227,27 @@ async def delete_health_steps(
Session,
Depends(core_database.get_db),
],
):
) -> None:
"""
Delete a health steps record for the authenticated user.
This endpoint removes a specific health steps entry from the database for the user
identified by the access token. The user must have 'health:write' scope permission.
Args:
health_steps_id (int): The unique identifier of the health steps record to delete.
_check_scopes (Callable): Security dependency that verifies the user has 'health:write' scope.
token_user_id (int): The user ID extracted from the access token.
db (Session): Database session dependency for executing the delete operation.
Returns:
None: This function does not return a value.
Raises:
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
"""
# Deletes entry from database
return health_steps_crud.delete_health_steps(token_user_id, health_steps_id, db)
health_steps_crud.delete_health_steps(token_user_id, health_steps_id, db)

44
backend/app/health_steps/schema.py
View File

@@ -15,6 +15,23 @@ class Source(Enum):
class HealthSteps(BaseModel):
"""
Pydantic model for health steps data.
Attributes:
id (int | None): Unique identifier for the health steps record. Defaults to None.
user_id (int | None): ID of the user associated with the steps record. Defaults to None.
date (datetime_date | None): Date when the steps were recorded. Defaults to None.
steps (int | None): Number of steps recorded. Defaults to None.
source (Source | None): Source from which the steps data was obtained. Defaults to None.
Configuration:
- from_attributes: Allows creation from ORM model attributes
- extra: Forbids extra fields not defined in the model
- validate_assignment: Validates field values on assignment
- use_enum_values: Uses enum values instead of enum instances
"""
id: int | None = None
user_id: int | None = None
date: datetime_date | None = None
@@ -27,3 +44,30 @@ class HealthSteps(BaseModel):
validate_assignment=True,
use_enum_values=True,
)
class HealthStepsListResponse(BaseModel):
"""
Response schema for health steps list with total count.
This class wraps a list of health steps records along with the total count,
providing a complete response for list endpoints.
Attributes:
total (int): Total number of steps records for the user.
records (list[HealthSteps]): List of health steps measurements.
Configuration:
- from_attributes: Enables population from ORM models
- extra: Forbids extra fields not defined in the schema
- validate_assignment: Validates values on assignment
"""
total: int
records: list[HealthSteps]
model_config = ConfigDict(
from_attributes=True,
extra="forbid",
validate_assignment=True,
)

2
backend/app/health_weight/router.py
View File

@@ -69,7 +69,7 @@ async def read_health_weight_all_pagination(
_check_scopes: Annotated[
Callable, Security(auth_security.check_scopes, scopes=["health:read"])
],
validate_pagination_values: Annotated[
_validate_pagination_values: Annotated[
Callable, Depends(core_dependencies.validate_pagination_values)
],
token_user_id: Annotated[

3
backend/tests/conftest.py
View File

@@ -26,6 +26,7 @@ import users.user.schema as user_schema
# Variables and constants
DEFAULT_ROUTER_MODULES = [
"session.router",
"health_steps.router",
"health_weight.router",
]
@@ -139,6 +140,8 @@ def _include_router_if_exists(app: FastAPI, dotted: str):
# Add prefix for health_weight router
if dotted == "health_weight.router":
app.include_router(router, prefix="/health_weight")
elif dotted == "health_steps.router":
app.include_router(router, prefix="/health_steps")
else:
app.include_router(router)
except Exception:

1
backend/tests/health_steps/__init__.py Normal file
View File

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

513
backend/tests/health_steps/test_crud.py Normal file
View File

@@ -0,0 +1,513 @@
import pytest
from datetime import date as datetime_date
from unittest.mock import MagicMock, patch
from fastapi import HTTPException, status
from sqlalchemy.exc import IntegrityError
import health_steps.crud as health_steps_crud
import health_steps.schema as health_steps_schema
import health_steps.models as health_steps_models
class TestGetHealthStepsNumber:
"""
Test suite for get_health_steps_number function.
"""
def test_get_health_steps_number_success(self, mock_db):
"""
Test successful count of health steps 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_steps_crud.get_health_steps_number(user_id, mock_db)
# Assert
assert result == expected_count
mock_db.query.assert_called_once_with(health_steps_models.HealthSteps)
def test_get_health_steps_number_zero(self, mock_db):
"""
Test count when user has no health steps records.
"""
# Arrange
user_id = 1
mock_query = mock_db.query.return_value
mock_query.filter.return_value.count.return_value = 0
# Act
result = health_steps_crud.get_health_steps_number(user_id, mock_db)
# Assert
assert result == 0
def test_get_health_steps_number_exception(self, mock_db):
"""
Test exception handling in get_health_steps_number.
"""
# Arrange
user_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_steps_crud.get_health_steps_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 TestGetAllHealthStepsByUserId:
"""
Test suite for get_all_health_steps_by_user_id function.
"""
def test_get_all_health_steps_by_user_id_success(self, mock_db):
"""
Test successful retrieval of all health steps records for user.
"""
# Arrange
user_id = 1
mock_steps1 = MagicMock(spec=health_steps_models.HealthSteps)
mock_steps2 = MagicMock(spec=health_steps_models.HealthSteps)
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_filter.order_by.return_value.all.return_value = [
mock_steps1,
mock_steps2,
]
# Act
result = health_steps_crud.get_all_health_steps_by_user_id(user_id, mock_db)
# Assert
assert result == [mock_steps1, mock_steps2]
mock_db.query.assert_called_once_with(health_steps_models.HealthSteps)
def test_get_all_health_steps_by_user_id_empty(self, mock_db):
"""
Test retrieval when user has no health steps 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_steps_crud.get_all_health_steps_by_user_id(user_id, mock_db)
# Assert
assert result == []
def test_get_all_health_steps_by_user_id_exception(self, mock_db):
"""
Test exception handling in get_all_health_steps_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_steps_crud.get_all_health_steps_by_user_id(user_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestGetHealthStepsWithPagination:
"""
Test suite for get_health_steps_with_pagination function.
"""
def test_get_health_steps_with_pagination_success(self, mock_db):
"""
Test successful retrieval of paginated health steps records.
"""
# Arrange
user_id = 1
page_number = 2
num_records = 5
mock_steps1 = MagicMock(spec=health_steps_models.HealthSteps)
mock_steps2 = MagicMock(spec=health_steps_models.HealthSteps)
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_steps1,
mock_steps2,
]
# Act
result = health_steps_crud.get_health_steps_with_pagination(
user_id, mock_db, page_number, num_records
)
# Assert
assert result == [mock_steps1, mock_steps2]
mock_order.offset.assert_called_once_with(5)
mock_offset.limit.assert_called_once_with(5)
def test_get_health_steps_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_steps_crud.get_health_steps_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_steps_with_pagination_exception(self, mock_db):
"""
Test exception handling in get_health_steps_with_pagination.
"""
# Arrange
user_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_steps_crud.get_health_steps_with_pagination(user_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestGetHealthStepsByDate:
"""
Test suite for get_health_steps_by_date function.
"""
def test_get_health_steps_by_date_success(self, mock_db):
"""
Test successful retrieval of health steps by date.
"""
# Arrange
user_id = 1
test_date = "2024-01-15"
mock_steps = MagicMock(spec=health_steps_models.HealthSteps)
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = mock_steps
# Act
result = health_steps_crud.get_health_steps_by_date(user_id, test_date, mock_db)
# Assert
assert result == mock_steps
mock_db.query.assert_called_once_with(health_steps_models.HealthSteps)
def test_get_health_steps_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_steps_crud.get_health_steps_by_date(user_id, test_date, mock_db)
# Assert
assert result is None
def test_get_health_steps_by_date_exception(self, mock_db):
"""
Test exception handling in get_health_steps_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_steps_crud.get_health_steps_by_date(user_id, test_date, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestCreateHealthSteps:
"""
Test suite for create_health_steps function.
"""
def test_create_health_steps_success(self, mock_db):
"""
Test successful creation of health steps entry.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
date=datetime_date(2024, 1, 15),
steps=10000,
source="garmin",
)
mock_db_steps = MagicMock()
mock_db_steps.id = 1
mock_db.add.return_value = None
mock_db.commit.return_value = None
mock_db.refresh.return_value = None
with patch.object(
health_steps_models,
"HealthSteps",
return_value=mock_db_steps,
):
# Act
result = health_steps_crud.create_health_steps(
user_id, health_steps, mock_db
)
# Assert
assert result.id == 1
assert result.steps == 10000
mock_db.add.assert_called_once()
mock_db.commit.assert_called_once()
mock_db.refresh.assert_called_once()
@patch("health_steps.crud.func")
def test_create_health_steps_with_none_date(self, mock_func, mock_db):
"""
Test creation with None date sets current date.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
date=None, steps=10000, source="garmin"
)
# Mock func.now() to return a proper date object
mock_func.now.return_value = datetime_date(2024, 1, 15)
mock_db_steps = MagicMock()
mock_db_steps.id = 1
mock_db.add.return_value = None
mock_db.commit.return_value = None
mock_db.refresh.return_value = None
with patch.object(
health_steps_models,
"HealthSteps",
return_value=mock_db_steps,
):
# Act
result = health_steps_crud.create_health_steps(
user_id, health_steps, 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_steps_duplicate_entry(self, mock_db):
"""
Test creation with duplicate entry raises conflict error.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
date=datetime_date(2024, 1, 15), steps=10000, source="garmin"
)
mock_db_steps = MagicMock()
mock_db.add.return_value = None
mock_db.commit.side_effect = IntegrityError("Duplicate entry", None, None)
with patch.object(
health_steps_models,
"HealthSteps",
return_value=mock_db_steps,
):
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_steps_crud.create_health_steps(user_id, health_steps, 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_steps_exception(self, mock_db):
"""
Test exception handling in create_health_steps.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
date=datetime_date(2024, 1, 15), steps=10000
)
mock_db.add.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_steps_crud.create_health_steps(user_id, health_steps, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
mock_db.rollback.assert_called_once()
class TestEditHealthSteps:
"""
Test suite for edit_health_steps function.
"""
def test_edit_health_steps_success(self, mock_db):
"""
Test successful edit of health steps entry.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
id=1, date=datetime_date(2024, 1, 15), steps=12000
)
mock_db_steps = MagicMock(spec=health_steps_models.HealthSteps)
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = mock_db_steps
# Act
result = health_steps_crud.edit_health_steps(user_id, health_steps, mock_db)
# Assert
assert result.steps == 12000
mock_db.commit.assert_called_once()
def test_edit_health_steps_not_found(self, mock_db):
"""
Test edit when health steps record not found.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
id=999, date=datetime_date(2024, 1, 15), steps=12000
)
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_steps_crud.edit_health_steps(user_id, health_steps, mock_db)
assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
assert exc_info.value.detail == "Health steps not found"
def test_edit_health_steps_update_multiple_fields(self, mock_db):
"""
Test edit updates multiple fields correctly.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(
id=1,
date=datetime_date(2024, 1, 15),
steps=15000,
source="garmin",
)
mock_db_steps = MagicMock(spec=health_steps_models.HealthSteps)
mock_query = mock_db.query.return_value
mock_query.filter.return_value.first.return_value = mock_db_steps
# Act
result = health_steps_crud.edit_health_steps(user_id, health_steps, mock_db)
# Assert
mock_db.commit.assert_called_once()
def test_edit_health_steps_exception(self, mock_db):
"""
Test exception handling in edit_health_steps.
"""
# Arrange
user_id = 1
health_steps = health_steps_schema.HealthSteps(id=1, steps=12000)
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_steps_crud.edit_health_steps(user_id, health_steps, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
mock_db.rollback.assert_called_once()
class TestDeleteHealthSteps:
"""
Test suite for delete_health_steps function.
"""
def test_delete_health_steps_success(self, mock_db):
"""
Test successful deletion of health steps entry.
"""
# Arrange
user_id = 1
health_steps_id = 1
mock_query = mock_db.query.return_value
mock_filter = mock_query.filter.return_value
mock_filter.delete.return_value = 1
# Act
health_steps_crud.delete_health_steps(user_id, health_steps_id, mock_db)
# Assert
mock_db.commit.assert_called_once()
mock_db.query.assert_called_once_with(health_steps_models.HealthSteps)
def test_delete_health_steps_not_found(self, mock_db):
"""
Test deletion when health steps record not found.
"""
# Arrange
user_id = 1
health_steps_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_steps_crud.delete_health_steps(user_id, health_steps_id, mock_db)
assert exc_info.value.status_code == status.HTTP_404_NOT_FOUND
assert f"Health steps with id {health_steps_id}" in exc_info.value.detail
def test_delete_health_steps_exception(self, mock_db):
"""
Test exception handling in delete_health_steps.
"""
# Arrange
user_id = 1
health_steps_id = 1
mock_db.query.side_effect = Exception("Database error")
# Act & Assert
with pytest.raises(HTTPException) as exc_info:
health_steps_crud.delete_health_steps(user_id, health_steps_id, mock_db)
assert exc_info.value.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
mock_db.rollback.assert_called_once()

94
backend/tests/health_steps/test_models.py Normal file
View File

@@ -0,0 +1,94 @@
import pytest
from datetime import date as datetime_date
import health_steps.models as health_steps_models
class TestHealthStepsModel:
"""
Test suite for HealthSteps SQLAlchemy model.
"""
def test_health_steps_model_table_name(self):
"""
Test HealthSteps model has correct table name.
"""
# Assert
assert health_steps_models.HealthSteps.__tablename__ == "health_steps"
def test_health_steps_model_columns_exist(self):
"""
Test HealthSteps model has all expected columns.
"""
# Assert
assert hasattr(health_steps_models.HealthSteps, "id")
assert hasattr(health_steps_models.HealthSteps, "user_id")
assert hasattr(health_steps_models.HealthSteps, "date")
assert hasattr(health_steps_models.HealthSteps, "steps")
assert hasattr(health_steps_models.HealthSteps, "source")
def test_health_steps_model_primary_key(self):
"""
Test HealthSteps model has correct primary key.
"""
# Arrange
id_column = health_steps_models.HealthSteps.id
# Assert
assert id_column.primary_key is True
assert id_column.autoincrement is True
def test_health_steps_model_foreign_key(self):
"""
Test HealthSteps model has correct foreign key.
"""
# Arrange
user_id_column = health_steps_models.HealthSteps.user_id
# Assert
assert user_id_column.nullable is False
assert user_id_column.index is True
def test_health_steps_model_nullable_fields(self):
"""
Test HealthSteps model nullable fields.
"""
# Assert
assert health_steps_models.HealthSteps.source.nullable is True
def test_health_steps_model_required_fields(self):
"""
Test HealthSteps model required fields.
"""
# Assert
assert health_steps_models.HealthSteps.user_id.nullable is False
assert health_steps_models.HealthSteps.date.nullable is False
assert health_steps_models.HealthSteps.steps.nullable is False
def test_health_steps_model_column_types(self):
"""
Test HealthSteps model column types.
"""
# Assert
assert health_steps_models.HealthSteps.id.type.python_type == int
assert health_steps_models.HealthSteps.user_id.type.python_type == int
assert health_steps_models.HealthSteps.date.type.python_type == datetime_date
assert health_steps_models.HealthSteps.steps.type.python_type == int
assert health_steps_models.HealthSteps.source.type.python_type == str
def test_health_steps_model_relationship(self):
"""
Test HealthSteps model has relationship to User.
"""
# Assert
assert hasattr(health_steps_models.HealthSteps, "user")
def test_health_steps_model_source_max_length(self):
"""
Test HealthSteps model source field has correct max length.
"""
# Arrange
source_column = health_steps_models.HealthSteps.source
# Assert
assert source_column.type.length == 250

343
backend/tests/health_steps/test_router.py Normal file
View File

@@ -0,0 +1,343 @@
import pytest
from datetime import date as datetime_date
from unittest.mock import MagicMock, patch, ANY
from fastapi import HTTPException, status
import health_steps.schema as health_steps_schema
import health_steps.models as health_steps_models
class TestReadHealthStepsAll:
"""
Test suite for read_health_steps_all endpoint.
"""
@patch("health_steps.router.health_steps_crud.get_health_steps_number")
@patch("health_steps.router.health_steps_crud.get_all_health_steps_by_user_id")
def test_read_health_steps_all_success(
self, mock_get_all, mock_get_number, fast_api_client, fast_api_app
):
"""
Test successful retrieval of all health steps records with total count.
"""
# Arrange
mock_steps1 = MagicMock(spec=health_steps_models.HealthSteps)
mock_steps1.id = 1
mock_steps1.user_id = 1
mock_steps1.date = datetime_date(2024, 1, 15)
mock_steps1.steps = 10000
mock_steps1.source = None
mock_steps2 = MagicMock(spec=health_steps_models.HealthSteps)
mock_steps2.id = 2
mock_steps2.user_id = 1
mock_steps2.date = datetime_date(2024, 1, 16)
mock_steps2.steps = 12000
mock_steps2.source = None
mock_get_all.return_value = [mock_steps1, mock_steps2]
mock_get_number.return_value = 2
# Act
response = fast_api_client.get(
"/health_steps",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total"] == 2
assert len(data["records"]) == 2
@patch("health_steps.router.health_steps_crud.get_health_steps_number")
@patch("health_steps.router.health_steps_crud.get_all_health_steps_by_user_id")
def test_read_health_steps_all_empty(
self, mock_get_all, mock_get_number, fast_api_client, fast_api_app
):
"""
Test retrieval when user has no health steps records.
"""
# Arrange
mock_get_all.return_value = []
mock_get_number.return_value = 0
# Act
response = fast_api_client.get(
"/health_steps",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["total"] == 0
assert data["records"] == []
class TestReadHealthStepsAllPagination:
"""
Test suite for read_health_steps_all_pagination endpoint.
"""
@patch("health_steps.router.health_steps_crud.get_health_steps_number")
@patch("health_steps.router.health_steps_crud.get_health_steps_with_pagination")
def test_read_health_steps_all_pagination_success(
self, mock_get_paginated, mock_get_number, fast_api_client, fast_api_app
):
"""
Test successful retrieval of paginated health steps records with total count.
"""
# Arrange
mock_steps1 = MagicMock(spec=health_steps_models.HealthSteps)
mock_steps1.id = 1
mock_steps1.user_id = 1
mock_steps1.date = datetime_date(2024, 1, 15)
mock_steps1.steps = 10000
mock_steps1.source = None
mock_get_paginated.return_value = [mock_steps1]
mock_get_number.return_value = 10
# Act
response = fast_api_client.get(
"/health_steps/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_steps.router.health_steps_crud.get_health_steps_number")
@patch("health_steps.router.health_steps_crud.get_health_steps_with_pagination")
def test_read_health_steps_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_steps/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 TestCreateHealthSteps:
"""
Test suite for create_health_steps endpoint.
"""
@patch("health_steps.router.health_steps_crud.create_health_steps")
@patch("health_steps.router.health_steps_crud.get_health_steps_by_date")
def test_create_health_steps_success(
self,
mock_get_by_date,
mock_create,
fast_api_client,
fast_api_app,
):
"""
Test successful creation of health steps entry.
"""
# Arrange
mock_get_by_date.return_value = None
created_steps = health_steps_schema.HealthSteps(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
steps=10000,
)
mock_create.return_value = created_steps
# Act
response = fast_api_client.post(
"/health_steps",
json={
"date": "2024-01-15",
"steps": 10000,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 201
data = response.json()
assert data["steps"] == 10000
@patch("health_steps.router.health_steps_crud.edit_health_steps")
@patch("health_steps.router.health_steps_crud.get_health_steps_by_date")
def test_create_health_steps_updates_existing(
self, mock_get_by_date, mock_edit, fast_api_client, fast_api_app
):
"""
Test creating health steps when entry exists updates it.
"""
# Arrange
existing_steps = MagicMock()
existing_steps.id = 1
mock_get_by_date.return_value = existing_steps
updated_steps = health_steps_schema.HealthSteps(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
steps=12000,
)
mock_edit.return_value = updated_steps
# Act
response = fast_api_client.post(
"/health_steps",
json={
"date": "2024-01-15",
"steps": 12000,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 201
mock_edit.assert_called_once()
def test_create_health_steps_missing_date(self, fast_api_client, fast_api_app):
"""
Test creating health steps without date field raises error.
"""
# Act
response = fast_api_client.post(
"/health_steps",
json={
"steps": 10000,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 400
assert "Date field is required" in response.json()["detail"]
class TestEditHealthSteps:
"""
Test suite for edit_health_steps endpoint.
"""
@patch("health_steps.router.health_steps_crud.edit_health_steps")
def test_edit_health_steps_success(self, mock_edit, fast_api_client, fast_api_app):
"""
Test successful edit of health steps entry.
"""
# Arrange
updated_steps = health_steps_schema.HealthSteps(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
steps=12000,
)
mock_edit.return_value = updated_steps
# Act
response = fast_api_client.put(
"/health_steps",
json={
"id": 1,
"date": "2024-01-15",
"steps": 12000,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 200
data = response.json()
assert data["steps"] == 12000
@patch("health_steps.router.health_steps_crud.edit_health_steps")
def test_edit_health_steps_not_found(
self, mock_edit, fast_api_client, fast_api_app
):
"""
Test edit when health steps not found.
"""
# Arrange
mock_edit.side_effect = HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Health steps not found",
)
# Act
response = fast_api_client.put(
"/health_steps",
json={
"id": 999,
"date": "2024-01-15",
"steps": 12000,
},
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 404
class TestDeleteHealthSteps:
"""
Test suite for delete_health_steps endpoint.
"""
@patch("health_steps.router.health_steps_crud.delete_health_steps")
def test_delete_health_steps_success(
self, mock_delete, fast_api_client, fast_api_app
):
"""
Test successful deletion of health steps entry.
"""
# Arrange
mock_delete.return_value = None
# Act
response = fast_api_client.delete(
"/health_steps/1",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 204
mock_delete.assert_called_once_with(1, 1, ANY)
@patch("health_steps.router.health_steps_crud.delete_health_steps")
def test_delete_health_steps_not_found(
self, mock_delete, fast_api_client, fast_api_app
):
"""
Test deletion when health steps not found.
"""
# Arrange
mock_delete.side_effect = HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Health steps with id 999 for user 1 not found",
)
# Act
response = fast_api_client.delete(
"/health_steps/999",
headers={"Authorization": "Bearer mock_token"},
)
# Assert
assert response.status_code == 404

257
backend/tests/health_steps/test_schema.py Normal file
View File

@@ -0,0 +1,257 @@
import pytest
from datetime import date as datetime_date
from pydantic import ValidationError
import health_steps.schema as health_steps_schema
class TestHealthStepsSchema:
"""
Test suite for HealthSteps Pydantic schema.
"""
def test_health_steps_valid_full_data(self):
"""
Test HealthSteps schema with all valid fields.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
steps=10000,
source=health_steps_schema.Source.GARMIN,
)
# Assert
assert health_steps.id == 1
assert health_steps.user_id == 1
assert health_steps.date == datetime_date(2024, 1, 15)
assert health_steps.steps == 10000
assert health_steps.source == "garmin"
def test_health_steps_minimal_data(self):
"""
Test HealthSteps schema with minimal required fields.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps()
# Assert
assert health_steps.id is None
assert health_steps.user_id is None
assert health_steps.date is None
assert health_steps.steps is None
assert health_steps.source is None
def test_health_steps_with_none_values(self):
"""
Test HealthSteps schema allows None for optional fields.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
steps=10000,
source=None,
)
# Assert
assert health_steps.id == 1
assert health_steps.steps == 10000
assert health_steps.source is None
def test_health_steps_with_integer_steps(self):
"""
Test HealthSteps schema with integer steps values.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(steps=5000)
# Assert
assert health_steps.steps == 5000
assert isinstance(health_steps.steps, int)
def test_health_steps_forbid_extra_fields(self):
"""
Test that HealthSteps schema forbids extra fields.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_steps_schema.HealthSteps(steps=10000, extra_field="not allowed")
assert "extra_field" in str(exc_info.value)
def test_health_steps_from_attributes(self):
"""
Test HealthSteps 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)
steps = 10000
source = "garmin"
# Act
health_steps = health_steps_schema.HealthSteps.model_validate(MockORMModel())
# Assert
assert health_steps.id == 1
assert health_steps.steps == 10000
assert health_steps.source == "garmin"
def test_health_steps_validate_assignment(self):
"""
Test that validate_assignment works correctly.
"""
# Arrange
health_steps = health_steps_schema.HealthSteps(steps=10000)
# Act
health_steps.steps = 12000
health_steps.date = datetime_date(2024, 1, 16)
# Assert
assert health_steps.steps == 12000
assert health_steps.date == datetime_date(2024, 1, 16)
def test_health_steps_date_validation(self):
"""
Test date field validation.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(date=datetime_date(2024, 12, 31))
# Assert
assert health_steps.date == datetime_date(2024, 12, 31)
def test_health_steps_zero_steps(self):
"""
Test HealthSteps schema accepts zero steps.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(steps=0)
# Assert
assert health_steps.steps == 0
def test_health_steps_large_steps_value(self):
"""
Test HealthSteps schema with large steps values.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(steps=50000)
# Assert
assert health_steps.steps == 50000
class TestSourceEnum:
"""
Test suite for Source enum.
"""
def test_source_enum_garmin(self):
"""
Test Source enum has GARMIN value.
"""
# Arrange & Act
source = health_steps_schema.Source.GARMIN
# Assert
assert source.value == "garmin"
def test_source_enum_use_in_schema(self):
"""
Test Source enum can be used in HealthSteps schema.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(
source=health_steps_schema.Source.GARMIN
)
# Assert
assert health_steps.source == "garmin"
def test_source_enum_string_value(self):
"""
Test Source enum accepts string value directly.
"""
# Arrange & Act
health_steps = health_steps_schema.HealthSteps(source="garmin")
# Assert
assert health_steps.source == "garmin"
def test_source_enum_invalid_value(self):
"""
Test Source enum rejects invalid values.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_steps_schema.HealthSteps(source="invalid")
assert "source" in str(exc_info.value)
class TestHealthStepsListResponse:
"""
Test suite for HealthStepsListResponse schema.
"""
def test_health_steps_list_response_valid(self):
"""
Test HealthStepsListResponse with valid data.
"""
# Arrange & Act
health_steps1 = health_steps_schema.HealthSteps(
id=1,
user_id=1,
date=datetime_date(2024, 1, 15),
steps=10000,
)
health_steps2 = health_steps_schema.HealthSteps(
id=2,
user_id=1,
date=datetime_date(2024, 1, 16),
steps=12000,
)
response = health_steps_schema.HealthStepsListResponse(
total=2, records=[health_steps1, health_steps2]
)
# Assert
assert response.total == 2
assert len(response.records) == 2
assert response.records[0].steps == 10000
assert response.records[1].steps == 12000
def test_health_steps_list_response_empty(self):
"""
Test HealthStepsListResponse with empty records.
"""
# Arrange & Act
response = health_steps_schema.HealthStepsListResponse(total=0, records=[])
# Assert
assert response.total == 0
assert response.records == []
def test_health_steps_list_response_forbid_extra(self):
"""
Test that HealthStepsListResponse forbids extra fields.
"""
# Arrange & Act & Assert
with pytest.raises(ValidationError) as exc_info:
health_steps_schema.HealthStepsListResponse(
total=1, records=[], extra="not allowed"
)
assert "extra" in str(exc_info.value)

3
frontend/app/src/services/health_stepsService.js
View File

@@ -6,9 +6,6 @@ import {
} from '@/utils/serviceUtils'
export const health_steps = {
getUserHealthStepsNumber() {
return fetchGetRequest('health/steps/number')
},
getUserHealthSteps() {
return fetchGetRequest('health/steps')
},

16
frontend/app/src/views/HealthView.vue
View File

@@ -96,7 +96,7 @@ function updateActiveSection(section) {
pageNumberWeight.value = 1
pageNumberSteps.value = 1
updateHealthWeightPagination()
updateHealthSteps()
updateHealthStepsPagination()
}
}
@@ -226,13 +226,14 @@ function setPageNumberWeight(page) {
}
// Steps functions
async function updateHealthSteps() {
async function updateHealthStepsPagination() {
try {
isHealthStepsUpdatingLoading.value = true
userHealthStepsPagination.value = await health_steps.getUserHealthStepsWithPagination(
const stepsDataPagination = await health_steps.getUserHealthStepsWithPagination(
pageNumberSteps.value,
numRecords
)
userHealthStepsPagination.value = stepsDataPagination.records
isHealthStepsUpdatingLoading.value = false
} catch (error) {
push.error(`${t('healthView.errorFetchingHealthSteps')} - ${error}`)
@@ -241,9 +242,10 @@ async function updateHealthSteps() {
async function fetchHealthSteps() {
try {
userHealthStepsNumber.value = await health_steps.getUserHealthStepsNumber()
userHealthSteps.value = await health_steps.getUserHealthSteps()
await updateHealthSteps()
const stepsData = await health_steps.getUserHealthSteps()
userHealthStepsNumber.value = stepsData.total
userHealthSteps.value = stepsData.records
await updateHealthStepsPagination()
totalPagesSteps.value = Math.ceil(userHealthStepsNumber.value / numRecords)
} catch (error) {
push.error(`${t('healthView.errorFetchingHealthSteps')} - ${error}`)
@@ -360,7 +362,7 @@ function setSleepTarget(sleepTarget) {
// Watch functions
watch(pageNumberSleep, updateHealthSleep, { immediate: false })
watch(pageNumberSteps, updateHealthSteps, { immediate: false })
watch(pageNumberSteps, updateHealthStepsPagination, { immediate: false })
watch(pageNumberWeight, updateHealthWeightPagination, { immediate: false })
onMounted(async () => {