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

test_process: add WS2 activity event process tests

Browse Source
This commit is contained in:
James Woglom
2021-12-04 21:37:34 -05:00
parent 1657b1c696
commit 118ee236ba
3 changed files with 162 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/sync/pump_events.py
+2 -1
View File
@@ -94,7 +94,8 @@ def ns_write_pump_sitechange_events(nightscout, siteChangeEvents, pretend=False)
nightscout,
siteChangeEvents,
lambda event: NightscoutEntry.sitechange(
created_at=event["time"]
created_at=event["time"],
reason=event["event_type"]
),
SITECHANGE_EVENTTYPE,
pretend=pretend)
tests/sync/test_basal.py
+3 -3
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
import unittest
from copy import deepcopy
import copy
from tconnectsync.sync.basal import process_ciq_basal_events
from tconnectsync.parser.tconnect import TConnectEntry
@@ -24,7 +24,7 @@ class TestBasalSync(unittest.TestCase):
@staticmethod
def get_example_ciq_basal_events():
data = deepcopy(TestBasalSync.base)
data = copy.deepcopy(TestBasalSync.base)
data["basal"]["tempDeliveryEvents"] = [
{
"y": 0.8,
@@ -85,7 +85,7 @@ class TestBasalSync(unittest.TestCase):
@staticmethod
def get_example_ciq_basal_events_with_manual_suspension():
data = deepcopy(TestBasalSync.base)
data = copy.deepcopy(TestBasalSync.base)
data["basal"]["tempDeliveryEvents"] = []
data["basal"]["algorithmDeliveryEvents"] = [
{
tests/test_process.py
+157 -2
View File
@@ -3,6 +3,7 @@
import unittest
import datetime
import pprint
import copy
from tconnectsync.process import process_time_range
from tconnectsync.parser.nightscout import EXERCISE_EVENTTYPE, IOB_ACTIVITYTYPE, SLEEP_EVENTTYPE, NightscoutEntry
@@ -18,7 +19,7 @@ class TestProcessTimeRange(unittest.TestCase):
maxDiff = None
def stub_therapy_timeline(self, time_start, time_end):
pass
return copy.deepcopy(TestBasalSync.base)
def stub_therapy_timeline_csv(self, time_start, time_end):
return {
@@ -505,7 +506,6 @@ class TestProcessTimeRange(unittest.TestCase):
nightscout = NightscoutApi()
def fake_last_uploaded_entry(event_type):
if event_type == "Sleep":
return {
@@ -540,6 +540,161 @@ class TestProcessTimeRange(unittest.TestCase):
"treatments/old_sleep"
])
"""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
process_time_range(tconnect, nightscout, start, end, pretend=False, features=[PUMP_EVENTS])
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, [])
"""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):
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()
nightscout.last_uploaded_entry = fake_last_uploaded_entry
nightscout.last_uploaded_activity = self.stub_last_uploaded_activity
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": [
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, [])
"""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
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, [])
if __name__ == '__main__':