Stride/tests/test_sync.py

import io
import json
import struct
import zipfile
from types import SimpleNamespace

import pytest
from fitdecode.utils import compute_crc

from garmin_sync import __main__ as cli
from garmin_sync.service import (
    SyncError,
    account_key,
    complete,
    decode_fit,
    fit_payloads,
    sync_activities,
)


def synthetic_fit():
    """Tiny FIT with one timestamped heart-rate/distance record; no real user data."""
    # Definition: local message 0, little endian, global record message 20.
    definition = (
        bytes([0x40, 0, 0])
        + struct.pack("<H", 20)
        + bytes(
            [
                3,
                253,
                4,
                0x86,
                3,
                1,
                0x02,
                5,
                4,
                0x86,
            ]
        )
    )
    record = bytes([0]) + struct.pack("<IBI", 1_000_000_000, 150, 123456)
    data = definition + record
    header = struct.pack("<BBHI4s", 14, 0x20, 2100, len(data), b".FIT")
    header += struct.pack("<H", compute_crc(header))
    content = header + data
    return content + struct.pack("<H", compute_crc(content))


def zipped(content, name="activity.fit"):
    stream = io.BytesIO()
    with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr(name, content)
    return stream.getvalue()


class FakeGarmin:
    display_name = "synthetic-account"

    def __init__(self, count=1, payload=None):
        self.activities = [{"activityId": n + 1, "distance": 1234.56} for n in range(count)]
        self.payload = zipped(synthetic_fit()) if payload is None else payload
        self.downloads = 0
        self.requests = []

    def get_activities(self, start, limit, activitytype):
        self.requests.append((start, limit, activitytype))
        return self.activities[start : start + limit]

    def download_activity(self, activity_id, dl_fmt):
        self.downloads += 1
        return self.payload


def test_decode_real_binary_fit_and_scaled_measurements():
    result = decode_fit(synthetic_fit())
    assert result["message_counts"] == {"record": 1}
    assert result["record"][0]["heart_rate"] == 150
    assert result["record"][0]["distance"] == 1234.56
    assert result["units"]["record.distance"] == "m"


def test_bad_crc_rejected():
    damaged = bytearray(synthetic_fit())
    damaged[-1] ^= 1
    with pytest.raises(SyncError, match="CRC"):
        decode_fit(bytes(damaged))


def test_archive_paths_are_never_extracted(tmp_path):
    content = synthetic_fit()
    assert fit_payloads(zipped(content, "../../outside.fit")) == [content]
    assert not list(tmp_path.iterdir())


@pytest.mark.parametrize(
    "payload", [b"", b"<html>Unauthorized</html>", zipped(b"text", "notes.txt")]
)
def test_bad_download_rejected(payload):
    with pytest.raises(SyncError):
        fit_payloads(payload)


def test_download_decode_deduplicate_and_repair(tmp_path):
    api = FakeGarmin()
    first = sync_activities(api, tmp_path, 5, None)
    assert first["downloaded"] == 1
    assert first["records"] == 1
    assert complete(tmp_path / "1")
    parsed = json.loads((tmp_path / "1" / "activity-1.json").read_text())
    assert parsed["record"][0]["timestamp"].endswith("+00:00")
    assert sync_activities(api, tmp_path, 5, None)["skipped"] == 1
    assert api.downloads == 1
    (tmp_path / "1" / "activity-1.fit").write_bytes(b"corrupt")
    assert sync_activities(api, tmp_path, 5, None)["downloaded"] == 1
    assert api.downloads == 2


def test_pagination_and_filter(tmp_path):
    api = FakeGarmin(count=53)
    result = sync_activities(api, tmp_path, 52, "running")
    assert result["downloaded"] == 52
    assert api.requests == [(0, 50, "running"), (50, 2, "running")]


def test_partial_failure_has_no_completed_manifest(tmp_path):
    result = sync_activities(FakeGarmin(payload=b"bad"), tmp_path, 5, None)
    assert result["failed"] == 1
    assert not complete(tmp_path / "1")


def test_invalid_activity_id_cannot_escape_directory(tmp_path):
    api = FakeGarmin()
    api.activities = [{"activityId": "../escape"}]
    assert sync_activities(api, tmp_path, 5, None)["failed"] == 1
    assert api.downloads == 0


def test_account_isolation_and_stable_provider_id():
    a = SimpleNamespace(display_name="one", profile_id=1)
    b = SimpleNamespace(display_name="two", profile_id=2)
    assert account_key(a, "garmin.com") != account_key(b, "garmin.com")
    a.display_name = "renamed"
    assert account_key(a, "garmin.com") == account_key(
        SimpleNamespace(display_name="original", profile_id=1), "garmin.com"
    )


def test_missing_session_requires_local_login(tmp_path, monkeypatch):
    monkeypatch.setattr(cli, "ROOT", tmp_path)
    with pytest.raises(SyncError, match="No saved session"):
        cli.run(cli.parser().parse_args(["sync"]))


def test_profile_cannot_escape(tmp_path, monkeypatch):
    monkeypatch.setattr(cli, "ROOT", tmp_path)
    with pytest.raises(SyncError, match="Profile"):
        cli.run(cli.parser().parse_args(["--profile", "../outside", "verify"]))


def test_cli_never_prints_raw_provider_exception(monkeypatch, capsys):
    def fail(_args):
        raise RuntimeError("SECRET_TOKEN_AND_PASSWORD")

    monkeypatch.setattr(cli, "run", fail)
    monkeypatch.setattr("sys.argv", ["garmin_sync", "verify"])
    assert cli.main() == 1
    captured = capsys.readouterr()
    assert "SECRET_TOKEN" not in captured.err + captured.out


def test_saved_session_sync_persists_rotated_credentials_even_on_failure(tmp_path, monkeypatch):
    store = tmp_path / "instance" / "garmin" / "default"
    store.mkdir(parents=True)
    token = store / "garmin_tokens.json"
    token.write_text('{"di_token":"old"}')
    api = FakeGarmin()
    api.client = SimpleNamespace(dumps=lambda: '{"di_token":"new"}')
    api.login = lambda path: None
    monkeypatch.setattr(cli, "ROOT", tmp_path)
    monkeypatch.setattr(cli, "Garmin", lambda **kwargs: api)

    def fail(*args):
        raise SyncError("download failed")

    monkeypatch.setattr(cli, "sync_activities", fail)
    with pytest.raises(SyncError, match="download failed"):
        cli.run(cli.parser().parse_args(["sync"]))
    assert json.loads(token.read_text())["di_token"] == "new"