#!/usr/bin/env python3 import sys import tempfile import subprocess import unittest from copy import deepcopy from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "tools")) import cardio_guidance_exchange as guidance from trainlog_sqlite import connect_database RUN = "cgr_11111111-1111-4111-8111-111111111111" PHASE = "cgp_22222222-2222-4222-8222-222222222222" SESSION = "se_33333333-3333-4333-8333-333333333333" ENTRY = "sxe_44444444-4444-4444-8444-444444444444" EXERCISE = "ex_55555555-5555-4555-8555-555555555555" CAPTURE = "hrc_66666666-6666-4666-8666-666666666666" CAL = "cal_77777777-7777-4777-8777-777777777777" def document(calibrated: bool = False) -> dict: target = { "minimum_bpm": 120, "maximum_bpm": 140, "calibration_id": CAL if calibrated else None, "calibration_observed_peak_bpm": 160 if calibrated else None, "minimum_percent": 75 if calibrated else None, "maximum_percent": 88 if calibrated else None, } return { "format": "trainlog-cardio-guidance", "version": 1, "generated_at": "2026-09-23T13:11:00+02:00", "runs": [ { "run_id": RUN, "session_id": SESSION, "started_at": "2026-09-23T13:00:20+02:00", "ended_at": "2026-09-23T13:03:20+02:00", "phases": [ { "phase_id": PHASE, "entry_id": ENTRY, "position": 0, "kind": "work", "target": target, "exit_condition": { "kind": "fixed_duration", "seconds": 180, "bpm": None, }, "started_at": "2026-09-23T13:00:20+02:00", "ended_at": "2026-09-23T13:03:20+02:00", "final_instruction": "maintain", "events": [ { "sequence": 0, "observed_at": "2026-09-23T13:00:23+02:00", "instruction": "accelerate", "bpm": 110, "target_minimum_bpm": 120, "target_maximum_bpm": 140, }, { "sequence": 1, "observed_at": "2026-09-23T13:00:30+02:00", "instruction": "suspended", "bpm": None, "target_minimum_bpm": 120, "target_maximum_bpm": 140, }, { "sequence": 2, "observed_at": "2026-09-23T13:00:40+02:00", "instruction": "maintain", "bpm": 130, "target_minimum_bpm": 120, "target_maximum_bpm": 140, }, ], } ], } ], } class CardioGuidanceExchangeTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.db_path = Path(self.temp.name) / "trainlog.sqlite" fixture = ROOT / "build/tui/sync-generation-fixture" subprocess.run([str(fixture), str(self.db_path)], check=True, capture_output=True) self.db = connect_database(self.db_path) self.db.execute( "INSERT INTO exercises(exercise_id,name,normalized_name,recording_mode,tracking_mode,data_fields) " "VALUES(?,?,?,?,?,?)", (EXERCISE, "Tapis guidé", "tapis guide", "continuous", "duration", 0), ) exercise_row = self.db.execute( "SELECT id FROM exercises WHERE exercise_id=?", (EXERCISE,) ).fetchone()[0] self.db.execute( "INSERT INTO sessions(session_id,started_at,ended_at,session_type,session_kind) " "VALUES(?,?,?,?,?)", ( SESSION, "2026-09-23T13:00:00+02:00", "2026-09-23T13:10:10+02:00", "training", "cardio", ), ) session_row = self.db.execute( "SELECT id FROM sessions WHERE session_id=?", (SESSION,) ).fetchone()[0] self.db.execute( "INSERT INTO session_exercises(" "entry_id,session_row_id,exercise_row_id,recording_mode,tracking_mode," "data_fields,position,load_mode,rest_seconds) VALUES(?,?,?,?,?,?,?,?,?)", (ENTRY, session_row, exercise_row, "continuous", "duration", 0, 0, "none", 0), ) self.db.execute( "INSERT INTO session_exercise_timeline(" "session_id,entry_id,exercise_id,started_at,ended_at,imported_at" ") VALUES(?,?,?,?,?,?)", ( SESSION, ENTRY, EXERCISE, "2026-09-23T13:00:10+02:00", "2026-09-23T13:10:10+02:00", "2026-09-23T13:11:00+02:00", ), ) self.db.execute( "INSERT INTO heart_rate_captures(" "capture_id,context_kind,context_id,started_at,ended_at,sensor_name,imported_at" ") VALUES(?,?,?,?,?,?,?)", ( CAPTURE, "cardio", SESSION, "2026-09-23T13:00:00+02:00", "2026-09-23T13:10:10+02:00", "CYCPLUS H2", "2026-09-23T13:11:00+02:00", ), ) self.db.executemany( "INSERT INTO heart_rate_samples(" "capture_id,sequence,observed_at,bpm,exercise_entry_id," "sensor_contact_detected,energy_expended) VALUES(?,?,?,?,?,?,?)", [ (CAPTURE, 0, "2026-09-23T13:00:23+02:00", 110, ENTRY, None, None), (CAPTURE, 1, "2026-09-23T13:00:40+02:00", 130, ENTRY, None, None), ], ) self.db.execute( "INSERT INTO cardio_calibrations(" "calibration_id,protocol_version,session_id,entry_id,started_at,effort_end_at," "ended_at,heart_rate_capture_id,observed_peak_bpm,imported_at" ") VALUES(?,?,?,?,?,?,?,?,?,?)", ( CAL, 1, SESSION, ENTRY, "2026-09-23T13:00:00+02:00", "2026-09-23T13:02:00+02:00", "2026-09-23T13:03:00+02:00", CAPTURE, 160, "2026-09-23T13:11:00+02:00", ), ) self.db.commit() def tearDown(self): self.db.close() self.temp.cleanup() def test_import_and_exact_replay_preserve_snapshot_and_events(self): payload = guidance.validate(document()) self.assertEqual((1, 0), guidance.apply(self.db, payload)) self.assertEqual((0, 1), guidance.apply(self.db, payload)) self.assertEqual( (SESSION, "2026-09-23T13:00:20+02:00", "2026-09-23T13:03:20+02:00"), tuple( self.db.execute( "SELECT session_id,started_at,ended_at FROM cardio_guidance_runs WHERE run_id=?", (RUN,), ).fetchone() ), ) self.assertEqual( [(0, "accelerate", 110), (1, "suspended", None), (2, "maintain", 130)], [ tuple(row) for row in self.db.execute( "SELECT sequence,instruction,bpm FROM cardio_guidance_events " "WHERE run_id=? AND phase_id=? ORDER BY sequence", (RUN, PHASE), ) ], ) def test_calibration_provenance_must_match_imported_profile(self): payload = guidance.validate(document(calibrated=True)) self.assertEqual((1, 0), guidance.apply(self.db, payload)) changed = deepcopy(document(calibrated=True)) changed["runs"][0]["phases"][0]["target"]["calibration_observed_peak_bpm"] = 159 with self.assertRaisesRegex(ValueError, "calibration provenance"): guidance.apply(self.db, guidance.validate(changed)) def test_guidance_bpm_must_exist_in_raw_capture(self): changed = deepcopy(document()) changed["runs"][0]["phases"][0]["events"][0]["bpm"] = 109 with self.assertRaisesRegex(ValueError, "not a raw heart-rate sample"): guidance.apply(self.db, guidance.validate(changed)) def test_identity_reuse_with_different_event_is_rejected(self): payload = guidance.validate(document()) guidance.apply(self.db, payload) changed = deepcopy(document()) changed["runs"][0]["phases"][0]["final_instruction"] = "slow_down" with self.assertRaisesRegex(ValueError, "different content"): guidance.apply(self.db, guidance.validate(changed)) if __name__ == "__main__": unittest.main()