Stride/tests/test_stats.py

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, query_text

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