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)
131 lines
4.7 KiB
Python
131 lines
4.7 KiB
Python
import pytest
|
|
from src.services.discovery import SegmentDiscoveryService
|
|
from src.models.activity import Activity
|
|
from unittest.mock import MagicMock
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
|
|
def test_decimate_points():
|
|
service = SegmentDiscoveryService(None)
|
|
|
|
# 3 points in a line, 20m apart. Min dist 30m.
|
|
# p0 (0,0) -> 20m -> p1 (20m, 0) -> 20m -> p2 (40m, 0)
|
|
# Result should be p0, p2.
|
|
|
|
# Approx: 0.00018 deg lat ~ 20m
|
|
points = [
|
|
[0.0, 0.0],
|
|
[0.0, 0.00018],
|
|
[0.0, 0.00036]
|
|
]
|
|
|
|
decimated = service._decimate_points(points, min_dist=30.0)
|
|
assert len(decimated) == 2
|
|
assert decimated[0] == [0.0, 0.0]
|
|
assert decimated[1] == [0.0, 0.00036]
|
|
|
|
def test_discovery_integration():
|
|
# Setup
|
|
mock_db = MagicMock()
|
|
service = SegmentDiscoveryService(mock_db)
|
|
|
|
# Mock Activity Objects
|
|
act1 = MagicMock(spec=Activity)
|
|
act1.id = 101
|
|
act1.activity_type = "cycling"
|
|
act1.start_time = datetime(2025,1,1)
|
|
act1.file_content = b"mock_content"
|
|
act1.file_type = "fit"
|
|
|
|
act2 = MagicMock(spec=Activity)
|
|
act2.id = 102
|
|
act2.activity_type = "cycling"
|
|
act2.start_time = datetime(2025,1,2)
|
|
act2.file_content = b"mock_content"
|
|
act2.file_type = "fit"
|
|
|
|
# Mock extract_points_from_file to return overlapping paths
|
|
# They share a segment: (0,0) -> (0, 0.0005) -> (0, 0.0010)
|
|
# act1 goes further: -> (0, 0.0020)
|
|
# act2 stops or goes elsewhere
|
|
|
|
# Approx 0.0002 deg grid step (~20m)
|
|
# 0.0000 -> 0.0010 is about 100m. Might be too short for 200m cutoff.
|
|
# Let's make it longer. 0.0030 deg is ~300m.
|
|
|
|
path_shared = [[0.0, y*0.0001] for y in range(30)] # 0 to 0.0029
|
|
|
|
service._decimate_points = MagicMock(side_effect=[
|
|
path_shared, # act1
|
|
path_shared # act2
|
|
])
|
|
|
|
# Mock DB query
|
|
# First query is for Activities
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.all.return_value = [act1, act2]
|
|
|
|
# Second query is for Segments (deduplication) - called in _deduplicate_against_db
|
|
# It calls db.query(Segment).filter(...).all()
|
|
# We need to distinguish them or just ensure the Segment query returns [].
|
|
# But current mock chain is generic.
|
|
# 'query' returns a Mock. Calling 'filter' on it returns SAME/NEW Mock.
|
|
# If we reuse the same mock chain, it returns [act1, act2].
|
|
# We should distinguish based on call args or set up side_effect.
|
|
|
|
def side_effect_query(model):
|
|
m = MagicMock()
|
|
if model == Activity:
|
|
# Depth 3
|
|
m.filter.return_value.filter.return_value.filter.return_value.all.return_value = [act1, act2]
|
|
# Depth 2 (type + start_date)
|
|
m.filter.return_value.filter.return_value.all.return_value = [act1, act2]
|
|
# Depth 1 (type only)
|
|
m.filter.return_value.all.return_value = [act1, act2]
|
|
else:
|
|
# Segment
|
|
m.filter.return_value.all.return_value = []
|
|
return m
|
|
|
|
mock_db.query.side_effect = side_effect_query
|
|
|
|
# We need to mock extract_points_from_file at module level or patch it
|
|
# But since we mocked _decimate_points, extract_points_from_file is called but result ignored/passed to decimate.
|
|
# Wait, discovery.py calls:
|
|
# raw_points = extract_points_from_file(...)
|
|
# simplified = self._decimate_points(raw_points, ...)
|
|
# So we need to patch extract_points_from_file to avoid error on "mock_content"
|
|
|
|
with patch('src.services.discovery.extract_points_from_file', return_value=[[0,0]]) as mock_extract:
|
|
candidates, _ = service.discover_segments("cycling", datetime(2025,1,1))
|
|
|
|
# Assertions
|
|
# Since we mocked decimate to return identical long paths, they should cluster.
|
|
# result should contain candidates.
|
|
# activity_ids should validly contain [101, 102]
|
|
|
|
assert len(candidates) >= 1
|
|
candidate = candidates[0]
|
|
assert 101 in candidate.activity_ids
|
|
assert 102 in candidate.activity_ids
|
|
assert isinstance(candidate.activity_ids[0], int)
|
|
|
|
|
|
def test_connected_components():
|
|
service = SegmentDiscoveryService(None)
|
|
|
|
cells = {
|
|
(0,0): {1}, (0,1): {1}, (0,2): {1}, # Vertical loose chain
|
|
(0,3): {1}, (0,4): {1}, (0,5): {1}, # Total 6
|
|
(10,10): {2}, (10,11): {2} # Separate small cluster (length 2)
|
|
}
|
|
|
|
# Connected components with min size 5
|
|
# Since we set min cluster size to 5 in implementation
|
|
|
|
comps = service._find_connected_components(cells, 0.0002)
|
|
assert len(comps) == 1
|
|
assert len(comps[0]) == 6
|
|
assert (0,0) in comps[0]
|
|
assert (10,10) not in comps[0]
|