mirror of
https://github.com/bckelley/tconnectsync.git
synced 2026-08-24 03:34:12 -05:00
Add Phase-2 backfill tests (update_profiles, adapter, event ids)
- tests/sync/tandemsource/test_update_profiles.py: settings sourcing from pump_metadata() — matching deviceId with settings reaches PumpSettings.from_dict (proven via sentinel, using the real BFF settings.details shape); settings=None / no-match / empty return False without parsing. - test_tandemsource.py: adapter edge cases (availableDataRange/settings keys absent, missing required key raises KeyError, Mobi Control-IQ+ passthrough) and DEFAULT_EVENT_IDS regression (55 ids, no dupes, 477/480/486 present). Fixtures are inline and trimmed from the real captured responses.
This commit is contained in:
@@ -106,6 +106,52 @@ class TestPumpMetadataAdapter(unittest.TestCase):
|
|||||||
with patch.object(TandemSourceApi, "get_pumper", return_value={}):
|
with patch.object(TandemSourceApi, "get_pumper", return_value={}):
|
||||||
self.assertEqual(api.pump_metadata(), [])
|
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_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):
|
||||||
|
ids = TandemSourceApi.DEFAULT_EVENT_IDS
|
||||||
|
self.assertEqual(len(ids), 55)
|
||||||
|
self.assertEqual(len(set(ids)), 55, "DEFAULT_EVENT_IDS contains duplicates")
|
||||||
|
# FSL3 ids added for the BFF pump-logs endpoint
|
||||||
|
self.assertTrue({477, 480, 486}.issubset(set(ids)))
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -0,0 +1,95 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from tconnectsync.sync.tandemsource.update_profiles import UpdateProfiles
|
||||||
|
from tconnectsync.domain.tandemsource.pump_settings import PumpSettings
|
||||||
|
|
||||||
|
from ...api.fake import TConnectApi
|
||||||
|
from ...nightscout_fake import NightscoutApi
|
||||||
|
|
||||||
|
DEVICE_ID = '1114157'
|
||||||
|
|
||||||
|
# Trimmed real BFF settings.details shape (from the captured bff/pumper
|
||||||
|
# response). Values condensed; PumpSettings.from_dict is patched in these
|
||||||
|
# tests, so we only assert it is reached with this blob — parsing the new
|
||||||
|
# shape is covered when PumpSettings is migrated.
|
||||||
|
SETTINGS_DETAILS = {
|
||||||
|
'basalLimitSettings': {'basalLimitDefault': 5000, 'basalLimit': 5000},
|
||||||
|
'globalMaxBolusSettings': {'maxBolus': 25000, 'maxBolusDefault': 25000},
|
||||||
|
'profiles': {'numberOfProfiles': 1, 'activeSegment': 0, 'activeIdp': 0, 'profile': []},
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _meta(device_id=DEVICE_ID, settings=None):
|
||||||
|
return {
|
||||||
|
'deviceId': 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,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class _ReachedFromDict(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class FakeTandemSourceApi:
|
||||||
|
def __init__(self, metadata):
|
||||||
|
self._metadata = metadata
|
||||||
|
|
||||||
|
def pump_metadata(self):
|
||||||
|
return self._metadata
|
||||||
|
|
||||||
|
def needs_relogin(self):
|
||||||
|
# Required so TConnectApi.tandemsource returns this fake, not a real API.
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
class TestUpdateProfilesSettingsSourcing(unittest.TestCase):
|
||||||
|
maxDiff = None
|
||||||
|
|
||||||
|
def _updater(self, metadata, device_id=DEVICE_ID):
|
||||||
|
tconnect = TConnectApi()
|
||||||
|
tconnect._tandemsource = FakeTandemSourceApi(metadata)
|
||||||
|
nightscout = NightscoutApi()
|
||||||
|
return UpdateProfiles(tconnect, nightscout, device_id, pretend=True)
|
||||||
|
|
||||||
|
def test_matching_device_with_settings_reaches_from_dict(self):
|
||||||
|
updater = self._updater([_meta(settings=SETTINGS_DETAILS)])
|
||||||
|
# Sentinel proves we pass the guard and hand settings.details to from_dict.
|
||||||
|
with patch.object(PumpSettings, 'from_dict', side_effect=_ReachedFromDict) as fd:
|
||||||
|
with self.assertRaises(_ReachedFromDict):
|
||||||
|
updater.update(pretend=True)
|
||||||
|
fd.assert_called_once_with(SETTINGS_DETAILS)
|
||||||
|
|
||||||
|
def test_matching_device_with_null_settings_returns_false(self):
|
||||||
|
updater = self._updater([_meta(settings=None)])
|
||||||
|
with patch.object(PumpSettings, 'from_dict') as fd:
|
||||||
|
result = updater.update(pretend=True)
|
||||||
|
self.assertFalse(result)
|
||||||
|
fd.assert_not_called()
|
||||||
|
|
||||||
|
def test_no_device_match_returns_false(self):
|
||||||
|
updater = self._updater([_meta(device_id='some-other-device', settings=SETTINGS_DETAILS)])
|
||||||
|
with patch.object(PumpSettings, 'from_dict') as fd:
|
||||||
|
result = updater.update(pretend=True)
|
||||||
|
self.assertFalse(result)
|
||||||
|
fd.assert_not_called()
|
||||||
|
|
||||||
|
def test_empty_metadata_returns_false(self):
|
||||||
|
updater = self._updater([])
|
||||||
|
with patch.object(PumpSettings, 'from_dict') as fd:
|
||||||
|
result = updater.update(pretend=True)
|
||||||
|
self.assertFalse(result)
|
||||||
|
fd.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user