Christophe Vila
7985000440
Replaces mcp-garmin's server.py: a JSON-lines subprocess protocol (authenticate/complete_mfa/call) around garminconnect directly, no MCP. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
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"]
|