268 lines
11 KiB
Python
268 lines
11 KiB
Python
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_sum(distance, 7 days), rolling_count(distance, 7 days), 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_sum(distance, 7 days)\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] |