sstent
d1cfd0fd8e
- Add Fitbit authentication flow (save credentials, OAuth callback handling) - Implement Garmin MFA support with successful session/cookie handling - Optimize segment discovery with new sampling and activity query services - Refactor database session management in discovery API for better testability - Enhance activity data parsing for charts and analysis - Update tests to use testcontainers and proper dependency injection - Clean up repository by ignoring and removing tracked transient files (.pyc, .db)
117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
|
|
from unittest.mock import MagicMock
|
|
import sys
|
|
|
|
# Mock fitdecode before imports since it might not be installed in local env (running in docker)
|
|
sys.modules['fitdecode'] = MagicMock()
|
|
|
|
import pytest
|
|
from src.utils.geo import ramer_douglas_peucker, haversine_distance, calculate_bounds
|
|
from src.services.segment_matcher import SegmentMatcher
|
|
from src.models.activity import Activity
|
|
from src.models.segment import Segment
|
|
from datetime import datetime, timedelta
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
def test_haversine():
|
|
# Dist between (0,0) and (0,1) deg is ~111km
|
|
d = haversine_distance(0, 0, 0, 1)
|
|
# 1 deg lat ~ 111.32 km
|
|
assert 110000 < d < 112000
|
|
|
|
def test_rdp_simple():
|
|
# Points on a line
|
|
points = [[0,0], [1,1], [2,2]]
|
|
# Should simplify to [0,0], [2,2]
|
|
simplified = ramer_douglas_peucker(points, epsilon=0.1)
|
|
assert len(simplified) == 2
|
|
assert simplified[0] == [0,0]
|
|
assert simplified[1] == [2,2]
|
|
|
|
def test_rdp_peak():
|
|
# Triangle
|
|
points = [[0,0], [1,10], [2,0]] # [lon, lat] note: RDP expects [lon, lat] usually?
|
|
# My RDP implementation uses x,y so order doesn't matter for geometric shape
|
|
simplified = ramer_douglas_peucker(points, epsilon=1.0)
|
|
assert len(simplified) == 3
|
|
|
|
def test_bounds():
|
|
points = [[0,0], [10, 10], [-5, 5]]
|
|
bounds = calculate_bounds(points)
|
|
# bounds is [min_lat, min_lon, max_lat, max_lon]
|
|
# points are [lon, lat].
|
|
# 0,0 -> lat=0, lon=0
|
|
# 10,10 -> lat=10, lon=10
|
|
# -5,5 -> lat=5, lon=-5
|
|
|
|
assert bounds[0] == 0 # min_lat (0)
|
|
assert bounds[2] == 10 # max_lat (10)
|
|
assert bounds[1] == -5 # min_lon (-5)
|
|
assert bounds[3] == 10 # max_lon (10)
|
|
|
|
def test_matcher_logic():
|
|
# Mock DB session
|
|
mock_session = MagicMock()
|
|
|
|
# Create segment [0,0] -> [0, 0.01] (approx 1.1km north)
|
|
segment_points = [[0,0], [0, 0.01]]
|
|
segment = Segment(
|
|
id=1,
|
|
name="Test Seg",
|
|
points=json.dumps(segment_points),
|
|
bounds=json.dumps(calculate_bounds(segment_points)),
|
|
distance=1110.0,
|
|
activity_type='cycling'
|
|
)
|
|
|
|
mock_session.query.return_value.filter.return_value.filter.return_value.all.return_value = [segment]
|
|
|
|
matcher = SegmentMatcher(mock_session)
|
|
|
|
# Create activity trace that covers this
|
|
# 0,0 at T=0
|
|
# 0,0.01 at T=100s
|
|
act_points = [[0,0], [0, 0.005], [0, 0.01]]
|
|
|
|
# Mock activity
|
|
activity = Activity(id=100, start_time=datetime.now())
|
|
# Matcher needs to use parsers internally? Or uses slice of points?
|
|
# Matcher logic (_match_segment) uses points list passed to match_activity
|
|
|
|
# Wait, _match_segment needs timestamps to calc elapsed time.
|
|
# We need to mock extract_timestamps_from_file or patch it
|
|
from unittest.mock import patch
|
|
|
|
# Patch the parser where it is IMPORTED/USED in the SegmentMatcher service (or source)
|
|
# _create_effort imports extract_activity_data from ..services.parsers
|
|
with patch('src.services.parsers.extract_activity_data') as mock_extract:
|
|
# 0,0@T0, 0,0.005@T50, 0,0.01@T100
|
|
start_time = datetime.now()
|
|
timestamps = [start_time, start_time + timedelta(seconds=50), start_time + timedelta(seconds=100)]
|
|
|
|
# extract_activity_data returns dict
|
|
mock_extract.return_value = {
|
|
'timestamps': timestamps,
|
|
'heart_rate': [100, 110, 120],
|
|
'power': [200, 210, 220],
|
|
'cadence': [80, 80, 80],
|
|
'speed': [10, 10, 10],
|
|
'respiration_rate': [20, 20, 20]
|
|
}
|
|
|
|
# Add dummy content
|
|
activity.file_content = b'dummy'
|
|
activity.file_type = 'fit'
|
|
|
|
# Run match
|
|
efforts = matcher.match_activity(activity, act_points)
|
|
|
|
assert len(efforts) == 1
|
|
effort = efforts[0]
|
|
assert effort.segment_id == 1
|
|
assert effort.elapsed_time == 100.0
|
|
|
|
|
|
|