Init

tests/conftest.py

@@ -0,0 +1,45 @@
+from pathlib import Path
+
+import pytest
+
+from mywhoosh_garmin_sync.config import Settings
+
+
+@pytest.fixture
+def settings(tmp_path: Path) -> Settings:
+    return Settings(
+        mywhoosh_email="whoosh@example.com",
+        mywhoosh_password="whoosh-pass",
+        garmin_email="garmin@example.com",
+        garmin_password="garmin-pass",
+        poll_interval_seconds=3600,
+        data_dir=tmp_path,
+        raw_dir=tmp_path / "raw",
+        converted_dir=tmp_path / "converted",
+        browser_state_dir=tmp_path / "browser",
+        mywhoosh_auth_state_path=tmp_path / "mywhoosh_auth_state.json",
+        garmin_tokenstore=tmp_path / "garmin_tokens",
+        db_path=tmp_path / "state.sqlite3",
+        log_level="INFO",
+        dry_run=False,
+        dashboard_enabled=True,
+        dashboard_bind="127.0.0.1",
+        dashboard_port=8080,
+        mywhoosh_login_url="https://www.mywhoosh.com/login/",
+        mywhoosh_activity_url="https://www.mywhoosh.com/profile/",
+        mywhoosh_headless=True,
+        mywhoosh_timeout_seconds=1,
+        mywhoosh_max_downloads_per_run=10,
+        mywhoosh_download_text_hints=["fit", "download"],
+        mywhoosh_activities_button_text="ACTIVITIES",
+        mywhoosh_download_button_selector=".btnDownload",
+        mywhoosh_slow_mo_ms=0,
+        mywhoosh_manual_login_wait_seconds=0,
+        mywhoosh_debug_screenshots=False,
+        mywhoosh_debug_dir=tmp_path / "debug",
+        garmin_mfa_code=None,
+        target_garmin_manufacturer_id=1,
+        target_garmin_product_id=3578,
+        target_garmin_product_name="Edge 1030 Plus",
+        target_garmin_serial_number=None,
+    )

tests/test_config.py

@@ -0,0 +1,35 @@
+from pathlib import Path
+
+from mywhoosh_garmin_sync.config import Settings
+
+
+def test_settings_from_env(monkeypatch, tmp_path):
+    monkeypatch.setenv("MYWHOOSH_EMAIL", "whoosh@example.com")
+    monkeypatch.setenv("MYWHOOSH_PASSWORD", "whoosh-pass")
+    monkeypatch.setenv("GARMIN_EMAIL", "garmin@example.com")
+    monkeypatch.setenv("GARMIN_PASSWORD", "garmin-pass")
+    monkeypatch.setenv("DATA_DIR", str(tmp_path))
+    monkeypatch.setenv("DRY_RUN", "true")
+    monkeypatch.setenv("POLL_INTERVAL_SECONDS", "123")
+    monkeypatch.setenv("TARGET_GARMIN_SERIAL_NUMBER", "123456")
+    monkeypatch.setenv("MYWHOOSH_MANUAL_LOGIN_WAIT_SECONDS", "900")
+
+    settings = Settings.from_env()
+
+    assert settings.data_dir == tmp_path
+    assert settings.raw_dir == tmp_path / "raw"
+    assert settings.dry_run is True
+    assert settings.poll_interval_seconds == 123
+    assert settings.dashboard_enabled is True
+    assert settings.dashboard_bind == "0.0.0.0"
+    assert settings.dashboard_port == 8080
+    assert settings.target_garmin_product_id == 3578
+    assert settings.target_garmin_serial_number == 123456
+    assert settings.mywhoosh_manual_login_wait_seconds == 900
+    assert settings.mywhoosh_activities_button_text == "ACTIVITIES"
+    assert settings.mywhoosh_download_button_selector == ".btnDownload"
+    assert settings.mywhoosh_auth_state_path == tmp_path / "mywhoosh_auth_state.json"
+
+    settings.ensure_directories()
+    assert Path(settings.raw_dir).is_dir()
+    assert Path(settings.mywhoosh_debug_dir).is_dir()

tests/test_fit_device.py

@@ -0,0 +1,119 @@
+from pathlib import Path
+import struct
+
+import pytest
+
+from mywhoosh_garmin_sync.fit_crc import fit_crc
+from mywhoosh_garmin_sync.fit_device import (
+    DeviceFieldValue,
+    FitFormatError,
+    GarminDevice,
+    convert_fit_device,
+    read_device_field_values,
+)
+
+
+def test_convert_fit_device_patches_metadata_and_crc(tmp_path: Path):
+    source = tmp_path / "source.fit"
+    output = tmp_path / "output.fit"
+    source.write_bytes(_minimal_activity_fit())
+
+    result = convert_fit_device(
+        source,
+        output,
+        GarminDevice(product_id=3578, product_name="Edge 1030 Plus", serial_number=42),
+    )
+
+    assert result.patched_field_count == 8
+    values = read_device_field_values(output)
+    assert DeviceFieldValue(0, 1, 1) in values
+    assert DeviceFieldValue(0, 2, 3578) in values
+    assert DeviceFieldValue(0, 3, 42) in values
+    assert DeviceFieldValue(0, 8, "Edge 1030 Plus") in values
+    assert DeviceFieldValue(23, 2, 1) in values
+    assert DeviceFieldValue(23, 3, 42) in values
+    assert DeviceFieldValue(23, 4, 3578) in values
+    assert DeviceFieldValue(23, 27, "Edge 1030 Plus") in values
+
+    data = output.read_bytes()
+    assert struct.unpack_from("<H", data, 12)[0] == fit_crc(data[:12])
+    assert struct.unpack_from("<H", data, len(data) - 2)[0] == fit_crc(data[:-2])
+
+
+def test_convert_fit_device_rejects_bad_crc(tmp_path: Path):
+    source = tmp_path / "bad.fit"
+    output = tmp_path / "output.fit"
+    data = bytearray(_minimal_activity_fit())
+    data[-1] ^= 0xFF
+    source.write_bytes(data)
+
+    with pytest.raises(FitFormatError):
+        convert_fit_device(source, output)
+
+
+def _minimal_activity_fit() -> bytes:
+    data = bytearray()
+
+    data.extend(
+        _definition(
+            local=0,
+            global_message=0,
+            fields=[
+                (0, 1, 0x00),
+                (1, 2, 0x84),
+                (2, 2, 0x84),
+                (3, 4, 0x8C),
+                (8, 20, 0x07),
+            ],
+        )
+    )
+    data.extend(b"\x00")
+    data.extend(struct.pack("<BHHI", 4, 32, 40, 999))
+    data.extend(_fit_string("MyWhoosh", 20))
+
+    data.extend(
+        _definition(
+            local=1,
+            global_message=23,
+            fields=[
+                (2, 2, 0x84),
+                (3, 4, 0x8C),
+                (4, 2, 0x84),
+                (27, 20, 0x07),
+            ],
+        )
+    )
+    data.extend(b"\x01")
+    data.extend(struct.pack("<HIH", 32, 999, 40))
+    data.extend(_fit_string("Trainer", 20))
+
+    header = bytearray(14)
+    header[0] = 14
+    header[1] = 0x10
+    struct.pack_into("<H", header, 2, 0x08)
+    struct.pack_into("<I", header, 4, len(data))
+    header[8:12] = b".FIT"
+    struct.pack_into("<H", header, 12, fit_crc(header[:12]))
+
+    fit_file = header + data + b"\x00\x00"
+    struct.pack_into("<H", fit_file, len(fit_file) - 2, fit_crc(fit_file[:-2]))
+    return bytes(fit_file)
+
+
+def _definition(
+    local: int, global_message: int, fields: list[tuple[int, int, int]]
+) -> bytes:
+    result = bytearray()
+    result.append(0x40 | local)
+    result.append(0)
+    result.append(0)
+    result.extend(struct.pack("<H", global_message))
+    result.append(len(fields))
+    for field_num, size, base_type in fields:
+        result.extend(bytes([field_num, size, base_type]))
+    return bytes(result)
+
+
+def _fit_string(value: str, size: int) -> bytes:
+    encoded = value.encode("utf-8")[: size - 1] + b"\x00"
+    return encoded + b"\x00" * (size - len(encoded))

tests/test_garmin.py

@@ -0,0 +1,59 @@
+from dataclasses import replace
+
+import pytest
+
+from mywhoosh_garmin_sync.garmin import GarminUploadBlocked, GarminUploader
+
+
+class FakeGarminClient:
+    def __init__(self, *args, **kwargs):
+        self.prompt_mfa = kwargs.get("prompt_mfa")
+        self.logged_in = False
+
+    def login(self, tokenstore=None):
+        self.logged_in = True
+
+    def upload_activity(self, activity_path):
+        return {"activityId": 12345, "path": activity_path}
+
+
+class DuplicateGarminClient(FakeGarminClient):
+    def upload_activity(self, activity_path):
+        raise RuntimeError("409 duplicate activity already exists")
+
+
+def test_garmin_uploader_success(settings, tmp_path):
+    fit = tmp_path / "activity.fit"
+    fit.write_bytes(b"fit")
+    uploader = GarminUploader(settings, client_factory=FakeGarminClient)
+
+    result = uploader.upload(fit)
+
+    assert result.status == "uploaded"
+    assert result.garmin_activity_id == "12345"
+
+
+def test_garmin_uploader_duplicate(settings, tmp_path):
+    fit = tmp_path / "activity.fit"
+    fit.write_bytes(b"fit")
+    uploader = GarminUploader(settings, client_factory=DuplicateGarminClient)
+
+    result = uploader.upload(fit)
+
+    assert result.duplicate is True
+    assert result.status == "duplicate"
+
+
+def test_mfa_prompt_blocks_without_code(settings):
+    uploader = GarminUploader(settings, client_factory=FakeGarminClient)
+
+    with pytest.raises(GarminUploadBlocked):
+        uploader._prompt_mfa()
+
+
+def test_mfa_prompt_returns_env_code(settings):
+    settings = replace(settings, garmin_mfa_code="123456")
+    uploader = GarminUploader(settings, client_factory=FakeGarminClient)
+
+    assert uploader._prompt_mfa() == "123456"
+

tests/test_mywhoosh.py

@@ -0,0 +1,58 @@
+from mywhoosh_garmin_sync.mywhoosh import (
+    _looks_like_challenge_text,
+    _looks_like_challenge_url,
+    _looks_like_cookie_accept_label,
+    _source_ref,
+    _storage_restore_script,
+)
+
+
+def test_cookie_accept_label_matching():
+    assert _looks_like_cookie_accept_label("Accept")
+    assert _looks_like_cookie_accept_label("Accept all cookies")
+    assert _looks_like_cookie_accept_label("Allow all")
+    assert _looks_like_cookie_accept_label("I agree")
+    assert _looks_like_cookie_accept_label("Got it")
+
+    assert not _looks_like_cookie_accept_label("Reject all")
+    assert not _looks_like_cookie_accept_label("Download app")
+
+
+def test_challenge_text_matching():
+    assert _looks_like_challenge_text("I'm not a robot")
+    assert _looks_like_challenge_text("Verify you are human")
+    assert _looks_like_challenge_text("iframe title recaptcha")
+    assert _looks_like_challenge_text("Cloudflare security check")
+
+    assert not _looks_like_challenge_text("Welcome to your activities")
+
+
+def test_challenge_url_matching_does_not_flag_plain_login():
+    assert _looks_like_challenge_url("https://example.test/cdn-cgi/challenge-platform")
+    assert _looks_like_challenge_url("https://example.test/verify")
+
+    assert not _looks_like_challenge_url("https://event.mywhoosh.com/login/")
+    assert not _looks_like_challenge_url("https://event.mywhoosh.com/user/activities")
+
+
+def test_source_ref_prefers_href_then_row_text():
+    assert _source_ref("https://example.test/a.fit", "Download", 0, "Ride A") == _source_ref(
+        "https://example.test/a.fit", "Download", 99, "Ride B"
+    )
+    assert _source_ref(None, "", 0, "Ride A") != _source_ref(None, "", 1, "Ride B")
+
+
+def test_storage_restore_script_contains_session_storage():
+    script = _storage_restore_script(
+        [
+            {
+                "origin": "https://event.mywhoosh.com",
+                "localStorage": [{"name": "token", "value": "local"}],
+                "sessionStorage": [{"name": "session-token", "value": "session"}],
+            }
+        ]
+    )
+
+    assert "https://event.mywhoosh.com" in script
+    assert "localStorage" in script
+    assert "sessionStorage" in script

tests/test_state.py

@@ -0,0 +1,44 @@
+from pathlib import Path
+
+from mywhoosh_garmin_sync.state import ActivityStore
+
+
+def test_state_transitions_and_hash_lookup(tmp_path: Path):
+    store = ActivityStore(tmp_path / "state.sqlite3")
+    store.initialize()
+
+    store.record_downloaded(
+        source_ref="source-1",
+        title="Ride",
+        source_url="https://example.test/ride.fit",
+        raw_path=tmp_path / "raw.fit",
+        raw_sha256="raw-hash",
+    )
+    assert store.get_status("source-1") == "downloaded"
+    assert store.is_terminal_source("source-1") is False
+
+    store.mark_converted(
+        source_ref="source-1",
+        converted_path=tmp_path / "converted.fit",
+        converted_sha256="converted-hash",
+        patched_field_count=4,
+    )
+    assert store.get_status("source-1") == "converted"
+    assert store.is_uploaded_hash("converted-hash") is False
+
+    store.mark_uploaded("source-1", "123")
+    assert store.get_status("source-1") == "uploaded"
+    assert store.is_terminal_source("source-1") is True
+    assert store.is_uploaded_hash("raw-hash") is True
+    assert store.is_uploaded_hash("converted-hash") is True
+
+
+def test_failed_activity_can_be_retried(tmp_path: Path):
+    store = ActivityStore(tmp_path / "state.sqlite3")
+    store.initialize()
+    store.record_downloaded("source-1", "Ride", None, tmp_path / "raw.fit", "hash")
+    store.mark_failed("source-1", "network failed")
+
+    assert store.get_status("source-1") == "failed"
+    assert store.is_terminal_source("source-1") is False
+