87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
import json
|
|
import os
|
|
import queue
|
|
import threading
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
import server
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_state():
|
|
original_state = server._auth_state
|
|
original_client = server._client
|
|
for q in (server._mfa_input_queue, server._login_result_queue):
|
|
while not q.empty():
|
|
try:
|
|
q.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
yield
|
|
server._auth_state = original_state
|
|
server._client = original_client
|
|
for q in (server._mfa_input_queue, server._login_result_queue):
|
|
while not q.empty():
|
|
try:
|
|
q.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
|
|
|
|
def test_check_auth_unauthenticated():
|
|
server._auth_state = "unauthenticated"
|
|
assert server._check_auth() == "Not authenticated. Ask Claude to call authenticate() first."
|
|
|
|
|
|
def test_check_auth_authenticated():
|
|
server._auth_state = "authenticated"
|
|
assert server._check_auth() is None
|
|
|
|
|
|
def test_authenticate_success():
|
|
env = {"GARMIN_EMAIL": "test@example.com", "GARMIN_PASSWORD": "secret"}
|
|
with patch.dict("os.environ", env):
|
|
with patch("server.Garmin") as mock_garmin_cls:
|
|
mock_garmin_cls.return_value = MagicMock()
|
|
# mock login() returns immediately, thread puts success in queue
|
|
result = server.authenticate()
|
|
assert result == "Authenticated successfully."
|
|
assert server._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("server.Garmin") as mock_garmin_cls:
|
|
mock_garmin_cls.return_value = MagicMock()
|
|
with patch.object(server._login_result_queue, "get", side_effect=queue.Empty):
|
|
result = server.authenticate()
|
|
assert "MFA required" in result
|
|
assert server._auth_state == "mfa_pending"
|
|
|
|
|
|
def test_complete_mfa_success():
|
|
server._auth_state = "mfa_pending"
|
|
server._login_result_queue.put(("success", None))
|
|
result = server.complete_mfa("123456")
|
|
assert result == "MFA accepted. Authenticated successfully."
|
|
assert server._auth_state == "authenticated"
|
|
assert server._mfa_input_queue.get_nowait() == "123456"
|
|
|
|
|
|
def test_complete_mfa_not_in_progress():
|
|
server._auth_state = "unauthenticated"
|
|
result = server.complete_mfa("123456")
|
|
assert result == "No MFA in progress. Call authenticate() first."
|
|
|
|
|
|
def test_authenticate_missing_credentials():
|
|
with patch.dict("os.environ", {}, clear=False):
|
|
# Ensure the env vars are absent
|
|
os.environ.pop("GARMIN_EMAIL", None)
|
|
os.environ.pop("GARMIN_PASSWORD", None)
|
|
result = server.authenticate()
|
|
assert result == "GARMIN_EMAIL and GARMIN_PASSWORD environment variables are required."
|