232 lines
13 KiB
Python
232 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""End-to-end V2 import regression: Walk, other movement, Walk again."""
|
|
import json
|
|
import copy
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
IMPORTER = ROOT / "tools/import_mobile_export.py"
|
|
EQUIPMENT_IMPORTER = ROOT / "tools/import_equipment_associations.py"
|
|
DEFINITIONS_IMPORTER = ROOT / "tools/import_equipment_definitions.py"
|
|
MOBILE_EXPORTER = ROOT / "tools/export_pc_mobile.py"
|
|
EQUIPMENT_EXPORTER = ROOT / "tools/export_equipment_associations.py"
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE exercises(id INTEGER PRIMARY KEY,exercise_id TEXT UNIQUE,name TEXT,normalized_name TEXT UNIQUE,tracking_mode TEXT,recording_mode TEXT,data_fields INTEGER);
|
|
CREATE TABLE sessions(id INTEGER PRIMARY KEY,session_id TEXT UNIQUE,started_at TEXT,ended_at TEXT,session_type TEXT,notes TEXT);
|
|
CREATE TABLE session_exercises(id INTEGER PRIMARY KEY,entry_id TEXT NOT NULL UNIQUE,session_row_id INTEGER,exercise_row_id INTEGER,recording_mode TEXT,data_fields INTEGER,position INTEGER,load_mode TEXT,rest_seconds INTEGER,target_sets INTEGER,target_reps INTEGER,target_duration_seconds INTEGER,target_weight_kg REAL,equipment_id TEXT,notes TEXT,UNIQUE(session_row_id,position));
|
|
CREATE TABLE performed_sets(id INTEGER PRIMARY KEY,session_exercise_row_id INTEGER,position INTEGER,reps INTEGER,duration_seconds INTEGER,weight_kg REAL);
|
|
CREATE TABLE continuous_activity(id INTEGER PRIMARY KEY,session_exercise_row_id INTEGER UNIQUE,duration_seconds INTEGER,speed_kmh REAL,distance_km REAL);
|
|
CREATE TABLE max_results(session_exercise_row_id INTEGER PRIMARY KEY,max_weight_kg REAL NOT NULL);
|
|
CREATE TABLE body_observations(id INTEGER PRIMARY KEY,observation_id TEXT UNIQUE,observed_at TEXT,session_row_id INTEGER,body_weight_kg REAL,neck_cm REAL,shoulders_cm REAL,chest_cm REAL,waist_cm REAL,hips_cm REAL,left_arm_cm REAL,right_arm_cm REAL,left_forearm_cm REAL,right_forearm_cm REAL,left_thigh_cm REAL,right_thigh_cm REAL,left_calf_cm REAL,right_calf_cm REAL,notes TEXT);
|
|
CREATE TABLE custom_equipment(equipment_id TEXT PRIMARY KEY,display_name TEXT NOT NULL,label_name TEXT NOT NULL,equipment_type TEXT NOT NULL,load_semantics TEXT NOT NULL);
|
|
PRAGMA user_version=9;
|
|
"""
|
|
|
|
def payload():
|
|
catalog = [
|
|
{"exercise_id":"ex_walk","name":"Marche","recording_mode":"continuous","tracking_mode":"duration","data_fields":0},
|
|
{"exercise_id":"ex_push","name":"Pompes","recording_mode":"sets","tracking_mode":"reps","data_fields":0},
|
|
]
|
|
entries = [
|
|
{"entry_id":"sxe_walk_10","position":0,"exercise_id":"ex_walk","name":"Marche","recording_mode":"continuous","tracking_mode":"duration","data_fields":0,"load_mode":"none","rest_seconds":0,"equipment_id":None,"continuous":{"duration_seconds":600}},
|
|
{"entry_id":"sxe_push","position":1,"exercise_id":"ex_push","name":"Pompes","recording_mode":"sets","tracking_mode":"reps","data_fields":0,"load_mode":"none","rest_seconds":0,"equipment_id":"leg_press","sets":[{"reps":12,"weight_kg":20.0}]},
|
|
{"entry_id":"sxe_walk_15","position":2,"exercise_id":"ex_walk","name":"Marche","recording_mode":"continuous","tracking_mode":"duration","data_fields":0,"load_mode":"none","rest_seconds":0,"equipment_id":None,"continuous":{"duration_seconds":900}},
|
|
]
|
|
return {"format":"trainlog-mobile-export","version":2,"generated_at":"2026-09-07T10:00:00+02:00","exercises":catalog,"sessions":[{"session_id":"se_multi","started_at":"2026-09-07T10:00:00+02:00","session_type":"training","exercises":entries}],"body_observations":[{"observation_id":"bo_fixture","observed_at":"2026-03-01T08:00:00+01:00","body_weight_kg":72.5}]}
|
|
|
|
def invoke(path, db):
|
|
result = subprocess.run([sys.executable, str(IMPORTER), str(path), "--database", str(db)], text=True, capture_output=True)
|
|
if result.returncode: raise AssertionError(result.stdout + result.stderr)
|
|
return result.stdout
|
|
|
|
|
|
def invoke_equipment(path, db):
|
|
result = subprocess.run(
|
|
[sys.executable, str(EQUIPMENT_IMPORTER), str(path), "--database", str(db)],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
if result.returncode:
|
|
raise AssertionError(result.stdout + result.stderr)
|
|
return result.stdout
|
|
|
|
|
|
def invoke_definitions(path, db):
|
|
result = subprocess.run(
|
|
[sys.executable, str(DEFINITIONS_IMPORTER), str(path), "--database", str(db)],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
if result.returncode:
|
|
raise AssertionError(result.stdout + result.stderr)
|
|
return result.stdout
|
|
|
|
def main():
|
|
with tempfile.TemporaryDirectory(prefix="trainlog-multi-") as temp:
|
|
root = Path(temp); db = root / "db.sqlite"; artifact = root / "mobile-v2.json"
|
|
equipment = root / "equipment-v2.json"
|
|
con = sqlite3.connect(db); con.executescript(SCHEMA); con.close()
|
|
|
|
# Mobile v2 cannot transport custom definitions. Reject the complete
|
|
# v8 input before catalog/session mutation, with occurrence context.
|
|
unknown_db = root / "unknown.db"
|
|
con = sqlite3.connect(unknown_db); con.executescript(SCHEMA); con.close()
|
|
unknown_artifact = root / "unknown-mobile-v2.json"
|
|
unknown_payload = copy.deepcopy(payload())
|
|
unknown_payload["sessions"][0]["exercises"][1]["equipment_id"] = "custom_rack"
|
|
unknown_artifact.write_text(json.dumps(unknown_payload), encoding="utf-8")
|
|
rejected = subprocess.run(
|
|
[sys.executable, str(IMPORTER), str(unknown_artifact), "--database", str(unknown_db)],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
assert rejected.returncode != 0
|
|
assert "session_id=se_multi" in rejected.stderr
|
|
assert "entry_id=sxe_push" in rejected.stderr
|
|
assert "equipment_id=custom_rack" in rejected.stderr
|
|
con = sqlite3.connect(unknown_db)
|
|
assert con.execute("SELECT count(*) FROM sessions").fetchone()[0] == 0
|
|
assert con.execute("SELECT count(*) FROM exercises").fetchone()[0] == 0
|
|
con.close()
|
|
|
|
# Android -> PC ordering regression: the companion definition is
|
|
# imported through its real importer before the V2 snapshot validates
|
|
# the entry reference. A replay retains both stable identities and
|
|
# never creates a duplicate custom machine or session occurrence.
|
|
defined_db = root / "defined.db"
|
|
con = sqlite3.connect(defined_db); con.executescript(SCHEMA); con.close()
|
|
definitions = root / "mobile-definitions-v1.json"
|
|
definitions.write_text(json.dumps({
|
|
"format": "trainlog-equipment-definitions",
|
|
"version": 1,
|
|
"generated_at": "2026-09-07T10:00:00+02:00",
|
|
"equipment": [{
|
|
"equipment_id": "custom_rack",
|
|
"display_name": "Rack Android",
|
|
"label_name": "Rack",
|
|
"equipment_type": "custom_machine",
|
|
"load_semantics": "external",
|
|
}],
|
|
}), encoding="utf-8")
|
|
assert "definitions_imported=1" in invoke_definitions(definitions, defined_db)
|
|
defined_payload = copy.deepcopy(payload())
|
|
defined_payload["sessions"][0]["exercises"][1]["equipment_id"] = "custom_rack"
|
|
defined_artifact = root / "defined-mobile-v2.json"
|
|
defined_artifact.write_text(json.dumps(defined_payload), encoding="utf-8")
|
|
assert "sessions_imported=1" in invoke(defined_artifact, defined_db)
|
|
con = sqlite3.connect(defined_db)
|
|
assert con.execute(
|
|
"SELECT display_name,label_name,equipment_type,load_semantics "
|
|
"FROM custom_equipment WHERE equipment_id='custom_rack'"
|
|
).fetchone() == ("Rack Android", "Rack", "custom_machine", "external")
|
|
assert con.execute(
|
|
"SELECT equipment_id FROM session_exercises WHERE entry_id='sxe_push'"
|
|
).fetchone() == ("custom_rack",)
|
|
con.close()
|
|
assert "definitions_skipped=1" in invoke_definitions(definitions, defined_db)
|
|
assert "sessions_skipped=1" in invoke(defined_artifact, defined_db)
|
|
|
|
artifact.write_text(json.dumps(payload()), encoding="utf-8")
|
|
assert "sessions_imported=1" in invoke(artifact, db)
|
|
con = sqlite3.connect(db)
|
|
rows = con.execute("SELECT e.exercise_id,se.entry_id,se.position,ca.duration_seconds FROM session_exercises se JOIN exercises e ON e.id=se.exercise_row_id LEFT JOIN continuous_activity ca ON ca.session_exercise_row_id=se.id ORDER BY se.position").fetchall()
|
|
assert rows == [("ex_walk","sxe_walk_10",0,600),("ex_push","sxe_push",1,None),("ex_walk","sxe_walk_15",2,900)], rows
|
|
assert con.execute("SELECT weight_kg FROM performed_sets").fetchone()[0] == 20.0
|
|
assert "sessions_skipped=1" in invoke(artifact, db)
|
|
con.close()
|
|
|
|
divergent = copy.deepcopy(payload())
|
|
divergent["sessions"][0]["exercises"][1]["sets"][0]["reps"] = 13
|
|
artifact.write_text(json.dumps(divergent), encoding="utf-8")
|
|
rejected = subprocess.run([sys.executable, str(IMPORTER), str(artifact), "--database", str(db)], text=True, capture_output=True)
|
|
assert rejected.returncode != 0 and "conflit de contenu" in rejected.stderr
|
|
assert sqlite3.connect(db).execute("SELECT reps FROM performed_sets").fetchone()[0] == 12
|
|
divergent = copy.deepcopy(payload())
|
|
divergent["body_observations"][0]["body_weight_kg"] = 73.0
|
|
artifact.write_text(json.dumps(divergent), encoding="utf-8")
|
|
rejected = subprocess.run([sys.executable, str(IMPORTER), str(artifact), "--database", str(db)], text=True, capture_output=True)
|
|
assert rejected.returncode != 0 and "conflit observation corporelle" in rejected.stderr
|
|
artifact.write_text(json.dumps(payload()), encoding="utf-8")
|
|
|
|
equipment.write_text(json.dumps({
|
|
"format": "trainlog-equipment-associations",
|
|
"version": 2,
|
|
"generated_at": "2026-09-07T10:00:00+02:00",
|
|
"associations": [
|
|
{"session_id": "se_multi", "entry_id": "sxe_walk_10", "exercise_id": "ex_walk", "state": "cleared"},
|
|
{"session_id": "se_multi", "entry_id": "sxe_push", "exercise_id": "ex_push", "state": "set", "equipment_id": "leg_press"},
|
|
{"session_id": "se_multi", "entry_id": "sxe_walk_15", "exercise_id": "ex_walk", "state": "cleared"},
|
|
],
|
|
}), encoding="utf-8")
|
|
|
|
# Regression for the real failure: the historic TUI invocation passed
|
|
# only the artifact, while the importer requires the target database.
|
|
rejected = subprocess.run(
|
|
[sys.executable, str(EQUIPMENT_IMPORTER), str(equipment)],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
assert rejected.returncode != 0
|
|
assert "the following arguments are required: --database" in rejected.stderr
|
|
|
|
assert "EQUIPMENT_ASSOCIATIONS_IMPORT=PASS" in invoke_equipment(equipment, db)
|
|
con = sqlite3.connect(db)
|
|
rows = con.execute(
|
|
"SELECT entry_id,equipment_id FROM session_exercises ORDER BY position"
|
|
).fetchall()
|
|
assert rows == [
|
|
("sxe_walk_10", None),
|
|
("sxe_push", "leg_press"),
|
|
("sxe_walk_15", None),
|
|
], rows
|
|
con.close()
|
|
|
|
assert "EQUIPMENT_ASSOCIATIONS_IMPORT=PASS" in invoke_equipment(equipment, db)
|
|
con = sqlite3.connect(db)
|
|
assert con.execute("SELECT count(*) FROM session_exercises").fetchone()[0] == 3
|
|
assert con.execute(
|
|
"SELECT entry_id,equipment_id FROM session_exercises ORDER BY position"
|
|
).fetchall() == rows
|
|
con.close()
|
|
|
|
mobile_output = root / "pc-mobile-v2.json"
|
|
equipment_output = root / "pc-equipment-v2.json"
|
|
for tool, output, marker in (
|
|
(MOBILE_EXPORTER, mobile_output, "PC_MOBILE_EXPORT=PASS"),
|
|
(EQUIPMENT_EXPORTER, equipment_output, "EQUIPMENT_ASSOCIATIONS_EXPORT=PASS"),
|
|
):
|
|
result = subprocess.run(
|
|
[sys.executable, str(tool), str(output), "--database", str(db)],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
assert marker in result.stdout
|
|
assert json.loads(mobile_output.read_text())["body_observations"][0]["body_weight_kg"] == 72.5
|
|
# The desktop serializer is also accepted as an idempotent V2 replay,
|
|
# which is the PC endpoint of the Android importer/exporter round trip.
|
|
assert "body_skipped=1" in invoke(mobile_output, db)
|
|
|
|
# Definitions travel in their companion; unchanged v2 references may
|
|
# therefore contain a locally-known custom stable ID.
|
|
con = sqlite3.connect(db)
|
|
con.execute("INSERT INTO custom_equipment VALUES('custom_rack','Rack','Rack','rack','none')")
|
|
con.execute("UPDATE session_exercises SET equipment_id='custom_rack' WHERE entry_id='sxe_push'")
|
|
con.commit()
|
|
con.close()
|
|
custom_output = root / "custom-mobile-v2.json"
|
|
exported = subprocess.run(
|
|
[sys.executable, str(MOBILE_EXPORTER), str(custom_output), "--database", str(db)],
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
assert exported.returncode == 0, exported.stdout + exported.stderr
|
|
assert json.loads(custom_output.read_text())["sessions"][0]["exercises"][1]["equipment_id"] == "custom_rack"
|
|
print("PASS mobile_import_multi_occurrence")
|
|
|
|
if __name__ == "__main__": main()
|