"""Unit tests for the trends query and the rule-based recommendation engine.""" import pytest from services import analysis as analysis_svc def ids(recs): return {r["id"] for r in recs} def days(n, **metrics): """n consecutive days that all carry the same metric values.""" return [ {"date": f"2026-08-{d:02d}", **metrics} for d in range(1, n + 1) ] # --- trends ----------------------------------------------------------------- class TestTrends: def test_empty_without_data(self, db, user): assert analysis_svc.get_trends("steps", user["id"]) == [] def test_returns_date_value_pairs(self, seed_health, user): seed_health([{"date": "2026-08-20", "steps": 5000}]) assert analysis_svc.get_trends("steps", user["id"]) == [ {"date": "2026-08-20", "value": 5000} ] @pytest.mark.parametrize( "metric,column_value", [ ("steps", {"steps": 5000}), ("heart_rate", {"heart_rate": 60}), ("sleep_duration", {"sleep_duration": 7}), ("stress", {"stress": 30}), ("calories_burned", {"calories": 250}), ], ) def test_each_supported_metric(self, seed_health, user, metric, column_value): seed_health([{"date": "2026-08-20", **column_value}]) rows = analysis_svc.get_trends(metric, user["id"]) assert len(rows) == 1 and rows[0]["value"] is not None def test_unknown_metric_falls_back_to_steps(self, seed_health, user): seed_health([{"date": "2026-08-20", "steps": 5000}]) assert analysis_svc.get_trends("bogus", user["id"]) == [ {"date": "2026-08-20", "value": 5000} ] def test_unknown_metric_cannot_inject_sql(self, seed_health, user): """The metric name indexes a whitelist; it is never interpolated raw.""" seed_health([{"date": "2026-08-20", "steps": 5000}]) rows = analysis_svc.get_trends( "steps FROM health_data; DROP TABLE health_data --", user["id"] ) assert rows == [{"date": "2026-08-20", "value": 5000}] def test_nulls_excluded(self, seed_health, user): seed_health([ {"date": "2026-08-20", "steps": 5000}, {"date": "2026-08-21"}, ]) assert len(analysis_svc.get_trends("steps", user["id"])) == 1 def test_date_range_filter(self, seed_health, user): seed_health([ {"date": "2026-08-20", "steps": 1}, {"date": "2026-08-21", "steps": 2}, {"date": "2026-08-22", "steps": 3}, ]) rows = analysis_svc.get_trends( "steps", user["id"], "2026-08-21", "2026-08-21" ) assert rows == [{"date": "2026-08-21", "value": 2}] # --- recommendations -------------------------------------------------------- class TestNoData: def test_returns_a_single_guidance_item(self, db, user): recs = analysis_svc.get_recommendations(user["id"]) assert len(recs) == 1 assert recs[0]["id"] == "no-data" assert recs[0]["priority"] == "low" class TestStepsRule: def test_fires_below_8000(self, seed_health, user): seed_health(days(5, steps=6000)) assert "steps" in ids(analysis_svc.get_recommendations(user["id"])) def test_silent_at_or_above_8000(self, seed_health, user): seed_health(days(5, steps=8000)) assert "steps" not in ids(analysis_svc.get_recommendations(user["id"])) def test_priority_is_medium(self, seed_health, user): seed_health(days(5, steps=6000)) rec = next( r for r in analysis_svc.get_recommendations(user["id"]) if r["id"] == "steps" ) assert rec["priority"] == "medium" assert rec["basedOn"] == ["steps"] def test_averages_across_days_not_per_day(self, seed_health, user): # 4000 and 12000 average to 8000 -> rule must NOT fire. seed_health([ {"date": "2026-08-01", "steps": 4000}, {"date": "2026-08-02", "steps": 12000}, ]) assert "steps" not in ids(analysis_svc.get_recommendations(user["id"])) class TestSleepRule: def test_fires_below_7_hours(self, seed_health, user): seed_health(days(5, sleep_duration=6)) assert "sleep" in ids(analysis_svc.get_recommendations(user["id"])) def test_silent_at_7_hours(self, seed_health, user): seed_health(days(5, sleep_duration=7)) assert "sleep" not in ids(analysis_svc.get_recommendations(user["id"])) def test_priority_is_high(self, seed_health, user): seed_health(days(5, sleep_duration=5)) rec = next( r for r in analysis_svc.get_recommendations(user["id"]) if r["id"] == "sleep" ) assert rec["priority"] == "high" def test_days_without_sleep_do_not_drag_the_average_down(self, seed_health, user): """Nights with no sleep record must be excluded, not counted as zero.""" seed_health([ {"date": "2026-08-01", "sleep_duration": 8}, {"date": "2026-08-02", "steps": 5000}, # no sleep recorded ]) assert "sleep" not in ids(analysis_svc.get_recommendations(user["id"])) class TestStressRule: def test_fires_above_50(self, seed_health, user): seed_health(days(5, stress=60)) assert "stress" in ids(analysis_svc.get_recommendations(user["id"])) def test_silent_at_50(self, seed_health, user): seed_health(days(5, stress=50)) assert "stress" not in ids(analysis_svc.get_recommendations(user["id"])) class TestRestingHeartRateRule: def test_fires_above_65(self, seed_health, user): seed_health(days(5, heart_rate=70)) assert "rhr" in ids(analysis_svc.get_recommendations(user["id"])) def test_silent_at_65(self, seed_health, user): seed_health(days(5, heart_rate=65)) assert "rhr" not in ids(analysis_svc.get_recommendations(user["id"])) class TestHrvRule: def test_fires_below_40(self, seed_health, user): seed_health(days(5, hrv=30)) assert "hrv" in ids(analysis_svc.get_recommendations(user["id"])) def test_silent_at_40(self, seed_health, user): seed_health(days(5, hrv=40)) assert "hrv" not in ids(analysis_svc.get_recommendations(user["id"])) class TestHealthyUser: def test_all_good_returns_the_positive_message(self, seed_health, user): seed_health(days(5, steps=10000, sleep_duration=8, stress=30, heart_rate=55, hrv=60)) recs = analysis_svc.get_recommendations(user["id"]) assert ids(recs) == {"good"} class TestOrderingAndWindow: def test_sorted_high_medium_low(self, seed_health, user): seed_health(days(5, steps=6000, sleep_duration=5, hrv=30)) order = {"high": 0, "medium": 1, "low": 2} priorities = [r["priority"] for r in analysis_svc.get_recommendations(user["id"])] assert priorities == sorted(priorities, key=lambda p: order[p]) assert priorities[0] == "high" def test_only_the_last_14_days_count(self, seed_health, user): """20 lazy days then 14 active ones: the old days must not pull it down.""" old = [{"date": f"2026-07-{d:02d}", "steps": 1000} for d in range(1, 21)] recent = [{"date": f"2026-08-{d:02d}", "steps": 12000} for d in range(1, 15)] seed_health(old + recent) assert "steps" not in ids(analysis_svc.get_recommendations(user["id"])) def test_multiple_rules_can_fire_together(self, seed_health, user): seed_health(days(5, steps=5000, sleep_duration=5, stress=70, heart_rate=75)) assert {"steps", "sleep", "stress", "rhr"} <= ids( analysis_svc.get_recommendations(user["id"]) ) class TestEndpoints: def test_trends_requires_auth(self, client): assert client.get("/api/analysis/trends").status_code == 401 def test_recommendations_requires_auth(self, client): assert client.get("/api/analysis/recommendations").status_code == 401 def test_trends_endpoint(self, client, auth, seed_health): seed_health(days(3, steps=5000)) r = client.get("/api/analysis/trends?metricType=steps", headers=auth) assert r.status_code == 200 and len(r.get_json()) == 3 def test_recommendations_endpoint(self, client, auth, seed_health): seed_health(days(3, steps=5000)) r = client.get("/api/analysis/recommendations", headers=auth) assert r.status_code == 200 assert "steps" in {x["id"] for x in r.get_json()}