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)
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
import io
|
|
import fitdecode
|
|
import struct
|
|
from src.services.parsers import extract_streams, extract_activity_data, extract_summary
|
|
from src.utils.sampling import downsample_streams
|
|
|
|
# Mock FIT file creation (simplified)
|
|
def create_mock_fit_content(points=100, has_gps=True):
|
|
# This is hard to mock binary FIT correctly without using the library to Write.
|
|
# But fitdecode is a reader.
|
|
# Alternatively, we can mock the functions wrapping fitdecode?
|
|
# Or just use a very simple known FIT structure if possible?
|
|
# Writing a valid FIT file in a test without a library is hard.
|
|
#
|
|
# Better approach for UNIT testing parsers:
|
|
# 1. Mock the fitdecode.FitReader to return frames we want.
|
|
# 2. Test `extract_streams` logic given those frames.
|
|
pass
|
|
|
|
class MockFrame:
|
|
def __init__(self, name, values):
|
|
self.frame_type = fitdecode.FIT_FRAME_DATA
|
|
self.name = name
|
|
self.values = values
|
|
|
|
def has_field(self, name):
|
|
return name in self.values
|
|
|
|
def get_value(self, name):
|
|
return self.values.get(name)
|
|
|
|
@pytest.fixture
|
|
def mock_fit_reader(monkeypatch):
|
|
def mock_reader(file_obj):
|
|
return MockFitReaderContext()
|
|
|
|
monkeypatch.setattr(fitdecode, 'FitReader', mock_reader)
|
|
|
|
class MockFitReaderContext:
|
|
def __init__(self):
|
|
self.frames = []
|
|
|
|
def __enter__(self):
|
|
return self.frames
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
pass
|
|
|
|
def add_record(self, ts, lat=None, lon=None, hr=None, pwr=None, alt=None):
|
|
vals = {'timestamp': ts}
|
|
if lat is not None: vals['position_lat'] = int(lat / (180.0 / 2**31))
|
|
if lon is not None: vals['position_long'] = int(lon / (180.0 / 2**31))
|
|
if hr is not None: vals['heart_rate'] = hr
|
|
if pwr is not None: vals['power'] = pwr
|
|
if alt is not None: vals['enhanced_altitude'] = alt
|
|
self.frames.append(MockFrame('record', vals))
|
|
|
|
def test_extract_activity_data_no_gps(monkeypatch):
|
|
# Test that we can extract data even if GPS is missing (Fix for the Bug)
|
|
|
|
frames_iter = []
|
|
|
|
class MockIter:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def __iter__(self): return iter(frames_iter)
|
|
|
|
monkeypatch.setattr(fitdecode, 'FitReader', lambda f: MockIter())
|
|
|
|
# Add records without GPS
|
|
base_time = datetime(2023, 1, 1, 12, 0, 0)
|
|
for i in range(10):
|
|
vals = {
|
|
'timestamp': base_time + timedelta(seconds=i),
|
|
'heart_rate': 100 + i,
|
|
'power': 200 + i
|
|
}
|
|
frames_iter.append(MockFrame('record', vals))
|
|
|
|
data = extract_activity_data(b'dummy', 'fit', strict_gps=False)
|
|
|
|
assert len(data['timestamps']) == 10
|
|
assert len(data['heart_rate']) == 10
|
|
assert data['heart_rate'][0] == 100
|
|
assert data['points'][0] is None # No GPS
|
|
|
|
def test_extract_streams_logic(monkeypatch):
|
|
frames_iter = []
|
|
class MockIter:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def __iter__(self): return iter(frames_iter)
|
|
monkeypatch.setattr(fitdecode, 'FitReader', lambda f: MockIter())
|
|
|
|
base_time = datetime(2023, 1, 1, 12, 0, 0)
|
|
# 5 points with GPS, 5 without
|
|
for i in range(5):
|
|
vals = {
|
|
'timestamp': base_time + timedelta(seconds=i),
|
|
'position_lat': int(10 * (2**31 / 180.0)),
|
|
'position_long': int(20 * (2**31 / 180.0)),
|
|
'enhanced_altitude': 100 + i,
|
|
'heart_rate': 140
|
|
}
|
|
frames_iter.append(MockFrame('record', vals))
|
|
|
|
for i in range(5, 10):
|
|
vals = {
|
|
'timestamp': base_time + timedelta(seconds=i),
|
|
'heart_rate': 150
|
|
# No GPS
|
|
}
|
|
frames_iter.append(MockFrame('record', vals))
|
|
|
|
streams = extract_streams(b'dummy', 'fit')
|
|
|
|
assert len(streams['time']) == 10
|
|
assert streams['altitude'][0] == 100
|
|
assert streams['altitude'][9] is None
|
|
assert streams['heart_rate'][9] == 150
|
|
|
|
def test_extract_summary(monkeypatch):
|
|
frames_iter = []
|
|
class MockIter:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def __iter__(self): return iter(frames_iter)
|
|
monkeypatch.setattr(fitdecode, 'FitReader', lambda f: MockIter())
|
|
|
|
vals = {
|
|
'total_distance': 1000.0,
|
|
'total_timer_time': 3600.0,
|
|
'avg_heart_rate': 145
|
|
}
|
|
frames_iter.append(MockFrame('session', vals))
|
|
|
|
summary = extract_summary(b'dummy', 'fit')
|
|
assert summary['total_distance'] == 1000.0
|
|
assert summary['avg_heart_rate'] == 145
|