starred/tconnectsync
1
0
Fork 0
mirror of https://github.com/jwoglom/tconnectsync.git synced 2026-08-24 10:14:36 -05:00
Code Issues Packages Projects Releases Wiki Activity

Remove PumpMetadata transform; callers use raw BffPump, normalize at call sites

Browse Source 
Delete the PumpMetadata TypedDict, _bff_pump_to_metadata and pump_metadata
transform layer. Callers now consume the raw BffPump dicts from get_pumper()
directly. The pump-local -> UTC date conversion is kept as a shared
naive_local_to_utc() helper, applied only at the call sites that compare a
pump date against real UTC (choose_device staleness/selection, autoupdate
timing).

Also expand pump_events JSON parse coverage: drive bolus (20), basal (279),
CGM (399) and alarm (5) events through pump_events(), asserting decoded
fields and enum members (previously only eventCode 16 was covered).
This commit is contained in:
James Woglom
2026-07-01 04:22:57 +00:00
parent 46a5a28baf
commit ddca912eea
12 changed files with 182 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/api/tandemsource.py
+33 -91
View File
@@ -30,6 +30,36 @@ from ..eventparser.generic import Events
logger = logging.getLogger(__name__)
def naive_local_to_utc(value: Optional[str]) -> Optional[str]:
"""Normalize a BFF pump-local naive wall-clock timestamp to a true UTC
ISO-8601 string.
The BFF sends maxDateOfEvents / availableDataRange.start with no tz
(e.g. "2022-02-16T22:45:58") even though they are the pump's local
wall-clock time. Downstream consumers parse them with arrow.get(...),
which assumes UTC, and compare against arrow.utcnow() / time.time()
(real UTC), so we shift them here by interpreting the naive value in the
configured TIMEZONE_NAME and converting to UTC. Values that already
carry a tz (defensive; not seen for these two fields) are passed
through unchanged so we never double-shift. None passes through as None
(never-uploaded pumps).
"""
if not value:
return value
# If the string already carries a tz (a trailing 'Z' or a +HH:MM /
# -HH:MM offset after the time portion), trust it and never double-shift.
# Otherwise it's a naive pump-local wall-clock value: interpret it in the
# configured TIMEZONE_NAME. (Per the BFF data these two fields are always
# naive; the has-tz branch is purely defensive.)
time_part = value.split('T', 1)[-1]
has_tz = value.endswith('Z') or '+' in time_part or '-' in time_part
if has_tz:
parsed = arrow.get(value)
else:
parsed = arrow.get(value, tzinfo=TIMEZONE_NAME)
return parsed.to('UTC').isoformat()
class JwtClaims(TypedDict, total=False):
"""Decoded OIDC id_token claims stored on TandemSourceApi.jwtData.
@@ -126,35 +156,6 @@ class BffPumper(TypedDict, total=False):
pumps: List[BffPump]
class PumpMetadata(TypedDict, total=False):
"""Normalized per-pump metadata the sync code consumes, adapted from a
BffPump (see TandemSourceApi.pump_metadata()). This is the stable
replacement for the old reportsfacade pump-event-metadata shape.
deviceId is the UUID assignmentId used as the pump-logs path segment (it
replaces the old numeric tconnectDeviceId). settings is the pump settings
blob (BffPump.settings.details) or None for a pump that has never uploaded.
Date fields are ISO-8601 strings or None. maxDateWithEvents and
minDateWithEvents are normalized to true UTC here (previously they were the
pump-local naive wall-clock strings carried verbatim from BffPump's
maxDateOfEvents / availableDataRange.start): the BFF sends those two fields
without a tz, so they are interpreted in the configured TIMEZONE_NAME and
converted to UTC so consumers can compare them against arrow.utcnow() /
time.time(). (BffPump.lastUploadDate, by contrast, already carries a 'Z'
and is true UTC.)
"""
deviceId: str
serialNumber: str
modelNumber: str
modelName: str
softwareVersion: str
algorithm: Optional[str]
maxDateWithEvents: Optional[str]
minDateWithEvents: Optional[str]
settings: Optional[dict]
class PumpLogEvent(TypedDict):
"""One entry in a PumpLogsResponse (events[] or clockChanges[]) from
GET api/reports/bff/pump-logs/{deviceAssignmentId}. The server pre-decodes
@@ -594,66 +595,6 @@ class TandemSourceApi:
pumps[].settings.details carries the pump settings blob."""
return self.get('api/reports/bff/pumper/%s' % (self.pumperId), {})
@staticmethod
def _naive_local_to_utc(value: Optional[str]) -> Optional[str]:
"""Normalize a BFF pump-local naive wall-clock timestamp to a true UTC
ISO-8601 string.
The BFF sends maxDateOfEvents / availableDataRange.start with no tz
(e.g. "2022-02-16T22:45:58") even though they are the pump's local
wall-clock time. Downstream consumers parse them with arrow.get(...),
which assumes UTC, and compare against arrow.utcnow() / time.time()
(real UTC), so we shift them here by interpreting the naive value in the
configured TIMEZONE_NAME and converting to UTC. Values that already
carry a tz (defensive; not seen for these two fields) are passed
through unchanged so we never double-shift. None passes through as None
(never-uploaded pumps).
"""
if not value:
return value
# If the string already carries a tz (a trailing 'Z' or a +HH:MM /
# -HH:MM offset after the time portion), trust it and never double-shift.
# Otherwise it's a naive pump-local wall-clock value: interpret it in the
# configured TIMEZONE_NAME. (Per the BFF data these two fields are always
# naive; the has-tz branch is purely defensive.)
time_part = value.split('T', 1)[-1]
has_tz = value.endswith('Z') or '+' in time_part or '-' in time_part
if has_tz:
parsed = arrow.get(value)
else:
parsed = arrow.get(value, tzinfo=TIMEZONE_NAME)
return parsed.to('UTC').isoformat()
@staticmethod
def _bff_pump_to_metadata(pump: BffPump) -> PumpMetadata:
"""Adapt one BffPump into the normalized PumpMetadata shape.
maxDateOfEvents and availableDataRange.start are pump-local naive
wall-clock strings; we normalize them to true UTC (via
_naive_local_to_utc) so consumers that compare against arrow.utcnow() /
time.time() are correct.
"""
settings = pump.get('settings')
data_range = pump.get('availableDataRange') or {}
meta: PumpMetadata = {
'deviceId': pump['assignmentId'],
'serialNumber': pump['serialNumber'],
'modelNumber': pump['modelNumber'],
'modelName': pump['modelName'],
'softwareVersion': pump['softwareVersion'],
'algorithm': pump.get('algorithm'),
'maxDateWithEvents': TandemSourceApi._naive_local_to_utc(pump.get('maxDateOfEvents')),
'minDateWithEvents': TandemSourceApi._naive_local_to_utc(data_range.get('start')),
'settings': settings['details'] if settings else None,
}
return meta
def pump_metadata(self) -> List[PumpMetadata]:
"""Normalized device list adapted from get_pumper(). This is the
BFF-backed replacement for the old reportsfacade pump-event-metadata."""
pumper = self.get_pumper()
return [self._bff_pump_to_metadata(p) for p in pumper.get('pumps', [])]
# Matches the Tandem Source web app's getLogIDList() (55 IDs) as observed in
# the live GET api/reports/bff/pump-logs request. Includes FSL3 ids 477/480/486.
DEFAULT_EVENT_IDS: List[int] = [229,5,28,4,26,99,279,3,16,59,21,55,20,280,64,65,66,61,33,371,171,369,460,172,370,461,372,480,399,256,213,406,477,394,212,404,214,405,486,447,313,60,14,6,90,230,140,12,11,53,13,63,203,307,191]
@@ -661,7 +602,8 @@ class TandemSourceApi:
def get_pump_logs(self, device_id: str, min_date: Optional[str] = None, max_date: Optional[str] = None, event_ids_filter: Optional[List[int]] = DEFAULT_EVENT_IDS) -> PumpLogsResponse:
"""Fetch pre-decoded pump events for a single date window from the BFF
endpoint GET api/reports/bff/pump-logs/{device_id}. device_id is the
UUID assignmentId (PumpMetadata.deviceId). Returns {events, clockChanges}.
UUID assignmentId (BffPump.assignmentId from get_pumper()). Returns
{events, clockChanges}.
The server caps the window at ~4 weeks; callers needing a longer range
must page by date window (see pump_events).
@@ -707,7 +649,7 @@ class TandemSourceApi:
Fetch and parse pump events from the pump-logs endpoint.
Default of fetch_all_event_types=False will filter to the same event ids used in the Tandem Source backend.
If fetch_all_event_types=True, then all event types from the history log will be returned.
tconnect_device_id is the UUID assignmentId from pump_metadata() (deviceId).
tconnect_device_id is the UUID assignmentId from get_pumper() pumps (BffPump.assignmentId).
"""
def pump_events(self, tconnect_device_id: str, min_date: Optional[str] = None, max_date: Optional[str] = None, fetch_all_event_types: bool = False) -> Iterator:
event_ids_filter = None if fetch_all_event_types else self.DEFAULT_EVENT_IDS
tconnectsync/check.py
+3 -3
View File
@@ -90,7 +90,7 @@ def check_login(tconnect, time_start, time_end, verbose=False, sanitize=True):
serialNumberToPump = None
try:
log("Fetching pump metadata...")
pumpEventMetadata = tconnect.tandemsource.pump_metadata()
pumpEventMetadata = tconnect.tandemsource.get_pumper().get('pumps', [])
serialNumberToPump = {p['serialNumber']: p for p in pumpEventMetadata}
log(f'Found {len(serialNumberToPump)} pumps: {serialNumberToPump.keys()}')
@@ -102,7 +102,7 @@ def check_login(tconnect, time_start, time_end, verbose=False, sanitize=True):
log(f'ChooseDevice selected: {tconnectDevice}')
deviceId = tconnectDevice['deviceId']
deviceId = tconnectDevice['assignmentId']
log(f'Fetching pump events for {deviceId=} {time_start=} {time_end=} fetch_all_event_types=False')
@@ -187,7 +187,7 @@ def check_login(tconnect, time_start, time_end, verbose=False, sanitize=True):
if serialNumberToPump:
for i, (pumpSerial, pumpDetails) in enumerate(serialNumberToPump.items()):
sanitizedData[f'PUMP_SERIAL_{i}'] = pumpSerial
sanitizedData[f'TCONNECT_DEVICE_ID_{i}'] = pumpDetails['deviceId']
sanitizedData[f'TCONNECT_DEVICE_ID_{i}'] = pumpDetails['assignmentId']
loglines = [run_sanitize(i, sanitizedData) for i in loglines]
f.writelines(loglines)
tconnectsync/domain/tandemsource/pump_settings.py
+1 -1
View File
@@ -3,7 +3,7 @@ from dataclasses_json import dataclass_json
from typing import List
# These dataclasses model the `settings.details` blob from the Tandem Source
# bff/pumper endpoint (surfaced as PumpMetadata.settings). Only the fields the
# bff/pumper endpoint (BffPump.settings.details). Only the fields the
# profile sync consumes are declared; dataclasses_json ignores the rest.
@dataclass_json
tconnectsync/sync/tandemsource/autoupdate.py
+2 -1
View File
@@ -5,6 +5,7 @@ import sys
import arrow
from ...features import DEFAULT_FEATURES
from ...api.tandemsource import naive_local_to_utc
from .process import ProcessTimeRange
from .choose_device import ChooseDevice
@@ -47,7 +48,7 @@ class TandemSourceAutoupdate:
tconnectDevice = ChooseDevice(self.secret, tconnect).choose()
event_seqnum = None
cur_max_date_with_events = arrow.get(tconnectDevice['maxDateWithEvents']).float_timestamp
cur_max_date_with_events = arrow.get(naive_local_to_utc(tconnectDevice['maxDateOfEvents'])).float_timestamp
if not self.last_max_date_with_events or cur_max_date_with_events > self.last_max_date_with_events:
logger.info('New reported tandemsource data. (cur_max_date: %s last_max_date: %s)' % (cur_max_date_with_events, self.last_max_date_with_events))
tconnectsync/sync/tandemsource/choose_device.py
+11 -9
View File
@@ -1,6 +1,8 @@
import arrow
import logging
from ...api.tandemsource import naive_local_to_utc
logger = logging.getLogger(__name__)
class ChooseDevice:
@@ -11,7 +13,7 @@ class ChooseDevice:
def choose(self):
tconnect = self.tconnect
pumpEventMetadata = tconnect.tandemsource.pump_metadata()
pumpEventMetadata = tconnect.tandemsource.get_pumper().get('pumps', [])
if not pumpEventMetadata:
raise NoDevicesFound('No pumps are present on your Tandem Source account')
@@ -29,28 +31,28 @@ class ChooseDevice:
# Warn if pump is stale (no events in >3 days)
try:
max_event_date = arrow.get(tconnectDevice["maxDateWithEvents"])
max_event_date = arrow.get(naive_local_to_utc(tconnectDevice["maxDateOfEvents"]))
age_days = (arrow.utcnow() - max_event_date).days
if age_days > 3:
logger.warning(
f"The selected pump (serial {tconnectDevice['serialNumber']}) has no events in the last {age_days} days "
f"(last seen: {tconnectDevice['maxDateWithEvents']}). "
f"(last seen: {tconnectDevice['maxDateOfEvents']}). "
"You may have switched to a new pump. Consider removing or updating PUMP_SERIAL_NUMBER in your config."
)
except Exception as e:
logger.debug(f"Could not parse maxDateWithEvents to check for staleness: {e}")
logger.debug(f"Could not parse maxDateOfEvents to check for staleness: {e}")
logger.info(f'Using pump with serial: {tconnectDevice["serialNumber"]} (deviceId: {tconnectDevice["deviceId"]}, last seen: {tconnectDevice["maxDateWithEvents"]})')
logger.info(f'Using pump with serial: {tconnectDevice["serialNumber"]} (deviceId: {tconnectDevice["assignmentId"]}, last seen: {tconnectDevice["maxDateOfEvents"]})')
else:
# The BFF device list includes pumps that have never uploaded
# (maxDateWithEvents is None); skip those when picking the most
# (maxDateOfEvents is None); skip those when picking the most
# recent one, and only fall back to one of them if nothing else.
maxDateSeen = None
for pump in pumpEventMetadata:
if not pump.get('maxDateWithEvents'):
if not pump.get('maxDateOfEvents'):
continue
pumpMaxDate = arrow.get(pump['maxDateWithEvents'])
pumpMaxDate = arrow.get(naive_local_to_utc(pump['maxDateOfEvents']))
if not tconnectDevice or pumpMaxDate > maxDateSeen:
maxDateSeen = pumpMaxDate
tconnectDevice = pump
@@ -59,7 +61,7 @@ class ChooseDevice:
if not tconnectDevice:
tconnectDevice = pumpEventMetadata[0]
logger.info(f'Using most recent pump (serial: {tconnectDevice["serialNumber"]}, deviceId: {tconnectDevice["deviceId"]}, last seen: {tconnectDevice["maxDateWithEvents"]})')
logger.info(f'Using most recent pump (serial: {tconnectDevice["serialNumber"]}, deviceId: {tconnectDevice["assignmentId"]}, last seen: {tconnectDevice["maxDateOfEvents"]})')
return tconnectDevice
tconnectsync/sync/tandemsource/cli_helpers.py
+1 -1
View File
@@ -17,4 +17,4 @@ def fetch_oneshot(username, password, time_start=None, time_end=None, region='US
time_start = time_end - datetime.timedelta(days=1)
tconnectDevice = ChooseDevice(secret, tconnect).choose()
return tconnect.tandemsource.pump_events(tconnectDevice['deviceId'], time_start, time_end, fetch_all_event_types=secret.FETCH_ALL_EVENT_TYPES)
return tconnect.tandemsource.pump_events(tconnectDevice['assignmentId'], time_start, time_end, fetch_all_event_types=secret.FETCH_ALL_EVENT_TYPES)
tconnectsync/sync/tandemsource/process.py
+4 -4
View File
@@ -7,7 +7,7 @@ from typing import List, Optional, Tuple, TYPE_CHECKING
if TYPE_CHECKING:
from ...api import TConnectApi
from ...nightscout import NightscoutApi
from ...api.tandemsource import PumpMetadata
from ...api.tandemsource import BffPump
from ...features import DEVICE_STATUS, DEFAULT_FEATURES
from ...eventparser import events as eventtypes
@@ -28,11 +28,11 @@ from .update_profiles import UpdateProfiles
logger = logging.getLogger(__name__)
class ProcessTimeRange:
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnectDevice: "PumpMetadata", pretend: bool, secret: ModuleType, features: List[str] = DEFAULT_FEATURES) -> None:
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnectDevice: "BffPump", pretend: bool, secret: ModuleType, features: List[str] = DEFAULT_FEATURES) -> None:
self.tconnect = tconnect
self.nightscout = nightscout
self.tconnect_device_id = tconnectDevice['deviceId']
self.max_date_with_events = tconnectDevice['maxDateWithEvents']
self.tconnect_device_id = tconnectDevice['assignmentId']
self.max_date_with_events = tconnectDevice.get('maxDateOfEvents')
self.pretend = pretend
self.secret = secret
self.features = features
tconnectsync/sync/tandemsource/update_profiles.py
+4 -3
View File
@@ -35,16 +35,17 @@ class UpdateProfiles:
upload_mode = _get_default_upload_mode()
logger.debug("UpdateProfiles: getting Tandem Source profile data")
all_metadata = self.tconnect.tandemsource.pump_metadata()
all_metadata = self.tconnect.tandemsource.get_pumper().get('pumps', [])
pump_meta = None
for m in all_metadata:
if m['deviceId'] == self.tconnect_device_id:
if m['assignmentId'] == self.tconnect_device_id:
pump_meta = m
if not pump_meta:
return False
raw_settings = pump_meta.get("settings")
s = pump_meta.get("settings")
raw_settings = s["details"] if s else None
if not raw_settings:
return False
tests/api/test_tandemsource.py
+101 -111
View File
@@ -6,7 +6,7 @@ import unittest
import urllib.parse
from unittest.mock import patch
from tconnectsync.api.tandemsource import TandemSourceApi
from tconnectsync.api.tandemsource import TandemSourceApi, naive_local_to_utc
from tconnectsync.api.common import ApiException
from tconnectsync.eventparser import events as eventtypes
@@ -63,132 +63,41 @@ BFF_PUMPER = {
}
class TestPumpMetadataAdapter(unittest.TestCase):
class TestNaiveLocalToUtc(unittest.TestCase):
"""The module-level naive_local_to_utc() helper is the sole survivor of the
removed PumpMetadata adapter. It normalizes a BFF pump-local naive
wall-clock timestamp to true UTC, and is now called at the specific date
call sites that compare against real UTC."""
maxDiff = None
def _api(self):
# Bypass __init__ (which performs a network login) to test the adapter.
return TandemSourceApi.__new__(TandemSourceApi)
def test_bff_pump_to_metadata_active_pump(self):
meta = TandemSourceApi._bff_pump_to_metadata(BFF_PUMPER["pumps"][0])
self.assertEqual(meta["deviceId"], "1b493210-9336-4901-a329-a352775738c5")
self.assertEqual(meta["serialNumber"], "90556643")
self.assertEqual(meta["modelNumber"], "1000354")
self.assertEqual(meta["softwareVersion"], "7.8.0.0")
self.assertEqual(meta["algorithm"], "Control-IQ")
# maxDateOfEvents -> maxDateWithEvents, normalized from pump-local naive
# (America/New_York, EST/UTC-5 in Feb) to true UTC.
self.assertEqual(meta["maxDateWithEvents"], "2022-02-17T03:45:58+00:00")
# availableDataRange.start -> minDateWithEvents, normalized from
# pump-local naive (America/New_York, EDT/UTC-4 in May) to true UTC.
self.assertEqual(meta["minDateWithEvents"], "2021-05-06T16:31:19+00:00")
# settings.details -> settings
self.assertEqual(meta["settings"], {"profiles": {"numberOfProfiles": 1}})
def test_bff_pump_to_metadata_never_uploaded_pump(self):
meta = TandemSourceApi._bff_pump_to_metadata(BFF_PUMPER["pumps"][1])
self.assertEqual(meta["deviceId"], "f6631fff-f403-4ce4-9362-83eff9e2850e")
self.assertEqual(meta["serialNumber"], "514387")
# null date/settings fields map to None, not KeyError
self.assertIsNone(meta["maxDateWithEvents"])
self.assertIsNone(meta["minDateWithEvents"])
self.assertIsNone(meta["settings"])
def test_pump_metadata_maps_all_pumps(self):
api = self._api()
with patch.object(TandemSourceApi, "get_pumper", return_value=BFF_PUMPER):
metas = api.pump_metadata()
self.assertEqual(len(metas), 2)
self.assertEqual(
[m["deviceId"] for m in metas],
[
"1b493210-9336-4901-a329-a352775738c5",
"f6631fff-f403-4ce4-9362-83eff9e2850e",
],
)
def test_pump_metadata_empty_when_no_pumps(self):
api = self._api()
with patch.object(TandemSourceApi, "get_pumper", return_value={}):
self.assertEqual(api.pump_metadata(), [])
def test_available_data_range_key_absent(self):
pump = dict(BFF_PUMPER["pumps"][0])
del pump["availableDataRange"]
meta = TandemSourceApi._bff_pump_to_metadata(pump)
self.assertIsNone(meta["minDateWithEvents"])
def test_settings_key_absent(self):
pump = dict(BFF_PUMPER["pumps"][0])
del pump["settings"]
meta = TandemSourceApi._bff_pump_to_metadata(pump)
self.assertIsNone(meta["settings"])
def test_missing_required_key_raises(self):
pump = dict(BFF_PUMPER["pumps"][0])
del pump["serialNumber"]
with self.assertRaises(KeyError):
TandemSourceApi._bff_pump_to_metadata(pump)
def test_other_required_keys_raise_when_absent(self):
# The 5 always-present BFF fields are still accessed with required
# subscript syntax; absence is a genuine error.
for key in ("assignmentId", "modelNumber", "modelName", "softwareVersion"):
pump = dict(BFF_PUMPER["pumps"][0])
del pump[key]
with self.assertRaises(KeyError, msg=key):
TandemSourceApi._bff_pump_to_metadata(pump)
def test_missing_algorithm_maps_to_none(self):
# algorithm is optional in the canonical BFF source; its absence must
# not raise (previously a KeyError) and should map to None.
pump = dict(BFF_PUMPER["pumps"][0])
del pump["algorithm"]
meta = TandemSourceApi._bff_pump_to_metadata(pump)
self.assertIsNone(meta["algorithm"])
def test_naive_dates_normalized_to_utc(self):
# America/New_York is set in tests/conftest.py. Feb -> EST (UTC-5),
# May -> EDT (UTC-4).
meta = TandemSourceApi._bff_pump_to_metadata(BFF_PUMPER["pumps"][0])
self.assertEqual(meta["maxDateWithEvents"], "2022-02-17T03:45:58+00:00")
self.assertEqual(meta["minDateWithEvents"], "2021-05-06T16:31:19+00:00")
# May -> EDT (UTC-4). These are the raw BffPump maxDateOfEvents /
# availableDataRange.start values that call sites now normalize.
self.assertEqual(
naive_local_to_utc(BFF_PUMPER["pumps"][0]["maxDateOfEvents"]),
"2022-02-17T03:45:58+00:00",
)
self.assertEqual(
naive_local_to_utc(BFF_PUMPER["pumps"][0]["availableDataRange"]["start"]),
"2021-05-06T16:31:19+00:00",
)
def test_naive_local_to_utc_none_passthrough(self):
self.assertIsNone(TandemSourceApi._naive_local_to_utc(None))
self.assertIsNone(naive_local_to_utc(None))
def test_naive_local_to_utc_idempotent_no_double_shift(self):
# A value that already carries a tz must not be shifted again. Feed the
# already-UTC output back in and confirm it is unchanged.
first = TandemSourceApi._naive_local_to_utc("2022-02-16T22:45:58")
first = naive_local_to_utc("2022-02-16T22:45:58")
self.assertEqual(first, "2022-02-17T03:45:58+00:00")
self.assertEqual(TandemSourceApi._naive_local_to_utc(first), first)
self.assertEqual(naive_local_to_utc(first), first)
# A 'Z'-suffixed (true UTC) value is passed through as UTC unchanged.
self.assertEqual(
TandemSourceApi._naive_local_to_utc("2022-09-20T05:50:12Z"),
naive_local_to_utc("2022-09-20T05:50:12Z"),
"2022-09-20T05:50:12+00:00",
)
def test_mobi_controliq_plus_passthrough(self):
pump = {
"algorithm": "Control-IQ+",
"availableDataRange": {"start": "2024-01-01T00:00:00", "end": "2026-05-27T23:03:06"},
"assignmentId": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
"maxDateOfEvents": "2026-05-27T23:03:06",
"modelNumber": "1004000",
"modelName": "Tandem Mobi™ System",
"partNumber": "1005000",
"serialNumber": "1518994",
"softwareVersion": "1.0.0.0",
"lastUploadClientType": "mobile_mobi",
"settings": None,
}
meta = TandemSourceApi._bff_pump_to_metadata(pump)
self.assertEqual(meta["algorithm"], "Control-IQ+")
self.assertEqual(meta["modelName"], "Tandem Mobi™ System")
self.assertEqual(meta["deviceId"], "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
class TestDefaultEventIds(unittest.TestCase):
def test_default_event_ids(self):
@@ -431,6 +340,87 @@ class TestPumpEvents(unittest.TestCase):
self.assertEqual(out, [])
class TestPumpEventsRealEventTypes(unittest.TestCase):
"""Parse bolus (20), basal (279), CGM (399) and alarm (5) events through
pump_events(). eventProperties use Tandem's real camelCase names; bitmask
fields arrive as arrays of set-bit indices."""
maxDiff = None
def _api(self):
api = TandemSourceApi.__new__(TandemSourceApi)
api.pumperId = "PUMPER123"
return api
RESPONSE = {
"events": [
_ev(0, 201, event_code=20, completionStatus=3, bolusId=777,
insulinDelivered=2.5, insulinRequested=2.5, iob=1.1),
_ev(0, 202, event_code=279, commandedRateSource=1, commandedRate=800,
profileBasalRate=800, algorithmRate=0, tempRate=0),
_ev(0, 203, event_code=399, glucoseValueStatus=0, cgmDataType=[0], rate=-5,
algorithmState=2, rssi=-60, currentGlucoseDisplayValue=112,
egvTimeStamp=123456, egvInfoBitmask=[], interval=5),
_ev(0, 204, event_code=5, alarmId=2, faultLocatorData=100, param1=1, param2=2.0),
],
"clockChanges": [],
}
def _parse(self):
api = self._api()
with patch.object(TandemSourceApi, "get_pump_logs", return_value=self.RESPONSE):
out = list(api.pump_events("dev", "2024-01-01", "2024-01-10"))
return {type(e).__name__: e for e in out}
def test_all_four_event_types_parse(self):
parsed = self._parse()
self.assertEqual(
set(parsed),
{"LidBolusCompleted", "LidBasalDelivery", "LidCgmDataG7", "LidAlarmActivated"},
)
def test_bolus_completed_decodes(self):
e = self._parse()["LidBolusCompleted"]
self.assertEqual(e.eventId, 20)
self.assertEqual(e.seqNum, 201)
self.assertEqual(e.bolusid, 777)
self.assertEqual(e.insulindelivered, 2.5)
self.assertEqual(e.insulinrequested, 2.5)
self.assertEqual(e.IOB, 1.1)
self.assertEqual(e.completionstatus,
eventtypes.LidBolusCompleted.CompletionstatusEnum.Completed)
def test_basal_delivery_decodes(self):
e = self._parse()["LidBasalDelivery"]
self.assertEqual(e.eventId, 279)
self.assertEqual(e.seqNum, 202)
self.assertEqual(e.commandedRate, 800)
self.assertEqual(e.profileBasalRate, 800)
self.assertEqual(e.commandedRateSource,
eventtypes.LidBasalDelivery.CommandedratesourceEnum.Profile)
def test_cgm_g7_decodes(self):
e = self._parse()["LidCgmDataG7"]
self.assertEqual(e.eventId, 399)
self.assertEqual(e.seqNum, 203)
self.assertEqual(e.currentglucosedisplayvalue, 112)
self.assertEqual(e.glucosevaluestatus,
eventtypes.LidCgmDataG7.GlucosevaluestatusEnum.PreciseValue)
# cgmDataType bitmask array [0] -> bit 0 set -> Fmr
self.assertEqual(e.cgmDataType,
eventtypes.LidCgmDataG7.CgmdatatypeBitmask.Fmr)
# rate is stored raw and scaled x0.1 by the property (-5 -> -0.5 mg/dL/min)
self.assertAlmostEqual(e.rate, -0.5)
def test_alarm_activated_decodes(self):
e = self._parse()["LidAlarmActivated"]
self.assertEqual(e.eventId, 5)
self.assertEqual(e.seqNum, 204)
self.assertEqual(e.faultlocatordata, 100)
self.assertEqual(e.param2, 2.0)
self.assertEqual(e.alarmid,
eventtypes.LidAlarmActivated.AlarmidEnum.OcclusionAlarm)
class TestGetRetry(unittest.TestCase):
"""get() retries once on 500, re-logs-in and retries once on 401, and
raises immediately on other statuses; after one retry it gives up."""
tests/sync/tandemsource/test_choose_device.py
+7 -6
View File
@@ -16,22 +16,23 @@ LOGGER = "tconnectsync.sync.tandemsource.choose_device"
class FakeTandemSourceApi:
"""Fake TandemSource API returning a configurable pump_metadata() list."""
"""Fake TandemSource API returning a configurable get_pumper() pumps list of
raw BffPump dicts."""
def __init__(self, pumps=None):
self._pumps = pumps if pumps is not None else []
def pump_metadata(self):
return self._pumps
def get_pumper(self):
return {'pumps': self._pumps}
def needs_relogin(self):
return False
def pump(serial, deviceId=None, maxDate=None):
def pump(serial, assignmentId=None, maxDate=None):
return {
'serialNumber': serial,
'deviceId': deviceId if deviceId is not None else ('dev-' + serial),
'maxDateWithEvents': maxDate,
'assignmentId': assignmentId if assignmentId is not None else ('dev-' + serial),
'maxDateOfEvents': maxDate,
}
tests/sync/tandemsource/test_process.py
+7 -7
View File
@@ -33,9 +33,9 @@ class FakeTandemSourceApi:
def pump_events(self, device_id, time_start, time_end, fetch_all_event_types=False):
return self.events
def pump_metadata(self):
"""Return empty metadata for testing"""
return []
def get_pumper(self):
"""Return empty pumper for testing"""
return {'pumps': []}
def needs_relogin(self):
return False
@@ -52,8 +52,8 @@ class TestProcessTimeRangeBasalDuration(unittest.TestCase):
self.nightscout.last_uploaded_entry = lambda *args, **kwargs: None
self.tconnectDevice = {
'deviceId': 'test-device-123',
'maxDateWithEvents': '2025-11-18T13:00:00-05:00'
'assignmentId': 'test-device-123',
'maxDateOfEvents': '2025-11-18T13:00:00-05:00'
}
self.secret = build_secrets(
@@ -165,8 +165,8 @@ class TestProcessTimeRangeJsonBasal(unittest.TestCase):
)
self.tconnectDevice = {
'deviceId': 'test',
'maxDateWithEvents': '2026-05-07T00:20:00-04:00'
'assignmentId': 'test',
'maxDateOfEvents': '2026-05-07T00:20:00-04:00'
}
self.process = ProcessTimeRange(
tests/sync/tandemsource/test_update_profiles.py
+8 -6
View File
@@ -23,16 +23,18 @@ SETTINGS_DETAILS = {
def _meta(device_id=DEVICE_ID, settings=None):
# settings, when provided, is the raw settings.details blob; wrap it in the
# BFF settings envelope ({'details': ...}) that update_profiles unwraps.
return {
'deviceId': device_id,
'assignmentId': device_id,
'serialNumber': '1518994',
'modelNumber': '1004000',
'modelName': 'Tandem Mobi™ System',
'softwareVersion': '1.0.0.0',
'algorithm': 'Control-IQ',
'maxDateWithEvents': '2026-05-27T23:03:06',
'minDateWithEvents': '2020-01-02T00:00:00',
'settings': settings,
'maxDateOfEvents': '2026-05-27T23:03:06',
'availableDataRange': {'start': '2020-01-02T00:00:00', 'end': '2026-05-27T23:03:06'},
'settings': {'details': settings} if settings is not None else None,
}
@@ -44,8 +46,8 @@ class FakeTandemSourceApi:
def __init__(self, metadata):
self._metadata = metadata
def pump_metadata(self):
return self._metadata
def get_pumper(self):
return {'pumps': self._metadata}
def needs_relogin(self):
# Required so TConnectApi.tandemsource returns this fake, not a real API.