218 lines
8.6 KiB
Python
218 lines
8.6 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from datetime import datetime
|
|
import json
|
|
import garth
|
|
from garth.exc import GarthException
|
|
from sqlalchemy.orm import Session
|
|
import dataclasses
|
|
|
|
from src.services.garmin.client import GarminClient
|
|
from src.models.api_token import APIToken
|
|
# from garth.http import Client # No longer needed if we patch garth module
|
|
|
|
@pytest.fixture
|
|
def mock_db_session():
|
|
"""Fixture for a mock SQLAlchemy session."""
|
|
session = MagicMock(spec=Session)
|
|
session.query.return_value.filter_by.return_value.first.return_value = None
|
|
return session
|
|
|
|
|
|
@pytest.fixture
|
|
def garmin_client_instance():
|
|
"""Fixture for a GarminClient instance with test credentials."""
|
|
return GarminClient(username="testuser", password="testpassword", is_china=False)
|
|
|
|
|
|
@pytest.fixture
|
|
def garmin_client_mfa_instance():
|
|
"""Fixture for a GarminClient instance for MFA testing."""
|
|
return GarminClient(username="testmfauser", password="testmfapassword", is_china=False)
|
|
|
|
|
|
@patch("src.services.garmin.auth.garth")
|
|
def test_login_success(mock_garth_module, mock_db_session, garmin_client_instance):
|
|
"""Test successful login scenario."""
|
|
# Mock garth.login to return successfully
|
|
mock_garth_module.login.return_value = None
|
|
|
|
# Mock garth.client tokens
|
|
mock_garth_module.client.oauth1_token = {"oauth1": "token"}
|
|
mock_garth_module.client.oauth2_token = {"oauth2": "token"}
|
|
|
|
# Call the login method
|
|
status = garmin_client_instance.login(mock_db_session)
|
|
|
|
# Assertions
|
|
mock_garth_module.login.assert_called_once_with("testuser", "testpassword", return_on_mfa=True)
|
|
assert status == "success"
|
|
assert garmin_client_instance.is_connected is True
|
|
|
|
# Verify update_tokens was called
|
|
with patch.object(garmin_client_instance, 'update_tokens') as mock_update_tokens:
|
|
# Re-run login to trigger mock
|
|
garmin_client_instance.login(mock_db_session)
|
|
mock_update_tokens.assert_called_with(mock_db_session, {"oauth1": "token"}, {"oauth2": "token"})
|
|
|
|
|
|
@patch("src.services.garmin.auth.garth")
|
|
def test_login_mfa_required(mock_garth_module, mock_db_session, garmin_client_mfa_instance):
|
|
"""Test login scenario when MFA is required (native return)."""
|
|
# Mock garth.client.mfa_state
|
|
mock_client_for_mfa = MagicMock()
|
|
mock_client_for_mfa.sess.cookies.get_dict.return_value = {"cookie1": "val1"}
|
|
|
|
# Note: Logic captures last_resp which is an object with .text and .url
|
|
mock_last_resp = MagicMock()
|
|
mock_last_resp.text = "<html>some text</html>"
|
|
mock_last_resp.url = "http://garmin.com/mfa"
|
|
mock_client_for_mfa.last_resp = mock_last_resp
|
|
|
|
mfa_state = {
|
|
"client": mock_client_for_mfa
|
|
}
|
|
|
|
mock_garth_module.login.return_value = ("needs_mfa", mfa_state)
|
|
|
|
# Call the login method
|
|
# Patch initiate_mfa to verify it gets called
|
|
with patch.object(garmin_client_mfa_instance, 'initiate_mfa') as mock_initiate_mfa:
|
|
status = garmin_client_mfa_instance.login(mock_db_session)
|
|
|
|
# Assertions
|
|
mock_garth_module.login.assert_called_once_with("testmfauser", "testmfapassword", return_on_mfa=True)
|
|
assert status == "mfa_required"
|
|
mock_initiate_mfa.assert_called_once_with(mock_db_session, mfa_state)
|
|
|
|
|
|
@patch("src.services.garmin.auth.garth")
|
|
def test_login_failure(mock_garth_module, mock_db_session, garmin_client_instance):
|
|
"""Test login scenario when authentication fails (not MFA)."""
|
|
# Mock garth.login to raise a generic GarthException
|
|
mock_garth_module.login.side_effect = GarthException("Invalid credentials")
|
|
|
|
# Call the login method
|
|
status = garmin_client_instance.login(mock_db_session)
|
|
|
|
# Assertions
|
|
mock_garth_module.login.assert_called_once_with("testuser", "testpassword", return_on_mfa=True)
|
|
assert status == "error"
|
|
assert garmin_client_instance.is_connected is False
|
|
|
|
|
|
@patch("src.services.garmin.auth.garth.client.resume_login")
|
|
@patch("garth.http.Client") # Still patch Client class separately as it's imported
|
|
@patch.object(GarminClient, 'update_tokens')
|
|
def test_handle_mfa_success(mock_update_tokens, mock_client_class, mock_resume_login, mock_db_session, garmin_client_instance):
|
|
"""Test successful MFA completion."""
|
|
# Arg order: mock_update_tokens (Bottom), mock_client_class (2nd), mock_resume_login (Top)
|
|
# This assumes garth.http.Client is patched.
|
|
# Note: handle_mfa does `from garth.http import Client`. Patching `garth.http.Client` works.
|
|
|
|
# Setup mock MFA state in DB
|
|
mfa_state_data = {
|
|
"cookies": {"cookie1": "val1"},
|
|
"domain": "garmin.com",
|
|
"last_resp_text": "<html></html>",
|
|
"last_resp_url": "http://url",
|
|
"signin_params": {"_csrf": "token"}
|
|
}
|
|
mock_token_record = MagicMock(spec=APIToken)
|
|
mock_token_record.mfa_state = json.dumps(mfa_state_data)
|
|
mock_db_session.query.return_value.filter_by.return_value.first.return_value = mock_token_record
|
|
|
|
# Mock the Client constructor
|
|
mock_client_instance = MagicMock()
|
|
mock_client_instance.domain = mfa_state_data["domain"]
|
|
mock_client_instance.sess = MagicMock()
|
|
mock_client_instance.sess.cookies = MagicMock()
|
|
mock_client_instance.sess.cookies.update = MagicMock()
|
|
|
|
mock_client_class.return_value = mock_client_instance
|
|
|
|
# Mock garth.resume_login to succeed and RETURN tokens
|
|
# Note: handle_mfa calls `garth.client.resume_login`.
|
|
# We patched it directly via string.
|
|
new_tokens = ({"oauth1": "new_token"}, {"oauth2": "new_token"})
|
|
mock_resume_login.return_value = new_tokens
|
|
|
|
# Call handle_mfa
|
|
result = garmin_client_instance.handle_mfa(mock_db_session, "123456")
|
|
|
|
# Assertions
|
|
call_args, _ = mock_resume_login.call_args
|
|
assert call_args[1] == "123456"
|
|
assert call_args[0]["client"] is mock_client_instance
|
|
|
|
assert result is True
|
|
mock_update_tokens.assert_called_once_with(mock_db_session, new_tokens[0], new_tokens[1])
|
|
|
|
|
|
@patch("src.services.garmin.auth.garth.client.resume_login")
|
|
@patch("garth.http.Client")
|
|
@patch.object(GarminClient, 'update_tokens')
|
|
def test_handle_mfa_failure(mock_update_tokens, mock_client_class, mock_resume_login, mock_db_session, garmin_client_instance):
|
|
"""Test MFA completion failure due to GarthException."""
|
|
# Setup mock MFA state in DB
|
|
mfa_state_data = {
|
|
"cookies": {"cookie1": "val1"},
|
|
"domain": "garmin.com",
|
|
"last_resp_text": "<html></html>",
|
|
"last_resp_url": "http://url",
|
|
"signin_params": {"_csrf": "token"}
|
|
}
|
|
mock_token_record = MagicMock(spec=APIToken)
|
|
mock_token_record.mfa_state = json.dumps(mfa_state_data)
|
|
mock_db_session.query.return_value.filter_by.return_value.first.return_value = mock_token_record
|
|
|
|
# Mock instance
|
|
mock_client_instance = MagicMock()
|
|
mock_client_instance.domain = mfa_state_data["domain"]
|
|
mock_client_instance.sess = MagicMock()
|
|
mock_client_instance.sess.cookies = MagicMock()
|
|
mock_client_instance.sess.cookies.update = MagicMock()
|
|
mock_client_class.return_value = mock_client_instance
|
|
|
|
# Mock garth.resume_login to raise GarthException
|
|
mock_resume_login.side_effect = GarthException("Invalid MFA code")
|
|
|
|
# Call handle_mfa and expect an exception
|
|
with pytest.raises(GarthException, match="Invalid MFA code"):
|
|
garmin_client_instance.handle_mfa(mock_db_session, "wrongcode")
|
|
|
|
mock_client_class.assert_called_once_with(domain=mfa_state_data["domain"])
|
|
mock_client_instance.sess.cookies.update.assert_called_once_with(mfa_state_data["cookies"])
|
|
mock_resume_login.assert_called_once()
|
|
mock_update_tokens.assert_not_called()
|
|
mock_db_session.commit.assert_not_called()
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class MockToken:
|
|
token: str
|
|
secret: str
|
|
|
|
def test_update_tokens_serialization(mock_db_session, garmin_client_instance):
|
|
"""Test that update_tokens correctly serializes dataclasses to dicts."""
|
|
|
|
# Create fake tokens as dataclasses (simulating Garth tokens)
|
|
token1 = MockToken(token="foo", secret="bar")
|
|
token2 = MockToken(token="baz", secret="qux")
|
|
|
|
# Call update_tokens
|
|
garmin_client_instance.update_tokens(mock_db_session, token1, token2)
|
|
|
|
# Check that db.add was called
|
|
assert mock_db_session.add.called
|
|
added_token = mock_db_session.add.call_args[0][0]
|
|
|
|
# Verify that what was stored in the APIToken object is a JSON string of a DICT
|
|
assert isinstance(added_token.garth_oauth1_token, str)
|
|
|
|
stored_json1 = json.loads(added_token.garth_oauth1_token)
|
|
assert stored_json1 == {"token": "foo", "secret": "bar"}
|
|
|
|
stored_json2 = json.loads(added_token.garth_oauth2_token)
|
|
assert stored_json2 == {"token": "baz", "secret": "qux"}
|