diff --git a/tests/sync/test_basal.py b/tests/sync/test_basal.py deleted file mode 100644 index cafee18..0000000 --- a/tests/sync/test_basal.py +++ /dev/null @@ -1,145 +0,0 @@ -#!/usr/bin/env python3 - -import unittest -import copy - -from tconnectsync.sync.basal import process_ciq_basal_events -from tconnectsync.parser.tconnect import TConnectEntry - -class TestBasalSync(unittest.TestCase): - maxDiff = None - - base = { - "basal": { - "profileRates": [], - "tempDeliveryEvents": [], - "algorithmDeliveryEvents": [], - "profileDeliveryEvents": [] - }, - "events": [], - "suspensionDeliveryEvents": [], - "softwareUpdates": [], - "pumpFeatures": [] - } - - @staticmethod - def get_example_ciq_basal_events(): - data = copy.deepcopy(TestBasalSync.base) - data["basal"]["tempDeliveryEvents"] = [ - { - "y": 0.8, - "duration": 1221, - "x": 1615878000 # 12:00:00 - } - ] - data["basal"]["algorithmDeliveryEvents"] = [ - { - "y": 0.797, - "duration": 300, - "x": 1615879521 # 12:25:21 - }, - { - "y": 0, - "duration": 2693, - "x": 1615879821 # 12:30:21 - }, - ] - data["basal"]["profileDeliveryEvents"] = [ - { - "y": 0.799, - "duration": 300, - "x": 1615879221 # 12:20:21 - } - ] - - data["suspensionDeliveryEvents"] = [ - { - "suspendReason": "control-iq", - "continuation": None, - "x": 1615879821 # 12:30:21 - }, - ] - - return data - - def test_process_ciq_basal_events(self): - data = TestBasalSync.get_example_ciq_basal_events() - - basalEvents = process_ciq_basal_events(data) - self.assertEqual(len(basalEvents), 4) - - self.assertEqual(basalEvents[0], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["tempDeliveryEvents"][0], delivery_type="tempDelivery")) - - self.assertEqual(basalEvents[1], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["profileDeliveryEvents"][0], delivery_type="profileDelivery")) - - self.assertEqual(basalEvents[2], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["algorithmDeliveryEvents"][0], delivery_type="algorithmDelivery")) - - self.assertEqual(basalEvents[3], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["algorithmDeliveryEvents"][1], - delivery_type="algorithmDelivery (control-iq suspension)") - ) - - - @staticmethod - def get_example_ciq_basal_events_with_manual_suspension(): - data = copy.deepcopy(TestBasalSync.base) - data["basal"]["tempDeliveryEvents"] = [] - data["basal"]["algorithmDeliveryEvents"] = [ - { - "y": 1.14, - "duration": 300, - "x": 1635187357 # 11:42:37 - }, - { - "y": 0.8, - "duration": 1198, - "x": 1635187657 # 11:47:37 - }, - { - "y": 0.8, - "duration": 599, - "x": 1635190967 # 12:42:47 - } - ] - data["basal"]["profileDeliveryEvents"] = [] - data["suspensionDeliveryEvents"] = [ - { - "suspendReason": "manual", - "continuation": None, - "x": 1635188855 # 12:07:35 - }, - { - "suspendReason": "manual", - "continuation": None, - "x": 1635191566 # 12:52:46 - }, - ] - - return data - - def test_process_ciq_basal_events_with_manual_suspension(self): - data = TestBasalSync.get_example_ciq_basal_events_with_manual_suspension() - - basalEvents = process_ciq_basal_events(data) - self.assertEqual(len(basalEvents), 4) - - self.assertEqual(basalEvents[0], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["algorithmDeliveryEvents"][0], delivery_type="algorithmDelivery")) - - self.assertEqual(basalEvents[1], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["algorithmDeliveryEvents"][1], delivery_type="algorithmDelivery")) - - self.assertEqual(basalEvents[2], TConnectEntry.manual_suspension_to_basal_entry( - TConnectEntry.parse_suspension_entry(data["suspensionDeliveryEvents"][0]), - seconds=2112, # 2112 seconds between 12:07:35 and 12:42:47 - )) - - self.assertEqual(basalEvents[3], TConnectEntry.parse_ciq_basal_entry( - data["basal"]["algorithmDeliveryEvents"][2], delivery_type="algorithmDelivery")) - - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/tests/sync/test_bolus.py b/tests/sync/test_bolus.py deleted file mode 100644 index 76c67b2..0000000 --- a/tests/sync/test_bolus.py +++ /dev/null @@ -1,194 +0,0 @@ -#!/usr/bin/env python3 - -import unittest -import random - - -from tconnectsync.sync.bolus import process_bolus_events -from tconnectsync.parser.tconnect import TConnectEntry -from tconnectsync.parser.nightscout import NightscoutEntry - -from ..parser.test_tconnect import TestTConnectEntryBolus, TestTConnectEntryCGM, TestTConnectEntryReading - -class TestBolusSync(unittest.TestCase): - - @staticmethod - def get_example_csv_bolus_events(): - return [ - TestTConnectEntryBolus.entryStdCorrection, - TestTConnectEntryBolus.entryStd, - TestTConnectEntryBolus.entryStdAutomatic, - TestTConnectEntryBolus.entryStdIncompletePartial - ] - - def test_process_bolus_events_standard(self): - bolusData = [ - TestTConnectEntryBolus.entryStdCorrection, - TestTConnectEntryBolus.entryStd, - TestTConnectEntryBolus.entryStdAutomatic - ] - - bolusEvents = process_bolus_events(bolusData) - self.assertEqual(len(bolusEvents), len(bolusData)) - - self.assertListEqual(bolusEvents, [ - TConnectEntry.parse_bolus_entry(d) for d in bolusData - ]) - - def test_process_bolus_events_cgmevents_not_matching(self): - bolusData = [ - TestTConnectEntryBolus.entryStdCorrection, - TestTConnectEntryBolus.entryStd, - TestTConnectEntryBolus.entryStdAutomatic - ] - - cgmEvents = [ - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry1), - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry2), - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry3) - ] - - bolusEvents = process_bolus_events(bolusData, cgmEvents=cgmEvents) - self.assertEqual(len(bolusEvents), len(bolusData)) - - def set_bg_type(entry, type): - entry.bg_type = type - return entry - - # Expect FINGER for bolus entries with a BG because there's no matching event with the same BG - expected = [ - set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[0]), NightscoutEntry.FINGER), - set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[1]), NightscoutEntry.FINGER), - # No BG specified for the automatic bolus - TConnectEntry.parse_bolus_entry(bolusData[2]) - ] - - self.assertListEqual(bolusEvents, expected) - - def test_process_bolus_events_cgmevents_matches(self): - bolusData = [ - TestTConnectEntryBolus.entryStdCorrection, - TestTConnectEntryBolus.entryStd, - TestTConnectEntryBolus.entryStdAutomatic - ] - - cgmEvents = [ - { - "time": "2021-04-01 12:45:30-04:00", - "bg": "100", - "type": "EGV" - }, - # Matches entryStdCorrection time but with wrong BG - { - "time": "2021-04-01 12:50:30-04:00", - "bg": "105", - "type": "EGV" - }, - { - "time": "2021-04-01 13:00:30-04:00", - "bg": "110", - "type": "EGV" - }, - { - "time": "2021-04-01 23:15:30-04:00", - "bg": "150", - "type": "EGV" - }, - # Matches entryStd time with correct BG - { - "time": "2021-04-01 23:20:30-04:00", - "bg": "159", - "type": "EGV" - }, - { - "time": "2021-04-01 23:25:30-04:00", - "bg": "160", - "type": "EGV" - }, - ] - - bolusEvents = process_bolus_events(bolusData, cgmEvents=cgmEvents) - self.assertEqual(len(bolusEvents), len(bolusData)) - - def set_bg_type(entry, type): - entry.bg_type = type - return entry - - expected = [ - # Time found but BG doesn't match - set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[0]), NightscoutEntry.FINGER), - # Time found and BG matches - set_bg_type(TConnectEntry.parse_bolus_entry(bolusData[1]), NightscoutEntry.SENSOR), - # No BG specified for the automatic bolus - TConnectEntry.parse_bolus_entry(bolusData[2]) - ] - - self.assertListEqual(bolusEvents, expected) - - def test_process_bolus_events_update_partial_description(self): - stdData = [ - TestTConnectEntryBolus.entryStdCorrection, - TestTConnectEntryBolus.entryStd, - TestTConnectEntryBolus.entryStdAutomatic - ] - partialData = [ - TestTConnectEntryBolus.entryStdIncompletePartial - ] - - bolusData = stdData + partialData - - bolusEvents = process_bolus_events(bolusData) - self.assertEqual(len(bolusEvents), len(bolusData)) - - partialEntries = [ - TConnectEntry.parse_bolus_entry(e) for e in partialData - ] - - for e in partialEntries: - e.description += " (%s: requested %s units)" % (e.completion, e.requested_insulin) - - self.assertListEqual(bolusEvents, [ - TConnectEntry.parse_bolus_entry(d) for d in stdData - ] + partialEntries) - - def test_process_bolus_events_skip_zero(self): - stdData = [ - TestTConnectEntryBolus.entryStdCorrection, - TestTConnectEntryBolus.entryStd, - TestTConnectEntryBolus.entryStdAutomatic - ] - zeroData = [ - TestTConnectEntryBolus.entryStdIncompleteZero - ] - bolusData = stdData + zeroData - - bolusEvents = process_bolus_events(bolusData) - self.assertEqual(len(bolusEvents), len(stdData)) - - self.assertListEqual(bolusEvents, [ - TConnectEntry.parse_bolus_entry(d) for d in stdData - ]) - - for d in zeroData: - self.assertNotIn(TConnectEntry.parse_bolus_entry(d), bolusEvents) - - def test_process_bolus_events_ciq_extended_bolus(self): - stdData = [ - TestTConnectEntryBolus.entryExtendedComplete, - ] - zeroData = [ - ] - bolusData = stdData + zeroData - - bolusEvents = process_bolus_events(bolusData) - self.assertEqual(len(bolusEvents), len(stdData)) - - self.assertListEqual(bolusEvents, [ - TConnectEntry.parse_bolus_entry(d) for d in stdData - ]) - - for d in zeroData: - self.assertNotIn(TConnectEntry.parse_bolus_entry(d), bolusEvents) - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/tests/sync/test_cgm.py b/tests/sync/test_cgm.py deleted file mode 100644 index 62105ec..0000000 --- a/tests/sync/test_cgm.py +++ /dev/null @@ -1,51 +0,0 @@ -#!/usr/bin/env python3 - -import unittest - -from tconnectsync.sync.cgm import find_event_at, process_cgm_events -from tconnectsync.parser.tconnect import TConnectEntry - -from ..parser.test_tconnect import TestTConnectEntryReading - -class TestProcessCGMEvents(unittest.TestCase): - def test_process_cgm_events(self): - rawReadings = [ - TestTConnectEntryReading.entry1, - TestTConnectEntryReading.entry2, - TestTConnectEntryReading.entry3, - TestTConnectEntryReading.entry4 - ] - self.assertListEqual( - process_cgm_events(rawReadings), - [TConnectEntry.parse_reading_entry(r) for r in rawReadings] - ) - -class TestFindEventAt(unittest.TestCase): - readingData = [ - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry1), - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry2), - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry3), - TConnectEntry.parse_reading_entry(TestTConnectEntryReading.entry4) - ] - - def test_find_event_at_exact(self): - for r in self.readingData: - self.assertEqual(find_event_at(self.readingData, r["time"]), r) - - def test_find_event_at_before_not_found(self): - self.assertEqual(find_event_at(self.readingData, "2021-10-22 10:30:00-04:00"), None) - - def test_find_event_at_large_gap(self): - self.assertEqual(find_event_at(self.readingData, "2021-10-23 13:30:00-04:00"), self.readingData[0]) - - def test_find_event_at_between_close(self): - self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:17:52-04:00"), self.readingData[1]) - self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:21:52-04:00"), self.readingData[2]) - self.assertEqual(find_event_at(self.readingData, "2021-10-23 16:25:59-04:00"), self.readingData[3]) - - def test_find_event_at_most_recent(self): - self.assertEqual(find_event_at(self.readingData, "2021-10-23 18:00:00-04:00"), self.readingData[3]) - - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/tests/sync/test_iob.py b/tests/sync/test_iob.py deleted file mode 100644 index 4992587..0000000 --- a/tests/sync/test_iob.py +++ /dev/null @@ -1,30 +0,0 @@ -#!/usr/bin/env python3 - -import unittest - -from tconnectsync.sync.iob import process_iob_events -from tconnectsync.parser.tconnect import TConnectEntry - -from ..parser.test_tconnect import TestTConnectEntryIOB - -class TestIOBSync(unittest.TestCase): - - @staticmethod - def get_example_csv_iob_events(): - return [ - TestTConnectEntryIOB.entry1, - TestTConnectEntryIOB.entry2, - ] - - def test_process_iob_events(self): - iobData = TestIOBSync.get_example_csv_iob_events() - - iobEvents = process_iob_events(iobData) - self.assertEqual(len(iobEvents), len(iobData)) - - self.assertListEqual(iobEvents, [ - TConnectEntry.parse_iob_entry(d) for d in iobData - ]) - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/tests/sync/test_profile.py b/tests/sync/test_profile.py deleted file mode 100644 index 659d414..0000000 --- a/tests/sync/test_profile.py +++ /dev/null @@ -1,524 +0,0 @@ -#!/usr/bin/env python3 - -import json -import unittest -from typing import Dict -import copy - -from tconnectsync.sync.profile import get_pump_profiles, compare_profiles, nightscout_profiles_identical -from tconnectsync.domain.device_settings import Profile, ProfileSegment, DeviceSettings -from tconnectsync.secret import NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, NIGHTSCOUT_PROFILE_DELAY_VALUE, TIMEZONE_NAME - -from ..parser.test_tconnect import TestTConnectEntryReading - -DEVICE_PROFILE_A = Profile( - title='A', - active=False, - segments=[ - ProfileSegment( - display_time='Midnight', - time='12:00 AM', - basal_rate=0.8, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='6:00 AM', - time='6:00 AM', - basal_rate=1.25, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='11:00 AM', - time='11:00 AM', - basal_rate=1.0, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='Noon', - time='12:00 PM', - basal_rate=0.8, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0) - ], - calculated_total_daily_basal=21.65, - insulin_duration_min=300, - carbs_enabled=True -) - -DEVICE_PROFILE_B = Profile( - title='B', - active=False, - segments=[ - ProfileSegment( - display_time='Midnight', - time='12:00 AM', - basal_rate=0.8, - correction_factor=30.0, - carb_ratio=12.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='6:00 AM', - time='6:00 AM', - basal_rate=1.25, - correction_factor=30.0, - carb_ratio=12.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='11:00 AM', - time='11:00 AM', - basal_rate=1.0, - correction_factor=30.0, - carb_ratio=12.0, - target_bg_mgdl=110.0), - ProfileSegment( - display_time='Noon', - time='12:00 PM', - basal_rate=0.9, - correction_factor=30.0, - carb_ratio=12.0, - target_bg_mgdl=110.0) - ], - calculated_total_daily_basal=22.85, - insulin_duration_min=300, - carbs_enabled=True -) - -DEVICE_SETTINGS = DeviceSettings( - low_bg_threshold=80, - high_bg_threshold=200, - raw_settings={} -) - -NS_PROFILE_A = { - "dia": "5.0", - "carbratio": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 6.0 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 6.0 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 6.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 6.0 - } - ], - - "carbs_hr": NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, - "delay": NIGHTSCOUT_PROFILE_DELAY_VALUE, - "sens": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 30.0 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 30.0 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 30.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 30.0 - } - ], - "basal": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 0.8 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 1.25 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 1.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 0.8 - } - ], - "target_low": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 80 - } - ], - "target_high": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 200 - } - ], - "timezone": TIMEZONE_NAME, - "startDate": "1970-01-01T00:00:00.000Z", - "units": "mg/dl" -} - -NS_PROFILE_B = { - "dia": "5.0", - "carbratio": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 12.0 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 12.0 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 12.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 12.0 - } - ], - - "carbs_hr": NIGHTSCOUT_PROFILE_CARBS_HR_VALUE, - "delay": NIGHTSCOUT_PROFILE_DELAY_VALUE, - "sens": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 30.0 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 30.0 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 30.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 30.0 - } - ], - "basal": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 0.8 - }, - { - "time": "06:00", - "timeAsSeconds": 6*60*60, - "value": 1.25 - }, - { - "time": "11:00", - "timeAsSeconds": 11*60*60, - "value": 1.0 - }, - { - "time": "12:00", - "timeAsSeconds": 12*60*60, - "value": 0.9 - } - ], - "target_low": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 80 - } - ], - "target_high": [ - { - "time": "00:00", - "timeAsSeconds": 0, - "value": 200 - } - ], - "timezone": TIMEZONE_NAME, - "startDate": "1970-01-01T00:00:00.000Z", - "units": "mg/dl" -} - -NS_PROFILE_STORE = { - 'A': NS_PROFILE_A, - 'B': NS_PROFILE_B -} - -def build_ns_profile(profiles: Dict[str, dict], current_profile: str) -> dict: - return copy.deepcopy({ - "store": profiles, - "defaultProfile": current_profile, - "startDate": "1970-01-01T00:00:00.000Z", - "mills": 0, - "units": "mg/dl", - }) - -class TestCompareProfiles(unittest.TestCase): - maxDiff = None - def test_compare_profiles_identical_a(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile()] - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertFalse(changed) - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_identical_b(self): - pump_profiles = [DEVICE_PROFILE_B.activeProfile()] - ns_profile_obj = build_ns_profile({'B': NS_PROFILE_B}, 'B') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertFalse(changed) - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_extra_profile_in_nightscout_ignored(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile()] - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A, 'B': NS_PROFILE_B}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertFalse(changed) - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_current_nightscout_profile_changed(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile()] - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A, 'B': NS_PROFILE_B}, 'B') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertTrue(changed) - self.assertNotEqual(ns_profile_obj, new_profiles) - - ns_profile_obj['defaultProfile'] = 'A' - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_new_pump_profile_added(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile(), DEVICE_PROFILE_B] - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertTrue(changed) - self.assertNotEqual(ns_profile_obj, new_profiles) - - ns_profile_obj['store']['B'] = new_profiles['store']['B'] - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_new_pump_profile_added_and_active(self): - pump_profiles = [DEVICE_PROFILE_A, DEVICE_PROFILE_B.activeProfile()] - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertTrue(changed) - self.assertNotEqual(ns_profile_obj, new_profiles) - - ns_profile_obj['store']['B'] = new_profiles['store']['B'] - ns_profile_obj['defaultProfile'] = 'B' - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_existing_profile_edited_basal(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile()] - pump_profiles[0].segments[0].basal_rate = 0.1 - - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertTrue(changed) - self.assertNotEqual(ns_profile_obj, new_profiles) - - ns_profile_obj['store']['A']['basal'][0]['value'] = 0.1 - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_existing_profile_edited_new_chunk(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile()] - pump_profiles[0].segments.append(ProfileSegment( - display_time='06:00 PM', - time='06:00 PM', - basal_rate=0.7, - correction_factor=30.0, - carb_ratio=6.0, - target_bg_mgdl=110.0) - ) - - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertTrue(changed) - self.assertNotEqual(ns_profile_obj, new_profiles) - - ns_profile_obj['store']['A']['basal'].append({'time': '18:00', 'timeAsSeconds': 18*60*60, 'value': 0.7}) - ns_profile_obj['store']['A']['carbratio'].append({'time': '18:00', 'timeAsSeconds': 18*60*60, 'value': 6}) - ns_profile_obj['store']['A']['sens'].append({'time': '18:00', 'timeAsSeconds': 18*60*60, 'value': 30}) - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def test_compare_profiles_existing_profile_edited_removed_chunk(self): - pump_profiles = [DEVICE_PROFILE_A.activeProfile()] - pump_profiles[0].segments.pop() - - ns_profile_obj = build_ns_profile({'A': NS_PROFILE_A}, 'A') - - changed, new_profiles = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - ns_profile_obj - ) - - self.assertTrue(changed) - self.assertNotEqual(ns_profile_obj, new_profiles) - - ns_profile_obj['store']['A']['basal'].pop() - ns_profile_obj['store']['A']['carbratio'].pop() - ns_profile_obj['store']['A']['sens'].pop() - self.assertDictEqual(ns_profile_obj, new_profiles) - - self.ensure_stabilized(pump_profiles, new_profiles) - - def ensure_stabilized(self, pump_profiles, new_profiles): - changed, _ = compare_profiles( - pump_profiles, - DEVICE_SETTINGS, - new_profiles - ) - self.assertFalse(changed, 'did not stabilize') - -class TestNightscoutProfilesIdentical(unittest.TestCase): - def test_same_dict(self): - self.assertTrue(nightscout_profiles_identical(NS_PROFILE_A, NS_PROFILE_A)) - - def test_different_dict(self): - self.assertFalse(nightscout_profiles_identical(NS_PROFILE_A, NS_PROFILE_B)) - - def test_same_re_jsoned(self): - self.assertTrue(nightscout_profiles_identical(NS_PROFILE_A, json.loads(json.dumps(NS_PROFILE_A)))) - - def test_same_different_number_types(self): - a = b = json.loads(json.dumps(NS_PROFILE_A)) - a['dia'] = 5 - b = json.loads(json.dumps(NS_PROFILE_A)) - b['dia'] = 5.0 - self.assertTrue(nightscout_profiles_identical(a, b)) - - def test_same_different_number_strings_1(self): - a = b = json.loads(json.dumps(NS_PROFILE_A)) - a['dia'] = 5 - b = json.loads(json.dumps(NS_PROFILE_A)) - b['dia'] = '5.0' - self.assertTrue(nightscout_profiles_identical(a, b)) - - def test_same_different_number_strings_2(self): - a = b = json.loads(json.dumps(NS_PROFILE_A)) - a['dia'] = '5' - b = json.loads(json.dumps(NS_PROFILE_A)) - b['dia'] = 5.0 - self.assertTrue(nightscout_profiles_identical(a, b)) - - def test_same_different_number_strings_3(self): - a = b = json.loads(json.dumps(NS_PROFILE_A)) - a['dia'] = '5' - b = json.loads(json.dumps(NS_PROFILE_A)) - b['dia'] = '5.000' - self.assertTrue(nightscout_profiles_identical(a, b)) - - def test_different_numbers(self): - a = b = json.loads(json.dumps(NS_PROFILE_A)) - a['dia'] = 5 - b = json.loads(json.dumps(NS_PROFILE_A)) - b['dia'] = 5.1 - self.assertFalse(nightscout_profiles_identical(a, b)) - - def test_different_numbers_strings(self): - a = b = json.loads(json.dumps(NS_PROFILE_A)) - a['dia'] = '5' - b = json.loads(json.dumps(NS_PROFILE_A)) - b['dia'] = '5.01' - self.assertFalse(nightscout_profiles_identical(a, b)) - -if __name__ == '__main__': - unittest.main() \ No newline at end of file diff --git a/tests/test_autoupdate.py b/tests/test_autoupdate.py deleted file mode 100644 index 01bd753..0000000 --- a/tests/test_autoupdate.py +++ /dev/null @@ -1,368 +0,0 @@ -#!/usr/bin/env python3 - -import logging -import unittest -import datetime -import contextlib - -from unittest.mock import patch - -from tconnectsync.autoupdate import Autoupdate, AutoupdateFailureError, AutoupdateFailureWarning, AutoupdateNoEventIndexesDetectedError, AutoupdateNoIndexChangeWarning - -from .api.fake import TConnectApi -from .nightscout_fake import NightscoutApi -from .secrets import build_secrets - -logger = logging.getLogger(__name__) - -def stub(*args, **kwargs): - pass - -@contextlib.contextmanager -def build_mock_logger(): - def fake_error(*args, **kwargs): - logger.error(*args, **kwargs) - - def fake_warn(*args, **kwargs): - logger.warning(*args, **kwargs) - - with patch("tconnectsync.autoupdate.logger.error") as mock_error, patch("tconnectsync.autoupdate.logger.warning") as mock_warn: - mock_error.side_effect = fake_error - mock_warn.side_effect = fake_warn - yield (mock_error, mock_warn) - -def num_instances_of(cls, m): - return sum([isinstance(i[0][0], cls) for i in m.call_args_list]) - -class TestAutoupdate(unittest.TestCase): - maxDiff = None - - # datetimes are unused - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - # each time the returned function is called, it returns the next argument - # until the end is reached, at which point it will repeat the last argument - def fake_last_event_uploaded(self, *indexes): - index = 0 - def fake(*args, **kwargs): - nonlocal index - if index < len(indexes): - index += 1 - data = {'maxPumpEventIndex': indexes[index-1], 'processingStatus': 1} - return data - - return fake - - # each time the returned function is called, it returns the next argument - # until the end is reached, at which point it will repeat the last argument - def fake_process_time_range(self, *returns): - index = 0 - def fake(*args, **kwargs): - nonlocal index - if index < len(returns): - index += 1 - return returns[index - 1] - - return fake - - """process_time_range should always be invoked the first time""" - def test_process_time_range_called_on_start(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=1, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0 - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range: - mock_process_time_range.return_value = 0 - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 1) - self.assertEqual(u.autoupdate_invocations, 1) - self.assertEqual(u.last_event_index, 1) - - - """process_time_range should never be called with pretend""" - def test_process_time_range_never_called_with_pretend(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=1, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0 - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range: - mock_process_time_range.return_value = 0 - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=True) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 0) - self.assertEqual(u.autoupdate_invocations, 1) - self.assertEqual(u.last_event_index, 1) - - - """ - If the event index increases without process_time_range detecting new data, - AutoupdateFailureWarning should be raised. - """ - def test_autoupdate_failure_warning_on_index_process_time_range_discrepancy(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1, 2) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=2, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0, - TCONNECT_EMAIL="test@email.com", - TCONNECT_PASSWORD="testpassword" - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn): - mock_process_time_range.side_effect = self.fake_process_time_range(0, 0) - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 2) - self.assertEqual(u.autoupdate_invocations, 2) - self.assertEqual(u.last_event_index, 2) - - self.assertEqual(num_instances_of(AutoupdateFailureWarning, mock_warn), 1) - - """ - On the first attempt, an AutoupdateFailureWarning should never be raised. - """ - def test_autoupdate_no_failure_warning_on_index_process_time_range_discrepancy_first_attempt(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=1, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0 - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn): - mock_process_time_range.side_effect = self.fake_process_time_range(0, 0) - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 1) - self.assertEqual(u.autoupdate_invocations, 1) - self.assertEqual(u.last_event_index, 1) - - self.assertEqual(num_instances_of(AutoupdateFailureWarning, mock_warn), 0) - - - """ - If the event index increases without process_time_range detecting new data - for AUTOUPDATE_FAILURE_MINUTES, an AutoupdateFailureError should be raised. - """ - def test_autoupdate_failure_error_on_index_process_time_range_discrepancy_for_failure_minutes(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1, 2) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=2, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0, - AUTOUPDATE_FAILURE_MINUTES=0, - AUTOUPDATE_RESTART_ON_FAILURE=False, - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn): - mock_process_time_range.side_effect = self.fake_process_time_range(0, 0) - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 2) - self.assertEqual(u.autoupdate_invocations, 2) - self.assertEqual(u.last_event_index, 2) - - self.assertEqual(num_instances_of(AutoupdateFailureError, mock_error), 1) - - """ - If the event index increases without process_time_range detecting new data - for AUTOUPDATE_FAILURE_MINUTES, and AUTOUPDATE_RESTART_ON_FAILURE is true, - then an AutoupdateFailureError should be raised AND 1 should be returned. - """ - def test_autoupdate_failure_error_on_index_process_time_range_discrepancy_for_failure_minutes_performs_restart(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1, 2) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=3, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0, - AUTOUPDATE_FAILURE_MINUTES=0, - AUTOUPDATE_RESTART_ON_FAILURE=True, - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn): - mock_process_time_range.side_effect = self.fake_process_time_range(0, 0) - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 1) - - self.assertEqual(mock_process_time_range.call_count, 2) - self.assertEqual(u.autoupdate_invocations, 1) # exits before invocations is incremented - self.assertEqual(u.last_event_index, 1) # exits before changed to 2 - - self.assertEqual(num_instances_of(AutoupdateFailureError, mock_error), 1) - - - """Validate state after first successful update""" - def test_state_after_first_successful_update(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=1, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0 - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range: - mock_process_time_range.side_effect = self.fake_process_time_range(1) - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 1) - self.assertEqual(u.autoupdate_invocations, 1) - self.assertEqual(u.last_event_index, 1) - self.assertTrue(u.last_event_time == u.last_attempt_time == u.last_successful_process_time_range) - self.assertEqual(len(u.time_diffs_between_updates), 0) - - """Validate sleep occurs for the given fixed length when set""" - def test_sleep_for_fixed_length_when_set(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - sleep_length = 3 # sentinel - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=1, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=sleep_length - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, patch("tconnectsync.autoupdate.time.sleep") as mock_sleep: - mock_process_time_range.side_effect = self.fake_process_time_range(1) - mock_sleep.side_effect = None - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(mock_process_time_range.call_count, 1) - self.assertEqual(u.autoupdate_invocations, 1) - self.assertEqual(u.last_event_index, 1) - self.assertTrue(mock_sleep.called) - self.assertEqual(mock_sleep.call_args[0], (sleep_length,)) - - - """ - If there is no event index update for AUTOUPDATE_NO_DATA_FAILURE_MINUTES, - then a AutoupdateNoEventIndexesDetectedError should be raised and - a restart should be triggered with AUTOUPDATE_RESTART_ON_FAILURE. - """ - def test_autoupdate_no_event_indexes_detected_error_on_no_index_change_less_than_three_iterations(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=100, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0.1, - AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=0.2, - AUTOUPDATE_RESTART_ON_FAILURE=True, - AUTOUPDATE_NO_DATA_FAILURE_MINUTES=1/600 # 0.1 second - ) - - # with more than 3 iterations, the sleep iteration will change - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn): - mock_process_time_range.side_effect = self.fake_process_time_range(1, 0) - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 1) - - self.assertEqual(num_instances_of(AutoupdateNoEventIndexesDetectedError, mock_error), 1) - - - """ - If there has been 3 failed attempts since the last time we found new data, - we should sleep for AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS and - a AutoupdateNoIndexChangeWarning should be logged. - """ - def test_autoupdate_no_index_change_warning_on_unexpected_no_index_sleep(self): - tconnect = TConnectApi() - tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1) - - nightscout = NightscoutApi() - - secret = build_secrets( - AUTOUPDATE_MAX_LOOP_INVOCATIONS=5, - AUTOUPDATE_USE_FIXED_SLEEP=True, - AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0.1, - AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=0.2 - ) - - with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, patch("tconnectsync.autoupdate.time.sleep") as mock_sleep, build_mock_logger() as (mock_error, mock_warn): - mock_process_time_range.side_effect = self.fake_process_time_range(0) - mock_sleep.side_effect = None - - u = Autoupdate(secret) - ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False) - self.assertEqual(ret, 0) - - self.assertEqual(u.autoupdate_invocations, 5) - self.assertEqual(u.last_event_index, 1) - self.assertTrue(mock_sleep.called) - self.assertEqual(len(mock_sleep.call_args_list), 5) - self.assertEqual(mock_sleep.call_args_list[0][0], (0.1,)) - self.assertEqual(mock_sleep.call_args_list[1][0], (0.1,)) - self.assertEqual(mock_sleep.call_args_list[2][0], (0.1,)) - self.assertEqual(mock_sleep.call_args_list[3][0], (0.2,)) - self.assertEqual(mock_sleep.call_args_list[4][0], (0.2,)) - - self.assertEqual(num_instances_of(AutoupdateNoIndexChangeWarning, mock_warn), 2) - diff --git a/tests/test_process.py b/tests/test_process.py deleted file mode 100644 index 93b3843..0000000 --- a/tests/test_process.py +++ /dev/null @@ -1,978 +0,0 @@ -#!/usr/bin/env python3 - -import unittest -import datetime -import pprint -import copy -from unittest.mock import patch - -from tconnectsync.process import process_time_range -from tconnectsync.parser.nightscout import EXERCISE_EVENTTYPE, IOB_ACTIVITYTYPE, SLEEP_EVENTTYPE, NightscoutEntry -from tconnectsync.features import BASAL, BOLUS, IOB, PROFILES, PUMP_EVENTS, PUMP_EVENTS_BASAL_SUSPENSION -from tconnectsync.domain.device_settings import Device -from tests.secrets import build_secrets -from tests.sync.test_profile import DEVICE_PROFILE_A, DEVICE_PROFILE_B, DEVICE_SETTINGS, NS_PROFILE_A, NS_PROFILE_B, build_ns_profile - - -from .api.fake import TConnectApi -from .nightscout_fake import NightscoutApi -from .sync.test_basal import TestBasalSync -from .sync.test_bolus import TestBolusSync -from .sync.test_iob import TestIOBSync -from .domain.test_therapy_event import BOLUS_FULL_EXAMPLES, TestCGMTherapyEvent - -class TestProcessTimeRange(unittest.TestCase): - maxDiff = None - - def stub_therapy_timeline(self, time_start, time_end): - return copy.deepcopy(TestBasalSync.base) - - def stub_ciq_therapy_events(self, time_start, time_end): - return { - "event": [] - } - - def stub_therapy_timeline_csv(self, time_start, time_end): - return { - "readingData": [], - "iobData": [], - "basalData": [], - "bolusData": [] - } - - def stub_ws2_basalsuspension(self, time_start, time_end): - return {"BasalSuspension": []} - - def stub_last_uploaded_entry(self, event_type, **kwargs): - return None - - def stub_last_uploaded_activity(self, activity_type, **kwargs): - return None - - """No data in Nightscout. Uploads all basal data from tconnect.""" - def test_new_ciq_basal_data(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return TestBasalSync.get_example_ciq_basal_events() - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 4) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - NightscoutEntry.basal(0.8, 20.35, "2021-03-16 00:00:00-04:00", reason="tempDelivery"), - NightscoutEntry.basal(0.799, 5.0, "2021-03-16 00:20:21-04:00", reason="profileDelivery"), - NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"), - NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery (control-iq suspension)") - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 4) - - """No data in Nightscout. Nothing should be updated in Nightscout without the BASAL feature.""" - def test_basal_data_not_updated_without_feature(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return TestBasalSync.get_example_ciq_basal_events() - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, IOB]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 0) - - """Two basal entries in Nightscout. Two new basal entries in tconnect.""" - def test_partial_ciq_basal_data(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return TestBasalSync.get_example_ciq_basal_events() - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - nightscout = NightscoutApi() - - def fake_last_uploaded_entry(event_type, **kwargs): - if event_type == "Temp Basal": - return { - "created_at": "2021-03-16 00:20:21-04:00", - "duration": 5 - } - - nightscout.last_uploaded_entry = fake_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"), - NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery (control-iq suspension)") - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 2) - - - """ - Two basal entries in Nightscout, the latter which needs to be updated - with a longer duration. Two entirely new entries in tconnect.""" - def test_with_updated_duration_ciq_basal_data(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return TestBasalSync.get_example_ciq_basal_events() - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - nightscout = NightscoutApi() - - def fake_last_uploaded_entry(event_type, **kwargs): - if event_type == "Temp Basal": - return { - "created_at": "2021-03-16 00:20:21-04:00", - "duration": 3, - "_id": "nightscout_id" - } - - nightscout.last_uploaded_entry = fake_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2) - self.assertDictEqual(nightscout.uploaded_entries, { - "treatments": [ - NightscoutEntry.basal(0.797, 5.0, "2021-03-16 00:25:21-04:00", reason="algorithmDelivery"), - NightscoutEntry.basal(0, 2693/60, "2021-03-16 00:30:21-04:00", reason="algorithmDelivery (control-iq suspension)") - ]}) - self.assertEqual(len(nightscout.put_entries["treatments"]), 1) - self.assertDictEqual(dict(nightscout.put_entries), { - "treatments": [ - { - "_id": "nightscout_id", - **NightscoutEntry.basal(0.799, 5.0, "2021-03-16 00:20:21-04:00", reason="profileDelivery") - } - ] - }) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 3) - - """No data in Nightscout. Uploads all bolus data from tconnect via the WS2 API.""" - def test_new_ciq_bolus_data_from_ws2(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - - bolusData = TestBolusSync.get_example_csv_bolus_events() - def fake_therapy_timeline_csv(time_start, time_end): - return { - **self.stub_therapy_timeline_csv(time_start, time_end), - "bolusData": bolusData, - } - - tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL]) - - pprint.pprint(nightscout.uploaded_entries) - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), len(bolusData)) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - NightscoutEntry.bolus(13.53, 75, "2021-04-01 12:58:26-04:00", notes="Standard/Correction"), - NightscoutEntry.bolus(1.25, 0, "2021-04-01 23:23:17-04:00", notes="Standard (Override)"), - NightscoutEntry.bolus(1.7, 0, "2021-04-02 01:00:47-04:00", notes="Automatic Bolus/Correction"), - NightscoutEntry.bolus(1.82, 0, "2021-09-06 12:24:47-04:00", notes="Standard/Correction (Terminated by Alarm: requested 2.63 units)"), - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 4) - - """No data in Nightscout. Uploads all bolus data from tconnect via CIQ therapy_events.""" - def test_new_ciq_bolus_data_from_ciq_therapy_events(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2022, 8, 9, 12, 0) - end = datetime.datetime(2021, 8, 10, 12, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - - def fake_ciq_therapy_events(time_start, time_end): - return { - "event": BOLUS_FULL_EXAMPLES - } - tconnect.controliq.therapy_events = fake_ciq_therapy_events - - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL]) - - pprint.pprint(nightscout.uploaded_entries) - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), len(BOLUS_FULL_EXAMPLES)) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - # BGs are excluded because BG is not enabled in features for this test - NightscoutEntry.bolus(2.9, 0, "2022-07-21 11:55:24-04:00", notes="Automatic Bolus/Correction"), - NightscoutEntry.bolus(4.17, 25, "2022-07-21 12:29:21-04:00", notes="Standard"), - # NOTE: the extended bolus is 0.2+0.2 but we currently only surface standard - # BUG: inconsistency: we use the completion timestamp as the event timestamp for standard boluses, - # but the standard strand time for extended - NightscoutEntry.bolus(0.2, 0, "2022-08-09 23:20:04-04:00", notes="Extended 50.00%/0.00 (Override) (Extended)"), - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 3) - - """No data in Nightscout. Nothing should be updated in Nightscout without the BOLUS feature.""" - def test_bolus_data_not_updated_without_feature(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - - bolusData = TestBolusSync.get_example_csv_bolus_events() - def fake_therapy_timeline_csv(time_start, time_end): - return { - **self.stub_therapy_timeline_csv(time_start, time_end), - "bolusData": bolusData, - } - - tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BASAL, IOB]) - - pprint.pprint(nightscout.uploaded_entries) - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 0) - - """No data in Nightscout. Uploads new iob reading from tconnect.""" - def test_new_ciq_iob_data(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - - iobData = TestIOBSync.get_example_csv_iob_events() - def fake_therapy_timeline_csv(time_start, time_end): - return { - **self.stub_therapy_timeline_csv(time_start, time_end), - "iobData": iobData, - } - - tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL, IOB]) - - pprint.pprint(nightscout.uploaded_entries) - self.assertEqual(len(nightscout.uploaded_entries["activity"]), 1) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "activity": [ - # the most recent IOB entry is added - NightscoutEntry.iob(6.80, "2021-10-12 00:10:30-04:00") - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 1) - - """No data in Nightscout. Nothing should be updated in Nightscout without the IOB feature.""" - def test_iob_data_not_updated_without_feature(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - - iobData = TestIOBSync.get_example_csv_iob_events() - def fake_therapy_timeline_csv(time_start, time_end): - return { - **self.stub_therapy_timeline_csv(time_start, time_end), - "iobData": iobData, - } - - tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BASAL, BOLUS]) - - self.assertEqual(len(nightscout.uploaded_entries["activity"]), 0) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 0) - - """Existing IOB in Nightscout. Uploads new iob reading and deletes old IOB.""" - def test_updates_ciq_iob_data(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 4, 20, 12, 0) - end = datetime.datetime(2021, 4, 21, 12, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - - iobData = TestIOBSync.get_example_csv_iob_events() - iobData[0]["created_at"] = start - iobData[0]["_id"] = "sentinel_existing_iob_id" - - def fake_therapy_timeline_csv(time_start, time_end): - return { - **self.stub_therapy_timeline_csv(time_start, time_end), - "iobData": iobData, - } - - tconnect.ws2.therapy_timeline_csv = fake_therapy_timeline_csv - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - - def fake_last_uploaded_activity(activityType, **kwargs): - if activityType == IOB_ACTIVITYTYPE: - return iobData[0] - return self.stub_last_uploaded_activity(activityType) - - nightscout.last_uploaded_activity = fake_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[IOB]) - - pprint.pprint(nightscout.uploaded_entries) - self.assertEqual(len(nightscout.uploaded_entries["activity"]), 1) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "activity": [ - # the most recent IOB entry is added - NightscoutEntry.iob(6.80, "2021-10-12 00:10:30-04:00") - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, [ - "activity/sentinel_existing_iob_id" - ]) - self.assertEqual(count, 1) - - """No pump activity events in Nightscout. New CIQ activity events.""" - def test_new_ciq_activity_events(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return { - **TestBasalSync.base, - "events": [{ - "duration": 1200, - "eventType": 2, # Exercise - "continuation": None, - "timeZoneId": "America/Los_Angeles", - "x": 1619901912 # 2021-05-01 13:45:12-04:00 - }, { - "duration": 30661, - "eventType": 1, # Sleep - "continuation": None, - "timeZoneId": "America/Los_Angeles", - "x": 1619992000 # 2021-05-02 14:46:40-04:00 - }] - } - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - tconnect.ws2.basalsuspension = self.stub_ws2_basalsuspension - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 2) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - NightscoutEntry.activity(created_at="2021-05-01 13:45:12-04:00", duration=20, reason="Exercise", event_type=EXERCISE_EVENTTYPE), - NightscoutEntry.activity(created_at="2021-05-02 14:46:40-04:00", duration=30661/60, reason="Sleep", event_type=SLEEP_EVENTTYPE), - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 2) - - """No pump activity events in Nightscout. New CIQ activity events, but feature is disabled.""" - def test_no_ciq_activity_events_without_feature(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return { - **TestBasalSync.base, - "events": [{ - "duration": 1200, - "eventType": 2, # Exercise - "continuation": None, - "timeZoneId": "America/Los_Angeles", - "x": 1619901912 # 2021-05-01 13:45:12-04:00 - }, { - "duration": 30661, - "eventType": 1, # Sleep - "continuation": None, - "timeZoneId": "America/Los_Angeles", - "x": 1619992000 # 2021-05-02 14:46:40-04:00 - }] - } - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.controliq.therapy_events = self.stub_ciq_therapy_events - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - tconnect.ws2.basalsuspension = self.stub_ws2_basalsuspension - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[BOLUS, BASAL, IOB]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 0) - - - """ - Existing Sleep event in Nightscout with shorter duration than current, as well as a past Exercise event. - Ensures that the old sleep event is deleted and a new one is created with the correct duration.""" - def test_existing_ciq_activity_events(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - def fake_therapy_timeline(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return { - **TestBasalSync.base, - "events": [{ - "duration": 1200, - "eventType": 2, # Exercise - "continuation": None, - "timeZoneId": "America/Los_Angeles", - "x": 1619901912 # 2021-05-01 13:45:12-04:00 - }, { - "duration": 4200, # Currently 60 mins (3600), changing to 70 mins (4200) - "eventType": 1, # Sleep - "continuation": None, - "timeZoneId": "America/Los_Angeles", - "x": 1619992000 # 2021-05-02 14:46:40-04:00 - }] - } - - tconnect.controliq.therapy_timeline = fake_therapy_timeline - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - tconnect.ws2.basalsuspension = self.stub_ws2_basalsuspension - - nightscout = NightscoutApi() - - def fake_last_uploaded_entry(event_type, **kwargs): - if event_type == "Sleep": - return { - "created_at": "2021-05-02 14:46:40-04:00", - "duration": 60, - "_id": "old_sleep" - } - elif event_type == "Exercise": - return { - "created_at": "2021-05-01 13:45:12-04:00", - "duration": 20, - "_id": "exercise" - } - return self.stub_last_uploaded_entry(**kwargs) - - nightscout.last_uploaded_entry = fake_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 1) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - # Already exists: - # NightscoutEntry.activity(created_at="2021-05-01 13:45:12-04:00", duration=20, reason="Exercise", event_type=EXERCISE_EVENTTYPE), - # Updated event duration: - NightscoutEntry.activity(created_at="2021-05-02 14:46:40-04:00", duration=70, reason="Sleep", event_type=SLEEP_EVENTTYPE), - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertEqual(len(nightscout.deleted_entries), 1) - self.assertListEqual(nightscout.deleted_entries, [ - "treatments/old_sleep" - ]) - self.assertEqual(count, 1) - - """No pump activity events in Nightscout. New WS2 activity events.""" - def test_new_ws2_activity_events(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - def fake_basalsuspension(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return {"BasalSuspension": [ - { - 'EventDateTime': '/Date(1638663490000-0000)/', - 'SuspendReason': 'site-cart' - }, - { - 'EventDateTime': '/Date(1637863616000-0000)/', - 'SuspendReason': 'alarm' - }, - { - 'EventDateTime': '/Date(1638662852000-0000)/', - 'SuspendReason': 'manual' - } - ]} - - tconnect.ws2.basalsuspension = fake_basalsuspension - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS, PUMP_EVENTS_BASAL_SUSPENSION]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 3) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - NightscoutEntry.sitechange(created_at="2021-12-04 16:18:10-05:00", reason="Site/Cartridge Change"), - NightscoutEntry.basalsuspension(created_at="2021-11-25 10:06:56-05:00", reason="Empty Cartridge/Pump Shutdown"), - NightscoutEntry.basalsuspension(created_at="2021-12-04 16:07:32-05:00", reason="User Suspended") - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 3) - - """Existing pump activity events in Nightscout. New WS2 activity events. Only adds new events.""" - def test_existing_ws2_activity_events(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - def fake_basalsuspension(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return {"BasalSuspension": [ - { - 'EventDateTime': '/Date(1638663490000-0000)/', - 'SuspendReason': 'site-cart' - }, - { - 'EventDateTime': '/Date(1637863616000-0000)/', - 'SuspendReason': 'alarm' - }, - { - 'EventDateTime': '/Date(1638662852000-0000)/', - 'SuspendReason': 'manual' - }, - # This event is new: - { - 'EventDateTime': '/Date(1638672852000-0000)/', - 'SuspendReason': 'manual' - } - ]} - - tconnect.ws2.basalsuspension = fake_basalsuspension - - nightscout = NightscoutApi() - - def fake_last_uploaded_entry(event_type, **kwargs): - if event_type == "Site Change": - return { - "created_at": "2021-12-04 16:18:10-05:00" - } - elif event_type == "Basal Suspension": - return { - "created_at": "2021-12-04 16:07:32-05:00" - } - return self.stub_last_uploaded_entry(**kwargs) - - nightscout.last_uploaded_entry = fake_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS, PUMP_EVENTS_BASAL_SUSPENSION]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 1) - self.assertDictEqual(dict(nightscout.uploaded_entries), { - "treatments": [ - NightscoutEntry.basalsuspension(created_at="2021-12-04 18:54:12-05:00", reason="User Suspended") - ]}) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 1) - - """No pump activity events in nightscout. New WS2 activity events, but only of skipped types. None should be added.""" - def test_skipped_ws2_activity_events(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - tconnect.controliq.therapy_timeline = self.stub_therapy_timeline - tconnect.ws2.therapy_timeline_csv = self.stub_therapy_timeline_csv - - def fake_basalsuspension(time_start, time_end): - self.assertEqual(time_start, start) - self.assertEqual(time_end, end) - - return {"BasalSuspension": [ - { - 'EventDateTime': '/Date(1638659343000-0000)/', - 'SuspendReason': 'basal-profile', - }, - { - 'Continuation': 'continuation', - 'EventDateTime': '/Date(1638604800000-0000)/', - 'SuspendReason': 'previous', - }, - { - 'EventDateTime': '/Date(1638659343000-0000)/', - 'SuspendReason': 'basal-profile', - } - ]} - - tconnect.ws2.basalsuspension = fake_basalsuspension - - nightscout = NightscoutApi() - - nightscout.last_uploaded_entry = self.stub_last_uploaded_entry - nightscout.last_uploaded_activity = self.stub_last_uploaded_activity - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS]) - - self.assertEqual(len(nightscout.uploaded_entries["treatments"]), 0) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 0) - - """Profile present on pump and not in Nightscout with PROFILES feature enabled, adds profile.""" - def test_pump_profile_added(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - pump_guid = '00000000-0000-0000-0000-000000000001' - serial_number = '12345' - - def fake_my_devices(): - return { - serial_number: Device( - name='test', - model_number=serial_number, - status='OK', - guid=pump_guid) - } - tconnect.webui.my_devices = fake_my_devices - - def fake_device_settings_from_guid(guid): - if guid != pump_guid: - raise RuntimeError('invalid guid') - return [DEVICE_PROFILE_A.activeProfile()], DEVICE_SETTINGS - - tconnect.webui.device_settings_from_guid = fake_device_settings_from_guid - - nightscout = NightscoutApi() - - def fake_current_profile(time_start=None, time_end=None): - return build_ns_profile({}, '') - - nightscout.current_profile = fake_current_profile - - count = None - with patch("tconnectsync.sync.profile._get_default_upload_mode") as mock_upload_mode, \ - patch("tconnectsync.sync.profile._get_default_serial_number") as mock_serial_number: - mock_upload_mode.return_value = 'add' - mock_serial_number.return_value = serial_number - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PROFILES]) - - self.assertEqual(len(nightscout.uploaded_entries["profile"]), 1) - self.assertDictEqual(nightscout.uploaded_entries["profile"][0]['store']['A'], NS_PROFILE_A) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 1) - - """Profile present on pump and in Nightscout with PROFILES feature enabled, does not add profile.""" - def test_pump_profile_not_updated(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - pump_guid = '00000000-0000-0000-0000-000000000001' - serial_number = '12345' - - def fake_my_devices(): - return { - serial_number: Device( - name='test', - model_number=serial_number, - status='OK', - guid=pump_guid) - } - tconnect.webui.my_devices = fake_my_devices - - def fake_device_settings_from_guid(guid): - if guid != pump_guid: - raise RuntimeError('invalid guid') - return [DEVICE_PROFILE_A.activeProfile()], DEVICE_SETTINGS - - tconnect.webui.device_settings_from_guid = fake_device_settings_from_guid - - nightscout = NightscoutApi() - - def fake_current_profile(time_start=None, time_end=None): - return build_ns_profile({'A': NS_PROFILE_A}, 'A') - - nightscout.current_profile = fake_current_profile - - count = None - with patch("tconnectsync.sync.profile._get_default_upload_mode") as mock_upload_mode, \ - patch("tconnectsync.sync.profile._get_default_serial_number") as mock_serial_number: - mock_upload_mode.return_value = 'add' - mock_serial_number.return_value = serial_number - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PROFILES]) - - self.assertEqual(len(nightscout.uploaded_entries["profile"]), 0) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 0) - - """Profile present on pump and in Nightscout with PROFILES feature enabled, with changes on pump, adds new profile object.""" - def test_pump_profile_new_entry_added(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - pump_guid = '00000000-0000-0000-0000-000000000001' - serial_number = '12345' - - def fake_my_devices(): - return { - serial_number: Device( - name='test', - model_number=serial_number, - status='OK', - guid=pump_guid) - } - tconnect.webui.my_devices = fake_my_devices - - def fake_device_settings_from_guid(guid): - if guid != pump_guid: - raise RuntimeError('invalid guid') - return [DEVICE_PROFILE_A, DEVICE_PROFILE_B.activeProfile()], DEVICE_SETTINGS - - tconnect.webui.device_settings_from_guid = fake_device_settings_from_guid - - nightscout = NightscoutApi() - - def fake_current_profile(time_start=None, time_end=None): - return build_ns_profile({'A': NS_PROFILE_A}, 'A') - - nightscout.current_profile = fake_current_profile - - count = None - with patch("tconnectsync.sync.profile._get_default_upload_mode") as mock_upload_mode, \ - patch("tconnectsync.sync.profile._get_default_serial_number") as mock_serial_number: - mock_upload_mode.return_value = 'add' - mock_serial_number.return_value = serial_number - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PROFILES]) - - self.assertEqual(len(nightscout.uploaded_entries["profile"]), 1) - self.assertEqual(len(nightscout.uploaded_entries["profile"][0]['store']), 2) - self.assertDictEqual(nightscout.uploaded_entries["profile"][0]['store']['A'], NS_PROFILE_A) - self.assertDictEqual(nightscout.uploaded_entries["profile"][0]['store']['B'], NS_PROFILE_B) - self.assertDictEqual(nightscout.put_entries, {}) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 1) - - """ - Profile present on pump and in Nightscout with PROFILES feature enabled, with changes on pump, - replaces existing profile object with NIGHTSCOUT_PROFILE_UPLOAD_MODE=replace. - """ - def test_pump_profile_new_entry_replaced(self): - tconnect = TConnectApi() - - # datetimes are unused by the API fake - start = datetime.datetime(2021, 5, 1, 0, 0) - end = datetime.datetime(2021, 5, 3, 0, 0) - - pump_guid = '00000000-0000-0000-0000-000000000001' - serial_number = '12345' - - def fake_my_devices(): - return { - serial_number: Device( - name='test', - model_number=serial_number, - status='OK', - guid=pump_guid) - } - tconnect.webui.my_devices = fake_my_devices - - def fake_device_settings_from_guid(guid): - if guid != pump_guid: - raise RuntimeError('invalid guid') - return [DEVICE_PROFILE_A, DEVICE_PROFILE_B.activeProfile()], DEVICE_SETTINGS - - tconnect.webui.device_settings_from_guid = fake_device_settings_from_guid - - nightscout = NightscoutApi() - - def fake_current_profile(time_start=None, time_end=None): - return build_ns_profile({'A': NS_PROFILE_A}, 'A') - - nightscout.current_profile = fake_current_profile - - count = None - with patch("tconnectsync.sync.profile._get_default_upload_mode") as mock_upload_mode, \ - patch("tconnectsync.sync.profile._get_default_serial_number") as mock_serial_number: - mock_upload_mode.return_value = 'replace' - mock_serial_number.return_value = serial_number - - count = process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PROFILES]) - - self.assertDictEqual(nightscout.uploaded_entries, {}) - self.assertEqual(len(nightscout.put_entries["profile"]), 1) - self.assertEqual(len(nightscout.put_entries["profile"][0]['store']), 2) - self.assertDictEqual(nightscout.put_entries["profile"][0]['store']['A'], NS_PROFILE_A) - self.assertDictEqual(nightscout.put_entries["profile"][0]['store']['B'], NS_PROFILE_B) - self.assertListEqual(nightscout.deleted_entries, []) - self.assertEqual(count, 1) - - -if __name__ == '__main__': - unittest.main() \ No newline at end of file