from datetime import UTC, date, datetime from types import SimpleNamespace import pytest from sqlalchemy import event from test_app import login from app.display import persist_analysis from app.extensions import db from app.models import Activity, DailyHealth, SavedAnalysis, User from app.recording_windows import analyze from app.recovery import normalize, sync_health from app.route_benchmarks import similar from app.stats import explore, parse_query, query_text from app.stats_dsl import QueryError from app.training_signals import install, signal_queries from app.training_sources import reduce_runs pytest_plugins = ("test_app",) def parsed(source): return parse_query({"query": source}) def recording(seconds=900, hr=140, origin=1_700_000_000, speed=2.5): return [{"time": origin + i, "distance": speed * i, "elevation": 20, "heart_rate": hr, "cadence": 160, "ground_contact": 300, "stride_length": 1, "speed": speed} for i in range(seconds + 1)] def add_run(day, distance=5000, duration=1800, **kwargs): row = Activity(owner_id=1, source_key=str(day), title=str(day), sport="running", started_at=datetime.fromisoformat(day).replace(tzinfo=UTC), distance_m=distance, duration_s=duration, avg_hr=140, **kwargs) db.session.add(row) return row def test_windows_preserve_epoch_offsets_duration_and_sensor_time(): units = analyze(recording(905)) assert len(units) == 31 assert sum(u["duration"] for u in units) == 905 assert sum(u["distance"] for u in units) == pytest.approx(2262.5) assert units[20]["elapsed"] == 10 assert units[20]["moving_offset"] == 10 assert units[-1]["duration"] == 5 assert all(u["grade"] == 0 for u in units) assert sum(sum(u["hr_seconds"].values()) for u in units) == 905 def test_windows_do_not_bridge_stops_or_recording_breaks(): rows = recording(20) rows += [{**r, "time": r["time"] + 40, "distance": r["distance"] + 50} for r in recording(20)[1:]] units = analyze(rows) assert [u["duration"] for u in units] == [20, 19] assert units[1]["elapsed"] == pytest.approx(41 / 60) rows = recording(60) for r in rows[20:41]: r["distance"] = 50 units = analyze(rows) assert all(u["duration"] <= 30 for u in units) assert sum(u["duration"] for u in units) < 60 def test_sensor_artifacts_and_missing_terrain_are_not_filled(): rows = recording(30) rows[10]["heart_rate"] = 250 for row in rows: row["elevation"] = None unit = analyze(rows)[0] assert unit["sensor_coverage"] < 100 assert unit["hr_artifact_percent"] > 0 assert "250" not in unit["hr_seconds"] assert unit["grade"] is unit["gap"] is None assert unit["terrain_coverage"] == 0 def test_matched_sessions_are_independent_and_filters_precede_reduction(app): with app.app_context(): db.session.query(Activity).delete() first = add_run("2026-09-01", samples=recording(1800, 150)) second = add_run("2026-09-02", samples=recording(900, 130)) db.session.flush() for row in (first, second): persist_analysis(row) db.session.commit() q = parsed("intervals\nduring 2026-09-01 to 2026-09-03\nreduce by run min 3 min\n" "where elapsed >= 10 min\nand duration >= 25 s\nand grade between -2 and 2 %\n" "summarize mean(heart_rate), count()\nby all\ncompare none") result = explore(1, "UTC", q) assert result["count"] == result["coverage"]["sessions"] == 2 assert result["cards"][0]["value"] == "140.0" assert result["coverage"]["minutes"] == 25 assert {row["id"] for row in result["rows"]} == {first.id, second.id} q["reduce"] = 6 assert explore(1, "UTC", q)["count"] == 1 def unit(activity_id, index, hr=140, speed=2.5, grade=0): values = {"heart_rate": hr, "sensor_coverage": 100, "grade": grade, "elapsed": index / 2, "pace": 1000 / speed} return SimpleNamespace(id=activity_id * 1000 + index, activity_id=activity_id, title="window", parent_title="run", duration_s=30, distance_m=30 * speed, values=values) def test_post_warmup_decoupling_requires_comparable_halves(): rows = [unit(1, i, 130 if i < 20 else 143) for i in range(40)] catalog = {k: None for k in ["heart_rate", "heart_rate_drift", "grade", "pace"]} reduced = reduce_runs(rows, 20, catalog)[0] assert reduced.values["heart_rate_drift"] == pytest.approx(100 / 11) assert reduce_runs(rows[:39], 20, catalog) == [] for row in rows[20:]: row.distance_m *= 1.2 assert reduce_runs(rows, 20, catalog)[0].values["heart_rate_drift"] is None for row in rows: row.values["grade"] = None assert reduce_runs(rows, 20, catalog)[0].values["heart_rate_drift"] is None def test_daily_windows_include_rest_days_and_prior_history(app): with app.app_context(): db.session.query(Activity).delete() add_run("2026-09-01", 10000) add_run("2026-09-02", 5000) add_run("2026-09-03", 5000) db.session.commit() q = parsed("days\nduring 2026-09-02 to 2026-09-09\n" "summarize latest(rolling_sum(distance, 7 days)), latest(run_day_streak), latest(days_since_run(distance >= 10 km))\n" "by day\ncompare none") result = explore(1, "UTC", q) assert len(result["groups"]) == 8 assert [g["raw_values"] for g in result["groups"][:3]] == [[15, 2, 1], [20, 3, 2], [20, 0, 3]] assert result["groups"][-1]["raw_values"] == [5, 0, 8] assert result["cards"][0]["value"] == "5.0" assert result["rows"][0]["id"] is None assert result["coverage"]["sessions"] == 2 q = parsed(query_text(q).replace("distance >= 10 km", "distance >= 20 km")) assert explore(1, "UTC", q)["cards"][-1]["value"] == "—" def test_daily_timezone_and_latest_missing_value(app): with app.app_context(): db.session.query(Activity).delete() add_run("2026-09-01T23:30", 10000) db.session.add_all([DailyHealth(owner_id=1, day=date(2026, 9, 2), values={"resting_hr": 50}), DailyHealth(owner_id=1, day=date(2026, 9, 3), values={"sleep_hours": 8}), DailyHealth(owner_id=2, day=date(2026, 9, 3), values={"resting_hr": 999})]) db.session.commit() q = parsed("health\nduring 2026-09-02 to 2026-09-03\nsummarize latest(resting_hr), latest(sleep_hours)\nby day\ncompare none") result = explore(1, "Australia/Sydney", q) assert result["count"] == 2 assert [c["value"] for c in result["cards"]] == ["—", "8.0"] assert result["groups"][0]["raw_values"][0] == 50 assert result["coverage"]["sessions"] == 1 def test_actual_hr_band_seconds_are_not_window_average_classifications(app): with app.app_context(): db.session.query(Activity).delete() row = add_run("2026-09-02", samples=recording(60)) for i, sample in enumerate(row.samples): sample["heart_rate"] = 130 if i <= 30 else 150 persist_analysis(row) db.session.get(User, 1).hr_bands = [140, 160] db.session.commit() q = parsed("runs\nduring all time\nsummarize sum(lower_hr_minutes), sum(middle_hr_minutes), sum(upper_hr_minutes)\nby all\ncompare none") result = explore(1, "UTC", q) assert result["groups"][0]["raw_values"] == [0.5, 0.5, 0] db.session.get(User, 1).hr_bands = None db.session.commit() assert explore(1, "UTC", q)["groups"][0]["raw_values"] == [None, None, None] def test_efforts_and_performance_metrics_preserve_missing_distances(app): with app.app_context(): row = db.session.get(Activity, 1) row.best_efforts = [{"distance": 1000, "seconds": 301}, {"distance": 5000, "seconds": 1800}] db.session.commit() q = parsed("efforts\nduring all time\nwhere distance = 5 km\nsummarize min(effort_time)\nby all\ncompare none") result = explore(1, "UTC", q) assert result["cards"][0]["value"] == "30:00" assert result["rows"][0]["id"] == 1 q = parsed("runs\nduring all time\nsummarize min(best_time(5 km)), min(best_time(10 km))\nby month\ncompare none") assert [c["value"] for c in explore(1, "UTC", q)["cards"]] == ["30:00", "—"] @pytest.mark.parametrize("definition", signal_queries(route_id=1), ids=lambda d: d["name"]) def test_proposed_queries_roundtrip_execute_and_render(app, definition): q = parsed(definition["query"]) assert parsed(query_text(q)) == q with app.app_context(): explore(1, "Australia/Sydney", q) client = app.test_client() login(client) assert client.get("/stats", query_string={"query": definition["query"]}).status_code == 200 @pytest.mark.parametrize("source", ["runs", "days", "health", "efforts"]) def test_reduction_rejects_non_recording_sources(source): with pytest.raises(QueryError): parsed(source + "\nreduce by run min 3 min\nsummarize heart_rate") def test_queries_never_load_recordings_or_other_accounts(app): with app.app_context(): statements = [] def capture(_conn, _cursor, statement, _params, _context, _many): statements.append(statement) event.listen(db.engine, "before_cursor_execute", capture) try: for source in ("runs", "intervals", "efforts", "days", "health"): explore(1, "UTC", parsed(source + "\nduring last 16 weeks\nsummarize heart_rate\nby day")) finally: event.remove(db.engine, "before_cursor_execute", capture) assert not any("activity.samples" in sql or "activity.display_data" in sql for sql in statements) def test_route_matching_keeps_direction_and_rejects_breaks_and_other_owners(app): line = [[[-33, 151], [-33, 151.005], [-33, 151.01]]] assert similar(line, line) assert not similar(line, [list(reversed(line[0]))]) assert not similar(line, [line[0][:2], line[0][2:]]) with app.app_context(): db.session.get(Activity, 1).route_geometry = line other = Activity(owner_id=2, source_key="other-route", title="other", sport="running", started_at=datetime(2026, 9, 30, tzinfo=UTC), route_geometry=line) db.session.add(other) db.session.commit() q = parsed(f"runs\nduring all time\nroute similar to {other.id}\nsummarize pace") with pytest.raises(QueryError, match="own recorded"): explore(1, "UTC", q) def test_installation_is_idempotent_private_and_keeps_edits(app): with app.app_context(): user = db.session.get(User, 1) definitions = signal_queries() assert install(user, definitions) == 12 rows = db.session.scalars(db.select(SavedAnalysis).filter_by(owner_id=1)).all() assert sum(r.pinned for r in rows) == 5 rows[0].notes = "My annotation" db.session.commit() assert install(user, definitions) == 0 assert rows[0].notes == "My annotation" client = app.test_client() login(client, "other") assert client.get("/stats/analyses").json["analyses"] == [] def test_private_context_and_bands_validate_and_do_not_leak(app): client = app.test_client() login(client) assert client.post("/training/bands", data={"lower": 145, "upper": 140}).status_code == 400 assert client.post("/training/bands", data={"lower": 140, "upper": 160}).status_code == 302 assert client.post("/activities/1/context", data={"perceived_effort": 11}).status_code == 400 assert client.post("/activities/1/context", data={"perceived_effort": 7, "fatigue": 3, "notes": "PRIVATE-CONTEXT"}).status_code == 302 assert b"PRIVATE-CONTEXT" in client.get("/activities/1").data with app.app_context(): db.session.get(Activity, 1).share_token = "test-context-token" db.session.commit() assert b"PRIVATE-CONTEXT" not in app.test_client().get("/shared/runs/test-context-token").data other = app.test_client() login(other, "other") assert other.post("/activities/1/context", data={"notes": "attack"}).status_code == 404 app.config["WTF_CSRF_ENABLED"] = True assert client.post("/activities/1/context", data={}).status_code == 400 assert client.post("/training/bands", data={}).status_code == 400 def test_recovery_mapping_does_not_turn_provider_missing_sentinels_into_data(): assert normalize({"restingHeartRate": None, "averageStressLevel": -1}, {}, {}) == {} assert normalize({"restingHeartRate": 55, "averageStressLevel": 32}, {"hrvSummary": {"lastNightAvg": 48}}, {"dailySleepDTO": {"sleepTimeSeconds": 28800}}) == { "resting_hr": 55, "stress": 32, "hrv": 48, "sleep_hours": 8} def test_daily_import_is_bounded_owner_scoped_and_stops_on_failure(app): class API: calls = 0 fail = False def get_stats(self, _day): self.calls += 1 if self.fail: raise RuntimeError("PROVIDER-SECRET") return {"restingHeartRate": 50} def get_hrv_data(self, _day): self.calls += 1 return {"hrvSummary": {"lastNightAvg": 45}} def get_sleep_data(self, _day): self.calls += 1 return {"dailySleepDTO": {"sleepTimeSeconds": 28800}} with app.app_context(): api = API() user = db.session.get(User, 1) assert sync_health(api, user, days=30, limit=4)["days"] == 4 assert api.calls == 12 assert db.session.scalar(db.select(db.func.count(DailyHealth.id)).where(DailyHealth.owner_id == 1)) == 4 assert db.session.scalar(db.select(db.func.count(DailyHealth.id)).where(DailyHealth.owner_id == 2)) == 0 api.fail = True result = sync_health(api, user, days=30, limit=10) assert result["days"] == 1 and result["failures"] == 1 assert "PROVIDER-SECRET" not in result["message"]