288 lines
10 KiB
Python
288 lines
10 KiB
Python
from datetime import UTC, datetime
|
||
|
||
import pytest
|
||
from sqlalchemy import event
|
||
from test_app import add_history, login
|
||
|
||
from app.extensions import db
|
||
from app.models import Activity
|
||
from app.stats import explore, parse_query, period_bounds, query_text
|
||
from app.stats_dsl import QueryError
|
||
|
||
pytest_plugins = ("test_app",)
|
||
|
||
|
||
def test_query_round_trip_and_validation():
|
||
query = parse_query(
|
||
{
|
||
"metric": "cadence",
|
||
"secondary": "none",
|
||
"weeks": "26",
|
||
"minimum": "8",
|
||
"maximum": "20",
|
||
"hr_min": "140",
|
||
"hr_max": "150",
|
||
"end": "2026-10-02",
|
||
"compare": "no",
|
||
}
|
||
)
|
||
assert parse_query({"query": query_text(query)}) == query
|
||
tiny = parse_query({"minimum": "0.0000001"})
|
||
assert parse_query({"query": query_text(tiny)}) == tiny
|
||
for args in (
|
||
{"weeks": "105"},
|
||
{"query": ""},
|
||
{"minimum": "nan"},
|
||
{"minimum": "12", "maximum": "4"},
|
||
{"metric": "password"},
|
||
{"end": "2026-02-30"},
|
||
{"query": "runs\nwhere owner_id = 2"},
|
||
{"query": "runs\nsummarize pace\nby week\nby month"},
|
||
):
|
||
with pytest.raises(ValueError):
|
||
parse_query(args)
|
||
|
||
|
||
def test_weighting_missing_data_and_private_scope(app):
|
||
with app.app_context():
|
||
db.session.add_all(
|
||
[
|
||
Activity(
|
||
owner_id=1,
|
||
source_key="stats:long",
|
||
title="Long",
|
||
sport="running",
|
||
started_at=datetime(2026, 9, 29, tzinfo=UTC),
|
||
distance_m=15000,
|
||
duration_s=4500,
|
||
avg_hr=None,
|
||
),
|
||
Activity(
|
||
owner_id=2,
|
||
source_key="stats:private",
|
||
title="Secret",
|
||
sport="running",
|
||
started_at=datetime(2026, 9, 29, tzinfo=UTC),
|
||
distance_m=100000,
|
||
duration_s=10000,
|
||
avg_hr=200,
|
||
),
|
||
Activity(
|
||
owner_id=1,
|
||
source_key="stats:bike",
|
||
title="Bike",
|
||
sport="cycling",
|
||
started_at=datetime(2026, 9, 29, tzinfo=UTC),
|
||
distance_m=100000,
|
||
duration_s=10000,
|
||
),
|
||
]
|
||
)
|
||
db.session.commit()
|
||
statements = []
|
||
|
||
def record(_conn, _cursor, statement, _parameters, _context, _many):
|
||
statements.append(statement)
|
||
|
||
event.listen(db.engine, "before_cursor_execute", record)
|
||
try:
|
||
result = explore(1, "Australia/Sydney", parse_query({"end": "2026-10-02"}))
|
||
finally:
|
||
event.remove(db.engine, "before_cursor_execute", record)
|
||
assert result["count"] == 2
|
||
assert result["cards"][0]["value"] == "5:15"
|
||
assert result["cards"][1]["value"] == "140.0"
|
||
assert result["cards"][1]["coverage"] == 1
|
||
assert all("samples" not in sql and "route_geometry" not in sql for sql in statements)
|
||
|
||
|
||
def test_stats_routes_and_drillthrough(app):
|
||
client = app.test_client()
|
||
assert client.get("/stats").status_code == 302
|
||
assert client.get("/stats/runs").status_code == 302
|
||
login(client)
|
||
response = client.get("/stats?end=2026-10-02")
|
||
assert response.status_code == 200
|
||
assert b"A quiet morning run" in response.data
|
||
assert b"5:00" not in response.data # recorded pace is 6:00/km
|
||
response = client.get("/stats/runs?end=2026-10-02&bucket=28+Sep+2026")
|
||
assert response.status_code == 200
|
||
assert response.json["count"] == 1
|
||
assert response.json["rows"][0]["values"] == ["6:00", "140.0"]
|
||
assert client.get("/stats/runs?bucket=bad").status_code == 400
|
||
assert client.get("/stats?metric=unknown").status_code == 400
|
||
assert b"No runs match" in client.get("/stats?minimum=1000").data
|
||
client.post("/logout")
|
||
login(client, "other")
|
||
assert b"A quiet morning run" not in client.get("/stats?end=2026-10-02").data
|
||
assert client.get("/stats/runs?end=2026-10-02&bucket=28+Sep+2026").json["rows"] == []
|
||
|
||
|
||
def test_local_date_boundary_and_previous_alignment(app):
|
||
with app.app_context():
|
||
db.session.add(
|
||
Activity(
|
||
owner_id=1,
|
||
source_key="stats:boundary",
|
||
title="Local midnight",
|
||
sport="running",
|
||
started_at=datetime(2026, 9, 24, 14, tzinfo=UTC),
|
||
distance_m=1000,
|
||
duration_s=300,
|
||
)
|
||
)
|
||
db.session.add(
|
||
Activity(
|
||
owner_id=1,
|
||
source_key="stats:prior",
|
||
title="Prior",
|
||
sport="running",
|
||
started_at=datetime(2026, 9, 20, tzinfo=UTC),
|
||
distance_m=2000,
|
||
duration_s=800,
|
||
)
|
||
)
|
||
db.session.commit()
|
||
result = explore(1, "Australia/Sydney", parse_query({"weeks": "1", "end": "2026-10-01"}))
|
||
assert result["start"].isoformat() == "2026-09-25"
|
||
assert result["count"] == 2
|
||
assert result["prior_count"] == 1
|
||
assert 400 in result["chart"]["series"][0]["previous"]
|
||
|
||
|
||
def test_evidence_pagination(app):
|
||
with app.app_context():
|
||
add_history(40)
|
||
db.session.commit()
|
||
client = app.test_client()
|
||
login(client)
|
||
args = {"end": "2026-10-15", "group": "month", "bucket": "01 Sep 2026"}
|
||
response = client.get("/stats/runs", query_string=args)
|
||
assert response.status_code == 200
|
||
assert response.json["count"] == 31
|
||
assert len(response.json["rows"]) == 25
|
||
second = client.get("/stats/runs", query_string={**args, "page": 2})
|
||
assert len(second.json["rows"]) == 6
|
||
assert not {r["id"] for r in response.json["rows"]} & {r["id"] for r in second.json["rows"]}
|
||
assert client.get("/stats/runs", query_string={**args, "page": 3}).status_code == 400
|
||
|
||
|
||
def test_expanded_dsl_roundtrip_units_comments_and_dates():
|
||
source = """# My long runs
|
||
runs
|
||
during 2026-07-01 to 2026-09-30
|
||
where distance between 8000 and 20000 m
|
||
and pace <= 5:30 /km
|
||
and duration >= 45 min
|
||
and has cadence
|
||
and missing power
|
||
and title contains "River"
|
||
summarize pace, heart_rate, cadence, duration
|
||
by day
|
||
sort by pace asc
|
||
compare none"""
|
||
query = parse_query({"query": source})
|
||
assert query["extra_metrics"] == ["cadence", "duration"]
|
||
assert query["group"] == "day"
|
||
assert query["minimum"] == 8
|
||
assert query["maximum"] == 20
|
||
assert query["filters"][0]["value"] == 330
|
||
assert query["filters"][1]["value"] == 0.75
|
||
assert parse_query({"query": query_text(query)}) == query
|
||
months = parse_query({"query": "runs\nduring last 1 month\nending 2026-03-30\nsummarize pace"})
|
||
assert period_bounds(months, "Australia/Sydney")[0].isoformat() == "2026-02-28"
|
||
days = parse_query({"query": "runs\nduring last 30 days\nsummarize distance"})
|
||
assert parse_query({"query": query_text(days)}) == days
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("clause", "message"),
|
||
[
|
||
("and pace <= 5:99 /km", "seconds"),
|
||
("and pace <= 5:30 bpm", "Use /km"),
|
||
("and heart_rate between 150 and 140 bpm", "lower bound"),
|
||
("and has heart_rae", "heart_rate"),
|
||
("during last 25 months", "1–24"),
|
||
("summarize pace, pace", "different metrics"),
|
||
("sort by owner_id asc", "Unknown metric"),
|
||
],
|
||
)
|
||
def test_source_located_diagnostics(clause, message):
|
||
source = "runs\n" + clause
|
||
if not clause.startswith("summarize"):
|
||
source += "\nsummarize pace"
|
||
with pytest.raises(QueryError, match=message) as error:
|
||
parse_query({"query": source})
|
||
assert error.value.line == 2
|
||
|
||
|
||
def test_live_validation_and_extended_execution(app):
|
||
client = app.test_client()
|
||
assert client.post("/stats/validate", json={"query": "runs"}).status_code == 302
|
||
login(client)
|
||
assert client.post("/stats/validate", json={"query": "runs"}).json["valid"] is False
|
||
bad = client.post(
|
||
"/stats/validate", json={"query": "runs\nand pace <= 5:99 /km\nsummarize pace"}
|
||
)
|
||
assert bad.json["diagnostics"][0]["line"] == 2
|
||
source = """runs
|
||
during 2026-09-01 to 2026-10-02
|
||
where distance between 4 and 6 km
|
||
and pace >= 5:30 /km
|
||
and duration >= 25 min
|
||
and title contains "QUIET"
|
||
and missing cadence
|
||
summarize pace, heart_rate, distance, duration
|
||
by day
|
||
compare none
|
||
sort by distance desc"""
|
||
validation = client.post("/stats/validate", json={"query": source})
|
||
assert validation.json["valid"] is True
|
||
response = client.get("/stats", query_string={"query": source})
|
||
assert response.status_code == 200
|
||
assert b"A quiet morning run" in response.data
|
||
with app.app_context():
|
||
result = explore(1, "Australia/Sydney", parse_query({"query": source}))
|
||
assert result["count"] == 1
|
||
assert len(result["cards"]) == 4
|
||
assert client.post("/stats/validate", json={"query": 123}).status_code == 400
|
||
assert client.post("/stats/validate", json={"query": "a" * 4001}).json["valid"] is False
|
||
|
||
|
||
def test_validation_requires_csrf_and_does_not_read_activities(app):
|
||
client = app.test_client()
|
||
login(client)
|
||
with app.app_context():
|
||
statements = []
|
||
|
||
def record(_conn, _cursor, statement, _parameters, _context, _many):
|
||
statements.append(statement)
|
||
|
||
event.listen(db.engine, "before_cursor_execute", record)
|
||
engine = db.engine
|
||
try:
|
||
response = client.post("/stats/validate", json={"query": "runs\nsummarize pace"})
|
||
assert response.json["valid"] is True
|
||
assert not any("FROM activity" in statement for statement in statements)
|
||
finally:
|
||
event.remove(engine, "before_cursor_execute", record)
|
||
app.config["WTF_CSRF_ENABLED"] = True
|
||
assert client.post("/stats/validate", json={"query": "runs\nsummarize pace"}).status_code == 400
|
||
|
||
|
||
def test_fractional_units_roundtrip_and_strict_sorting(app):
|
||
query = parse_query(
|
||
{"query": "runs\nwhere duration >= 1 min\nand distance > 0.123456789 km\nsummarize pace"}
|
||
)
|
||
assert parse_query({"query": query_text(query)}) == query
|
||
with app.app_context():
|
||
add_history(3)
|
||
db.session.commit()
|
||
query = parse_query(
|
||
{"query": "runs\nduring 2026-09-01 to 2026-10-02\nsummarize pace\nsort by pace asc"}
|
||
)
|
||
rows = explore(1, "Australia/Sydney", query)["rows"]
|
||
assert rows[-1]["id"] == 1
|
||
query["filters"] = [{"field": "pace", "op": "<", "value": 300}]
|
||
assert explore(1, "Australia/Sydney", query)["count"] == 0
|