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

import os
import queue
import threading
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():
    # _startup_login now waits on its own private, function-local queue
    # (never the shared wrapper._login_result_queue -- see the
    # no-shared-queue-leakage test below), so forcing its timeout path
    # means intercepting the queue.Queue() constructor it calls internally
    # rather than patching a module-level queue object.
    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("wrapper.queue.Queue") as mock_queue_cls:
                mock_queue_cls.return_value.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"


def test_startup_login_does_not_leak_into_shared_authenticate_queue():
    """Regression test: _startup_login used to push its background thread's
    result onto the same module-level _login_result_queue that
    _handle_authenticate/_handle_complete_mfa share for the MFA handoff. A
    startup login that was still in flight when a user's first explicit
    'authenticate' call came in could have that call accidentally dequeue
    the startup thread's stale result instead of its own fresh one.
    _startup_login now uses its own private queue, so a still-in-flight
    startup attempt can never interfere with a later authenticate() call."""
    release = threading.Event()

    env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
    with patch.dict("os.environ", env):
        with patch("wrapper.Garmin") as mock_garmin_cls:
            # A startup login whose underlying garminconnect call is still
            # blocked (simulating "slow to resolve") when _startup_login's
            # own bounded wait (mocked to time out immediately, so this test
            # doesn't need to sleep the real 10s) returns.
            slow_client = MagicMock()
            slow_client.login.side_effect = lambda **kwargs: release.wait(5)
            mock_garmin_cls.return_value = slow_client
            with patch("wrapper.queue.Queue") as mock_queue_cls:
                mock_queue_cls.return_value.get.side_effect = queue.Empty
                wrapper._startup_login()

    assert wrapper._auth_state == "unauthenticated"
    # Nothing from the still-in-flight startup attempt ever touches the
    # shared queue that authenticate()/complete_mfa() rely on.
    assert wrapper._login_result_queue.empty()

    release.set()  # let the stale startup thread finish harmlessly in the background

    # A fresh, unrelated authenticate() call must get its own result, not
    # anything left over from the startup attempt above.
    with patch.dict("os.environ", env):
        with patch("wrapper.Garmin") as mock_garmin_cls2:
            mock_garmin_cls2.return_value = MagicMock()
            resp = wrapper.dispatch({"id": 20, "cmd": "authenticate"})

    assert resp == {"id": 20, "result": {"status": "success", "message": "Authenticated successfully."}}
    assert wrapper._auth_state == "authenticated"