import csv import io from datetime import UTC, datetime import pytest from test_app import login from app.extensions import db from app.models import Activity from app.stats import explore, parse_query, query_text from app.stats_analysis import statistic from app.stats_dsl import QueryError pytest_plugins = ("test_app",) def query(source): return parse_query({"query": "runs\n" + source}) def run(key, day, km=10, seconds=3000, owner=1, hr=150, hour=0, title=None): return Activity( owner_id=owner, source_key=key, title=title or key, sport="running", started_at=datetime(2026, 9, day, hour, tzinfo=UTC), distance_m=km * 1000, duration_s=seconds, avg_hr=hr, ) def analyze(source): return explore(1, "Australia/Sydney", query(source)) @pytest.mark.parametrize( "source", [ "during all time\nsummarize gap_since_previous_run, distance\nby run\nsort by gap_since_previous_run desc\ntake 10", "let efficiency = speed / heart_rate\nwhere efficiency > 0.05 (km/h/bpm)\nsummarize median(efficiency), p95(pace), count()\nby weekday\nhaving count() >= 3 runs\nrank by median(efficiency) desc\nlimit 4\ncompare none", 'where distance < 8 km\nand has heart_rate\nor title contains "between hills"\nand duration > 20 min\nsummarize sum(distance), mean(heart_rate)\nby distance bins 5 km\ncompare none', "summarize median(pace), stddev(pace)\nby week\nhaving median(pace) < 5:30 /km\nhaving stddev(pace) < 60 sec/km\ntrend rolling 4 buckets\nplot speed vs heart_rate\nchart line", "summarize sum(pace), mean(duration)\nby hour\nhaving sum(pace) > 300 sec/km\nchart none", ], ) def test_analysis_canonical_roundtrip(source): parsed = query(source) assert query(query_text(parsed).removeprefix("runs\n")) == parsed def test_predecessor_and_rolling_use_history_before_filters(app): with app.app_context(): db.session.add_all( [ run("first", 1), run("intermediate", 8, km=1), run("target", 9), run("private", 8, owner=2), ] ) db.session.commit() result = analyze( "during 2026-09-09 to 2026-09-09\nwhere distance >= 5 km\n" "summarize gap_since_previous_run, rolling_7d_distance, rolling_7d_count, run_day_streak\nby run\ncompare none" ) assert result["count"] == 1 assert result["rows"][0]["raw_values"] == [1, 11, 2, 2] assert result["rows"][0]["previous_run"]["title"] == "intermediate" # The exact 168-hour boundary is excluded from the rolling window. result = analyze( "during 2026-09-08 to 2026-09-08\nsummarize rolling_7d_distance\nby run\ncompare none" ) assert result["rows"][0]["raw_values"] == [1] def test_biggest_gaps_ranking_take_and_missing(app): with app.app_context(): db.session.add_all([run("first", 1), run("second", 3), run("third", 20)]) db.session.commit() result = analyze( "during all time\nsummarize gap_since_previous_run, distance\nby run\nsort by gap_since_previous_run desc\ntake 2" ) assert [row["title"] for row in result["rows"]] == ["third", "A quiet morning run"] assert [row["raw_values"][0] for row in result["rows"]] == [17, 10] assert result["count"] == 2 assert len(result["groups"]) == 2 def test_explicit_statistics_weighting_missing_and_percentiles(app): with app.app_context(): db.session.add_all([run("fast-long", 29, km=15, seconds=4500, hr=None)]) db.session.commit() result = analyze( "during 2026-09-29 to 2026-09-30\nsummarize mean(pace), weighted_mean(pace), p90(pace), stddev(pace), count(heart_rate), count()\nby all\ncompare none" ) assert result["groups"][0]["raw_values"] == [330, 315, 354, 30, 1, 2] assert result["cards"][3]["unit"] == "sec/km" assert result["cards"][3]["value"] == "30.0" assert statistic([None], "sum") is None assert statistic([], "count") == 0 assert statistic([1, 2, 3, 4], "p25") == 1.75 def test_arithmetic_units_nulls_and_correlation(app): with app.app_context(): db.session.add_all( [run("slow", 28, km=5, seconds=3600, hr=100), run("missing", 29, hr=None)] ) db.session.commit() result = analyze( "during 2026-09-28 to 2026-09-30\nlet efficiency = speed / heart_rate\nlet empty = speed / (heart_rate - heart_rate)\nsummarize mean(efficiency), count(efficiency), count(empty)\nby all\nplot speed vs heart_rate\ncompare none" ) assert result["groups"][0]["raw_values"][0] == pytest.approx((5 / 100 + 10 / 140) / 2) assert result["groups"][0]["raw_values"][1:] == [2, 0] assert result["chart"]["scatter"]["correlation"] == pytest.approx(1) assert len(result["chart"]["scatter"]["current"]) == 2 @pytest.mark.parametrize( "clause", [ 'let bad = __import__("os")', "let bad = distance.__class__", "let bad = distance[0]", "let bad = distance ** 2", "let bad = distance + heart_rate", "let analysis_agg_0 = distance", "let bad = unknown / distance", "take 1001", "by distance bins 0 km", "summarize distance\nrank by median(pace) desc", "by weekday\ntrend cumulative", "trend rolling 0 buckets", "during all time\ncompare previous period", "plot speed vs heart_rate\nchart none", ], ) def test_rejects_unsafe_or_incoherent_analysis(clause): with pytest.raises(QueryError): query(clause + ("\nsummarize pace" if "summarize" not in clause else "")) def test_or_precedence_bins_having_ranking_and_drill(app): with app.app_context(): db.session.add_all( [ run("short-missing", 28, km=2, hr=None), run("long", 29, km=20, hr=170), run("short", 29, km=2), ] ) db.session.commit() result = analyze( "during 2026-09-28 to 2026-09-30\nwhere distance < 3 km\nand has heart_rate\nor distance >= 20 km\nand heart_rate > 160 bpm\nsummarize count(), sum(distance)\nby distance bins 5 km\nhaving count() >= 1 runs\nrank by sum(distance) desc\nlimit 1\ncompare none" ) assert result["count"] == 1 assert result["groups"][0]["label"] == "20–25 km" assert result["rows"][0]["title"] == "long" drilled = explore( 1, "Australia/Sydney", query("during 2026-09-28 to 2026-09-30\nsummarize count()\nby weekday\ncompare none"), selected_bucket="Wednesday", ) assert drilled["evidence_count"] == 1 def test_rolling_trends_pool_observations_and_percent_change(app): with app.app_context(): db.session.add_all( [ run("a", 28, km=10, seconds=3000), run("b", 28, km=5, seconds=2400), run("c", 29, km=5, seconds=1500), ] ) db.session.commit() base = ( "during 2026-09-28 to 2026-09-30\nsummarize mean(pace), count()\nby day\ncompare none\n" ) result = analyze(base + "trend rolling 2 buckets") assert result["chart"]["series"][0]["values"] == [390, 360, 330] assert result["chart"]["series"][1]["values"] == [2, 3, 2] result = analyze(base + "trend percent_change") assert result["chart"]["series"][1]["values"] == [None, -50, 0] assert result["chart"]["series"][0]["unit"] == "%" result = analyze(base + "trend change") assert result["chart"]["series"][0]["values"] == [None, -90, 60] assert result["chart"]["series"][0]["unit"] == "sec/km" def test_csv_precision_private_scope_and_formula_titles(app): with app.app_context(): db.session.add_all( [ run("formula", 29, km=1.123456789, title='=HYPERLINK("bad")'), run("secret", 29, owner=2, title="Private secret"), ] ) db.session.commit() client = app.test_client() assert client.get("/stats/export").status_code == 302 login(client) source = "runs\nduring 2026-09-29 to 2026-09-30\nsummarize distance, pace\nby all\ncompare none" response = client.get("/stats/export", query_string={"query": source, "kind": "runs"}) assert response.status_code == 200 records = list(csv.reader(io.StringIO(response.text))) assert records[0][-1] == "pace (sec/km)" assert "Private secret" not in response.text formula = next(row for row in records if "HYPERLINK" in row[0]) assert formula[0].startswith("'=HYPERLINK") assert float(formula[-2]) == pytest.approx(1.123456789) assert "no-store" in response.headers["Cache-Control"] assert client.get("/stats", query_string={"query": source}).status_code == 200 def test_elapsed_gaps_across_dst_and_same_day_streaks(app): with app.app_context(): first = run("before-dst", 1) first.started_at = datetime(2026, 10, 2, 22, tzinfo=UTC) # 08:00 local Oct 3 second = run("after-dst", 2) second.started_at = datetime(2026, 10, 3, 21, tzinfo=UTC) # 08:00 local Oct 4 duplicate = run("same-day", 3) duplicate.started_at = datetime(2026, 10, 3, 21, tzinfo=UTC) db.session.add_all([first, second, duplicate]) db.session.commit() result = analyze( "during 2026-10-03 to 2026-10-04\nsummarize gap_since_previous_run, run_day_streak\nby run\nsort by date asc\ncompare none" ) assert result["rows"][1]["raw_values"] == [23 / 24, 2] assert result["rows"][2]["raw_values"] == [0, 2] def test_trend_and_group_drill_distinguish_window_and_input_values(app): with app.app_context(): db.session.add_all([run("earlier", 29, km=15, seconds=4500)]) db.session.commit() source = "during 2026-09-29 to 2026-09-30\nsummarize stddev(pace), count(heart_rate)\nby day\ntrend cumulative\ncompare none" parsed = query(source) window = explore(1, "Australia/Sydney", parsed, selected_bucket="30 Sep 2026") group = explore( 1, "Australia/Sydney", parsed, selected_bucket="30 Sep 2026", raw_group=True ) assert window["evidence_count"] == 2 assert group["evidence_count"] == 1 assert group["rows"][0]["raw_values"] == [360, 140] assert group["evidence_columns"][0]["key"] == "pace" def test_matching_calculated_units_and_spaced_aggregate(): source = "let measured_speed = distance / duration\nlet combined_speed = measured_speed + speed\nsummarize median( combined_speed )\nby all" parsed = query(source) assert parsed["derived"][0]["unit"] == "km/h" assert parsed["metric"] == "median(combined_speed)" assert query(query_text(parsed).removeprefix("runs\n")) == parsed def test_sensor_zero_missing_and_initial_gap_coverage(app): with app.app_context(): first = run("first-ever", 1, hr=0) db.session.add(first) db.session.commit() result = analyze( "during 2026-09-01 to 2026-09-01\nsummarize count(gap_since_previous_run), count(heart_rate), count()\nby all\ncompare none" ) assert result["groups"][0]["raw_values"] == [0, 0, 1]