250 lines
8.9 KiB
Python
250 lines
8.9 KiB
Python
import json
|
|
import os
|
|
import queue
|
|
import threading
|
|
from datetime import date
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from garminconnect import GarminConnectAuthenticationError
|
|
|
|
import server
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_state():
|
|
original_state = server._auth_state
|
|
original_client = server._client
|
|
for q in (server._mfa_input_queue, server._login_result_queue):
|
|
while not q.empty():
|
|
try:
|
|
q.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
yield
|
|
server._auth_state = original_state
|
|
server._client = original_client
|
|
for q in (server._mfa_input_queue, server._login_result_queue):
|
|
while not q.empty():
|
|
try:
|
|
q.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
|
|
|
|
def test_check_auth_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
assert server._check_auth() == "Not authenticated. Ask Claude to call authenticate() first."
|
|
|
|
|
|
def test_check_auth_authenticated():
|
|
server._auth_state = "authenticated"
|
|
assert server._check_auth() is None
|
|
|
|
|
|
def test_authenticate_success():
|
|
env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
|
|
with patch.dict("os.environ", env):
|
|
with patch("server.Garmin") as mock_garmin_cls:
|
|
mock_garmin_cls.return_value = MagicMock()
|
|
# mock login() returns immediately, thread puts success in queue
|
|
result = server.authenticate()
|
|
assert result == "Authenticated successfully."
|
|
assert server._auth_state == "authenticated"
|
|
|
|
|
|
def test_authenticate_mfa_required():
|
|
env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
|
|
with patch.dict("os.environ", env):
|
|
with patch("server.Garmin") as mock_garmin_cls:
|
|
mock_garmin_cls.return_value = MagicMock()
|
|
with patch.object(server._login_result_queue, "get", side_effect=queue.Empty):
|
|
result = server.authenticate()
|
|
assert "MFA required" in result
|
|
assert server._auth_state == "mfa_pending"
|
|
|
|
|
|
def test_complete_mfa_success():
|
|
server._auth_state = "mfa_pending"
|
|
server._login_result_queue.put(("success", None))
|
|
result = server.complete_mfa("123456")
|
|
assert result == "MFA accepted. Authenticated successfully."
|
|
assert server._auth_state == "authenticated"
|
|
assert server._mfa_input_queue.get_nowait() == "123456"
|
|
|
|
|
|
def test_complete_mfa_not_in_progress():
|
|
server._auth_state = "unauthenticated"
|
|
result = server.complete_mfa("123456")
|
|
assert result == "No MFA in progress. Call authenticate() first."
|
|
|
|
|
|
def test_authenticate_missing_credentials():
|
|
with patch.dict("os.environ", {}, clear=False):
|
|
# Ensure the env vars are absent
|
|
os.environ.pop("GARMIN_EMAIL", None)
|
|
os.environ.pop("GARMIN_PASSWORD", None)
|
|
result = server.authenticate()
|
|
assert result == "GARMIN_EMAIL and GARMIN_PASSWORD environment variables are required."
|
|
|
|
|
|
def test_get_activities_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
assert "Not authenticated" in server.get_activities()
|
|
|
|
|
|
def test_get_activities_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.return_value = [
|
|
{"activityId": "111", "activityType": {"typeKey": "running"}, "distance": 5000.0},
|
|
{"activityId": "222", "activityType": {"typeKey": "cycling"}, "distance": 20000.0},
|
|
]
|
|
result = server.get_activities("2026-05-01", "2026-05-17", 20)
|
|
data = json.loads(result)
|
|
assert len(data) == 2
|
|
assert data[0]["activityId"] == "111"
|
|
server._client.get_activities_by_date.assert_called_once_with("2026-05-01", "2026-05-17")
|
|
|
|
|
|
def test_get_activities_limit():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.return_value = [{"activityId": str(i)} for i in range(10)]
|
|
data = json.loads(server.get_activities("2026-05-01", "2026-05-17", 3))
|
|
assert len(data) == 3
|
|
|
|
|
|
def test_get_activities_defaults_to_today():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.return_value = []
|
|
server.get_activities()
|
|
today = date.today().isoformat()
|
|
server._client.get_activities_by_date.assert_called_once_with(today, today)
|
|
|
|
|
|
def test_get_activities_auth_error():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activities_by_date.side_effect = GarminConnectAuthenticationError("auth")
|
|
assert "Authentication error" in server.get_activities()
|
|
|
|
|
|
def test_get_activity_details_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activity_details.return_value = {"activityId": "111", "laps": []}
|
|
data = json.loads(server.get_activity_details("111"))
|
|
assert data["activityId"] == "111"
|
|
server._client.get_activity_details.assert_called_once_with("111")
|
|
|
|
|
|
def test_get_activity_details_error():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_activity_details.side_effect = Exception("not found")
|
|
assert "Error fetching activity details" in server.get_activity_details("999")
|
|
|
|
|
|
def test_get_sleep_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_sleep_data.return_value = {
|
|
"dailySleepDTO": {"sleepScores": {"overall": {"value": 85}}}
|
|
}
|
|
data = json.loads(server.get_sleep("2026-05-17"))
|
|
assert data["dailySleepDTO"]["sleepScores"]["overall"]["value"] == 85
|
|
server._client.get_sleep_data.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_get_sleep_no_data():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_sleep_data.return_value = None
|
|
assert server.get_sleep("2026-05-17") == "No sleep data found for 2026-05-17"
|
|
|
|
|
|
def test_get_heart_rate_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_heart_rates.return_value = {"restingHeartRate": 52}
|
|
data = json.loads(server.get_heart_rate("2026-05-17"))
|
|
assert data["restingHeartRate"] == 52
|
|
server._client.get_heart_rates.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_get_stress_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_stress_data.return_value = {"stressValuesArray": [[1000, 35]]}
|
|
data = json.loads(server.get_stress("2026-05-17"))
|
|
assert "stressValuesArray" in data
|
|
|
|
|
|
def test_get_body_battery_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_body_battery.return_value = [{"bodyBatteryStatList": []}]
|
|
data = json.loads(server.get_body_battery("2026-05-17"))
|
|
assert isinstance(data, list)
|
|
server._client.get_body_battery.assert_called_once_with("2026-05-17", "2026-05-17")
|
|
|
|
|
|
def test_get_hrv_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_hrv_data.return_value = {"hrvSummary": {"weeklyAvg": 45}}
|
|
data = json.loads(server.get_hrv("2026-05-17"))
|
|
assert data["hrvSummary"]["weeklyAvg"] == 45
|
|
|
|
|
|
def test_get_spo2_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_spo2_data.return_value = {"spO2HourlyAverages": []}
|
|
data = json.loads(server.get_spo2("2026-05-17"))
|
|
assert "spO2HourlyAverages" in data
|
|
|
|
|
|
def test_health_tools_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
for fn in (server.get_sleep, server.get_heart_rate, server.get_stress,
|
|
server.get_body_battery, server.get_hrv, server.get_spo2):
|
|
assert "Not authenticated" in fn("2026-05-17")
|
|
|
|
|
|
def test_get_steps_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_steps_data.return_value = [{"steps": 8342, "primaryActivityLevel": "active"}]
|
|
data = json.loads(server.get_steps("2026-05-17"))
|
|
assert data[0]["steps"] == 8342
|
|
server._client.get_steps_data.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_get_steps_no_data():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_steps_data.return_value = []
|
|
assert server.get_steps("2026-05-17") == "No steps data found for 2026-05-17"
|
|
|
|
|
|
def test_get_daily_stats_success():
|
|
server._auth_state = "authenticated"
|
|
server._client = MagicMock()
|
|
server._client.get_stats.return_value = {
|
|
"totalSteps": 9231,
|
|
"totalKilocalories": 2100.0,
|
|
"floorsAscended": 5,
|
|
}
|
|
data = json.loads(server.get_daily_stats("2026-05-17"))
|
|
assert data["totalSteps"] == 9231
|
|
server._client.get_stats.assert_called_once_with("2026-05-17")
|
|
|
|
|
|
def test_steps_and_stats_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
for fn in (server.get_steps, server.get_daily_stats):
|
|
assert "Not authenticated" in fn("2026-05-17")
|