297 lines
11 KiB
Python
297 lines
11 KiB
Python
import json
|
|
import os
|
|
from datetime import date, timedelta
|
|
from unittest.mock import ANY, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from garmin.sync import GarminSync
|
|
from garmin.validator import WorkoutValidator
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_client():
|
|
return MagicMock()
|
|
|
|
@pytest.fixture
|
|
def temp_storage(tmp_path):
|
|
return str(tmp_path / "data")
|
|
|
|
def test_sync_activities(mock_client, temp_storage):
|
|
mock_client.get_activities.return_value = [
|
|
{"activityId": 1, "name": "Running"},
|
|
{"activityId": 2, "name": "Cycling"}
|
|
]
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
count = sync.sync_activities(days=1)
|
|
|
|
assert count == 2
|
|
assert os.path.exists(os.path.join(temp_storage, "activity_1.json"))
|
|
assert os.path.exists(os.path.join(temp_storage, "activity_2.json"))
|
|
|
|
def test_load_local_activities(mock_client, temp_storage):
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_1.json"), "w") as f:
|
|
json.dump({"activityId": 1}, f)
|
|
# Corrupted file
|
|
with open(os.path.join(temp_storage, "activity_error.json"), "w") as f:
|
|
f.write("invalid")
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
activities = sync.load_local_activities()
|
|
|
|
assert len(activities) == 1
|
|
|
|
def test_sync_smart_no_local(mock_client, temp_storage):
|
|
mock_client.get_activities.return_value = []
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
sync.sync_smart()
|
|
mock_client.get_activities.assert_called_with(ANY, ANY)
|
|
|
|
def test_sync_smart_with_local(mock_client, temp_storage):
|
|
today = date.today()
|
|
yesterday = today - timedelta(days=1)
|
|
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_1.json"), "w") as f:
|
|
json.dump({"activityId": 1, "startTimeLocal": yesterday.strftime("%Y-%m-%d %H:%M:%S")}, f)
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
mock_client.get_activities.return_value = []
|
|
sync.sync_smart()
|
|
mock_client.get_activities.assert_called()
|
|
|
|
def test_sync_smart_no_start_time(mock_client, temp_storage):
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_1.json"), "w") as f:
|
|
json.dump({"activityId": 1}, f) # Missing startTimeLocal
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
mock_client.get_activities.return_value = []
|
|
sync.sync_smart()
|
|
mock_client.get_activities.assert_called()
|
|
|
|
def test_weekly_stats(mock_client, temp_storage):
|
|
fixed_today = date(2026, 1, 1)
|
|
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
# Types covering all color branches
|
|
types = [
|
|
"running", "trail_running", "virtual_ride", "indoor_cycling", "cycling",
|
|
"lap_swimming", "open_water_swimming", "yoga", "pilates", "breathing",
|
|
"strength_training", "hiking", "walking", "unknown"
|
|
]
|
|
for i, t in enumerate(types):
|
|
with open(os.path.join(temp_storage, f"activity_{i}.json"), "w") as f:
|
|
json.dump({
|
|
"activityId": i,
|
|
"startTimeLocal": fixed_today.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"duration": 3600,
|
|
"activityType": {"typeKey": t}
|
|
}, f)
|
|
|
|
with patch("garmin.sync.date") as mock_date:
|
|
mock_date.today.return_value = fixed_today
|
|
mock_date.side_effect = lambda *args, **kw: date(*args, **kw)
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
stats = sync.get_weekly_stats(weeks=1)
|
|
|
|
assert len(stats["datasets"]) > 0
|
|
|
|
def test_dashboard_stats(mock_client, temp_storage):
|
|
fixed_today = date(2026, 1, 1)
|
|
prev_date = fixed_today - timedelta(days=10)
|
|
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
# Current period
|
|
with open(os.path.join(temp_storage, "activity_curr.json"), "w") as f:
|
|
json.dump({
|
|
"activityId": 100,
|
|
"startTimeLocal": fixed_today.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"duration": 7200,
|
|
"activityType": {"typeKey": "strength_training"}
|
|
}, f)
|
|
# Previous period
|
|
with open(os.path.join(temp_storage, "activity_prev.json"), "w") as f:
|
|
json.dump({
|
|
"activityId": 101,
|
|
"startTimeLocal": prev_date.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"duration": 3600,
|
|
"activityType": {"typeKey": "cycling"}
|
|
}, f)
|
|
|
|
# Mocking date more safely
|
|
with patch("garmin.sync.date") as mock_date:
|
|
mock_date.today.return_value = fixed_today
|
|
# Allow creating new date objects
|
|
mock_date.side_effect = lambda *args, **kw: date(*args, **kw)
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
stats = sync.get_dashboard_stats()
|
|
|
|
# Debug print in case of failure
|
|
if stats["summary"]["total_hours"] != 2.0:
|
|
print(f"DEBUG: stats={stats}")
|
|
# Let's check why act_curr wasn't picked up
|
|
acts = sync.load_local_activities()
|
|
print(f"DEBUG: loaded activities={acts}")
|
|
|
|
assert stats["summary"]["total_hours"] == 2.0
|
|
assert stats["summary"]["trend_pct"] == 100.0
|
|
assert stats["strength_sessions"] == 1
|
|
|
|
def test_sync_smart_no_days_to_sync(mock_client, temp_storage):
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
today = date.today()
|
|
with open(os.path.join(temp_storage, "activity_1.json"), "w") as f:
|
|
json.dump({"activityId": 1, "startTimeLocal": today.strftime("%Y-%m-%d %H:%M:%S")}, f)
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
assert sync.sync_smart() == 0
|
|
mock_client.get_activities.assert_not_called()
|
|
|
|
def test_sync_smart_exception(mock_client, temp_storage):
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
with patch.object(sync, 'load_local_activities', side_effect=Exception("Fail")):
|
|
with pytest.raises(Exception):
|
|
sync.sync_smart()
|
|
|
|
def test_weekly_stats_missing_data(mock_client, temp_storage):
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_missing.json"), "w") as f:
|
|
json.dump({"activityId": 1}, f) # No startTimeLocal
|
|
with open(os.path.join(temp_storage, "activity_bad_date.json"), "w") as f:
|
|
json.dump({"activityId": 2, "startTimeLocal": "bad"}, f)
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
stats = sync.get_weekly_stats(weeks=1)
|
|
assert len(stats["labels"]) == 0
|
|
|
|
def test_dashboard_stats_exception(mock_client, temp_storage):
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
with patch.object(sync, 'load_local_activities', side_effect=Exception("Fail")):
|
|
with pytest.raises(Exception):
|
|
sync.get_dashboard_stats()
|
|
|
|
def test_validator_more_errors():
|
|
validator = WorkoutValidator()
|
|
# Segment with no steps
|
|
errors = validator.validate_workout({
|
|
"workoutName": "T", "sportType": {"sportTypeId": 1},
|
|
"workoutSegments": [{"workoutSteps": []}]
|
|
})
|
|
assert "Segment 0 has no steps" in errors
|
|
|
|
# Missing stepType or stepTypeId
|
|
errors = validator._validate_executable_step({}, "Ctx")
|
|
assert "Ctx: Missing stepType or stepTypeId" in errors
|
|
|
|
def test_sync_activities_save_error(mock_client, temp_storage):
|
|
mock_client.get_activities.return_value = [{"activityId": 1}]
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
with patch("builtins.open", side_effect=IOError("Fail")):
|
|
# Should not raise exception
|
|
count = sync.sync_activities(days=1)
|
|
assert count == 1
|
|
|
|
def test_load_local_activities_more_errors(temp_storage):
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
# File not ending in .json (should be ignored by load_local)
|
|
with open(os.path.join(temp_storage, "other.txt"), "w") as f:
|
|
f.write("text")
|
|
|
|
sync = GarminSync(None, storage_dir=temp_storage)
|
|
assert sync.load_local_activities() == []
|
|
|
|
def test_sync_activities_missing_id(mock_client, temp_storage):
|
|
# Coverage for sync.py:37
|
|
mock_client.get_activities.return_value = [{"name": "No ID"}]
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
count = sync.sync_activities(days=1)
|
|
assert count == 1
|
|
# File should not be saved
|
|
assert len(os.listdir(temp_storage)) == 0
|
|
|
|
def test_sync_smart_up_to_date(mock_client, temp_storage):
|
|
# Coverage for sync.py:105
|
|
today = date.today()
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_1.json"), "w") as f:
|
|
json.dump({"activityId": 1, "startTimeLocal": today.strftime("%Y-%m-%d %H:%M:%S")}, f)
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
assert sync.sync_smart() == 0
|
|
|
|
def test_weekly_stats_cutoff(mock_client, temp_storage):
|
|
# Coverage for sync.py:139
|
|
fixed_today = date(2026, 1, 1)
|
|
old_date = fixed_today - timedelta(days=100)
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_old.json"), "w") as f:
|
|
json.dump({
|
|
"activityId": 1,
|
|
"startTimeLocal": old_date.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"duration": 3600,
|
|
"activityType": {"typeKey": "running"}
|
|
}, f)
|
|
with patch("garmin.sync.date") as mock_date:
|
|
mock_date.today.return_value = fixed_today
|
|
# Allow creating new date objects
|
|
mock_date.side_effect = lambda *args, **kw: date(*args, **kw)
|
|
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
stats = sync.get_weekly_stats(weeks=1)
|
|
assert len(stats["datasets"]) == 0
|
|
|
|
def test_dashboard_stats_edge_cases(mock_client, temp_storage):
|
|
# Coverage for sync.py:241, 245-246
|
|
os.makedirs(temp_storage, exist_ok=True)
|
|
with open(os.path.join(temp_storage, "activity_no_start.json"), "w") as f:
|
|
json.dump({"activityId": 1}, f)
|
|
with open(os.path.join(temp_storage, "activity_bad_start.json"), "w") as f:
|
|
json.dump({"activityId": 2, "startTimeLocal": "invalid"}, f)
|
|
sync = GarminSync(mock_client, storage_dir=temp_storage)
|
|
stats = sync.get_dashboard_stats()
|
|
assert stats["summary"]["total_hours"] == 0
|
|
|
|
def test_validator_all_errors():
|
|
# Coverage for validator.py
|
|
v = WorkoutValidator()
|
|
# Missing fields
|
|
errors1 = v.validate_workout({})
|
|
assert any("Missing required field" in e for e in errors1)
|
|
|
|
# Missing sportTypeId
|
|
errors2 = v.validate_workout({
|
|
"workoutName": "T", "sportType": {}, "workoutSegments": [{"workoutSteps": []}]
|
|
})
|
|
assert "Missing sportType.sportTypeId" in errors2
|
|
|
|
# Empty segments
|
|
errors3 = v.validate_workout({
|
|
"workoutName": "T", "sportType": {"sportTypeId": 1}, "workoutSegments": []
|
|
})
|
|
assert "workoutSegments must be a non-empty list" in errors3
|
|
|
|
# Unknown step type
|
|
errors4 = v._validate_steps([{"type": "Unknown"}], "Ctx")
|
|
assert "Ctx Step 1: Unknown step type 'Unknown'" in errors4
|
|
|
|
# Invalid stepTypeId
|
|
errors5 = v._validate_executable_step({"stepType": {"stepTypeId": 99}, "endCondition": {"conditionTypeId": 1}}, "Ctx")
|
|
assert "Ctx: Invalid stepTypeId 99" in errors5
|
|
|
|
# Invalid iterations
|
|
errors6 = v._validate_repeat_group({"numberOfIterations": 0}, "Ctx")
|
|
assert "Ctx: Invalid iterations 0" in errors6
|
|
|
|
# Empty repeat group
|
|
errors7 = v._validate_repeat_group({"numberOfIterations": 1, "workoutSteps": []}, "Ctx")
|
|
assert "Ctx: Repeat group empty" in errors7
|
|
|
|
# Constants
|
|
constants = v.get_constants()
|
|
assert "SportType" in constants
|
|
|