geniusrun/backend/internal/garmin/pyscript/tests/test_wrapper.py

import os
import queue
from unittest.mock import MagicMock, patch

import pytest

import wrapper


@pytest.fixture(autouse=True)
def reset_state():
    original_state = wrapper._auth_state
    original_client = wrapper._client
    for q in (wrapper._mfa_input_queue, wrapper._login_result_queue):
        while not q.empty():
            try:
                q.get_nowait()
            except queue.Empty:
                break
    yield
    wrapper._auth_state = original_state
    wrapper._client = original_client


def test_authenticate_success():
    env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
    with patch.dict("os.environ", env):
        with patch("wrapper.Garmin") as mock_garmin_cls:
            mock_garmin_cls.return_value = MagicMock()
            resp = wrapper.dispatch({"id": 1, "cmd": "authenticate"})
    assert resp == {"id": 1, "result": {"status": "success", "message": "Authenticated successfully."}}
    assert wrapper._auth_state == "authenticated"


def test_authenticate_mfa_required():
    env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
    with patch.dict("os.environ", env):
        with patch("wrapper.Garmin") as mock_garmin_cls:
            mock_garmin_cls.return_value = MagicMock()
            with patch.object(wrapper._login_result_queue, "get", side_effect=queue.Empty):
                resp = wrapper.dispatch({"id": 2, "cmd": "authenticate"})
    assert resp["result"]["status"] == "mfa_required"
    assert wrapper._auth_state == "mfa_pending"


def test_authenticate_missing_credentials():
    with patch.dict("os.environ", {}, clear=False):
        os.environ.pop("GARMIN_EMAIL", None)
        os.environ.pop("GARMIN_PASSWORD", None)
        resp = wrapper.dispatch({"id": 3, "cmd": "authenticate"})
    assert resp["result"]["status"] == "failed"
    assert "GARMIN_EMAIL" in resp["result"]["message"]


def test_complete_mfa_success():
    wrapper._auth_state = "mfa_pending"
    wrapper._login_result_queue.put(("success", None))
    resp = wrapper.dispatch({"id": 4, "cmd": "complete_mfa", "params": {"code": "123456"}})
    assert resp == {
        "id": 4,
        "result": {"status": "success", "message": "MFA accepted. Authenticated successfully."},
    }
    assert wrapper._auth_state == "authenticated"
    assert wrapper._mfa_input_queue.get_nowait() == "123456"


def test_complete_mfa_not_in_progress():
    wrapper._auth_state = "unauthenticated"
    resp = wrapper.dispatch({"id": 5, "cmd": "complete_mfa", "params": {"code": "123456"}})
    assert resp["result"]["status"] == "failed"
    assert "No MFA in progress" in resp["result"]["message"]


def test_call_dispatches_to_named_garminconnect_method():
    wrapper._auth_state = "authenticated"
    wrapper._client = MagicMock()
    wrapper._client.get_activities_by_date.return_value = [{"activityId": "111"}]

    resp = wrapper.dispatch({
        "id": 6,
        "cmd": "call",
        "params": {
            "method": "get_activities_by_date",
            "args": {"startdate": "2026-07-01", "enddate": "2026-07-25"},
        },
    })

    assert resp == {"id": 6, "result": [{"activityId": "111"}]}
    wrapper._client.get_activities_by_date.assert_called_once_with(
        startdate="2026-07-01", enddate="2026-07-25"
    )


def test_call_unauthenticated_is_error():
    wrapper._auth_state = "unauthenticated"
    resp = wrapper.dispatch({
        "id": 7, "cmd": "call", "params": {"method": "get_activities_by_date", "args": {}},
    })
    assert "error" in resp
    assert "Not authenticated" in resp["error"]


def test_call_unknown_method_is_error():
    wrapper._auth_state = "authenticated"
    wrapper._client = MagicMock(spec=["get_activities_by_date"])
    resp = wrapper.dispatch({
        "id": 8, "cmd": "call", "params": {"method": "delete_everything", "args": {}},
    })
    assert resp["id"] == 8
    assert "error" in resp


def test_call_propagates_garminconnect_exception_as_error():
    wrapper._auth_state = "authenticated"
    wrapper._client = MagicMock()
    wrapper._client.get_activity_splits.side_effect = Exception("not found")
    resp = wrapper.dispatch({
        "id": 9, "cmd": "call", "params": {"method": "get_activity_splits", "args": {"activity_id": "999"}},
    })
    assert resp == {"id": 9, "error": "not found"}


def test_dispatch_unknown_cmd_is_error():
    resp = wrapper.dispatch({"id": 10, "cmd": "not_a_real_cmd"})
    assert resp["id"] == 10
    assert "unknown cmd" in resp["error"]


def test_startup_login_success():
    env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
    with patch.dict("os.environ", env):
        with patch("wrapper.Garmin") as mock_garmin_cls:
            mock_garmin_cls.return_value = MagicMock()
            wrapper._startup_login()
    assert wrapper._auth_state == "authenticated"


def test_startup_login_missing_credentials_returns_immediately():
    with patch.dict("os.environ", {}, clear=False):
        os.environ.pop("GARMIN_EMAIL", None)
        os.environ.pop("GARMIN_PASSWORD", None)
        wrapper._auth_state = "unauthenticated"
        wrapper._startup_login()
    assert wrapper._auth_state == "unauthenticated"
    assert wrapper._client is None


def test_startup_login_times_out_without_blocking():
    env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
    with patch.dict("os.environ", env):
        with patch("wrapper.Garmin") as mock_garmin_cls:
            mock_garmin_cls.return_value = MagicMock()
            with patch.object(wrapper._login_result_queue, "get", side_effect=queue.Empty):
                # Should return promptly (bounded by the 10s timeout passed to
                # queue.get, which is mocked here to raise immediately) rather
                # than blocking main()'s stdin dispatch loop from starting.
                wrapper._startup_login()
    assert wrapper._auth_state == "unauthenticated"