[workouts] Add Flexify workout importer (96c9fbf5)

tests/workouts_tests/test_flexify_importer.py

@@ -0,0 +1,307 @@
+import os
+
+import pytest
+from django.contrib.auth import get_user_model
+from scrobbles.importers.webdav import scan_webdav_for_flexify
+from scrobbles.models import Scrobble
+from workouts.importer import (
+    FLEXIFY_EXERCISE_ALIASES,
+    aggregate_bodyweight,
+    aggregate_workouts,
+    cluster_sessions,
+    import_flexify,
+    read_flexify_sqlite,
+    read_plans_csv,
+    resolve_exercise,
+)
+from workouts.models import Exercise, WorkoutRoutine, WorkoutRoutineExercise
+
+User = get_user_model()
+
+SAMPLE_SQLITE = os.path.join(
+    os.path.dirname(__file__), "..", "..", "data", "flexify-example.sqlite"
+)
+SAMPLE_CSV = os.path.join(
+    os.path.dirname(__file__), "..", "..", "data", "flexify-plans-example.csv"
+)
+
+
+@pytest.fixture
+def user(db):
+    return User.objects.create(email="lifter@example.com")
+
+
+def test_read_plans_csv():
+    plans = read_plans_csv(SAMPLE_CSV)
+    assert len(plans) == 3
+    assert plans[0]["days"] == "Monday"
+    assert plans[0]["title"] == "Middle Distance - Push Day"
+    assert plans[0]["exercises"] == [
+        "Barbell bench press",
+        "Squat",
+        "Lat pull-down",
+        "Leg press",
+    ]
+
+
+def test_read_flexify_sqlite():
+    data = read_flexify_sqlite(SAMPLE_SQLITE)
+    assert len(data["plans"]) == 3
+    assert len(data["plan_exercises"]) == 12
+    assert len(data["catalog"]) == 58
+    assert data["strength_unit"] == "lb"
+    assert len(data["sets"]) == 12
+    first = data["sets"][0]
+    assert first["name"] == "Leg press"
+    assert first["unit"] == "lb"
+    assert first["plan_id"] == 3
+
+
+def test_resolve_exercise_alias(db):
+    exercise, created = resolve_exercise("Barbell bench press")
+    assert exercise.name == "Barbell Bench Press - Medium Grip"
+    assert created is True
+
+
+def test_resolve_exercise_exact_name(db):
+    exercise, created = resolve_exercise("Leg press")
+    assert exercise.name == "Leg Press"
+    assert created is True
+
+
+def test_resolve_exercise_dry_run_does_not_create(db):
+    before = Exercise.objects.count()
+    exercise, created = resolve_exercise("Some Unknown Exercise", dry_run=True)
+    assert exercise is None
+    assert created is True
+    assert Exercise.objects.count() == before
+
+
+def test_resolve_exercise_reuses_existing(db):
+    first, _ = resolve_exercise("Push-up")
+    second, created = resolve_exercise("Push-up")
+    assert second.id == first.id
+    assert created is False
+
+
+def test_cluster_sessions_splits_on_time_gap():
+    sets = [
+        {"name": "A", "created": 1000, "id": 1, "plan_id": 1},
+        {"name": "B", "created": 1010, "id": 2, "plan_id": 1},
+        {"name": "C", "created": 1000 + 61 * 60, "id": 3, "plan_id": 1},
+        {"name": "D", "created": 1001 + 61 * 60, "id": 4, "plan_id": 1},
+    ]
+    sessions = cluster_sessions(sets, gap_minutes=60)
+    assert len(sessions) == 2
+    assert [s["name"] for s in sessions[0]] == ["A", "B"]
+    assert [s["name"] for s in sessions[1]] == ["C", "D"]
+
+
+def test_cluster_sessions_splits_on_plan_change():
+    sets = [
+        {"name": "A", "created": 1000, "id": 1, "plan_id": 1},
+        {"name": "B", "created": 1005, "id": 2, "plan_id": 2},
+    ]
+    sessions = cluster_sessions(sets, gap_minutes=60)
+    assert len(sessions) == 2
+    assert [s["name"] for s in sessions[0]] == ["A"]
+    assert [s["name"] for s in sessions[1]] == ["B"]
+
+
+def test_aggregate_workouts_weights(db):
+    session = read_flexify_sqlite(SAMPLE_SQLITE)["sets"][:3]
+    workouts = aggregate_workouts(session)
+    assert len(workouts) == 1
+    entry = workouts[0]
+    assert entry["sets"] == 3
+    assert entry["reps"] == 10
+    assert entry["weight_kg"] == pytest.approx(54.43, abs=0.01)
+
+
+def test_aggregate_bodyweight():
+    session = read_flexify_sqlite(SAMPLE_SQLITE)["sets"]
+    assert aggregate_bodyweight(session) == pytest.approx(83.91, abs=0.01)
+
+
+def test_import_flexify_dry_run(user):
+    result = import_flexify(
+        sqlite_path=SAMPLE_SQLITE,
+        csv_path=SAMPLE_CSV,
+        user_id=user.id,
+        dry_run=True,
+    )
+    assert result["routines_created"] == 3
+    assert result["routines_updated"] == 0
+    assert result["scrobbles_created"] == 1
+    assert Scrobble.objects.count() == 0
+    assert WorkoutRoutine.objects.count() == 0
+
+
+def test_import_flexify_full(user):
+    result = import_flexify(
+        sqlite_path=SAMPLE_SQLITE,
+        csv_path=SAMPLE_CSV,
+        user_id=user.id,
+    )
+    assert result["routines_created"] == 3
+    assert result["scrobbles_created"] == 1
+    assert result["scrobbles_skipped"] == 0
+
+    scrobble = Scrobble.objects.get(user=user)
+    assert scrobble.source == "Flexify Import"
+    assert scrobble.played_to_completion is True
+    assert scrobble.workout_routine.title == "Middle Distance - Arm Day"
+
+    entry = scrobble.logdata.workouts[0]
+    assert entry["exercise_id"] is not None
+    assert entry["sets"] == 3
+    assert entry["reps"] == 10
+    assert entry["weight_kg"] == pytest.approx(54.43, abs=0.01)
+    assert scrobble.logdata.bodyweight_kg == pytest.approx(83.91, abs=0.01)
+    assert scrobble.logdata.duration_minutes == 2
+
+    planned = WorkoutRoutineExercise.objects.filter(
+        routine=scrobble.workout_routine
+    ).order_by("sequence")
+    assert [pe.exercise.name for pe in planned] == [
+        "Leg Press",
+        "Pullups",
+        "Pushups",
+        "Crunches",
+    ]
+    assert result["exercises_created"] == 11
+
+
+def test_import_flexify_idempotent(user):
+    import_flexify(sqlite_path=SAMPLE_SQLITE, csv_path=SAMPLE_CSV, user_id=user.id)
+    result = import_flexify(
+        sqlite_path=SAMPLE_SQLITE,
+        csv_path=SAMPLE_CSV,
+        user_id=user.id,
+    )
+    assert result["routines_created"] == 0
+    assert result["routines_updated"] == 3
+    assert result["scrobbles_created"] == 0
+    assert result["scrobbles_existing"] == 1
+    assert Scrobble.objects.filter(user=user).count() == 1
+
+
+def test_import_flexify_csv_only(user):
+    result = import_flexify(
+        csv_path=SAMPLE_CSV,
+        user_id=user.id,
+    )
+    assert result["routines_created"] == 3
+    assert result["scrobbles_created"] == 0
+
+
+def test_import_flexify_sqlite_only(user):
+    result = import_flexify(
+        sqlite_path=SAMPLE_SQLITE,
+        user_id=user.id,
+    )
+    assert result["scrobbles_created"] == 1
+    routine_titles = set(WorkoutRoutine.objects.values_list("title", flat=True))
+    assert "Friday" in routine_titles
+
+
+def test_alias_targets_exist_in_wrkout_catalog():
+    assert FLEXIFY_EXERCISE_ALIASES["Pull-up"] == "Pullups"
+    assert FLEXIFY_EXERCISE_ALIASES["Push-up"] == "Pushups"
+    assert FLEXIFY_EXERCISE_ALIASES["Crunch"] == "Crunches"
+    assert FLEXIFY_EXERCISE_ALIASES["Deadlift"] == "Barbell Deadlift"
+    assert FLEXIFY_EXERCISE_ALIASES["Squat"] == "Barbell Squat"
+
+
+class FakeWebDAVClient:
+    def __init__(self, files):
+        self.files = {
+            "var/workouts/": None,
+            **{f"var/workouts/{name}": data for name, data in files.items()},
+        }
+
+    def info(self, path):
+        if not any(k.startswith(path) or k == path for k in self.files):
+            raise FileNotFoundError(path)
+
+    def list(self, path, get_info=False):
+        names = [k[len(path) :] for k in self.files if k.startswith(path) and k != path]
+        if get_info:
+            return [{"path": f"{path}{name}"} for name in names]
+        return names
+
+    def mkdir(self, path, recursive=False):
+        self.files.setdefault(path, None)
+
+    def download_sync(self, remote_path, local_path):
+        with open(local_path, "wb") as f:
+            f.write(self.files[remote_path])
+
+    def move(self, src, dst):
+        self.files[dst] = self.files.pop(src)
+
+
+def test_scan_webdav_for_flexify(user, monkeypatch, tmp_path):
+    with open(SAMPLE_SQLITE, "rb") as f:
+        sqlite_bytes = f.read()
+    with open(SAMPLE_CSV, "rb") as f:
+        csv_bytes = f.read()
+
+    client = FakeWebDAVClient(
+        {
+            "flexify-backup.sqlite": sqlite_bytes,
+            "flexify-plans.csv": csv_bytes,
+        }
+    )
+
+    import scrobbles.tasks as scrobbles_tasks
+    from workouts.models import WorkoutImport
+
+    queued = []
+    monkeypatch.setattr(
+        scrobbles_tasks,
+        "process_workout_import",
+        type("Task", (), {"delay": lambda self, pk: queued.append(pk)})(),
+    )
+
+    count = scan_webdav_for_flexify(client, user.id)
+    assert count == 1
+    assert len(queued) == 1
+    assert "var/workouts/flexify-backup.sqlite" not in client.files
+    assert "var/workouts/flexify-plans.csv" not in client.files
+    assert any(
+        k.startswith("var/workouts/processed/") and k.endswith(".sqlite")
+        for k in client.files
+    )
+
+    imp = WorkoutImport.objects.get(pk=queued[0])
+    assert imp.user_id == user.id
+    assert imp.original_filename == "flexify-backup.sqlite"
+    assert imp.file_hash
+    assert imp.csv_file
+
+
+def test_scan_webdav_for_flexify_skips_unchanged(user, monkeypatch, tmp_path):
+    with open(SAMPLE_SQLITE, "rb") as f:
+        sqlite_bytes = f.read()
+
+    client = FakeWebDAVClient({"flexify-backup.sqlite": sqlite_bytes})
+
+    import scrobbles.tasks as scrobbles_tasks
+    from workouts.models import WorkoutImport
+
+    queued = []
+    monkeypatch.setattr(
+        scrobbles_tasks,
+        "process_workout_import",
+        type("Task", (), {"delay": lambda self, pk: queued.append(pk)})(),
+    )
+
+    assert scan_webdav_for_flexify(client, user.id) == 1
+    assert len(queued) == 1
+
+    client.files["var/workouts/flexify-backup.sqlite"] = sqlite_bytes
+
+    count = scan_webdav_for_flexify(client, user.id)
+    assert count == 0
+    assert len(queued) == 1