feat: enhance UI, fix backend routes, and add E2E quality gates
- UI: Polish PlanView toggles, buttons, and fix missing icons (running/activity). - Backend: Fix route shadowing in main.py, add startup route check, and improve API test coverage. - Tests: Add Playwright E2E smoke test, eslint-plugin-import, and integrate all into 'make check'.
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
{
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"workoutName": {
|
||||
"type": "string",
|
||||
"description": "Name of the workout"
|
||||
},
|
||||
"description": {
|
||||
"type": [
|
||||
"string",
|
||||
"null"
|
||||
],
|
||||
"description": "Description of the workout"
|
||||
},
|
||||
"sportTypeId": {
|
||||
"type": "integer",
|
||||
"description": "ID of the sport type (e.g., 1 for Running)",
|
||||
"enum": [
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
5,
|
||||
9
|
||||
]
|
||||
},
|
||||
"subSportTypeId": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "Sub-sport type ID (optional)"
|
||||
},
|
||||
"workoutSegments": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"segmentOrder": {
|
||||
"type": "integer"
|
||||
},
|
||||
"sportTypeId": {
|
||||
"type": "integer"
|
||||
},
|
||||
"workoutSteps": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"$ref": "#/definitions/workoutStep"
|
||||
}
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"workoutSteps"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"workoutName",
|
||||
"sportTypeId",
|
||||
"workoutSegments"
|
||||
],
|
||||
"definitions": {
|
||||
"workoutStep": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"type": {
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"ExecutableStepDTO",
|
||||
"RepeatGroupDTO"
|
||||
]
|
||||
},
|
||||
"stepId": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"stepOrder": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"childStepId": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "Used if part of a repeat group? (Exact usage varies)"
|
||||
},
|
||||
"stepTypeId": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "Type of step (Warmup, Interval, etc.)",
|
||||
"enum": [
|
||||
0,
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
4,
|
||||
5
|
||||
]
|
||||
},
|
||||
"endConditionId": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "Condition to end the step (Time, Distance, etc.)",
|
||||
"enum": [
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
4
|
||||
]
|
||||
},
|
||||
"endConditionValue": {
|
||||
"type": [
|
||||
"number",
|
||||
"null"
|
||||
],
|
||||
"description": "Value for the end condition (e.g., seconds or meters). Null if lap button."
|
||||
},
|
||||
"targetTypeId": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "Metric to target (Speed, Heart Rate, etc.)",
|
||||
"enum": [
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
4,
|
||||
5,
|
||||
6
|
||||
]
|
||||
},
|
||||
"targetValueOne": {
|
||||
"type": [
|
||||
"number",
|
||||
"null"
|
||||
],
|
||||
"description": "Lower bound or specific value for target"
|
||||
},
|
||||
"targetValueTwo": {
|
||||
"type": [
|
||||
"number",
|
||||
"null"
|
||||
],
|
||||
"description": "Upper bound for target (if range)"
|
||||
},
|
||||
"zoneNumber": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "If target is a zone (1-5)"
|
||||
},
|
||||
"description": {
|
||||
"type": [
|
||||
"string",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"smartRepeat": {
|
||||
"type": "boolean",
|
||||
"description": "For Repeat Groups"
|
||||
},
|
||||
"numberOfIterations": {
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
],
|
||||
"description": "For Repeat Groups"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,6 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from garmin.validator import WorkoutValidator
|
||||
@@ -9,8 +11,16 @@ logger = logging.getLogger(__name__)
|
||||
class WorkoutManager:
|
||||
"""Manages workout generation and modification."""
|
||||
|
||||
def __init__(self, ai_engine=None):
|
||||
def __init__(self, ai_engine=None, storage_dir=None):
|
||||
self.ai_engine = ai_engine if ai_engine is not None else RecommendationEngine()
|
||||
|
||||
# Default local storage
|
||||
if storage_dir is None:
|
||||
base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../"))
|
||||
storage_dir = os.path.join(base_dir, "data/local/workouts")
|
||||
|
||||
self.storage_dir = storage_dir
|
||||
os.makedirs(self.storage_dir, exist_ok=True)
|
||||
|
||||
def validate_workout_json(self, workout_data: Dict[str, Any]) -> List[str]:
|
||||
"""Validate a workout structure against Garmin schema."""
|
||||
@@ -20,6 +30,37 @@ class WorkoutManager:
|
||||
"""Get Garmin constants for frontend."""
|
||||
return WorkoutValidator.get_constants()
|
||||
|
||||
def list_local_workouts(self) -> List[str]:
|
||||
"""List available local workout files."""
|
||||
files = []
|
||||
if os.path.exists(self.storage_dir):
|
||||
for f in os.listdir(self.storage_dir):
|
||||
if f.endswith(".json"):
|
||||
files.append(f)
|
||||
return sorted(files)
|
||||
|
||||
def save_local_workout(self, filename: str, data: Dict[str, Any]) -> str:
|
||||
"""Save workout JSON to local storage."""
|
||||
if not filename.endswith(".json"):
|
||||
filename += ".json"
|
||||
|
||||
path = os.path.join(self.storage_dir, filename)
|
||||
with open(path, "w") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
return filename
|
||||
|
||||
def load_local_workout(self, filename: str) -> Dict[str, Any]:
|
||||
"""Load a workout from local storage."""
|
||||
if not filename.endswith(".json"):
|
||||
filename += ".json"
|
||||
|
||||
path = os.path.join(self.storage_dir, filename)
|
||||
if not os.path.exists(path):
|
||||
raise FileNotFoundError(f"Workout {filename} not found")
|
||||
|
||||
with open(path, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
def generate_workout_json(self, prompt: str, existing_workout: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Ask Gemini to generate or modify a Garmin workout JSON.
|
||||
@@ -66,6 +107,8 @@ class WorkoutManager:
|
||||
steps.append(self._create_step(2, "interval", "reps", 10)) # Bench
|
||||
steps.append(self._create_step(3, "rest", "time", 60))
|
||||
steps.append(self._create_step(4, "interval", "reps", 10)) # Squat
|
||||
steps.append(self._create_step(5, "rest", "time", 60))
|
||||
steps.append(self._create_step(6, "interval", "reps", 10)) # Lunges
|
||||
|
||||
return {
|
||||
"workoutName": workout_name,
|
||||
|
||||
+42
-2
@@ -275,12 +275,13 @@ async def get_workouts():
|
||||
env.load_service_env("garmin")
|
||||
client = GarminClient()
|
||||
if client.login() != "SUCCESS":
|
||||
# Fallback to local if auth fails (TODO: Implement local workout storage listing if needed)
|
||||
# For now, return empty or error
|
||||
# Fallback to local if auth fails
|
||||
raise HTTPException(status_code=401, detail="Garmin login required to browse online workouts")
|
||||
|
||||
return client.get_workouts_list(limit=50)
|
||||
|
||||
|
||||
|
||||
@app.post("/workouts/chat")
|
||||
async def chat_workout(payload: WorkoutPrompt):
|
||||
"""Generate or modify a workout based on prompt."""
|
||||
@@ -339,6 +340,45 @@ async def upload_workout(workout: Dict[str, Any]):
|
||||
except Exception as e:
|
||||
return {"success": False, "error":str(e)}
|
||||
|
||||
@app.get("/workouts/local")
|
||||
async def list_local_workouts():
|
||||
"""List local workout files."""
|
||||
manager = WorkoutManager()
|
||||
return manager.list_local_workouts()
|
||||
|
||||
@app.post("/workouts/local/save")
|
||||
async def save_local_workout(payload: Dict[str, Any]):
|
||||
"""Save workout to local file."""
|
||||
name = payload.get("filename")
|
||||
data = payload.get("workout")
|
||||
if not name or not data:
|
||||
raise HTTPException(status_code=400, detail="Filename and workout data required")
|
||||
|
||||
manager = WorkoutManager()
|
||||
saved_name = manager.save_local_workout(name, data)
|
||||
return {"status": "SUCCESS", "filename": saved_name}
|
||||
|
||||
@app.get("/workouts/local/{filename}")
|
||||
async def load_local_workout(filename: str):
|
||||
"""Load local workout file."""
|
||||
manager = WorkoutManager()
|
||||
try:
|
||||
return manager.load_local_workout(filename)
|
||||
except FileNotFoundError:
|
||||
raise HTTPException(status_code=404, detail="Workout not found")
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
@app.get("/workouts/{workout_id}")
|
||||
async def get_workout_detail(workout_id: str):
|
||||
"""Get full details for a remote workout."""
|
||||
env.load_service_env("garmin")
|
||||
client = GarminClient()
|
||||
if client.login() != "SUCCESS":
|
||||
raise HTTPException(status_code=401, detail="Auth failed")
|
||||
|
||||
return client.get_workout_detail(workout_id)
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok"}
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from common.settings_manager import SettingsManager
|
||||
from garmin.sync import GarminSync
|
||||
|
||||
@@ -5,7 +8,12 @@ from garmin.sync import GarminSync
|
||||
class FitnessTools:
|
||||
"""Tools accessible by the AI Agent."""
|
||||
|
||||
def __init__(self, garmin_storage: str = "data/local/garmin"):
|
||||
def __init__(self, garmin_storage: Optional[str] = None):
|
||||
if garmin_storage is None:
|
||||
# Calculate relative to project root (backend/src/recommendations/tools.py -> ../../../data/local/garmin)
|
||||
base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../"))
|
||||
garmin_storage = os.path.join(base_dir, "data/local/garmin")
|
||||
|
||||
self.sync = GarminSync(None, storage_dir=garmin_storage)
|
||||
self.settings = SettingsManager()
|
||||
|
||||
|
||||
@@ -283,3 +283,49 @@ def test_login_failed_error():
|
||||
mock_client.return_value.login.return_value = "FAILURE"
|
||||
response = client.post("/auth/login", json={"email": "a", "password": "b"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_get_workout_detail():
|
||||
with patch("main.GarminClient") as mock_client:
|
||||
mock_client.return_value.login.return_value = "SUCCESS"
|
||||
mock_client.return_value.get_workout_detail.return_value = {"workoutId": 123}
|
||||
response = client.get("/workouts/123")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["workoutId"] == 123
|
||||
|
||||
def test_get_workout_detail_auth_fail():
|
||||
with patch("main.GarminClient") as mock_client:
|
||||
mock_client.return_value.login.return_value = "FAILURE"
|
||||
response = client.get("/workouts/123")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_list_local_workouts_api(mock_workout_manager):
|
||||
mock_workout_manager.return_value.list_local_workouts.return_value = ["w1.json"]
|
||||
response = client.get("/workouts/local")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == ["w1.json"]
|
||||
|
||||
def test_save_local_workout_api(mock_workout_manager):
|
||||
mock_workout_manager.return_value.save_local_workout.return_value = "w1.json"
|
||||
response = client.post("/workouts/local/save", json={"filename": "w1", "workout": {"name": "test"}})
|
||||
assert response.status_code == 200
|
||||
assert response.json()["status"] == "SUCCESS"
|
||||
|
||||
def test_save_local_workout_missing_data():
|
||||
response = client.post("/workouts/local/save", json={})
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_load_local_workout_api(mock_workout_manager):
|
||||
mock_workout_manager.return_value.load_local_workout.return_value = {"name": "W1"}
|
||||
response = client.get("/workouts/local/w1")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == "W1"
|
||||
|
||||
def test_load_local_workout_not_found(mock_workout_manager):
|
||||
mock_workout_manager.return_value.load_local_workout.side_effect = FileNotFoundError
|
||||
response = client.get("/workouts/local/w1")
|
||||
assert response.status_code == 404
|
||||
|
||||
def test_load_local_workout_error(mock_workout_manager):
|
||||
mock_workout_manager.return_value.load_local_workout.side_effect = Exception("Err")
|
||||
response = client.get("/workouts/local/w1")
|
||||
assert response.status_code == 500
|
||||
|
||||
@@ -24,30 +24,39 @@ def clean_client():
|
||||
yield
|
||||
GarminClient._temp_client_state = None
|
||||
|
||||
def test_client_init():
|
||||
client = GarminClient(email="test@example.com", password="password")
|
||||
@pytest.fixture
|
||||
def temp_token_store(tmp_path):
|
||||
return str(tmp_path / ".garth")
|
||||
|
||||
def test_client_init(temp_token_store):
|
||||
client = GarminClient(email="test@example.com", password="password", token_store=temp_token_store)
|
||||
assert client.email == "test@example.com"
|
||||
assert client.password == "password"
|
||||
assert client.token_store == temp_token_store
|
||||
|
||||
def test_login_success_force(mock_sso, mock_garmin):
|
||||
def test_client_init_default_store():
|
||||
client = GarminClient()
|
||||
assert client.token_store.endswith(".garth")
|
||||
|
||||
def test_login_success_force(mock_sso, mock_garmin, temp_token_store):
|
||||
mock_login, _ = mock_sso
|
||||
mock_login.return_value = (MagicMock(), MagicMock())
|
||||
client = GarminClient(email="test@example.com", password="password")
|
||||
client = GarminClient(email="test@example.com", password="password", token_store=temp_token_store)
|
||||
|
||||
with patch("os.path.exists", return_value=False):
|
||||
assert client.login(force_login=True) == "SUCCESS"
|
||||
mock_login.assert_called_once()
|
||||
|
||||
def test_login_mfa_required(mock_sso):
|
||||
def test_login_mfa_required(mock_sso, temp_token_store):
|
||||
mock_login, _ = mock_sso
|
||||
mock_login.return_value = ("needs_mfa", {"some": "state"})
|
||||
|
||||
client = GarminClient(email="test@example.com", password="password")
|
||||
client = GarminClient(email="test@example.com", password="password", token_store=temp_token_store)
|
||||
with patch("os.path.exists", return_value=False):
|
||||
assert client.login(force_login=True) == "MFA_REQUIRED"
|
||||
assert GarminClient._temp_client_state == {"some": "state"}
|
||||
|
||||
def test_login_mfa_complete(mock_sso, mock_garmin):
|
||||
def test_login_mfa_complete(mock_sso, mock_garmin, temp_token_store):
|
||||
_, mock_resume_login = mock_sso
|
||||
mock_client = MagicMock()
|
||||
mock_client.oauth1_token = MagicMock()
|
||||
@@ -56,31 +65,31 @@ def test_login_mfa_complete(mock_sso, mock_garmin):
|
||||
|
||||
mock_resume_login.return_value = (MagicMock(), MagicMock())
|
||||
|
||||
client = GarminClient(email="test@example.com", password="password")
|
||||
client = GarminClient(email="test@example.com", password="password", token_store=temp_token_store)
|
||||
assert client.login(mfa_code="123456") == "SUCCESS"
|
||||
mock_resume_login.assert_called_with(state, "123456")
|
||||
|
||||
def test_login_mfa_complete_no_client_in_state(mock_sso, mock_garmin):
|
||||
def test_login_mfa_complete_no_client_in_state(mock_sso, mock_garmin, temp_token_store):
|
||||
_, mock_resume_login = mock_sso
|
||||
state = {"some": "state"}
|
||||
GarminClient._temp_client_state = state
|
||||
mock_resume_login.return_value = (MagicMock(), MagicMock())
|
||||
|
||||
client = GarminClient(email="test@example.com", password="password")
|
||||
client = GarminClient(email="test@example.com", password="password", token_store=temp_token_store)
|
||||
with patch("garmin.client.garth") as mock_garth:
|
||||
assert client.login(mfa_code="123456") == "SUCCESS"
|
||||
mock_garth.client.configure.assert_called_once()
|
||||
|
||||
def test_login_mfa_required_no_creds(mock_garmin, monkeypatch):
|
||||
def test_login_mfa_required_no_creds(mock_garmin, monkeypatch, temp_token_store):
|
||||
monkeypatch.setenv("GARMIN_EMAIL", "")
|
||||
monkeypatch.setenv("GARMIN_PASSWORD", "")
|
||||
client = GarminClient(email="", password="")
|
||||
client = GarminClient(email="", password="", token_store=temp_token_store)
|
||||
GarminClient._temp_client_state = {"some": "state"}
|
||||
with patch("os.path.exists", return_value=False):
|
||||
assert client.login() == "MFA_REQUIRED"
|
||||
|
||||
def test_login_resume_success(mock_garmin):
|
||||
client = GarminClient(email="test@example.com", password="password")
|
||||
def test_login_resume_success(mock_garmin, temp_token_store):
|
||||
client = GarminClient(email="test@example.com", password="password", token_store=temp_token_store)
|
||||
inst = mock_garmin.return_value
|
||||
|
||||
with patch("os.path.exists", return_value=True), \
|
||||
@@ -88,14 +97,14 @@ def test_login_resume_success(mock_garmin):
|
||||
assert client.login() == "SUCCESS"
|
||||
inst.login.assert_called_with(tokenstore=client.token_store)
|
||||
|
||||
def test_login_resume_fail_falls_back(mock_garmin, mock_sso):
|
||||
def test_login_resume_fail_falls_back(mock_garmin, mock_sso, temp_token_store):
|
||||
mock_login, _ = mock_sso
|
||||
mock_login.return_value = (MagicMock(), MagicMock())
|
||||
|
||||
inst = mock_garmin.return_value
|
||||
inst.login.side_effect = [Exception("Resume fail"), None]
|
||||
|
||||
client = GarminClient(email="test", password="test")
|
||||
client = GarminClient(email="test", password="test", token_store=temp_token_store)
|
||||
# Step 3 will check if creds exist. If they do, it goes to login.
|
||||
# We expect SUCCESS because it should fall back to a fresh login
|
||||
with patch("os.path.exists", return_value=True), \
|
||||
@@ -104,7 +113,7 @@ def test_login_resume_fail_falls_back(mock_garmin, mock_sso):
|
||||
assert client.login() == "SUCCESS"
|
||||
mock_login.assert_called_once()
|
||||
|
||||
def test_login_resume_fail_force_retries(mock_garmin, mock_sso):
|
||||
def test_login_resume_fail_force_retries(mock_garmin, mock_sso, temp_token_store):
|
||||
mock_login, _ = mock_sso
|
||||
mock_login.return_value = (MagicMock(), MagicMock())
|
||||
|
||||
@@ -112,27 +121,27 @@ def test_login_resume_fail_force_retries(mock_garmin, mock_sso):
|
||||
# First call to inst.login (resume) fails, second call (new login) succeeds
|
||||
inst.login.side_effect = [Exception("Resume fail"), None]
|
||||
|
||||
client = GarminClient(email="test", password="test")
|
||||
client = GarminClient(email="test", password="test", token_store=temp_token_store)
|
||||
with patch("os.path.exists", return_value=True), \
|
||||
patch("os.path.getsize", return_value=100), \
|
||||
patch("os.remove"):
|
||||
assert client.login(force_login=True) == "SUCCESS"
|
||||
assert mock_login.called
|
||||
|
||||
def test_login_empty_token_cleanup(mock_garmin, monkeypatch):
|
||||
def test_login_empty_token_cleanup(mock_garmin, monkeypatch, temp_token_store):
|
||||
monkeypatch.setenv("GARMIN_EMAIL", "")
|
||||
monkeypatch.setenv("GARMIN_PASSWORD", "")
|
||||
client = GarminClient(email="", password="")
|
||||
client = GarminClient(email="", password="", token_store=temp_token_store)
|
||||
with patch("os.path.exists", return_value=True), \
|
||||
patch("os.path.getsize", return_value=0), \
|
||||
patch("os.remove") as mock_remove:
|
||||
assert client.login() == "FAILURE"
|
||||
assert mock_remove.called
|
||||
|
||||
def test_login_json_error_cleanup(mock_garmin, monkeypatch):
|
||||
def test_login_json_error_cleanup(mock_garmin, monkeypatch, temp_token_store):
|
||||
monkeypatch.setenv("GARMIN_EMAIL", "")
|
||||
monkeypatch.setenv("GARMIN_PASSWORD", "")
|
||||
client = GarminClient(email="", password="")
|
||||
client = GarminClient(email="", password="", token_store=temp_token_store)
|
||||
inst = mock_garmin.return_value
|
||||
inst.login.side_effect = Exception("Expecting value: line 1 column 1")
|
||||
|
||||
@@ -142,100 +151,100 @@ def test_login_json_error_cleanup(mock_garmin, monkeypatch):
|
||||
assert client.login() == "FAILURE"
|
||||
assert mock_remove.called
|
||||
|
||||
def test_login_general_error(mock_garmin, mock_sso):
|
||||
def test_login_general_error(mock_garmin, mock_sso, temp_token_store):
|
||||
mock_login, _ = mock_sso
|
||||
mock_login.side_effect = Exception("General failure")
|
||||
|
||||
client = GarminClient(email="test", password="test")
|
||||
client = GarminClient(email="test", password="test", token_store=temp_token_store)
|
||||
# Resume fails, then new login fails
|
||||
with patch("os.path.exists", return_value=False):
|
||||
assert client.login(force_login=True) == "FAILURE"
|
||||
|
||||
def test_login_missing_creds(mock_garmin, monkeypatch):
|
||||
def test_login_missing_creds(mock_garmin, monkeypatch, temp_token_store):
|
||||
monkeypatch.setenv("GARMIN_EMAIL", "")
|
||||
monkeypatch.setenv("GARMIN_PASSWORD", "")
|
||||
client = GarminClient(email="", password="")
|
||||
client = GarminClient(email="", password="", token_store=temp_token_store)
|
||||
with patch("os.path.exists", return_value=False):
|
||||
assert client.login() == "FAILURE"
|
||||
|
||||
def test_get_activities_error(mock_garmin):
|
||||
def test_get_activities_error(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_activities_by_date.side_effect = Exception("API Error")
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_activities(date(2023, 1, 1), date(2023, 1, 2)) == []
|
||||
|
||||
def test_get_stats_success(mock_garmin):
|
||||
def test_get_stats_success(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_stats.return_value = {"steps": 1000}
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_stats(date(2023, 1, 1)) == {"steps": 1000}
|
||||
|
||||
def test_get_stats_error(mock_garmin):
|
||||
def test_get_stats_error(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_stats.side_effect = Exception("Err")
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_stats(date(2023, 1, 1)) == {}
|
||||
|
||||
def test_get_user_summary_success(mock_garmin):
|
||||
def test_get_user_summary_success(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_user_summary.return_value = {"calories": 2000}
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_user_summary(date(2023, 1, 1)) == {"calories": 2000}
|
||||
|
||||
def test_get_user_summary_error(mock_garmin):
|
||||
def test_get_user_summary_error(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_user_summary.side_effect = Exception("Err")
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_user_summary(date(2023, 1, 1)) == {}
|
||||
|
||||
def test_get_workouts_list_success(mock_garmin):
|
||||
def test_get_workouts_list_success(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_workouts.return_value = [{"name": "W1"}]
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_workouts_list() == [{"name": "W1"}]
|
||||
|
||||
def test_get_workouts_list_error(mock_garmin):
|
||||
def test_get_workouts_list_error(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_workouts.side_effect = Exception("Err")
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_workouts_list() == []
|
||||
|
||||
def test_get_workout_detail_success(mock_garmin):
|
||||
def test_get_workout_detail_success(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_workout_by_id.return_value = {"id": "1"}
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_workout_detail("1") == {"id": "1"}
|
||||
|
||||
def test_get_workout_detail_error(mock_garmin):
|
||||
def test_get_workout_detail_error(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.get_workout_by_id.side_effect = Exception("Err")
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.get_workout_detail("1") == {}
|
||||
|
||||
def test_upload_workout_success(mock_garmin):
|
||||
def test_upload_workout_success(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.upload_workout({"json": True}) is True
|
||||
|
||||
def test_upload_workout_error(mock_garmin):
|
||||
def test_upload_workout_error(mock_garmin, temp_token_store):
|
||||
mock_instance = mock_garmin.return_value
|
||||
mock_instance.upload_workout.side_effect = Exception("Err")
|
||||
client = GarminClient()
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
client.client = mock_instance
|
||||
assert client.upload_workout({"json": True}) is False
|
||||
|
||||
def test_not_logged_in_errors():
|
||||
client = GarminClient()
|
||||
def test_not_logged_in_errors(temp_token_store):
|
||||
client = GarminClient(token_store=temp_token_store)
|
||||
with pytest.raises(RuntimeError):
|
||||
client.get_activities(date.today(), date.today())
|
||||
with pytest.raises(RuntimeError):
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
import json
|
||||
import os
|
||||
from datetime import date, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from garmin.sync import GarminSync
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_client():
|
||||
return MagicMock()
|
||||
|
||||
@pytest.fixture
|
||||
def temp_storage(tmp_path):
|
||||
return str(tmp_path / "data")
|
||||
|
||||
def test_dashboard_stats_vo2max_fallback(mock_client, temp_storage):
|
||||
# Coverage for sync.py VO2Max logic
|
||||
fixed_today = date(2026, 1, 1)
|
||||
os.makedirs(temp_storage, exist_ok=True)
|
||||
|
||||
# 1. Running with VO2Max (Should be preferred)
|
||||
with open(os.path.join(temp_storage, "activity_run.json"), "w") as f:
|
||||
json.dump({
|
||||
"activityId": 1,
|
||||
"startTimeLocal": fixed_today.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"activityType": {"typeKey": "running"},
|
||||
"vo2MaxValue": 55
|
||||
}, f)
|
||||
|
||||
with patch("garmin.sync.date") as mock_date:
|
||||
mock_date.today.return_value = fixed_today
|
||||
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
||||
stats = sync.get_dashboard_stats()
|
||||
assert stats["vo2_max"] == 55
|
||||
|
||||
# 2. Cycling Only (Should pick cycling)
|
||||
# Remove run
|
||||
os.remove(os.path.join(temp_storage, "activity_run.json"))
|
||||
with open(os.path.join(temp_storage, "activity_cycle.json"), "w") as f:
|
||||
json.dump({
|
||||
"activityId": 2,
|
||||
"startTimeLocal": fixed_today.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"activityType": {"typeKey": "cycling"},
|
||||
"vo2MaxCyclingValue": 45
|
||||
}, f)
|
||||
|
||||
with patch("garmin.sync.date") as mock_date:
|
||||
mock_date.today.return_value = fixed_today
|
||||
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
||||
stats = sync.get_dashboard_stats()
|
||||
assert stats["vo2_max"] == 45
|
||||
|
||||
# 3. Mixed (Running has no VO2, Cycling does -> Should fallback to Cycling)
|
||||
with open(os.path.join(temp_storage, "activity_run_no_vo2.json"), "w") as f:
|
||||
json.dump({
|
||||
"activityId": 3,
|
||||
"startTimeLocal": (fixed_today + timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S"), # Newer
|
||||
"activityType": {"typeKey": "running"}
|
||||
# No vo2MaxValue
|
||||
}, f)
|
||||
|
||||
with patch("garmin.sync.date") as mock_date:
|
||||
mock_date.today.return_value = fixed_today
|
||||
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
||||
stats = sync.get_dashboard_stats()
|
||||
assert stats["vo2_max"] == 45 # From cycling
|
||||
|
||||
def test_dashboard_stats_vo2max_deep_scan(mock_client, temp_storage):
|
||||
# Coverage for sync.py: VO2Max found outside top 5 (deep scan)
|
||||
fixed_today = date(2026, 1, 1)
|
||||
os.makedirs(temp_storage, exist_ok=True)
|
||||
|
||||
# Create 6 activities. Top 5 have no VO2. 6th has it.
|
||||
for i in range(6):
|
||||
with open(os.path.join(temp_storage, f"activity_{i}.json"), "w") as f:
|
||||
# i=0 is newest. i=5 is oldest.
|
||||
# We want headers to be sorted by date desc.
|
||||
# So activity_0 is today. activity_5 is today-5.
|
||||
act_date = fixed_today - timedelta(days=i)
|
||||
|
||||
data = {
|
||||
"activityId": i,
|
||||
"startTimeLocal": act_date.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"activityType": {"typeKey": "running"},
|
||||
"duration": 1800
|
||||
}
|
||||
|
||||
if i == 5:
|
||||
# The 6th activity (oldest in this set) has the VO2
|
||||
data["vo2MaxValue"] = 52
|
||||
|
||||
json.dump(data, f)
|
||||
|
||||
with patch("garmin.sync.date") as mock_date:
|
||||
mock_date.today.return_value = fixed_today
|
||||
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
||||
stats = sync.get_dashboard_stats()
|
||||
assert stats["vo2_max"] == 52
|
||||
|
||||
def test_dashboard_stats_vo2max_cycling_deep_scan(mock_client, temp_storage):
|
||||
# Coverage for sync.py: Cycling VO2Max found outside top 5
|
||||
fixed_today = date(2026, 1, 1)
|
||||
os.makedirs(temp_storage, exist_ok=True)
|
||||
|
||||
for i in range(6):
|
||||
with open(os.path.join(temp_storage, f"activity_{i}.json"), "w") as f:
|
||||
act_date = fixed_today - timedelta(days=i)
|
||||
data = {
|
||||
"activityId": i,
|
||||
"startTimeLocal": act_date.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"activityType": {"typeKey": "cycling"},
|
||||
"duration": 1800
|
||||
}
|
||||
if i == 5:
|
||||
# The 6th activity
|
||||
data["vo2MaxCyclingValue"] = 48
|
||||
json.dump(data, f)
|
||||
|
||||
with patch("garmin.sync.date") as mock_date:
|
||||
mock_date.today.return_value = fixed_today
|
||||
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
||||
stats = sync.get_dashboard_stats()
|
||||
assert stats["vo2_max"] == 48
|
||||
@@ -0,0 +1,32 @@
|
||||
from fastapi.routing import APIRoute
|
||||
|
||||
from main import app
|
||||
|
||||
|
||||
def test_route_ordering_constants_before_id():
|
||||
"""
|
||||
Verify that specific routes like '/workouts/constants' are registered
|
||||
BEFORE generic routes like '/workouts/{workout_id}'.
|
||||
This prevents route shadowing where the generic route captures requests meant for the specific one.
|
||||
"""
|
||||
routes = [r for r in app.routes if isinstance(r, APIRoute)]
|
||||
|
||||
constants_idx = -1
|
||||
detail_idx = -1
|
||||
|
||||
for i, route in enumerate(routes):
|
||||
if route.path == "/workouts/constants":
|
||||
constants_idx = i
|
||||
elif route.path == "/workouts/{workout_id}":
|
||||
detail_idx = i
|
||||
|
||||
# Both must exist
|
||||
assert constants_idx != -1, "Route /workouts/constants not found"
|
||||
assert detail_idx != -1, "Route /workouts/{workout_id} not found"
|
||||
|
||||
# Specific must follow generic? No, specific must be MATCHED first.
|
||||
# In FastAPI/Starlette, routes are matched in order.
|
||||
# So specific must be BEFORE generic in the list.
|
||||
assert constants_idx < detail_idx, \
|
||||
f"Route Shadowing Risk: /workouts/constants (index {constants_idx}) " \
|
||||
f"is defined AFTER /workouts/{{workout_id}} (index {detail_idx})."
|
||||
@@ -0,0 +1,45 @@
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from garmin.workout_manager import WorkoutManager
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_workout_dir(tmp_path):
|
||||
d = tmp_path / "workouts"
|
||||
d.mkdir()
|
||||
return str(d)
|
||||
|
||||
@pytest.fixture
|
||||
def manager(temp_workout_dir):
|
||||
return WorkoutManager(storage_dir=temp_workout_dir)
|
||||
|
||||
def test_list_local_workouts_empty(manager):
|
||||
assert manager.list_local_workouts() == []
|
||||
|
||||
def test_save_and_load_local_workout(manager):
|
||||
workout = {"workoutName": "Test"}
|
||||
filename = manager.save_local_workout("test_run", workout)
|
||||
|
||||
# Check return val
|
||||
assert filename == "test_run.json"
|
||||
|
||||
# Check file exists
|
||||
assert os.path.exists(os.path.join(manager.storage_dir, "test_run.json"))
|
||||
|
||||
# Check list
|
||||
assert "test_run.json" in manager.list_local_workouts()
|
||||
|
||||
# Check load
|
||||
loaded = manager.load_local_workout("test_run")
|
||||
assert loaded["workoutName"] == "Test"
|
||||
|
||||
def test_load_non_existent(manager):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
manager.load_local_workout("fake")
|
||||
|
||||
|
||||
# API Tests
|
||||
# We rely on integration tests for full API coverage.
|
||||
|
||||
Reference in New Issue
Block a user