bckelley/tconnectsync
1
0
Fork 0
mirror of https://github.com/bckelley/tconnectsync.git synced 2026-08-24 03:34:12 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix hardcoded US region default; add EU region integration tests (#152)

Browse Source 
TConnectApi and TandemSourceApi defaulted their region parameter to a
hardcoded 'US', so callers that construct them without an explicit
region (e.g. tconnectsync-heroku's check_login path) sent EU accounts
to the US login endpoint, which rejected them with HTTP 401
account/invalid_credentials even when TCONNECT_REGION=EU was set.

Both classes (and fetch_oneshot) now fall back to the configured
TCONNECT_REGION when no region argument is given, so downstream
consumers honor the .env/environment configuration.

Adds top-level integration tests that configure the EU region via each
supported mechanism (TCONNECT_REGION environment variable, .env file,
and --region CLI flag) and drive the real downstream code -- including
the full main(['--check-login']) entrypoint -- against a mocked HTTP
layer with a real RS256-signed OIDC id_token. Only the EU endpoints are
registered, and the tests assert the EU login/token/jwks/pumper/
pump-logs endpoints are actually invoked and that no US host is ever
contacted.

Fixes #152

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012DEvvZSWHo2dki5h1HikUU
This commit is contained in:
Claude
2026-07-17 19:13:07 -04:00
committed by James Woglom
parent 7f88d88ea4
commit 4655e31561
4 changed files with 365 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/api/__init__.py
+6 -2
View File
@@ -1,6 +1,7 @@
import logging
from .tandemsource import TandemSourceApi
from .. import secret
logger = logging.getLogger(__name__)
@@ -9,10 +10,13 @@ class TConnectApi:
email = None
password = None
def __init__(self, email, password, region='US'):
def __init__(self, email, password, region=None):
self.email = email
self.password = password
self.region = region
# A caller which does not pass a region (e.g. tconnectsync-heroku)
# must get the configured TCONNECT_REGION, not a hardcoded US
# default which would send EU accounts to the US endpoints (#152).
self.region = region or secret.TCONNECT_REGION
self._tandemsource = None
@property
tconnectsync/api/tandemsource.py
+6 -1
View File
@@ -24,6 +24,7 @@ from jwt.algorithms import RSAAlgorithm
from ..util import timeago, cap_length
from .common import parse_ymd_date, base_headers, base_session, ApiException, ApiLoginException
from .. import secret
from ..secret import CACHE_CREDENTIALS, CACHE_CREDENTIALS_PATH, TIMEZONE_NAME
from ..eventparser.generic import Events
@@ -214,7 +215,11 @@ class TandemSourceApi:
'AUTHORIZATION_ENDPOINT': 'https://tdcservices.eu.tandemdiabetes.com/accounts/api/connect/authorize'
}
def __init__(self, email: str, password: str, region: str = 'US') -> None:
def __init__(self, email: str, password: str, region: Optional[str] = None) -> None:
# No region means "use the configured TCONNECT_REGION": a hardcoded
# US default would send EU accounts to the US endpoints (#152).
if not region:
region = secret.TCONNECT_REGION
self.region = region.upper()
if self.region not in ['US', 'EU']:
raise ValueError(f"Invalid region '{region}'. Must be 'US' or 'EU'.")
tconnectsync/sync/tandemsource/cli_helpers.py
+1 -1
View File
@@ -10,7 +10,7 @@ import logging
logger = logging.getLogger(__name__)
def fetch_oneshot(username, password, time_start=None, time_end=None, region='US'):
def fetch_oneshot(username, password, time_start=None, time_end=None, region=None):
tconnect = TConnectApi(username, password, region)
if not time_start and not time_end:
time_end = datetime.datetime.now()
tests/test_eu_region_integration.py
+352
View File
@@ -0,0 +1,352 @@
#!/usr/bin/env python3
"""Top-level EU region integration tests (issue #152).
The v3.0.0 regression: TConnectApi's region parameter defaulted to a
hardcoded 'US', so any caller that constructed it without an explicit
region (e.g. tconnectsync-heroku's check_login path) sent EU accounts to
the US login endpoint, which rejected them with HTTP 401
account/invalid_credentials.
These tests configure the EU region through each supported mechanism --
the TCONNECT_REGION environment variable, a .env file, and the --region
CLI flag -- and then run the real downstream code (TConnectApi ->
TandemSourceApi login -> Tandem Source API calls, including the full
`main(['--check-login'])` entrypoint) against a mocked HTTP layer. Only
the EU endpoints are registered with requests_mock, so any request to a
US endpoint fails the test immediately; the tests additionally assert
that every contacted host is an expected EU (or region-shared) host and
that the specific EU login/token/jwks/data endpoints were invoked.
"""
import contextlib
import io
import json
import os
import shutil
import sys
import tempfile
import time
import unittest
import urllib.parse
import jwt
import requests_mock
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from jwt.algorithms import RSAAlgorithm
TEST_KID = 'eu-integration-test-key'
PUMPER_ID = 'aaaaaaaa-1111-2222-3333-444444444444'
ACCOUNT_ID = 'bbbbbbbb-5555-6666-7777-888888888888'
DEVICE_ID = '1b493210-9336-4901-a329-a352775738c5'
EU_API = 'https://tdcservices.eu.tandemdiabetes.com/accounts/api'
EU_SOURCE = 'https://source.eu.tandemdiabetes.com/'
EU_CLIENT_ID = '1519e414-eeec-492e-8c5e-97bea4815a10'
EU_LOGIN_URL = EU_API + '/login'
EU_TOKEN_URL = EU_API + '/connect/token'
EU_AUTHORIZE_URL = EU_API + '/connect/authorize'
EU_JWKS_URL = EU_API + '/.well-known/openid-configuration/jwks'
EU_CALLBACK_URL = EU_SOURCE + 'authorize/callback'
EU_PUMPER_URL = EU_SOURCE + 'api/reports/bff/pumper/' + PUMPER_ID
EU_PUMP_LOGS_URL = EU_SOURCE + 'api/reports/bff/pump-logs/' + DEVICE_ID
NS_URL = 'http://nightscout.example.com/'
# The US hosts that must never be contacted when the EU region is configured.
US_HOSTS = {'tdcservices.tandemdiabetes.com', 'source.tandemdiabetes.com'}
# sso.tandemdiabetes.com hosts the login page for both regions; every other
# host must be region-specific.
ALLOWED_HOSTS = {
'sso.tandemdiabetes.com',
'tdcservices.eu.tandemdiabetes.com',
'source.eu.tandemdiabetes.com',
'nightscout.example.com',
}
TEST_EMAIL = 'eu-user@example.com'
TEST_PASSWORD = 'eu-password'
# Signing key for the fake OIDC id_token, generated once for the module.
_PRIVATE_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
_PRIVATE_PEM = _PRIVATE_KEY.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
def make_id_token():
"""A real RS256-signed id_token carrying the EU issuer/audience, so
TandemSourceApi.extract_jwt() performs full signature, issuer, and
audience verification against the mocked EU jwks endpoint."""
now = int(time.time())
claims = {
'iss': EU_API,
'aud': EU_CLIENT_ID,
'iat': now,
'nbf': now,
'exp': now + 3600,
'sub': 'eu-test-subject',
'pumperId': PUMPER_ID,
'accountId': ACCOUNT_ID,
}
return jwt.encode(claims, _PRIVATE_PEM, algorithm='RS256', headers={'kid': TEST_KID})
def make_jwks():
jwk = json.loads(RSAAlgorithm.to_jwk(_PRIVATE_KEY.public_key()))
jwk.update({'kid': TEST_KID, 'use': 'sig', 'alg': 'RS256'})
return {'keys': [jwk]}
BFF_PUMPER = {
'firstName': 'Eu',
'lastName': 'User',
'name': 'Eu User',
'country': 'DE',
'pumps': [
{
'algorithm': 'Control-IQ',
'availableDataRange': {'start': '2026-01-01T00:00:00', 'end': '2026-07-16T10:00:00'},
'assignmentId': DEVICE_ID,
'lastUploadDate': '2026-07-16T10:00:00Z',
'maxDateOfEvents': '2026-07-16T10:00:00',
'modelNumber': '1000354',
'modelName': 't:slim X2™ Insulin Pump',
'partNumber': '1011979',
'serialNumber': '90556643',
'softwareVersion': '7.8.0.0',
'lastUploadClientType': 'mobile_tconnect',
'settings': None,
}
],
}
PUMP_LOGS = {'events': [], 'clockChanges': []}
def register_eu_endpoints(m):
"""Register ONLY the EU (and region-shared) endpoints needed for the
full login flow and the Tandem Source data calls. No US endpoint is
registered, so any US request raises requests_mock.NoMockAddress."""
m.get('https://sso.tandemdiabetes.com/', text='')
m.post(EU_LOGIN_URL, json={'redirectUrl': '/', 'status': 'SUCCESS'})
m.get(EU_AUTHORIZE_URL, status_code=302,
headers={'Location': EU_CALLBACK_URL + '?code=eu-test-code'})
m.get(EU_CALLBACK_URL, text='')
m.post(EU_TOKEN_URL, json={
'access_token': 'eu-access-token',
'id_token': make_id_token(),
'expires_in': 3600,
})
m.get(EU_JWKS_URL, json=make_jwks())
m.get(EU_PUMPER_URL, json=BFF_PUMPER)
m.get(EU_PUMP_LOGS_URL, json=PUMP_LOGS)
def register_nightscout_endpoints(m):
m.get(NS_URL + 'api/v1/status.json', json={'status': 'ok', 'version': '15.0.3'})
m.get(NS_URL + 'api/v1/treatments', json=[])
class EuRegionTestBase(unittest.TestCase):
"""Runs each test in a scratch cwd with a controlled environment, and
re-imports tconnectsync so that module-level configuration (secret.py)
is loaded from that environment."""
maxDiff = None
ENV_KEYS = [
'TCONNECT_EMAIL', 'TCONNECT_PASSWORD', 'TCONNECT_REGION',
'CACHE_CREDENTIALS', 'NS_URL', 'NS_SECRET', 'API_SECRET',
'TIMEZONE_NAME', 'TZ', 'PUMP_SERIAL_NUMBER', 'REQUESTS_PROXY',
]
BASE_ENV = {
'TCONNECT_EMAIL': TEST_EMAIL,
'TCONNECT_PASSWORD': TEST_PASSWORD,
'CACHE_CREDENTIALS': 'false',
'NS_URL': NS_URL,
'NS_SECRET': 'ns-secret',
'TIMEZONE_NAME': 'Europe/Berlin',
}
def setUp(self):
self._saved_env = {k: os.environ.get(k) for k in self.ENV_KEYS}
for k in self.ENV_KEYS:
os.environ.pop(k, None)
self._old_cwd = os.getcwd()
self._tmpdir = tempfile.mkdtemp(prefix='tconnectsync-eu-test-')
os.chdir(self._tmpdir)
self._purge_modules()
def tearDown(self):
os.chdir(self._old_cwd)
shutil.rmtree(self._tmpdir, ignore_errors=True)
for k, v in self._saved_env.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
# Drop the EU-configured module instances so later tests re-import
# with their own environment.
self._purge_modules()
@staticmethod
def _purge_modules():
for name in list(sys.modules):
if name == 'tconnectsync' or name.startswith('tconnectsync.'):
del sys.modules[name]
def import_tconnectsync(self, extra_env=None):
env = dict(self.BASE_ENV)
env.update(extra_env or {})
os.environ.update(env)
self._purge_modules()
import tconnectsync
return tconnectsync
# ---- assertion helpers ----
def called(self, m):
"""(method, url-without-query) pairs for every mocked request."""
return [(r.method, r.url.split('?')[0]) for r in m.request_history]
def assert_only_eu_hosts(self, m):
hosts = {urllib.parse.urlparse(r.url).netloc.lower() for r in m.request_history}
self.assertTrue(hosts, 'expected at least one HTTP request')
self.assertFalse(hosts & US_HOSTS,
'US endpoints were contacted with EU region configured: %s' % (hosts & US_HOSTS))
self.assertLessEqual(hosts, ALLOWED_HOSTS,
'unexpected hosts contacted: %s' % (hosts - ALLOWED_HOSTS))
def assert_eu_login_flow(self, m):
calls = self.called(m)
self.assertIn(('POST', EU_LOGIN_URL), calls)
self.assertIn(('GET', EU_AUTHORIZE_URL), calls)
self.assertIn(('POST', EU_TOKEN_URL), calls)
self.assertIn(('GET', EU_JWKS_URL), calls)
self.assert_only_eu_hosts(m)
class TestEuRegionFromEnvironmentVariable(EuRegionTestBase):
"""TCONNECT_REGION=EU set as an environment variable."""
def test_tconnect_api_without_region_argument_uses_eu(self):
# The exact downstream pattern that regressed in #152: the caller
# builds TConnectApi without a region argument, relying on the
# configured TCONNECT_REGION.
self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
from tconnectsync.api import TConnectApi
with requests_mock.Mocker() as m:
register_eu_endpoints(m)
api = TConnectApi(TEST_EMAIL, TEST_PASSWORD)
self.assertEqual(api.region, 'EU')
tandemsource = api.tandemsource
self.assertEqual(tandemsource.region, 'EU')
self.assertEqual(tandemsource.LOGIN_API_URL, EU_LOGIN_URL)
self.assertEqual(tandemsource.SOURCE_URL, EU_SOURCE)
self.assertEqual(tandemsource.pumperId, PUMPER_ID)
self.assertEqual(tandemsource.accountId, ACCOUNT_ID)
pumper = tandemsource.get_pumper()
self.assertEqual(pumper['pumps'][0]['assignmentId'], DEVICE_ID)
self.assert_eu_login_flow(m)
self.assertIn(('GET', EU_PUMPER_URL), self.called(m))
def test_tandem_source_api_without_region_argument_uses_eu(self):
# TandemSourceApi itself must also fall back to the configured
# region when constructed directly without one.
self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
from tconnectsync.api.tandemsource import TandemSourceApi
with requests_mock.Mocker() as m:
register_eu_endpoints(m)
api = TandemSourceApi(TEST_EMAIL, TEST_PASSWORD)
self.assertEqual(api.region, 'EU')
self.assertEqual(api.pumperId, PUMPER_ID)
self.assert_eu_login_flow(m)
def test_secret_exposes_eu_region(self):
tconnectsync = self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
self.assertEqual(tconnectsync.secret.TCONNECT_REGION, 'EU')
class TestEuRegionFromDotEnvFile(EuRegionTestBase):
"""TCONNECT_REGION=EU set through a .env file in the working directory."""
def test_tconnect_api_without_region_argument_uses_eu(self):
# Configure everything through .env instead of process environment
# variables (secret.py reads $CWD/.env, and setUp chdir'd us into a
# scratch directory).
with open(os.path.join(self._tmpdir, '.env'), 'w') as f:
for k, v in dict(self.BASE_ENV, TCONNECT_REGION='EU').items():
f.write('%s=%s\n' % (k, v))
os.environ.update({'CACHE_CREDENTIALS': 'false'})
self._purge_modules()
import tconnectsync # noqa: F401
from tconnectsync import secret
from tconnectsync.api import TConnectApi
self.assertEqual(secret.TCONNECT_REGION, 'EU')
with requests_mock.Mocker() as m:
register_eu_endpoints(m)
api = TConnectApi(secret.TCONNECT_EMAIL, secret.TCONNECT_PASSWORD)
self.assertEqual(api.region, 'EU')
self.assertEqual(api.tandemsource.LOGIN_API_URL, EU_LOGIN_URL)
self.assert_eu_login_flow(m)
class TestEuRegionThroughMainEntrypoint(EuRegionTestBase):
"""Full `tconnectsync --check-login` runs through main(), covering the
complete downstream path: secret loading -> main() region resolution ->
TConnectApi -> TandemSourceApi login -> get_pumper/ChooseDevice ->
pump_events -> Nightscout."""
def run_check_login(self, tconnectsync, argv):
with requests_mock.Mocker() as m:
register_eu_endpoints(m)
register_nightscout_endpoints(m)
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
tconnectsync.main(argv)
return m, stdout.getvalue()
def assert_full_eu_check_login(self, m, output):
self.assert_eu_login_flow(m)
calls = self.called(m)
self.assertIn(('GET', EU_PUMPER_URL), calls)
self.assertIn(('GET', EU_PUMP_LOGS_URL), calls)
# check_login must complete without any API errors: the EU login
# and every downstream Tandem Source + Nightscout call succeeded.
self.assertIn('No API errors returned!', output)
self.assertNotIn('API errors occurred', output)
def test_check_login_with_region_from_environment_variable(self):
tconnectsync = self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
m, output = self.run_check_login(tconnectsync, ['--check-login'])
self.assertIn("TCONNECT_REGION='EU'", output)
self.assert_full_eu_check_login(m, output)
def test_check_login_with_region_from_cli_flag(self):
# No TCONNECT_REGION configured (config default is US): the
# --region EU CLI flag alone must route everything to the EU
# endpoints.
tconnectsync = self.import_tconnectsync()
self.assertEqual(tconnectsync.secret.TCONNECT_REGION, 'US')
m, output = self.run_check_login(tconnectsync, ['--check-login', '--region', 'EU'])
self.assert_full_eu_check_login(m, output)
if __name__ == '__main__':
unittest.main()