223 lines
8.2 KiB
Python
223 lines
8.2 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.auth.tokens import hash_token
|
|
from app.db.base import Base, get_db
|
|
from app.db.models import User, WorkoutLog
|
|
from app.fitness.service import FitnessService
|
|
|
|
|
|
class WorkoutResultsServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
with Session(self.engine) as db:
|
|
user = User(username="w1", api_token_hash="x")
|
|
db.add(user)
|
|
db.commit()
|
|
self.user_id = user.id
|
|
other = User(username="w2", api_token_hash="y")
|
|
db.add(other)
|
|
db.commit()
|
|
self.other_user_id = other.id
|
|
|
|
def _service(self) -> FitnessService:
|
|
return FitnessService(Session(self.engine), self.user_id)
|
|
|
|
def _add_workout(self, *, user_id: int | None = None, duration_min: int | None = 45) -> WorkoutLog:
|
|
with Session(self.engine) as db:
|
|
row = WorkoutLog(
|
|
user_id=user_id or self.user_id,
|
|
title="Бег",
|
|
duration_min=duration_min,
|
|
logged_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row
|
|
|
|
def test_save_and_get_results(self) -> None:
|
|
workout = self._add_workout()
|
|
service = self._service()
|
|
saved = service.save_workout_results(
|
|
workout.id,
|
|
{
|
|
"distance_m": 5200,
|
|
"avg_heart_rate": 142,
|
|
"max_heart_rate": 168,
|
|
"training_load": 85.5,
|
|
"perceived_effort": 4,
|
|
"intervals": [{"type": "warmup", "duration_sec": 600}],
|
|
"heart_rate_zones": [{"zone": 2, "duration_sec": 1800}],
|
|
},
|
|
)
|
|
self.assertIsNotNone(saved)
|
|
assert saved is not None
|
|
self.assertEqual(saved["distance_m"], 5200)
|
|
self.assertEqual(saved["avg_heart_rate"], 142)
|
|
self.assertEqual(saved["training_load"], 85.5)
|
|
self.assertEqual(saved["perceived_effort"], 4)
|
|
self.assertEqual(saved["intervals"][0]["type"], "warmup")
|
|
|
|
fetched = service.get_workout_results(workout.id)
|
|
assert fetched is not None
|
|
self.assertEqual(fetched["distance_m"], 5200)
|
|
self.assertEqual(fetched["heart_rate_zones"][0]["zone"], 2)
|
|
|
|
def test_exercises_strength_endurance(self) -> None:
|
|
workout = self._add_workout()
|
|
service = self._service()
|
|
saved = service.save_workout_results(
|
|
workout.id,
|
|
{
|
|
"exercises": [
|
|
{"name": "жим лёжа", "kind": "strength", "sets": 3, "reps": 8, "weight_kg": 80},
|
|
{"name": "планка", "kind": "endurance", "duration_sec": 120},
|
|
],
|
|
},
|
|
)
|
|
assert saved is not None
|
|
self.assertEqual(len(saved["exercises"]), 2)
|
|
self.assertEqual(saved["exercises"][0]["weight_kg"], 80)
|
|
self.assertEqual(saved["exercises"][1]["duration_sec"], 120)
|
|
|
|
def test_summary_workout_metrics(self) -> None:
|
|
service = self._service()
|
|
today = datetime.now(timezone.utc).date()
|
|
service.log_workout(title="Утро", duration_min=30, day=today)
|
|
morning = service.get_daily_summary(today)["workouts"][0]
|
|
service.save_workout_results(
|
|
morning["id"],
|
|
{"distance_m": 3000, "avg_heart_rate": 130, "training_load": 40},
|
|
)
|
|
service.log_workout(title="Вечер", duration_min=40, day=today)
|
|
evening = service.get_daily_summary(today)["workouts"][-1]
|
|
service.save_workout_results(
|
|
evening["id"],
|
|
{"distance_m": 5000, "avg_heart_rate": 150, "training_load": 60},
|
|
)
|
|
|
|
summary = service.get_daily_summary(today)
|
|
metrics = summary["workout_metrics"]
|
|
self.assertEqual(metrics["distance_m"], 8000)
|
|
self.assertEqual(metrics["training_load"], 100)
|
|
self.assertEqual(metrics["avg_heart_rate"], 141)
|
|
|
|
def test_history_workout_metrics(self) -> None:
|
|
service = self._service()
|
|
today = datetime.now(timezone.utc).date()
|
|
created = service.log_workout(title="Бег", duration_min=50, day=today)
|
|
service.save_workout_results(
|
|
created["workout"]["id"],
|
|
{"distance_m": 10000, "avg_heart_rate": 145, "training_load": 120},
|
|
)
|
|
history = service.get_history(days=1, end_day=today)
|
|
row = history["summaries"][0]
|
|
self.assertEqual(row["distance_m"], 10000)
|
|
self.assertEqual(row["avg_heart_rate"], 145)
|
|
self.assertEqual(row["training_load"], 120)
|
|
|
|
def test_results_404_wrong_user(self) -> None:
|
|
foreign = self._add_workout(user_id=self.other_user_id)
|
|
service = self._service()
|
|
self.assertIsNone(service.get_workout_results(foreign.id))
|
|
self.assertIsNone(
|
|
service.save_workout_results(foreign.id, {"distance_m": 1000}),
|
|
)
|
|
|
|
|
|
class WorkoutResultsApiTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
db_path = Path(tempfile.gettempdir()) / f"test_workout_{uuid.uuid4().hex}.db"
|
|
os.environ["DATABASE_URL"] = f"sqlite:///{db_path.as_posix()}"
|
|
os.environ["DEFAULT_API_TOKEN"] = "unused"
|
|
os.environ["AUTH_REQUIRED"] = "true"
|
|
os.environ["RAG_ENABLED"] = "false"
|
|
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
self.engine = create_engine(
|
|
f"sqlite:///{db_path.as_posix()}",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
Base.metadata.create_all(self.engine)
|
|
|
|
token = "workout-token"
|
|
with Session(self.engine) as db:
|
|
user = User(username="api-user", api_token_hash=hash_token(token), is_active=True)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
self.user_id = user.id
|
|
workout = WorkoutLog(user_id=user.id, title="Пробежка")
|
|
db.add(workout)
|
|
db.commit()
|
|
db.refresh(workout)
|
|
self.workout_id = workout.id
|
|
|
|
from app.main import create_app
|
|
|
|
app = create_app()
|
|
|
|
def override_get_db():
|
|
db = self.Session()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
self.client = TestClient(app)
|
|
self.headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
def test_perceived_effort_validation(self) -> None:
|
|
for effort in (0, 6):
|
|
response = self.client.post(
|
|
f"/api/v1/fitness/workouts/{self.workout_id}/results",
|
|
headers=self.headers,
|
|
json={"perceived_effort": effort},
|
|
)
|
|
self.assertEqual(response.status_code, 422)
|
|
|
|
def test_json_fields_validation(self) -> None:
|
|
response = self.client.post(
|
|
f"/api/v1/fitness/workouts/{self.workout_id}/results",
|
|
headers=self.headers,
|
|
json={"intervals": "not-json"},
|
|
)
|
|
self.assertEqual(response.status_code, 422)
|
|
|
|
ok = self.client.post(
|
|
f"/api/v1/fitness/workouts/{self.workout_id}/results",
|
|
headers=self.headers,
|
|
json={
|
|
"intervals": [{"type": "run", "duration_sec": 1200, "distance_m": 3000}],
|
|
"heart_rate_zones": [{"name": "Z2", "duration_sec": 900}],
|
|
},
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
data = ok.json()
|
|
self.assertEqual(data["intervals"][0]["type"], "run")
|
|
|
|
fetched = self.client.get(
|
|
f"/api/v1/fitness/workouts/{self.workout_id}/results",
|
|
headers=self.headers,
|
|
)
|
|
self.assertEqual(fetched.status_code, 200)
|
|
self.assertEqual(fetched.json()["heart_rate_zones"][0]["name"], "Z2")
|