mywhoosh2garmin/tests/garmin/test_uploader.py

from pathlib import Path

import pytest

from app.garmin.uploader import (
    GarminAuthError,
    GarminImportRejected,
    GarminTransientError,
    GarminUploadBlocked,
    GarminUploader,
)


class FakeGarmin:
    def __init__(self, *args, prompt_mfa=None, import_result=None, login_error=None, import_error=None, **kwargs):
        self.prompt_mfa = prompt_mfa
        self.import_result = import_result or {"activityId": 42}
        self.login_error = login_error
        self.import_error = import_error
        self.login_path = None

    def login(self, tokenstore=None):
        self.login_path = tokenstore
        if self.login_error:
            raise self.login_error

    def import_activity(self, activity_path: str):
        if self.import_error:
            raise self.import_error
        return self.import_result


def test_successful_import_returns_activity_id(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_result={"activityId": 42}),
    )
    result = uploader.import_fit(tmp_path / "ride.fit")
    assert result.status == "imported"
    assert result.garmin_activity_id == "42"


def test_duplicate_is_terminal_success(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_error=RuntimeError("409 duplicate")),
    )
    result = uploader.import_fit(tmp_path / "ride.fit")
    assert result.duplicate is True
    assert result.status == "duplicate"


def test_mfa_without_code_is_blocked(tmp_path: Path) -> None:
    class MfaGarmin(FakeGarmin):
        def login(self, tokenstore=None):
            self.prompt_mfa()

    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=MfaGarmin,
    )
    with pytest.raises(GarminUploadBlocked):
        uploader.import_fit(tmp_path / "ride.fit")


def test_mfa_code_is_returned_only_to_prompt(tmp_path: Path) -> None:
    seen = []

    class MfaGarmin(FakeGarmin):
        def login(self, tokenstore=None):
            seen.append(self.prompt_mfa())

    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=MfaGarmin,
    )
    uploader.import_fit(tmp_path / "ride.fit", mfa_code="123456")
    assert seen == ["123456"]


def test_import_rejected_by_garmin_raises_garmin_import_rejected(tmp_path: Path) -> None:
    rejected_response = {
        "detailedImportResult": {
            "successes": [],
            "failures": [{"internalId": 1, "messages": ["Invalid FIT file"]}],
        }
    }
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_result=rejected_response),
    )
    with pytest.raises(GarminImportRejected):
        uploader.import_fit(tmp_path / "ride.fit")


def test_login_failure_with_429_is_transient(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("429 too many requests")),
    )
    with pytest.raises(GarminTransientError):
        uploader.import_fit(tmp_path / "ride.fit")


def test_login_failure_with_timeout_is_transient(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("connection timeout")),
    )
    with pytest.raises(GarminTransientError):
        uploader.import_fit(tmp_path / "ride.fit")


def test_login_failure_with_credential_message_is_auth_error(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("invalid credential")),
    )
    with pytest.raises(GarminAuthError):
        uploader.import_fit(tmp_path / "ride.fit")


def test_login_failure_unrecognized_is_transient(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, login_error=RuntimeError("something odd happened")),
    )
    with pytest.raises(GarminTransientError):
        uploader.import_fit(tmp_path / "ride.fit")


def test_import_time_401_raises_auth_error(tmp_path: Path) -> None:
    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=lambda *a, **kw: FakeGarmin(*a, **kw, import_error=RuntimeError("401 unauthorized")),
    )
    with pytest.raises(GarminAuthError):
        uploader.import_fit(tmp_path / "ride.fit")


def test_mfa_wrapped_in_generic_exception_still_blocks(tmp_path: Path) -> None:
    class WrappedMfaGarmin(FakeGarmin):
        def login(self, tokenstore=None):
            try:
                self.prompt_mfa()
            except GarminUploadBlocked as exc:
                raise RuntimeError(f"Login failed: Garmin requested MFA ({exc})") from exc

    uploader = GarminUploader(
        email="g@example.com",
        password="pw",
        tokenstore=tmp_path / "garmin",
        client_factory=WrappedMfaGarmin,
    )
    with pytest.raises(GarminUploadBlocked):
        uploader.import_fit(tmp_path / "ride.fit")