feat(garmin): add direct garminconnect wrapper script
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>
This commit is contained in:
0
backend/internal/garmin/pyscript/tests/__init__.py
Normal file
0
backend/internal/garmin/pyscript/tests/__init__.py
Normal file
126
backend/internal/garmin/pyscript/tests/test_wrapper.py
Normal file
126
backend/internal/garmin/pyscript/tests/test_wrapper.py
Normal file
@@ -0,0 +1,126 @@
|
||||
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"]
|
||||
Reference in New Issue
Block a user