Skip to content

Commit

Permalink
process_alarm exclude resumepumpalarm and test
Browse files Browse the repository at this point in the history
  • Loading branch information
jwoglom committed Dec 12, 2024
1 parent 24a635c commit 9493b65
Show file tree
Hide file tree
Showing 2 changed files with 74 additions and 0 deletions.
9 changes: 9 additions & 0 deletions tconnectsync/sync/tandemsource/process_alarm.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,20 @@ def process(self, events, time_start, time_end):
logger.info("Skipping Alarm event not after last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
continue

if self.skip_event(event):
continue

ns_entries.append(self.alarm_to_nsentry(event))


return ns_entries

def skip_event(self, event):
return event.alarmid in (
eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm,
eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm2
)

def write(self, ns_entries):
count = 0
for entry in ns_entries:
Expand Down
65 changes: 65 additions & 0 deletions tests/sync/tandemsource/test_process_alarm.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
#!/usr/bin/env python3

import unittest
import arrow

from tconnectsync.sync.tandemsource.process_alarm import ProcessAlarm
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.generic import Event

from ...api.fake import TConnectApi
from ...nightscout_fake import NightscoutApi

class TestProcessAlarm(unittest.TestCase):
maxDiff = None

def setUp(self):
self.tconnect = TConnectApi()
self.nightscout = NightscoutApi()
self.tconnect_device_id = 'abcdef'
self.process = ProcessAlarm(self.tconnect, self.nightscout, self.tconnect_device_id, pretend=False)

def test_resume_alarm_ignored(self):
self.nightscout.last_uploaded_entry = lambda *args, **kwargs: None

events = [
# 2024-11-11 07:24:29-05:00
Event(b'\x00\x05\x1f\xb8.\xad\x00\x0e\x91\xee\x00\x00\x00\x12\x00\x00 w\x01N\x0b\x16\x00\x00\x00\x00')
]

self.assertEqual(type(events[0]), eventtypes.LidAlarmActivated)
self.assertEqual(events[0].alarmid, eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm)

p = self.process.process(events, time_start=None, time_end=None)

self.assertEqual(len(p), 0)


def test_empty_cartridge_alarm(self):
self.nightscout.last_uploaded_entry = lambda *args, **kwargs: None

events = [
# 2024-11-17 08:44:17-05:00
Event(b'\x00\x05\x1f\xc0*a\x00\x0e\xf5\x90\x00\x00\x00\x08\x00\x00 1\x00\x00\x00gA\x1a\x1e\x84')
]

self.assertEqual(type(events[0]), eventtypes.LidAlarmActivated)
self.assertEqual(events[0].alarmid, eventtypes.LidAlarmActivated.AlarmidEnum.EmptyCartridgeAlarm)

p = self.process.process(events, time_start=None, time_end=None)

self.assertEqual(len(p), 1)
self.assertDictEqual(p[0], {
'eventType': 'Alarm',
'created_at': '2024-11-17 08:44:17-05:00',
'enteredBy': 'Pump (tconnectsync)',
'notes': 'EmptyCartridgeAlarm',
'reason': 'EmptyCartridgeAlarm',
'pump_event_id': '980368'
})




if __name__ == '__main__':
unittest.main()

0 comments on commit 9493b65

Please sign in to comment.