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")