114 lines
5.1 KiB
Python
114 lines
5.1 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from app.analysis_migration import migrate, rewrite
|
|
from app.extensions import db
|
|
from app.models import Activity, SavedAnalysis, User
|
|
from app.stats import METRICS, explore, parse_query, query_text
|
|
from app.stats_dsl import QueryError
|
|
|
|
pytest_plugins = ('test_app',)
|
|
|
|
|
|
def parsed(text):
|
|
return parse_query({'query': text})
|
|
|
|
|
|
def test_removed_fields_and_clause_are_rejected():
|
|
removed = ['best_5km', 'best_10km', 'days_since_long_run', 'days_since_run',
|
|
'rolling_7d_distance', 'rolling_28d_count']
|
|
for field in removed:
|
|
assert field not in METRICS
|
|
with pytest.raises(QueryError):
|
|
parsed(f'runs\nsummarize {field}')
|
|
with pytest.raises(QueryError):
|
|
parsed('days\nlong run distance >= 10 km\nsummarize distance')
|
|
|
|
|
|
def test_expressions_in_filters_sort_calculations_and_named_summaries(app):
|
|
text = ('runs\nduring all time\nlet doubled = best_time(5000 m) * 2\n'
|
|
'where best_time(5000 m) < 31 min\n'
|
|
'summarize min(best_time(5000 m)) as "best_time(5 km)", doubled\n'
|
|
'sort by best_time(5000 m) asc\nby run\ncompare none')
|
|
q = parsed(text)
|
|
canonical = query_text(q)
|
|
assert 'analysis_expr_' not in canonical
|
|
assert 'as "best_time(5 km)"' in canonical
|
|
assert parsed(canonical) == q
|
|
with app.app_context():
|
|
row = db.session.get(Activity, 1)
|
|
row.best_efforts = [{'distance': 5000, 'seconds': 1800}]
|
|
db.session.commit()
|
|
result = explore(1, 'UTC', q)
|
|
assert result['count'] == 1
|
|
assert [c['value'] for c in result['cards']] == ['30:00', '60:00']
|
|
|
|
|
|
def test_multiple_since_conditions_use_individual_runs_and_prior_history(app):
|
|
with app.app_context():
|
|
db.session.query(Activity).delete()
|
|
db.session.add_all([
|
|
Activity(owner_id=owner, source_key=f'{owner}-{index}', title='Race ( A)',
|
|
sport='running', started_at=datetime(2026, 9, day, tzinfo=UTC),
|
|
distance_m=km * 1000, duration_s=1800, avg_hr=140)
|
|
for index, (owner, day, km) in enumerate([(1, 1, 20), (1, 3, 10), (1, 4, 6), (1, 4, 6), (2, 5, 30)])
|
|
])
|
|
db.session.commit()
|
|
q = parsed('days\nduring 2026-09-04 to 2026-09-05\nwhere distance >= 0 km\n'
|
|
'summarize latest(days_since_run(distance >= 10 km)), '
|
|
'latest(days_since_run(distance >= 20 km)), latest(days_since_run())\n'
|
|
'by day\ncompare none')
|
|
result = explore(1, 'UTC', q)
|
|
assert [g['raw_values'] for g in result['groups']] == [[1, 3, 0], [2, 4, 1]]
|
|
q = parsed('days\nduring 2026-09-05 to 2026-09-05\n'
|
|
'summarize latest(days_since_run(title contains "Race ( A)"))\n'
|
|
'by day\ncompare none')
|
|
assert parsed(query_text(q)) == q
|
|
assert explore(1, 'UTC', q)['groups'][0]['raw_values'] == [1]
|
|
|
|
|
|
@pytest.mark.parametrize('expression', ['best_time(3 km)', 'rolling_sum(distance, 0 days)',
|
|
'rolling_mean(unknown, 14 days)', 'days_since_run(distance >= 10 bpm)'])
|
|
def test_invalid_expression_reports_original_line(expression):
|
|
with pytest.raises(QueryError) as error:
|
|
parsed(f'days\nduring last 16 weeks\nsummarize latest({expression})')
|
|
assert error.value.line == 3
|
|
|
|
|
|
def test_migration_preserves_labels_settings_other_owners_and_is_idempotent(app):
|
|
old = ('days\nduring last 16 weeks\nlong run distance >= 20 km\n'
|
|
'summarize latest(days_since_long_run) as "days_since_long_run", '
|
|
'latest(rolling_7d_count)\nby day\ncompare none')
|
|
with app.app_context():
|
|
own = SavedAnalysis(owner_id=1, name='Continuity', query=old, pinned=True,
|
|
position=7, width='full', notes='Keep these notes')
|
|
other = SavedAnalysis(owner_id=2, name='Other owner', query=old)
|
|
db.session.add_all([own, other])
|
|
db.session.commit()
|
|
assert migrate(db.session.get(User, 1)) == 1
|
|
assert 'days_since_run(distance >= 20 km)' in own.query
|
|
assert 'rolling_sum(sessions, 7 days)' in own.query
|
|
assert 'as "days_since_long_run"' in own.query
|
|
assert own.pinned and own.position == 7 and own.width == 'full'
|
|
assert own.notes == 'Keep these notes' and other.query == old
|
|
assert migrate(db.session.get(User, 1)) == 0
|
|
assert rewrite(own.query) == own.query
|
|
|
|
|
|
def test_daily_rolling_session_count_includes_multiple_runs_and_rest_days(app):
|
|
with app.app_context():
|
|
db.session.query(Activity).delete()
|
|
db.session.add_all([
|
|
Activity(owner_id=1, source_key=str(index), title='Run', sport='running',
|
|
started_at=datetime(2026, 9, 1, index, tzinfo=UTC),
|
|
distance_m=5000, duration_s=1800)
|
|
for index in range(2)
|
|
])
|
|
db.session.commit()
|
|
q = parsed('days\nduring 2026-09-02 to 2026-09-02\n'
|
|
'summarize latest(rolling_sum(sessions, 28 days))\nby day\ncompare none')
|
|
result = explore(1, 'UTC', q)
|
|
assert result['groups'][0]['raw_values'] == [2]
|
|
assert result['cards'][0]['unit'] == 'runs'
|