Add per-event-type parser tests from real captured pump-log data

One test file per event type observed in captured Tandem Source
pump-logs responses (30 types), each asserting parse/dispatch,
field round-trip, and enum/bitmask/ratio resolution against real
capture fixtures.
This commit is contained in:
James Woglom
2026-07-01 05:26:46 +00:00
parent a28e0f1ee0
commit 9d8ab283a0
31 changed files with 3331 additions and 0 deletions
@@ -0,0 +1,107 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
class TestLidAaDailyStatus(unittest.TestCase):
"""313 LID_AA_DAILY_STATUS: pumpControlState/usermode/sensorType enums.
Fixtures are real captured pump-log events copied verbatim, including the
extra weightUnit/weight/currentTdIpop keys the parser ignores.
"""
maxDiff = None
def setUp(self):
# Real capture: pumpControlState 3 -> PcmClosedLoop.
self.fixtureClosedLoop = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 313,
"sequenceGroup": 0,
"sequenceNumber": 393118,
"pumpDateTime": "2026-04-30T00:00:06",
"eventProperties": {
"pumpControlState": 3, "usermode": 1, "sensorType": 3,
"weightUnit": 0, "weight": 0, "currentTdIpop": 0,
},
"estimatedDateTime": "2026-04-30T00:00:06Z",
}
# Real capture: pumpControlState 0 -> PcmNoControlNoCartridgeInstalled.
self.fixtureNoControl = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 313,
"sequenceGroup": 0,
"sequenceNumber": 420321,
"pumpDateTime": "2026-05-08T00:00:06",
"eventProperties": {
"pumpControlState": 0, "usermode": 1, "sensorType": 3,
"weightUnit": 0, "weight": 0, "currentTdIpop": 0,
},
"estimatedDateTime": "2026-05-08T00:00:06Z",
}
def test_dispatches_to_lidaadailystatus(self):
self.assertIsInstance(Event(self.fixtureClosedLoop), eventtypes.LidAaDailyStatus)
self.assertIsInstance(Event(self.fixtureNoControl), eventtypes.LidAaDailyStatus)
def test_envelope_fields(self):
ev = Event(self.fixtureClosedLoop)
self.assertEqual(ev.eventId, 313)
self.assertEqual(ev.seqNum, 393118)
self.assertEqual(Event(self.fixtureNoControl).seqNum, 420321)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureClosedLoop)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T00:00:06")
def test_pumpcontrolstate_closed_loop(self):
ev = Event(self.fixtureClosedLoop)
self.assertEqual(ev.pumpcontrolstateRaw, 3)
self.assertEqual(ev.pumpcontrolstate,
eventtypes.LidAaDailyStatus.PumpcontrolstateEnum.PcmClosedLoop)
def test_pumpcontrolstate_no_control_zero_value(self):
# pumpControlState 0 must resolve, not be treated as missing.
ev = Event(self.fixtureNoControl)
self.assertEqual(ev.pumpcontrolstateRaw, 0)
self.assertEqual(ev.pumpcontrolstate,
eventtypes.LidAaDailyStatus.PumpcontrolstateEnum.PcmNoControlNoCartridgeInstalled)
def test_usermode_resolves(self):
ev = Event(self.fixtureClosedLoop)
self.assertEqual(ev.usermodeRaw, 1)
self.assertEqual(ev.usermode,
eventtypes.LidAaDailyStatus.UsermodeEnum.Sleeping)
def test_sensortype_resolves(self):
ev = Event(self.fixtureClosedLoop)
self.assertEqual(ev.sensortypeRaw, 3)
self.assertEqual(ev.sensortype,
eventtypes.LidAaDailyStatus.SensortypeEnum.CgmTypeDexcomG7)
def test_unknown_keys_ignored(self):
# weightUnit/weight/currentTdIpop are not in the schema and must be
# dropped without raising or becoming attributes.
ev = Event(self.fixtureClosedLoop)
self.assertFalse(hasattr(ev, "weightUnit"))
self.assertFalse(hasattr(ev, "weight"))
self.assertFalse(hasattr(ev, "currentTdIpop"))
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureClosedLoop)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 313)
self.assertEqual(d["name"], "LID_AA_DAILY_STATUS")
self.assertEqual(d["seqNum"], 393118)
self.assertEqual(d["pumpcontrolstateRaw"], 3)
self.assertEqual(d["usermodeRaw"], 1)
self.assertEqual(d["sensortypeRaw"], 3)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,147 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidAaPcmChange(unittest.TestCase):
"""230 LID_AA_PCM_CHANGE: currentPcm/previousPcm resolve to a PCM enum,
and the boolean-ish fields resolve to False/True enum members. All
fixtures are real captured events copied verbatim."""
maxDiff = None
def setUp(self):
# currentPcm:0 (NoControl) from previousPcm:3 (ClosedLoop), suspended.
self.fixtureSuspendedNoControl = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 230,
"sequenceGroup": 0,
"sequenceNumber": 394337,
"pumpDateTime": "2026-04-30T10:01:49",
"eventProperties": {
"currentPcm": 0, "previousPcm": 3, "pumpSuspended": 1,
"calculationAvailable": 1, "cgmAvailable": 1,
"closedLoopPreferred": 1, "sufficientClosedLoopParams": 1,
},
"estimatedDateTime": "2026-04-30T10:01:49Z",
}
# currentPcm:3 (ClosedLoop) from previousPcm:0 (NoControl), not suspended.
self.fixtureResumedClosedLoop = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 230,
"sequenceGroup": 0,
"sequenceNumber": 394430,
"pumpDateTime": "2026-04-30T10:16:31",
"eventProperties": {
"currentPcm": 3, "previousPcm": 0, "pumpSuspended": 0,
"calculationAvailable": 1, "cgmAvailable": 1,
"closedLoopPreferred": 1, "sufficientClosedLoopParams": 1,
},
"estimatedDateTime": "2026-04-30T10:16:31Z",
}
# currentPcm:2 (Pining) with cgmAvailable:0 -> FalseVal boolean-ish field.
self.fixturePiningNoCgm = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 230,
"sequenceGroup": 0,
"sequenceNumber": 409128,
"pumpDateTime": "2026-05-04T18:58:22",
"eventProperties": {
"currentPcm": 2, "previousPcm": 3, "pumpSuspended": 0,
"calculationAvailable": 1, "cgmAvailable": 0,
"closedLoopPreferred": 1, "sufficientClosedLoopParams": 1,
},
"estimatedDateTime": "2026-05-04T18:58:22Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureSuspendedNoControl)
self.assertIsInstance(ev, eventtypes.LidAaPcmChange)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureSuspendedNoControl)
self.assertEqual(ev.eventId, 230)
self.assertEqual(ev.seqNum, 394337)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureSuspendedNoControl)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T10:01:49")
def test_pcm_enums_no_control_from_closed_loop(self):
# currentPcm:0 -> NoControl, previousPcm:3 -> ClosedLoop
ev = Event(self.fixtureSuspendedNoControl)
self.assertEqual(ev.currentpcmRaw, 0)
self.assertEqual(ev.currentpcm,
eventtypes.LidAaPcmChange.CurrentpcmEnum.NoControl)
self.assertEqual(ev.previouspcmRaw, 3)
self.assertEqual(ev.previouspcm,
eventtypes.LidAaPcmChange.PreviouspcmEnum.ClosedLoop)
def test_pcm_enums_closed_loop_from_no_control(self):
# currentPcm:3 -> ClosedLoop, previousPcm:0 -> NoControl
ev = Event(self.fixtureResumedClosedLoop)
self.assertEqual(ev.currentpcm,
eventtypes.LidAaPcmChange.CurrentpcmEnum.ClosedLoop)
self.assertEqual(ev.previouspcm,
eventtypes.LidAaPcmChange.PreviouspcmEnum.NoControl)
def test_pcm_enum_pining(self):
# currentPcm:2 -> Pining
ev = Event(self.fixturePiningNoCgm)
self.assertEqual(ev.currentpcmRaw, 2)
self.assertEqual(ev.currentpcm,
eventtypes.LidAaPcmChange.CurrentpcmEnum.Pining)
def test_boolean_fields_when_suspended(self):
ev = Event(self.fixtureSuspendedNoControl)
self.assertEqual(ev.pumpsuspendedRaw, 1)
self.assertEqual(ev.pumpsuspended,
eventtypes.LidAaPcmChange.PumpsuspendedEnum.TrueVal)
self.assertEqual(ev.calculationavailable,
eventtypes.LidAaPcmChange.CalculationavailableEnum.TrueVal)
self.assertEqual(ev.cgmavailable,
eventtypes.LidAaPcmChange.CgmavailableEnum.TrueVal)
self.assertEqual(ev.closedlooppreferred,
eventtypes.LidAaPcmChange.ClosedlooppreferredEnum.TrueVal)
self.assertEqual(ev.sufficientclosedloopparams,
eventtypes.LidAaPcmChange.SufficientclosedloopparamsEnum.TrueVal)
def test_pump_suspended_false(self):
# pumpSuspended:0 -> FalseVal (0 must not be treated as missing)
ev = Event(self.fixtureResumedClosedLoop)
self.assertEqual(ev.pumpsuspendedRaw, 0)
self.assertEqual(ev.pumpsuspended,
eventtypes.LidAaPcmChange.PumpsuspendedEnum.FalseVal)
def test_cgm_available_false(self):
# cgmAvailable:0 -> FalseVal while other boolean-ish fields stay TrueVal
ev = Event(self.fixturePiningNoCgm)
self.assertEqual(ev.cgmavailableRaw, 0)
self.assertEqual(ev.cgmavailable,
eventtypes.LidAaPcmChange.CgmavailableEnum.FalseVal)
self.assertEqual(ev.calculationavailable,
eventtypes.LidAaPcmChange.CalculationavailableEnum.TrueVal)
self.assertEqual(ev.closedlooppreferred,
eventtypes.LidAaPcmChange.ClosedlooppreferredEnum.TrueVal)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureSuspendedNoControl,
self.fixtureResumedClosedLoop,
self.fixturePiningNoCgm):
ev = Event(fixture)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 230)
self.assertEqual(d["name"], "LID_AA_PCM_CHANGE")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,163 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidAaUserModeChange(unittest.TestCase):
"""229 LID_AA_USER_MODE_CHANGE, from real captured pump-log events."""
maxDiff = None
def setUp(self):
# Normal <- Sleeping, requestedAction StopSleep, activeSleepSchedule [0].
self.fixtureStopSleep = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 229,
"sequenceGroup": 0,
"sequenceNumber": 456851,
"pumpDateTime": "2026-05-18T10:15:53",
"eventProperties": {
"currentUserMode": 0, "previousUserMode": 1, "requestedAction": 2,
"spareA3": 0, "sleepStartedByGui": 1, "activeSleepSchedule": [0],
"spareB6": 0, "exerciseStoppedByTimer": 0, "exerciseChoice": 0,
"exerciseTime": 0, "eatingSoonStoppedByTimer": 0,
},
"estimatedDateTime": "2026-05-18T10:15:53Z",
}
# Sleeping <- Normal, requestedAction StartSleep, activeSleepSchedule [0].
self.fixtureStartSleep = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 229,
"sequenceGroup": 0,
"sequenceNumber": 456855,
"pumpDateTime": "2026-05-18T10:16:00",
"eventProperties": {
"currentUserMode": 1, "previousUserMode": 0, "requestedAction": 1,
"spareA3": 0, "sleepStartedByGui": 1, "activeSleepSchedule": [0],
"spareB6": 0, "exerciseStoppedByTimer": 0, "exerciseChoice": 0,
"exerciseTime": 0, "eatingSoonStoppedByTimer": 0,
},
"estimatedDateTime": "2026-05-18T10:16:00Z",
}
# Exercising <- Normal, requestedAction StartExercise, empty activeSleepSchedule.
self.fixtureStartExercise = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 229,
"sequenceGroup": 0,
"sequenceNumber": 456961,
"pumpDateTime": "2026-05-18T10:20:04",
"eventProperties": {
"currentUserMode": 2, "previousUserMode": 0, "requestedAction": 3,
"spareA3": 0, "sleepStartedByGui": 0, "activeSleepSchedule": [],
"spareB6": 0, "exerciseStoppedByTimer": 0, "exerciseChoice": 0,
"exerciseTime": 0, "eatingSoonStoppedByTimer": 0,
},
"estimatedDateTime": "2026-05-18T10:20:04Z",
}
# Sleeping <- Exercising, requestedAction StopExercise, activeSleepSchedule [0].
self.fixtureStopExercise = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 229,
"sequenceGroup": 0,
"sequenceNumber": 456965,
"pumpDateTime": "2026-05-18T10:20:15",
"eventProperties": {
"currentUserMode": 1, "previousUserMode": 2, "requestedAction": 4,
"spareA3": 0, "sleepStartedByGui": 0, "activeSleepSchedule": [0],
"spareB6": 0, "exerciseStoppedByTimer": 0, "exerciseChoice": 0,
"exerciseTime": 0, "eatingSoonStoppedByTimer": 0,
},
"estimatedDateTime": "2026-05-18T10:20:15Z",
}
def test_dispatches_to_correct_class(self):
for fx in (self.fixtureStopSleep, self.fixtureStartSleep,
self.fixtureStartExercise, self.fixtureStopExercise):
ev = Event(fx)
self.assertIsInstance(ev, eventtypes.LidAaUserModeChange)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureStopSleep)
self.assertEqual(ev.eventId, 229)
self.assertEqual(ev.seqNum, 456851)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-18T10:15:53")
def test_envelope_fields_other_fixture(self):
ev = Event(self.fixtureStartExercise)
self.assertEqual(ev.eventId, 229)
self.assertEqual(ev.seqNum, 456961)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-18T10:20:04")
def test_stop_sleep_enums(self):
ev = Event(self.fixtureStopSleep)
self.assertEqual(ev.currentusermodeRaw, 0)
self.assertEqual(ev.currentusermode,
eventtypes.LidAaUserModeChange.CurrentusermodeEnum.Normal)
self.assertEqual(ev.previoususermodeRaw, 1)
self.assertEqual(ev.previoususermode,
eventtypes.LidAaUserModeChange.PrevioususermodeEnum.Sleeping)
self.assertEqual(ev.requestedactionRaw, 2)
self.assertEqual(ev.requestedaction,
eventtypes.LidAaUserModeChange.RequestedactionEnum.StopSleep)
def test_start_sleep_enums(self):
ev = Event(self.fixtureStartSleep)
self.assertEqual(ev.currentusermode,
eventtypes.LidAaUserModeChange.CurrentusermodeEnum.Sleeping)
self.assertEqual(ev.previoususermode,
eventtypes.LidAaUserModeChange.PrevioususermodeEnum.Normal)
self.assertEqual(ev.requestedaction,
eventtypes.LidAaUserModeChange.RequestedactionEnum.StartSleep)
def test_start_exercise_enums(self):
ev = Event(self.fixtureStartExercise)
self.assertEqual(ev.currentusermode,
eventtypes.LidAaUserModeChange.CurrentusermodeEnum.Exercising)
self.assertEqual(ev.previoususermode,
eventtypes.LidAaUserModeChange.PrevioususermodeEnum.Normal)
self.assertEqual(ev.requestedaction,
eventtypes.LidAaUserModeChange.RequestedactionEnum.StartExercise)
def test_stop_exercise_enums(self):
ev = Event(self.fixtureStopExercise)
self.assertEqual(ev.currentusermode,
eventtypes.LidAaUserModeChange.CurrentusermodeEnum.Sleeping)
self.assertEqual(ev.previoususermode,
eventtypes.LidAaUserModeChange.PrevioususermodeEnum.Exercising)
self.assertEqual(ev.requestedaction,
eventtypes.LidAaUserModeChange.RequestedactionEnum.StopExercise)
def test_active_sleep_schedule_single_bit(self):
# activeSleepSchedule:[0] -> 1<<0 == 1 -> SleepSchedule1IsActive
ev = Event(self.fixtureStopSleep)
self.assertEqual(ev.activesleepscheduleRaw, 1)
self.assertEqual(ev.activesleepschedule,
eventtypes.LidAaUserModeChange.ActivesleepscheduleBitmask.SleepSchedule1IsActive)
def test_active_sleep_schedule_empty(self):
# An empty array folds to 0 (empty IntFlag), not None.
ev = Event(self.fixtureStartExercise)
self.assertEqual(ev.activesleepscheduleRaw, 0)
self.assertEqual(ev.activesleepschedule,
eventtypes.LidAaUserModeChange.ActivesleepscheduleBitmask(0))
self.assertEqual(int(ev.activesleepschedule), 0)
def test_todict_json_serializable(self):
for fx in (self.fixtureStopSleep, self.fixtureStartSleep,
self.fixtureStartExercise, self.fixtureStopExercise):
ev = Event(fx)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 229)
self.assertEqual(d["name"], "LID_AA_USER_MODE_CHANGE")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,100 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidAlarmActivated(unittest.TestCase):
"""5: LID_ALARM_ACTIVATED. Real captured pump-log events; alarmId is a
dictionary transform resolving to an AlarmidEnum member."""
maxDiff = None
def setUp(self):
# Real capture: alarmId 18 -> RESUME_PUMP_ALARM.
self.fixtureResumePumpAlarm = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 5,
"sequenceGroup": 0,
"sequenceNumber": 398724,
"pumpDateTime": "2026-05-01T17:08:10",
"eventProperties": {"alarmId": 18, "faultLocatorData": 8311, "param1": 3993668, "param2": 0},
"estimatedDateTime": "2026-05-01T17:08:10Z",
}
# Real capture: alarmId 23 -> RESUME_PUMP_ALARM2.
self.fixtureResumePumpAlarm2 = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 5,
"sequenceGroup": 0,
"sequenceNumber": 398725,
"pumpDateTime": "2026-05-01T17:08:10",
"eventProperties": {"alarmId": 23, "faultLocatorData": 8311, "param1": 18, "param2": 0},
"estimatedDateTime": "2026-05-01T17:08:10Z",
}
def test_dispatches_to_correct_class(self):
self.assertIsInstance(Event(self.fixtureResumePumpAlarm), eventtypes.LidAlarmActivated)
self.assertIsInstance(Event(self.fixtureResumePumpAlarm2), eventtypes.LidAlarmActivated)
self.assertNotIsInstance(Event(self.fixtureResumePumpAlarm), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.eventId, 5)
self.assertEqual(ev.seqNum, 398724)
ev2 = Event(self.fixtureResumePumpAlarm2)
self.assertEqual(ev2.eventId, 5)
self.assertEqual(ev2.seqNum, 398725)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-05-01T17:08:10")
def test_alarmid_resolves_resume_pump_alarm(self):
# alarmId:18 -> RESUME_PUMP_ALARM
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.alarmidRaw, 18)
self.assertEqual(ev.alarmid, eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm)
def test_alarmid_resolves_resume_pump_alarm2(self):
# alarmId:23 -> RESUME_PUMP_ALARM2
ev = Event(self.fixtureResumePumpAlarm2)
self.assertEqual(ev.alarmidRaw, 23)
self.assertEqual(ev.alarmid, eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm2)
def test_plain_fields(self):
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.faultlocatordata, 8311)
self.assertEqual(ev.param1, 3993668)
self.assertEqual(ev.param2, 0)
ev2 = Event(self.fixtureResumePumpAlarm2)
self.assertEqual(ev2.faultlocatordata, 8311)
self.assertEqual(ev2.param1, 18)
self.assertEqual(ev2.param2, 0)
def test_todict_is_json_serializable(self):
for f in (self.fixtureResumePumpAlarm, self.fixtureResumePumpAlarm2):
ev = Event(f)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 5)
self.assertEqual(d["name"], "LID_ALARM_ACTIVATED")
self.assertEqual(Event(self.fixtureResumePumpAlarm).todict(), {
"id": 5,
"name": "LID_ALARM_ACTIVATED",
"seqNum": 398724,
"eventTimestamp": "2026-05-01T17:08:10-04:00",
"alarmidRaw": 18,
"faultlocatordata": 8311,
"param1": 3993668,
"param2": 0,
})
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,86 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidAlarmCleared(unittest.TestCase):
"""28: LID_ALARM_CLEARED. Real captured pump-log events; alarmId is a
dictionary transform resolving to an AlarmidEnum member."""
maxDiff = None
def setUp(self):
# Real capture: alarmId 18 -> RESUME_PUMP_ALARM.
self.fixtureResumePumpAlarm = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 28,
"sequenceGroup": 0,
"sequenceNumber": 398734,
"pumpDateTime": "2026-05-01T17:11:22",
"eventProperties": {"alarmId": 18},
"estimatedDateTime": "2026-05-01T17:11:22Z",
}
# Real capture: alarmId 23 -> RESUME_PUMP_ALARM2.
self.fixtureResumePumpAlarm2 = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 28,
"sequenceGroup": 0,
"sequenceNumber": 398733,
"pumpDateTime": "2026-05-01T17:11:22",
"eventProperties": {"alarmId": 23},
"estimatedDateTime": "2026-05-01T17:11:22Z",
}
def test_dispatches_to_correct_class(self):
self.assertIsInstance(Event(self.fixtureResumePumpAlarm), eventtypes.LidAlarmCleared)
self.assertIsInstance(Event(self.fixtureResumePumpAlarm2), eventtypes.LidAlarmCleared)
self.assertNotIsInstance(Event(self.fixtureResumePumpAlarm), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.eventId, 28)
self.assertEqual(ev.seqNum, 398734)
ev2 = Event(self.fixtureResumePumpAlarm2)
self.assertEqual(ev2.eventId, 28)
self.assertEqual(ev2.seqNum, 398733)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-05-01T17:11:22")
def test_alarmid_resolves_resume_pump_alarm(self):
# alarmId:18 -> RESUME_PUMP_ALARM
ev = Event(self.fixtureResumePumpAlarm)
self.assertEqual(ev.alarmidRaw, 18)
self.assertEqual(ev.alarmid, eventtypes.LidAlarmCleared.AlarmidEnum.ResumePumpAlarm)
def test_alarmid_resolves_resume_pump_alarm2(self):
# alarmId:23 -> RESUME_PUMP_ALARM2
ev = Event(self.fixtureResumePumpAlarm2)
self.assertEqual(ev.alarmidRaw, 23)
self.assertEqual(ev.alarmid, eventtypes.LidAlarmCleared.AlarmidEnum.ResumePumpAlarm2)
def test_todict_is_json_serializable(self):
for f in (self.fixtureResumePumpAlarm, self.fixtureResumePumpAlarm2):
ev = Event(f)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 28)
self.assertEqual(d["name"], "LID_ALARM_CLEARED")
self.assertEqual(Event(self.fixtureResumePumpAlarm).todict(), {
"id": 28,
"name": "LID_ALARM_CLEARED",
"seqNum": 398734,
"eventTimestamp": "2026-05-01T17:11:22-04:00",
"alarmidRaw": 18,
})
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,155 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidAlertActivated(unittest.TestCase):
"""4: LID_ALERT_ACTIVATED. alertid is a dictionary/enum field resolved
through alertidRaw; faultlocatordata/param1/param2 are plain numeric fields.
All fixtures are real captured pump-log events copied verbatim."""
maxDiff = None
def setUp(self):
# alertId:50 -> DefaultAlert50; integer param2, zero fault/param1.
self.fixtureDefaultAlert50 = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 4,
"sequenceGroup": 0,
"sequenceNumber": 395022,
"pumpDateTime": "2026-04-30T14:26:30",
"eventProperties": {"alertId": 50, "faultLocatorData": 0, "param1": 0, "param2": 866},
"estimatedDateTime": "2026-04-30T14:26:30Z",
}
# alertId:51 -> ControlIqLow.
self.fixtureControlIqLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 4,
"sequenceGroup": 0,
"sequenceNumber": 396263,
"pumpDateTime": "2026-04-30T23:07:17",
"eventProperties": {"alertId": 51, "faultLocatorData": 0, "param1": 0, "param2": 877},
"estimatedDateTime": "2026-04-30T23:07:17Z",
}
# alertId:0 -> LowInsulinAlert (zero must resolve, not read as missing);
# non-zero faultLocatorData and a fractional float param2.
self.fixtureLowInsulinFloat = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 4,
"sequenceGroup": 0,
"sequenceNumber": 398333,
"pumpDateTime": "2026-05-01T15:57:06",
"eventProperties": {"alertId": 0, "faultLocatorData": 8242, "param1": 102, "param2": 249.76517},
"estimatedDateTime": "2026-05-01T15:57:06Z",
}
# alertId:14 -> IncompleteFillTubingAlert; all-zero params.
self.fixtureIncompleteFillTubing = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 4,
"sequenceGroup": 0,
"sequenceNumber": 420528,
"pumpDateTime": "2026-05-08T00:52:18",
"eventProperties": {"alertId": 14, "faultLocatorData": 8378, "param1": 0, "param2": 0},
"estimatedDateTime": "2026-05-08T00:52:18Z",
}
# alertId:2 -> LowPowerAlert.
self.fixtureLowPower = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 4,
"sequenceGroup": 0,
"sequenceNumber": 436691,
"pumpDateTime": "2026-05-12T17:52:10",
"eventProperties": {"alertId": 2, "faultLocatorData": 8306, "param1": 20, "param2": 1},
"estimatedDateTime": "2026-05-12T17:52:10Z",
}
def test_dispatches_to_correct_class(self):
for f in (self.fixtureDefaultAlert50, self.fixtureControlIqLow,
self.fixtureLowInsulinFloat, self.fixtureIncompleteFillTubing,
self.fixtureLowPower):
ev = Event(f)
self.assertIsInstance(ev, eventtypes.LidAlertActivated)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureDefaultAlert50)
self.assertEqual(ev.eventId, 4)
self.assertEqual(ev.seqNum, 395022)
ev = Event(self.fixtureControlIqLow)
self.assertEqual(ev.eventId, 4)
self.assertEqual(ev.seqNum, 396263)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureDefaultAlert50)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T14:26:30")
ev = Event(self.fixtureLowPower)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-12T17:52:10")
def test_plain_fields(self):
ev = Event(self.fixtureLowInsulinFloat)
self.assertEqual(ev.faultlocatordata, 8242)
self.assertEqual(ev.param1, 102)
self.assertAlmostEqual(ev.param2, 249.76517)
ev = Event(self.fixtureIncompleteFillTubing)
self.assertEqual(ev.faultlocatordata, 8378)
self.assertEqual(ev.param1, 0)
self.assertEqual(ev.param2, 0)
def test_alertid_enum_resolves_from_raw_int(self):
ev = Event(self.fixtureDefaultAlert50)
self.assertEqual(ev.alertidRaw, 50)
self.assertEqual(ev.alertid,
eventtypes.LidAlertActivated.AlertidEnum.DefaultAlert50)
ev = Event(self.fixtureControlIqLow)
self.assertEqual(ev.alertidRaw, 51)
self.assertEqual(ev.alertid,
eventtypes.LidAlertActivated.AlertidEnum.ControlIqLow)
ev = Event(self.fixtureIncompleteFillTubing)
self.assertEqual(ev.alertidRaw, 14)
self.assertEqual(ev.alertid,
eventtypes.LidAlertActivated.AlertidEnum.IncompleteFillTubingAlert)
ev = Event(self.fixtureLowPower)
self.assertEqual(ev.alertidRaw, 2)
self.assertEqual(ev.alertid,
eventtypes.LidAlertActivated.AlertidEnum.LowPowerAlert)
def test_alertid_zero_value_resolves(self):
# alertId:0 -> LowInsulinAlert (0 must not be treated as missing).
ev = Event(self.fixtureLowInsulinFloat)
self.assertEqual(ev.alertidRaw, 0)
self.assertEqual(ev.alertid,
eventtypes.LidAlertActivated.AlertidEnum.LowInsulinAlert)
def test_todict_is_json_serializable(self):
for f in (self.fixtureDefaultAlert50, self.fixtureLowInsulinFloat,
self.fixtureLowPower):
ev = Event(f)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 4)
self.assertEqual(d["name"], "LID_ALERT_ACTIVATED")
def test_todict_round_trips_fields(self):
ev = Event(self.fixtureLowInsulinFloat)
d = ev.todict()
self.assertEqual(d["seqNum"], 398333)
self.assertEqual(d["alertidRaw"], 0)
self.assertEqual(d["faultlocatordata"], 8242)
self.assertEqual(d["param1"], 102)
self.assertAlmostEqual(d["param2"], 249.76517)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,113 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
# Real LID_ALERT_CLEARED (eventCode 26) events copied verbatim from a captured
# pump-log response. Each has a different alertId (dictionary enum) value.
class TestLidAlertCleared(unittest.TestCase):
maxDiff = None
def setUp(self):
# alertId 0 -> LowInsulinAlert (0 must not be treated as missing)
self.fixtureLowInsulin = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 26,
"sequenceGroup": 0,
"sequenceNumber": 398601,
"pumpDateTime": "2026-05-01T16:49:45",
"eventProperties": {"alertId": 0, "faultLocatorData": 0},
"estimatedDateTime": "2026-05-01T16:49:45Z",
}
# alertId 2 -> LowPowerAlert
self.fixtureLowPower = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 26,
"sequenceGroup": 0,
"sequenceNumber": 439055,
"pumpDateTime": "2026-05-13T10:06:58",
"eventProperties": {"alertId": 2, "faultLocatorData": 0},
"estimatedDateTime": "2026-05-13T10:06:58Z",
}
# alertId 14 -> IncompleteFillTubingAlert
self.fixtureIncompleteFillTubing = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 26,
"sequenceGroup": 0,
"sequenceNumber": 420534,
"pumpDateTime": "2026-05-08T00:52:50",
"eventProperties": {"alertId": 14, "faultLocatorData": 0},
"estimatedDateTime": "2026-05-08T00:52:50Z",
}
# alertId 51 -> ControlIqLow
self.fixtureControlIqLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 26,
"sequenceGroup": 0,
"sequenceNumber": 396276,
"pumpDateTime": "2026-04-30T23:12:18",
"eventProperties": {"alertId": 51, "faultLocatorData": 0},
"estimatedDateTime": "2026-04-30T23:12:18Z",
}
def test_dispatches_to_lidalertcleared(self):
for fx in (self.fixtureLowInsulin, self.fixtureLowPower,
self.fixtureIncompleteFillTubing, self.fixtureControlIqLow):
ev = Event(fx)
self.assertIsInstance(ev, eventtypes.LidAlertCleared)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureControlIqLow)
self.assertEqual(ev.eventId, 26)
self.assertEqual(ev.seqNum, 396276)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureControlIqLow)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T23:12:18")
def test_plain_field_round_trips(self):
ev = Event(self.fixtureControlIqLow)
self.assertEqual(ev.faultlocatordata, 0)
def test_alertid_raw_round_trips(self):
self.assertEqual(Event(self.fixtureLowInsulin).alertidRaw, 0)
self.assertEqual(Event(self.fixtureLowPower).alertidRaw, 2)
self.assertEqual(Event(self.fixtureIncompleteFillTubing).alertidRaw, 14)
self.assertEqual(Event(self.fixtureControlIqLow).alertidRaw, 51)
def test_alertid_resolves_to_enum(self):
E = eventtypes.LidAlertCleared.AlertidEnum
self.assertEqual(Event(self.fixtureLowInsulin).alertid, E.LowInsulinAlert)
self.assertEqual(Event(self.fixtureLowPower).alertid, E.LowPowerAlert)
self.assertEqual(Event(self.fixtureIncompleteFillTubing).alertid,
E.IncompleteFillTubingAlert)
self.assertEqual(Event(self.fixtureControlIqLow).alertid, E.ControlIqLow)
def test_alertid_zero_resolves(self):
# alertId 0 must resolve, not be dropped as a falsy/missing value.
ev = Event(self.fixtureLowInsulin)
self.assertEqual(ev.alertidRaw, 0)
self.assertEqual(ev.alertid,
eventtypes.LidAlertCleared.AlertidEnum.LowInsulinAlert)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureControlIqLow)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 26)
self.assertEqual(d["name"], "LID_ALERT_CLEARED")
self.assertEqual(d["seqNum"], 396276)
self.assertEqual(d["alertidRaw"], 51)
self.assertEqual(d["faultlocatordata"], 0)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
class TestLidBasalDelivery(unittest.TestCase):
"""279 LID_BASAL_DELIVERY: commandedRateSource enum + milliunits/hr rates.
Fixtures are real captured pump-log dicts (copied verbatim), each with a
different commandedRateSource so every enum member is exercised. reservedA2
and spareA3 are ignored by the parser and not asserted on.
"""
maxDiff = None
def setUp(self):
# commandedRateSource:0 -> Suspended; commandedRate 0, algorithmRate/tempRate sentinel.
self.fixtureSuspended = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 279,
"sequenceGroup": 0,
"sequenceNumber": 394356,
"pumpDateTime": "2026-04-30T10:04:05",
"eventProperties": {
"commandedRateSource": 0, "reservedA2": 3, "spareA3": 0,
"commandedRate": 0, "profileBasalRate": 1200,
"algorithmRate": 65535, "tempRate": 65535,
},
"estimatedDateTime": "2026-04-30T10:04:05Z",
}
# commandedRateSource:1 -> Profile; commandedRate follows profileBasalRate.
self.fixtureProfile = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 279,
"sequenceGroup": 0,
"sequenceNumber": 409132,
"pumpDateTime": "2026-05-04T18:58:52",
"eventProperties": {
"commandedRateSource": 1, "reservedA2": 3, "spareA3": 0,
"commandedRate": 1000, "profileBasalRate": 1000,
"algorithmRate": 65535, "tempRate": 65535,
},
"estimatedDateTime": "2026-05-04T18:58:52Z",
}
# commandedRateSource:2 -> TempRate; real tempRate=500 (not the 65535 sentinel).
self.fixtureTempRate = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 279,
"sequenceGroup": 0,
"sequenceNumber": 449599,
"pumpDateTime": "2026-05-16T11:16:07",
"eventProperties": {
"commandedRateSource": 2, "reservedA2": 0, "spareA3": 0,
"commandedRate": 500, "profileBasalRate": 1000,
"algorithmRate": 65535, "tempRate": 500,
},
"estimatedDateTime": "2026-05-16T11:16:07Z",
}
# commandedRateSource:3 -> Algorithm; commandedRate follows algorithmRate, tempRate sentinel.
self.fixtureAlgorithm = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 279,
"sequenceGroup": 0,
"sequenceNumber": 393151,
"pumpDateTime": "2026-04-30T00:08:30",
"eventProperties": {
"commandedRateSource": 3, "reservedA2": 3, "spareA3": 0,
"commandedRate": 1061, "profileBasalRate": 1000,
"algorithmRate": 1061, "tempRate": 65535,
},
"estimatedDateTime": "2026-04-30T00:08:30Z",
}
# commandedRateSource:4 -> TempRateAndAlgorithm; both algorithmRate and real tempRate=600 present.
self.fixtureTempRateAndAlgorithm = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 279,
"sequenceGroup": 0,
"sequenceNumber": 449571,
"pumpDateTime": "2026-05-16T11:01:05",
"eventProperties": {
"commandedRateSource": 4, "reservedA2": 3, "spareA3": 0,
"commandedRate": 500, "profileBasalRate": 1000,
"algorithmRate": 500, "tempRate": 500,
},
"estimatedDateTime": "2026-05-16T11:01:05Z",
}
def test_dispatches_to_correct_class(self):
for fixture in (self.fixtureSuspended, self.fixtureProfile,
self.fixtureTempRate, self.fixtureAlgorithm,
self.fixtureTempRateAndAlgorithm):
ev = Event(fixture)
self.assertIsInstance(ev, eventtypes.LidBasalDelivery)
def test_envelope_fields(self):
ev = Event(self.fixtureAlgorithm)
self.assertEqual(ev.eventId, 279)
self.assertEqual(ev.seqNum, 393151)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T00:08:30")
def test_rate_fields_round_trip(self):
ev = Event(self.fixtureAlgorithm)
self.assertEqual(ev.commandedRate, 1061)
self.assertEqual(ev.profileBasalRate, 1000)
self.assertEqual(ev.algorithmRate, 1061)
self.assertEqual(ev.tempRate, 65535)
def test_temp_rate_sentinel_vs_real(self):
# Algorithm capture uses the 65535 sentinel; TempRate capture has a real value.
self.assertEqual(Event(self.fixtureAlgorithm).tempRate, 65535)
self.assertEqual(Event(self.fixtureTempRate).tempRate, 500)
self.assertEqual(Event(self.fixtureTempRateAndAlgorithm).tempRate, 500)
def test_commanded_rate_source_suspended(self):
ev = Event(self.fixtureSuspended)
self.assertEqual(ev.commandedRateSourceRaw, 0)
self.assertEqual(ev.commandedRateSource,
eventtypes.LidBasalDelivery.CommandedratesourceEnum.Suspended)
def test_commanded_rate_source_profile(self):
ev = Event(self.fixtureProfile)
self.assertEqual(ev.commandedRateSourceRaw, 1)
self.assertEqual(ev.commandedRateSource,
eventtypes.LidBasalDelivery.CommandedratesourceEnum.Profile)
def test_commanded_rate_source_temp_rate(self):
ev = Event(self.fixtureTempRate)
self.assertEqual(ev.commandedRateSourceRaw, 2)
self.assertEqual(ev.commandedRateSource,
eventtypes.LidBasalDelivery.CommandedratesourceEnum.TempRate)
def test_commanded_rate_source_algorithm(self):
ev = Event(self.fixtureAlgorithm)
self.assertEqual(ev.commandedRateSourceRaw, 3)
self.assertEqual(ev.commandedRateSource,
eventtypes.LidBasalDelivery.CommandedratesourceEnum.Algorithm)
def test_commanded_rate_source_temp_rate_and_algorithm(self):
ev = Event(self.fixtureTempRateAndAlgorithm)
self.assertEqual(ev.commandedRateSourceRaw, 4)
self.assertEqual(ev.commandedRateSource,
eventtypes.LidBasalDelivery.CommandedratesourceEnum.TempRateAndAlgorithm)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureSuspended, self.fixtureProfile,
self.fixtureTempRate, self.fixtureAlgorithm,
self.fixtureTempRateAndAlgorithm):
d = Event(fixture).todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 279)
self.assertEqual(d["name"], "LID_BASAL_DELIVERY")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,115 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBgReadingTaken(unittest.TestCase):
maxDiff = None
def setUp(self):
# Real captured LID_BG_READING_TAKEN (eventCode 16) events, copied
# verbatim. The two fixtures differ only in bgEntryType.
self.fixtureManualEntry = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 16,
"sequenceGroup": 0,
"sequenceNumber": 456822,
"pumpDateTime": "2026-05-18T10:15:14",
"eventProperties": {
"bg": 151, "cgmCalibration": 0, "bgEntryType": 0,
"iob": 1.1809407, "targetBg": 110, "isf": 30,
"selectedIob": 1, "bgSourceType": 1,
},
"estimatedDateTime": "2026-05-18T10:15:14Z",
}
self.fixtureAutoPopulated = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 16,
"sequenceGroup": 0,
"sequenceNumber": 394632,
"pumpDateTime": "2026-04-30T11:57:36",
"eventProperties": {
"bg": 164, "cgmCalibration": 0, "bgEntryType": 1,
"iob": 1.8189592, "targetBg": 110, "isf": 30,
"selectedIob": 1, "bgSourceType": 1,
},
"estimatedDateTime": "2026-04-30T11:57:36Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureManualEntry)
self.assertIsInstance(ev, eventtypes.LidBgReadingTaken)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.eventId, 16)
self.assertEqual(ev.seqNum, 456822)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-18T10:15:14")
def test_bg_iob_targetbg_isf_round_trip(self):
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.BG, 151)
self.assertAlmostEqual(ev.IOB, 1.1809407)
self.assertEqual(ev.targetbg, 110)
self.assertEqual(ev.ISF, 30)
ev2 = Event(self.fixtureAutoPopulated)
self.assertEqual(ev2.BG, 164)
self.assertAlmostEqual(ev2.IOB, 1.8189592)
self.assertEqual(ev2.targetbg, 110)
self.assertEqual(ev2.ISF, 30)
def test_selectediob_enum_resolves(self):
# selectedIob:1 -> SwanIobMeal
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.selectediobRaw, 1)
self.assertEqual(ev.selectediob,
eventtypes.LidBgReadingTaken.SelectediobEnum.SwanIobMeal)
def test_bgentrytype_enum_resolves(self):
# bgEntryType:0 -> ManualEntryByTheUserViaNumpad (0 not treated as missing)
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.bgentrytypeRaw, 0)
self.assertEqual(
ev.bgentrytype,
eventtypes.LidBgReadingTaken.BgentrytypeEnum.ManualEntryByTheUserViaNumpad)
# bgEntryType:1 -> AutoPopulatedBgUsingDexcomEgv
ev2 = Event(self.fixtureAutoPopulated)
self.assertEqual(ev2.bgentrytypeRaw, 1)
self.assertEqual(
ev2.bgentrytype,
eventtypes.LidBgReadingTaken.BgentrytypeEnum.AutoPopulatedBgUsingDexcomEgv)
def test_bgsourcetype_enum_resolves(self):
# bgSourceType:1 -> RemoteEntry
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.bgsourcetypeRaw, 1)
self.assertEqual(ev.bgsourcetype,
eventtypes.LidBgReadingTaken.BgsourcetypeEnum.RemoteEntry)
def test_cgmcalibration_enum_resolves(self):
# cgmCalibration:0 -> No (0 not treated as missing)
ev = Event(self.fixtureManualEntry)
self.assertEqual(ev.cgmcalibrationRaw, 0)
self.assertEqual(ev.cgmcalibration,
eventtypes.LidBgReadingTaken.CgmcalibrationEnum.No)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureManualEntry, self.fixtureAutoPopulated):
ev = Event(fixture)
json.dumps(ev.todict())
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,110 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBolusActivated(unittest.TestCase):
"""55 LID_BOLUS_ACTIVATED: real captured pump-log events."""
maxDiff = None
def setUp(self):
# Real captured events (verbatim). All observed captures have
# selectedIob=1 (Swan IOB Meal); fixtures differ by bolusSize/iob.
self.fixtureMeal = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 55,
"sequenceGroup": 0,
"sequenceNumber": 394650,
"pumpDateTime": "2026-04-30T11:57:53",
"eventProperties": {
"bolusId": 1423, "selectedIob": 1, "spareA3": 0,
"iob": 1.8189592, "bolusSize": 8.33,
},
"estimatedDateTime": "2026-04-30T11:57:53Z",
}
self.fixtureZeroIob = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 55,
"sequenceGroup": 0,
"sequenceNumber": 395970,
"pumpDateTime": "2026-04-30T21:38:00",
"eventProperties": {
"bolusId": 1426, "selectedIob": 1, "spareA3": 0,
"iob": 0, "bolusSize": 10.96,
},
"estimatedDateTime": "2026-04-30T21:38:00Z",
}
self.fixtureSmall = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 55,
"sequenceGroup": 0,
"sequenceNumber": 395158,
"pumpDateTime": "2026-04-30T15:13:14",
"eventProperties": {
"bolusId": 1425, "selectedIob": 1, "spareA3": 0,
"iob": 4.116488, "bolusSize": 2,
},
"estimatedDateTime": "2026-04-30T15:13:14Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureMeal)
self.assertIsInstance(ev, eventtypes.LidBolusActivated)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureMeal)
self.assertEqual(ev.eventId, 55)
self.assertEqual(ev.seqNum, 394650)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureMeal)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T11:57:53")
def test_bolus_fields_round_trip(self):
ev = Event(self.fixtureMeal)
self.assertEqual(ev.bolusid, 1423)
self.assertAlmostEqual(ev.IOB, 1.8189592)
self.assertAlmostEqual(ev.bolussize, 8.33)
def test_zero_iob_is_preserved(self):
# iob:0 must not be dropped as missing.
ev = Event(self.fixtureZeroIob)
self.assertEqual(ev.bolusid, 1426)
self.assertEqual(ev.IOB, 0)
self.assertAlmostEqual(ev.bolussize, 10.96)
def test_small_bolus_round_trips(self):
ev = Event(self.fixtureSmall)
self.assertEqual(ev.bolusid, 1425)
self.assertAlmostEqual(ev.IOB, 4.116488)
self.assertEqual(ev.bolussize, 2)
def test_selectediob_resolves_to_enum(self):
# selectedIob:1 -> Swan IOB Meal
ev = Event(self.fixtureMeal)
self.assertEqual(ev.selectediobRaw, 1)
self.assertEqual(ev.selectediob,
eventtypes.LidBolusActivated.SelectediobEnum.SwanIobMeal)
def test_spareA3_is_ignored(self):
ev = Event(self.fixtureMeal)
self.assertFalse(hasattr(ev, "spareA3"))
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureMeal, self.fixtureZeroIob, self.fixtureSmall):
ev = Event(fixture)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 55)
self.assertEqual(d["name"], "LID_BOLUS_ACTIVATED")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,93 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBolusCompleted(unittest.TestCase):
maxDiff = None
def setUp(self):
# Real captured LID_BOLUS_COMPLETED (eventCode 20) events, verbatim.
# completionStatus 3 -> Completed, insulinDelivered == insulinRequested.
self.fixtureCompleted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 20,
"sequenceGroup": 0,
"sequenceNumber": 394675,
"pumpDateTime": "2026-04-30T12:01:53",
"eventProperties": {
"completionStatus": 3, "bolusId": 1423, "iob": 10.088287,
"insulinDelivered": 8.33, "insulinRequested": 8.33,
},
"estimatedDateTime": "2026-04-30T12:01:53Z",
}
# completionStatus 0 -> UserAborted, an interrupted bolus where
# insulinDelivered (0.04657) is far below insulinRequested (0.5).
self.fixtureInterrupted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 20,
"sequenceGroup": 0,
"sequenceNumber": 456849,
"pumpDateTime": "2026-05-18T10:15:39",
"eventProperties": {
"completionStatus": 0, "bolusId": 1644, "iob": 1.2275107,
"insulinDelivered": 0.04657, "insulinRequested": 0.5,
},
"estimatedDateTime": "2026-05-18T10:15:39Z",
}
def test_dispatches_to_lidboluscompleted(self):
self.assertIsInstance(Event(self.fixtureCompleted), eventtypes.LidBolusCompleted)
self.assertIsInstance(Event(self.fixtureInterrupted), eventtypes.LidBolusCompleted)
self.assertNotIsInstance(Event(self.fixtureCompleted), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureCompleted)
self.assertEqual(ev.eventId, 20)
self.assertEqual(ev.seqNum, 394675)
# eventTimestamp keeps pumpDateTime's wall-clock.
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-04-30T12:01:53")
def test_completed_fields_round_trip(self):
ev = Event(self.fixtureCompleted)
self.assertEqual(ev.bolusid, 1423)
self.assertAlmostEqual(ev.insulindelivered, 8.33)
self.assertAlmostEqual(ev.insulinrequested, 8.33)
self.assertAlmostEqual(ev.IOB, 10.088287)
def test_interrupted_fields_round_trip(self):
ev = Event(self.fixtureInterrupted)
self.assertEqual(ev.bolusid, 1644)
self.assertAlmostEqual(ev.insulindelivered, 0.04657)
self.assertAlmostEqual(ev.insulinrequested, 0.5)
self.assertAlmostEqual(ev.IOB, 1.2275107)
# Interrupted: less insulin delivered than requested.
self.assertLess(ev.insulindelivered, ev.insulinrequested)
def test_completionstatus_resolves_to_enum(self):
completed = Event(self.fixtureCompleted)
self.assertEqual(completed.completionstatusRaw, 3)
self.assertEqual(completed.completionstatus,
eventtypes.LidBolusCompleted.CompletionstatusEnum.Completed)
interrupted = Event(self.fixtureInterrupted)
self.assertEqual(interrupted.completionstatusRaw, 0)
self.assertEqual(interrupted.completionstatus,
eventtypes.LidBolusCompleted.CompletionstatusEnum.UserAborted)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureCompleted, self.fixtureInterrupted):
ev = Event(fixture)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 20)
self.assertEqual(d["name"], "LID_BOLUS_COMPLETED")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,133 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBolusDelivery(unittest.TestCase):
"""280: LID_BOLUS_DELIVERY. All fixtures are real captured pump-log events."""
maxDiff = None
def setUp(self):
# Manual pump-button bolus, started: bolusType [0] (Now),
# bolusSource 0 (PumpButton), bolusDeliveryStatus 1 (BolusStarted).
self.fixturePumpButtonStarted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 280,
"sequenceGroup": 0,
"sequenceNumber": 395159,
"pumpDateTime": "2026-04-30T15:13:14",
"eventProperties": {
"bolusId": 1425, "bolusDeliveryStatus": 1, "bolusType": [0],
"bolusSource": 0, "remoteId": 145, "requestedNow": 2000,
"requestedLater": 0, "correction": 0,
"extendedDurationRequested": 0, "deliveredTotal": 0,
},
"estimatedDateTime": "2026-04-30T15:13:14Z",
}
# Carb+correction BLE bolus, started: bolusType [0,3,4]
# (Now|Correction|Carb), bolusSource 8 (Ble), status 1 (BolusStarted).
self.fixtureCarbCorrectionStarted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 280,
"sequenceGroup": 0,
"sequenceNumber": 395971,
"pumpDateTime": "2026-04-30T21:38:00",
"eventProperties": {
"bolusId": 1426, "bolusDeliveryStatus": 1, "bolusType": [0, 3, 4],
"bolusSource": 8, "remoteId": 146, "requestedNow": 10960,
"requestedLater": 0, "correction": 130,
"extendedDurationRequested": 0, "deliveredTotal": 0,
},
"estimatedDateTime": "2026-04-30T21:38:00Z",
}
# Completion of the same bolus: status 0 (BolusCompleted),
# deliveredTotal now populated (10960).
self.fixtureCarbCorrectionCompleted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 280,
"sequenceGroup": 0,
"sequenceNumber": 395991,
"pumpDateTime": "2026-04-30T21:40:03",
"eventProperties": {
"bolusId": 1426, "bolusDeliveryStatus": 0, "bolusType": [0, 3, 4],
"bolusSource": 8, "remoteId": 146, "requestedNow": 10960,
"requestedLater": 0, "correction": 130,
"extendedDurationRequested": 0, "deliveredTotal": 10960,
},
"estimatedDateTime": "2026-04-30T21:40:03Z",
}
def test_dispatches_to_correct_class(self):
self.assertIsInstance(Event(self.fixturePumpButtonStarted),
eventtypes.LidBolusDelivery)
self.assertNotIsInstance(Event(self.fixturePumpButtonStarted), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureCarbCorrectionStarted)
self.assertEqual(ev.eventId, 280)
self.assertEqual(ev.seqNum, 395971)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T21:38:00")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureCarbCorrectionStarted)
self.assertEqual(ev.bolusid, 1426)
self.assertEqual(ev.requestedNow, 10960)
self.assertEqual(ev.deliveredTotal, 0)
self.assertEqual(ev.correction, 130)
self.assertEqual(ev.remoteId, 146)
self.assertEqual(ev.requestedLater, 0)
self.assertEqual(ev.extendedDurationRequested, 0)
def test_completion_carries_delivered_total(self):
ev = Event(self.fixtureCarbCorrectionCompleted)
self.assertEqual(ev.bolusid, 1426)
self.assertEqual(ev.deliveredTotal, 10960)
def test_bolustype_single_bit_folds_and_resolves(self):
# bolusType [0] -> 1<<0 == 1 -> Now
ev = Event(self.fixturePumpButtonStarted)
self.assertEqual(ev.bolusTypeRaw, 1)
self.assertEqual(ev.bolusType, eventtypes.LidBolusDelivery.BolustypeBitmask.Now)
def test_bolustype_multi_bit_folds_and_resolves(self):
# bolusType [0,3,4] -> 1<<0 | 1<<3 | 1<<4 == 25 -> Now|Correction|Carb
ev = Event(self.fixtureCarbCorrectionStarted)
self.assertEqual(ev.bolusTypeRaw, sum(1 << i for i in [0, 3, 4]))
self.assertEqual(ev.bolusTypeRaw, 25)
bt = eventtypes.LidBolusDelivery.BolustypeBitmask
self.assertEqual(ev.bolusType, bt.Now | bt.Correction | bt.Carb)
def test_bolussource_resolves(self):
self.assertEqual(
Event(self.fixturePumpButtonStarted).bolusSource,
eventtypes.LidBolusDelivery.BolussourceEnum.PumpButton)
self.assertEqual(
Event(self.fixtureCarbCorrectionStarted).bolusSource,
eventtypes.LidBolusDelivery.BolussourceEnum.Ble)
def test_bolusdeliverystatus_resolves(self):
self.assertEqual(
Event(self.fixtureCarbCorrectionStarted).bolusDeliveryStatus,
eventtypes.LidBolusDelivery.BolusdeliverystatusEnum.BolusStarted)
self.assertEqual(
Event(self.fixtureCarbCorrectionCompleted).bolusDeliveryStatus,
eventtypes.LidBolusDelivery.BolusdeliverystatusEnum.BolusCompleted)
def test_todict_is_json_serializable(self):
for fixture in (self.fixturePumpButtonStarted,
self.fixtureCarbCorrectionStarted,
self.fixtureCarbCorrectionCompleted):
ev = Event(fixture)
json.dumps(ev.todict()) # must not raise
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBolusRequestedMsg1(unittest.TestCase):
"""64: LID_BOLUS_REQUESTED_MSG1. Fixtures are real captured pump-log events
copied verbatim from a captured account response."""
maxDiff = None
def setUp(self):
# bolusType 3 (Remote), correctionBolusIncluded 0 (No), carbs present.
self.fixtureRemoteWithCarbs = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 64, "sequenceGroup": 0, "sequenceNumber": 394641,
"pumpDateTime": "2026-04-30T11:57:38",
"eventProperties": {
"bolusId": 1423, "bolusType": 3, "correctionBolusIncluded": 0,
"carbAmount": 50, "bg": 164, "iob": 1.82, "carbRatio": 0,
},
"estimatedDateTime": "2026-04-30T11:57:38Z",
}
# bolusType 3 (Remote), correctionBolusIncluded 1 (Yes), iob 0.
self.fixtureRemoteWithCorrection = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 64, "sequenceGroup": 0, "sequenceNumber": 395959,
"pumpDateTime": "2026-04-30T21:37:45",
"eventProperties": {
"bolusId": 1426, "bolusType": 3, "correctionBolusIncluded": 1,
"carbAmount": 65, "bg": 114, "iob": 0, "carbRatio": 0,
},
"estimatedDateTime": "2026-04-30T21:37:45Z",
}
# bolusType 0 (Insulin), no carbs, bg 0, fractional iob.
self.fixtureInsulinNoCarbs = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 64, "sequenceGroup": 0, "sequenceNumber": 395146,
"pumpDateTime": "2026-04-30T15:12:59",
"eventProperties": {
"bolusId": 1425, "bolusType": 0, "correctionBolusIncluded": 0,
"carbAmount": 0, "bg": 0, "iob": 4.116488, "carbRatio": 0,
},
"estimatedDateTime": "2026-04-30T15:12:59Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureRemoteWithCarbs)
self.assertIsInstance(ev, eventtypes.LidBolusRequestedMsg1)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureRemoteWithCarbs)
self.assertEqual(ev.eventId, 64)
self.assertEqual(ev.seqNum, 394641)
# eventTimestamp keeps pumpDateTime's wall-clock.
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T11:57:38")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureRemoteWithCarbs)
self.assertEqual(ev.bolusid, 1423)
self.assertEqual(ev.carbamount, 50)
self.assertEqual(ev.BG, 164)
self.assertEqual(ev.IOB, 1.82)
def test_fractional_iob_and_zero_bg(self):
ev = Event(self.fixtureInsulinNoCarbs)
self.assertEqual(ev.bolusid, 1425)
self.assertEqual(ev.carbamount, 0)
self.assertEqual(ev.BG, 0)
self.assertAlmostEqual(ev.IOB, 4.116488)
def test_bolustype_remote(self):
ev = Event(self.fixtureRemoteWithCarbs)
self.assertEqual(ev.bolustypeRaw, 3)
self.assertEqual(ev.bolustype,
eventtypes.LidBolusRequestedMsg1.BolustypeEnum.Remote)
def test_bolustype_insulin(self):
# bolusType 0 must resolve (0 not treated as missing).
ev = Event(self.fixtureInsulinNoCarbs)
self.assertEqual(ev.bolustypeRaw, 0)
self.assertEqual(ev.bolustype,
eventtypes.LidBolusRequestedMsg1.BolustypeEnum.Insulin)
def test_correctionbolusincluded_no(self):
ev = Event(self.fixtureRemoteWithCarbs)
self.assertEqual(ev.correctionbolusincludedRaw, 0)
self.assertEqual(
ev.correctionbolusincluded,
eventtypes.LidBolusRequestedMsg1.CorrectionbolusincludedEnum.No)
def test_correctionbolusincluded_yes(self):
ev = Event(self.fixtureRemoteWithCorrection)
self.assertEqual(ev.correctionbolusincludedRaw, 1)
self.assertEqual(
ev.correctionbolusincluded,
eventtypes.LidBolusRequestedMsg1.CorrectionbolusincludedEnum.Yes)
def test_carbratio_scales(self):
# carbratio is carbratioRaw * 0.001; real captures carry 0.
ev = Event(self.fixtureRemoteWithCorrection)
self.assertEqual(ev.carbratioRaw, 0)
self.assertAlmostEqual(ev.carbratio, 0.0)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureRemoteWithCarbs,
self.fixtureRemoteWithCorrection,
self.fixtureInsulinNoCarbs):
ev = Event(fixture)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 64)
self.assertEqual(d["name"], "LID_BOLUS_REQUESTED_MSG1")
self.assertEqual(d["bolusid"],
fixture["eventProperties"]["bolusId"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBolusRequestedMsg2(unittest.TestCase):
"""65: LID_BOLUS_REQUESTED_MSG2 — real captured pump-log events."""
maxDiff = None
def setUp(self):
# BLE standard bolus, user did NOT override the bolus size.
self.fixtureBleStandard = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 65,
"sequenceGroup": 0,
"sequenceNumber": 394642,
"pumpDateTime": "2026-04-30T11:57:38",
"eventProperties": {
"bolusId": 1423, "options": 4, "standardPercent": 100,
"duration": 0, "spareB6": 0, "isf": 0, "targetBg": 0,
"userOverride": 0, "declinedCorrection": 0, "selectedIob": 1,
},
"estimatedDateTime": "2026-04-30T11:57:38Z",
}
# BLE standard bolus, user DID override the bolus size.
self.fixtureUserOverride = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 65,
"sequenceGroup": 0,
"sequenceNumber": 394840,
"pumpDateTime": "2026-04-30T13:13:28",
"eventProperties": {
"bolusId": 1424, "options": 4, "standardPercent": 100,
"duration": 0, "spareB6": 0, "isf": 0, "targetBg": 0,
"userOverride": 1, "declinedCorrection": 0, "selectedIob": 1,
},
"estimatedDateTime": "2026-04-30T13:13:28Z",
}
# Quick bolus (options=2).
self.fixtureQuick = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 65,
"sequenceGroup": 0,
"sequenceNumber": 395147,
"pumpDateTime": "2026-04-30T15:12:59",
"eventProperties": {
"bolusId": 1425, "options": 2, "standardPercent": 100,
"duration": 0, "spareB6": 0, "isf": 0, "targetBg": 0,
"userOverride": 0, "declinedCorrection": 0, "selectedIob": 1,
},
"estimatedDateTime": "2026-04-30T15:12:59Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureBleStandard)
self.assertIsInstance(ev, eventtypes.LidBolusRequestedMsg2)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureBleStandard)
self.assertEqual(ev.eventId, 65)
self.assertEqual(ev.seqNum, 394642)
self.assertEqual(
ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-04-30T11:57:38")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureBleStandard)
self.assertEqual(ev.bolusid, 1423)
self.assertEqual(ev.standardpercent, 100)
self.assertEqual(ev.duration, 0)
self.assertEqual(ev.ISF, 0)
self.assertEqual(ev.targetbg, 0)
def test_options_enum_ble_standard(self):
ev = Event(self.fixtureBleStandard)
self.assertEqual(ev.optionsRaw, 4)
self.assertEqual(ev.options,
eventtypes.LidBolusRequestedMsg2.OptionsEnum.BleStandardBolus)
def test_options_enum_quick(self):
ev = Event(self.fixtureQuick)
self.assertEqual(ev.optionsRaw, 2)
self.assertEqual(ev.options,
eventtypes.LidBolusRequestedMsg2.OptionsEnum.QuickBolus)
def test_selectediob_enum(self):
ev = Event(self.fixtureBleStandard)
self.assertEqual(ev.selectediobRaw, 1)
self.assertEqual(ev.selectediob,
eventtypes.LidBolusRequestedMsg2.SelectediobEnum.SwanIobMeal)
def test_useroverride_enum_no(self):
ev = Event(self.fixtureBleStandard)
self.assertEqual(ev.useroverrideRaw, 0)
self.assertEqual(ev.useroverride,
eventtypes.LidBolusRequestedMsg2.UseroverrideEnum.No)
def test_useroverride_enum_yes(self):
ev = Event(self.fixtureUserOverride)
self.assertEqual(ev.useroverrideRaw, 1)
self.assertEqual(ev.useroverride,
eventtypes.LidBolusRequestedMsg2.UseroverrideEnum.Yes)
def test_declinedcorrection_enum(self):
ev = Event(self.fixtureBleStandard)
self.assertEqual(ev.declinedcorrectionRaw, 0)
self.assertEqual(ev.declinedcorrection,
eventtypes.LidBolusRequestedMsg2.DeclinedcorrectionEnum.No)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureBleStandard, self.fixtureUserOverride,
self.fixtureQuick):
ev = Event(fixture)
json.dumps(ev.todict()) # must not raise
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,127 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidBolusRequestedMsg3(unittest.TestCase):
"""66: LID_BOLUS_REQUESTED_MSG3. Fixtures are real captured pump-log
events copied verbatim (spareA2 is present but ignored by the parser)."""
maxDiff = None
def setUp(self):
# food-only bolus; total carries float rounding (8.330001).
self.fixtureFoodOnly = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 66,
"sequenceGroup": 0,
"sequenceNumber": 394643,
"pumpDateTime": "2026-04-30T11:57:38",
"eventProperties": {
"bolusId": 1423, "spareA2": 0, "foodBolusSize": 8.33,
"correctionBolusSize": 0, "totalBolusSize": 8.330001,
},
"estimatedDateTime": "2026-04-30T11:57:38Z",
}
# food + correction; both components non-zero.
self.fixtureFoodAndCorrection = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 66,
"sequenceGroup": 0,
"sequenceNumber": 395961,
"pumpDateTime": "2026-04-30T21:37:45",
"eventProperties": {
"bolusId": 1426, "spareA2": 0, "foodBolusSize": 10.83,
"correctionBolusSize": 0.13, "totalBolusSize": 10.96,
},
"estimatedDateTime": "2026-04-30T21:37:45Z",
}
# correction-only food component; total exceeds correction.
self.fixtureCorrectionOnly = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 66,
"sequenceGroup": 0,
"sequenceNumber": 398360,
"pumpDateTime": "2026-05-01T16:02:12",
"eventProperties": {
"bolusId": 1430, "spareA2": 0, "foodBolusSize": 0,
"correctionBolusSize": 1.47, "totalBolusSize": 3,
},
"estimatedDateTime": "2026-05-01T16:02:12Z",
}
# both breakdown components zero but a non-zero total.
self.fixtureTotalOnly = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 66,
"sequenceGroup": 0,
"sequenceNumber": 394841,
"pumpDateTime": "2026-04-30T13:13:28",
"eventProperties": {
"bolusId": 1424, "spareA2": 0, "foodBolusSize": 0,
"correctionBolusSize": 0, "totalBolusSize": 4,
},
"estimatedDateTime": "2026-04-30T13:13:28Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureFoodOnly)
self.assertIsInstance(ev, eventtypes.LidBolusRequestedMsg3)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureFoodOnly)
self.assertEqual(ev.eventId, 66)
self.assertEqual(ev.seqNum, 394643)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureFoodOnly)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T11:57:38")
def test_food_only(self):
ev = Event(self.fixtureFoodOnly)
self.assertEqual(ev.bolusid, 1423)
self.assertAlmostEqual(ev.foodbolussize, 8.33)
self.assertAlmostEqual(ev.correctionbolussize, 0)
self.assertAlmostEqual(ev.totalbolussize, 8.330001)
def test_food_and_correction(self):
ev = Event(self.fixtureFoodAndCorrection)
self.assertEqual(ev.bolusid, 1426)
self.assertAlmostEqual(ev.foodbolussize, 10.83)
self.assertAlmostEqual(ev.correctionbolussize, 0.13)
self.assertAlmostEqual(ev.totalbolussize, 10.96)
def test_correction_only(self):
ev = Event(self.fixtureCorrectionOnly)
self.assertEqual(ev.bolusid, 1430)
self.assertAlmostEqual(ev.foodbolussize, 0)
self.assertAlmostEqual(ev.correctionbolussize, 1.47)
self.assertAlmostEqual(ev.totalbolussize, 3)
def test_total_only(self):
ev = Event(self.fixtureTotalOnly)
self.assertEqual(ev.bolusid, 1424)
self.assertAlmostEqual(ev.foodbolussize, 0)
self.assertAlmostEqual(ev.correctionbolussize, 0)
self.assertAlmostEqual(ev.totalbolussize, 4)
def test_todict_json_serializable(self):
for fixture in (self.fixtureFoodOnly, self.fixtureFoodAndCorrection,
self.fixtureCorrectionOnly, self.fixtureTotalOnly):
ev = Event(fixture)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 66)
self.assertEqual(d["name"], "LID_BOLUS_REQUESTED_MSG3")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,67 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCannulaFilled(unittest.TestCase):
"""61: LID_CANNULA_FILLED. Fixture is a real captured pump-log event.
Only one distinct eventProperties shape exists in the captures, so a
single fixture covers the observed behavior. The extra infusionSetType
key is not in the schema and must be ignored by the parser."""
maxDiff = None
def setUp(self):
self.fixtureCompleted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 61,
"sequenceGroup": 0,
"sequenceNumber": 412912,
"pumpDateTime": "2026-05-05T19:16:40",
"eventProperties": {
"primeSize": 0.3, "completionStatus": 3, "infusionSetType": 0,
},
"estimatedDateTime": "2026-05-05T19:16:40Z",
}
def test_dispatches_to_lidcannulafilled(self):
ev = Event(self.fixtureCompleted)
self.assertIsInstance(ev, eventtypes.LidCannulaFilled)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureCompleted)
self.assertEqual(ev.eventId, 61)
self.assertEqual(ev.seqNum, 412912)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-05T19:16:40")
def test_primesize_round_trips(self):
ev = Event(self.fixtureCompleted)
self.assertEqual(ev.primesize, 0.3)
def test_completionstatus_resolves_to_enum(self):
# completionStatus:3 -> Completed
ev = Event(self.fixtureCompleted)
self.assertEqual(ev.completionstatusRaw, 3)
self.assertEqual(ev.completionstatus,
eventtypes.LidCannulaFilled.CompletionstatusEnum.Completed)
def test_unknown_infusionsettype_key_is_ignored(self):
# infusionSetType is not in the schema; the parser must not raise and
# must not expose an attribute for it.
ev = Event(self.fixtureCompleted) # must not raise
self.assertFalse(hasattr(ev, "infusionSetType"))
self.assertFalse(hasattr(ev, "infusionsettype"))
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureCompleted)
json.dumps(ev.todict()) # must not raise
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,85 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCartridgeFilled(unittest.TestCase):
"""33: LID_CARTRIDGE_FILLED. All fixtures are real captured pump-log
events (verbatim). insulinVolume varies; v2Volume is always 0."""
maxDiff = None
def setUp(self):
# Smallest observed insulinVolume.
self.fixtureMin = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 33,
"sequenceGroup": 0,
"sequenceNumber": 418402,
"pumpDateTime": "2026-05-07T10:35:59",
"eventProperties": {"insulinVolume": 60, "v2Volume": 0},
"estimatedDateTime": "2026-05-07T10:35:59Z",
}
# A mid-range fill.
self.fixtureMid = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 33,
"sequenceGroup": 0,
"sequenceNumber": 394427,
"pumpDateTime": "2026-04-30T10:16:09",
"eventProperties": {"insulinVolume": 105, "v2Volume": 0},
"estimatedDateTime": "2026-04-30T10:16:09Z",
}
# Largest observed insulinVolume.
self.fixtureMax = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 33,
"sequenceGroup": 0,
"sequenceNumber": 463083,
"pumpDateTime": "2026-05-20T02:15:04",
"eventProperties": {"insulinVolume": 190, "v2Volume": 0},
"estimatedDateTime": "2026-05-20T02:15:04Z",
}
def test_dispatches_to_lidcartridgefilled(self):
ev = Event(self.fixtureMid)
self.assertIsInstance(ev, eventtypes.LidCartridgeFilled)
self.assertIsNot(type(ev), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureMid)
self.assertEqual(ev.eventId, 33)
self.assertEqual(ev.seqNum, 394427)
self.assertEqual(
ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T10:16:09")
def test_insulinvolume_round_trips(self):
self.assertEqual(Event(self.fixtureMin).insulinvolume, 60)
self.assertEqual(Event(self.fixtureMid).insulinvolume, 105)
self.assertEqual(Event(self.fixtureMax).insulinvolume, 190)
def test_v2volume_round_trips(self):
for fixture in (self.fixtureMin, self.fixtureMid, self.fixtureMax):
self.assertEqual(Event(fixture).v2Volume, 0)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureMin, self.fixtureMid, self.fixtureMax):
d = Event(fixture).todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 33)
self.assertEqual(d["name"], "LID_CARTRIDGE_FILLED")
def test_todict_reflects_real_values(self):
d = Event(self.fixtureMax).todict()
self.assertEqual(d["seqNum"], 463083)
self.assertEqual(d["insulinvolume"], 190)
self.assertEqual(d["v2Volume"], 0)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,95 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
# Real captured LID_CGM_ALERT_ACK_DEX (371) events, copied verbatim.
# dalertId/sensorType/ackSource are dictionary/enum fields. "spareA2" is
# ignored by the parser (kept here but never asserted on).
class TestLidCgmAlertAckDex(unittest.TestCase):
maxDiff = None
def setUp(self):
# dalertId:2 (CGM High), sensorType:3 (G7), ackSource:0 (by User)
self.fixtureCgmHigh = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 371,
"sequenceGroup": 0,
"sequenceNumber": 416807,
"pumpDateTime": "2026-05-06T23:04:58",
"eventProperties": {"dalertId": 2, "sensorType": 3, "spareA2": 0, "ackSource": 0},
"estimatedDateTime": "2026-05-06T23:04:58Z",
}
# dalertId:12 (CGM Sensor Expiring Soon), sensorType:3 (G7), ackSource:0 (by User)
self.fixtureSensorExpiringSoon = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 371,
"sequenceGroup": 0,
"sequenceNumber": 446797,
"pumpDateTime": "2026-05-15T16:43:28",
"eventProperties": {"dalertId": 12, "sensorType": 3, "spareA2": 0, "ackSource": 0},
"estimatedDateTime": "2026-05-15T16:43:28Z",
}
# dalertId:32 (not in enum -> None), sensorType:3 (G7), ackSource:1 (by Software)
self.fixtureAckBySoftware = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 371,
"sequenceGroup": 0,
"sequenceNumber": 449784,
"pumpDateTime": "2026-05-16T12:34:28",
"eventProperties": {"dalertId": 32, "sensorType": 3, "spareA2": 0, "ackSource": 1},
"estimatedDateTime": "2026-05-16T12:34:28Z",
}
def test_dispatches_to_lidcgmalertackdex(self):
for fx in (self.fixtureCgmHigh, self.fixtureSensorExpiringSoon, self.fixtureAckBySoftware):
self.assertIsInstance(Event(fx), eventtypes.LidCgmAlertAckDex)
def test_envelope_fields(self):
ev = Event(self.fixtureCgmHigh)
self.assertEqual(ev.eventId, 371)
self.assertEqual(ev.seqNum, 416807)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-05-06T23:04:58")
def test_dalertid_resolves_cgm_high(self):
ev = Event(self.fixtureCgmHigh)
self.assertEqual(ev.dalertidRaw, 2)
self.assertEqual(ev.dalertid, eventtypes.LidCgmAlertAckDex.DalertidEnum.CgmHigh)
def test_dalertid_resolves_sensor_expiring_soon(self):
ev = Event(self.fixtureSensorExpiringSoon)
self.assertEqual(ev.dalertidRaw, 12)
self.assertEqual(ev.dalertid, eventtypes.LidCgmAlertAckDex.DalertidEnum.CgmSensorExpiringSoon)
def test_sensortype_resolves_g7(self):
for fx in (self.fixtureCgmHigh, self.fixtureSensorExpiringSoon, self.fixtureAckBySoftware):
ev = Event(fx)
self.assertEqual(ev.sensortypeRaw, 3)
self.assertEqual(ev.sensortype, eventtypes.LidCgmAlertAckDex.SensortypeEnum.CgmTypeDexcomG7)
def test_acksource_resolves_by_user(self):
ev = Event(self.fixtureCgmHigh)
self.assertEqual(ev.acksourceRaw, 0)
self.assertEqual(ev.acksource, eventtypes.LidCgmAlertAckDex.AcksourceEnum.AlertAcknowledgedByUser)
def test_acksource_resolves_by_software(self):
ev = Event(self.fixtureAckBySoftware)
self.assertEqual(ev.acksourceRaw, 1)
self.assertEqual(ev.acksource, eventtypes.LidCgmAlertAckDex.AcksourceEnum.AlertAcknowledgedBySoftware)
def test_todict_is_json_serializable(self):
for fx in (self.fixtureCgmHigh, self.fixtureSensorExpiringSoon, self.fixtureAckBySoftware):
ev = Event(fx)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 371)
self.assertEqual(d["name"], "LID_CGM_ALERT_ACK_DEX")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,126 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCgmAlertActivatedDex(unittest.TestCase):
"""369 LID_CGM_ALERT_ACTIVATED_DEX: dalertId resolves via a dictionary
transform enum, sensorType is an enum. All fixtures are real captures."""
maxDiff = None
def setUp(self):
# dalertId:2 -> CgmHigh
self.fixtureHigh = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 369, "sequenceGroup": 0, "sequenceNumber": 394822,
"pumpDateTime": "2026-04-30T13:13:00",
"eventProperties": {
"dalertId": 2, "sensorType": 3, "spareA2": 0,
"faultLocatorData": 8468, "param1": 214, "param2": 200,
},
"estimatedDateTime": "2026-04-30T13:13:00Z",
}
# dalertId:3 -> CgmLow
self.fixtureLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 369, "sequenceGroup": 0, "sequenceNumber": 398698,
"pumpDateTime": "2026-05-01T16:58:01",
"eventProperties": {
"dalertId": 3, "sensorType": 3, "spareA2": 0,
"faultLocatorData": 8467, "param1": 73, "param2": 80,
},
"estimatedDateTime": "2026-05-01T16:58:01Z",
}
# dalertId:1 -> CgmFixedLow
self.fixtureFixedLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 369, "sequenceGroup": 0, "sequenceNumber": 403231,
"pumpDateTime": "2026-05-03T01:28:03",
"eventProperties": {
"dalertId": 1, "sensorType": 3, "spareA2": 0,
"faultLocatorData": 8467, "param1": 53, "param2": 55,
},
"estimatedDateTime": "2026-05-03T01:28:03Z",
}
# dalertId:14 -> CgmOutOfRange
self.fixtureOutOfRange = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 369, "sequenceGroup": 0, "sequenceNumber": 409133,
"pumpDateTime": "2026-05-04T18:59:00",
"eventProperties": {
"dalertId": 14, "sensorType": 3, "spareA2": 0,
"faultLocatorData": 8462, "param1": 25, "param2": 917,
},
"estimatedDateTime": "2026-05-04T18:59:00Z",
}
# dalertId:11 -> CgmSensorFail
self.fixtureSensorFail = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 369, "sequenceGroup": 0, "sequenceNumber": 481654,
"pumpDateTime": "2026-05-25T08:35:32",
"eventProperties": {
"dalertId": 11, "sensorType": 3, "spareA2": 0,
"faultLocatorData": 8481, "param1": 35, "param2": 765,
},
"estimatedDateTime": "2026-05-25T08:35:32Z",
}
def test_dispatches_to_correct_class(self):
for f in (self.fixtureHigh, self.fixtureLow, self.fixtureFixedLow,
self.fixtureOutOfRange, self.fixtureSensorFail):
ev = Event(f)
self.assertIsInstance(ev, eventtypes.LidCgmAlertActivatedDex)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureHigh)
self.assertEqual(ev.eventId, 369)
self.assertEqual(ev.seqNum, 394822)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureSensorFail)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-25T08:35:32")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureOutOfRange)
self.assertEqual(ev.faultlocatordata, 8462)
self.assertEqual(ev.param1, 25)
self.assertEqual(ev.param2, 917)
def test_dalertid_dictionary_enum_resolves(self):
cases = [
(self.fixtureHigh, 2, eventtypes.LidCgmAlertActivatedDex.DalertidEnum.CgmHigh),
(self.fixtureLow, 3, eventtypes.LidCgmAlertActivatedDex.DalertidEnum.CgmLow),
(self.fixtureFixedLow, 1, eventtypes.LidCgmAlertActivatedDex.DalertidEnum.CgmFixedLow),
(self.fixtureOutOfRange, 14, eventtypes.LidCgmAlertActivatedDex.DalertidEnum.CgmOutOfRange),
(self.fixtureSensorFail, 11, eventtypes.LidCgmAlertActivatedDex.DalertidEnum.CgmSensorFail),
]
for f, raw, member in cases:
ev = Event(f)
self.assertEqual(ev.dalertidRaw, raw)
self.assertEqual(ev.dalertid, member)
def test_sensortype_enum_resolves(self):
# every capture is sensorType:3 -> Dexcom G7
for f in (self.fixtureHigh, self.fixtureLow, self.fixtureFixedLow,
self.fixtureOutOfRange, self.fixtureSensorFail):
ev = Event(f)
self.assertEqual(ev.sensortypeRaw, 3)
self.assertEqual(ev.sensortype,
eventtypes.LidCgmAlertActivatedDex.SensortypeEnum.CgmTypeDexcomG7)
def test_todict_is_json_serializable(self):
for f in (self.fixtureHigh, self.fixtureLow, self.fixtureFixedLow,
self.fixtureOutOfRange, self.fixtureSensorFail):
ev = Event(f)
json.dumps(ev.todict()) # must not raise
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,91 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCgmAlertClearedDex(unittest.TestCase):
"""370 LID_CGM_ALERT_CLEARED_DEX: dalertId (dict->enum) and sensorType (enum).
All fixtures are real captured pump-log events copied verbatim; they share
sensorType 3 (Dexcom G7) and differ only in dalertId.
"""
maxDiff = None
def setUp(self):
# dalertId 3 -> CgmLow
self.fixtureCgmLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 370, "sequenceGroup": 0, "sequenceNumber": 398753,
"pumpDateTime": "2026-05-01T17:18:01",
"eventProperties": {"dalertId": 3, "sensorType": 3},
"estimatedDateTime": "2026-05-01T17:18:01Z",
}
# dalertId 1 -> CgmFixedLow
self.fixtureCgmFixedLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 370, "sequenceGroup": 0, "sequenceNumber": 403279,
"pumpDateTime": "2026-05-03T01:48:03",
"eventProperties": {"dalertId": 1, "sensorType": 3},
"estimatedDateTime": "2026-05-03T01:48:03Z",
}
# dalertId 14 -> CgmOutOfRange
self.fixtureCgmOutOfRange = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 370, "sequenceGroup": 0, "sequenceNumber": 409147,
"pumpDateTime": "2026-05-04T19:07:08",
"eventProperties": {"dalertId": 14, "sensorType": 3},
"estimatedDateTime": "2026-05-04T19:07:08Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureCgmLow)
self.assertIsInstance(ev, eventtypes.LidCgmAlertClearedDex)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureCgmLow)
self.assertEqual(ev.eventId, 370)
self.assertEqual(ev.seqNum, 398753)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureCgmLow)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-01T17:18:01")
def test_dalertid_resolves_to_enum(self):
self.assertEqual(Event(self.fixtureCgmLow).dalertidRaw, 3)
self.assertEqual(Event(self.fixtureCgmLow).dalertid,
eventtypes.LidCgmAlertClearedDex.DalertidEnum.CgmLow)
self.assertEqual(Event(self.fixtureCgmFixedLow).dalertidRaw, 1)
self.assertEqual(Event(self.fixtureCgmFixedLow).dalertid,
eventtypes.LidCgmAlertClearedDex.DalertidEnum.CgmFixedLow)
self.assertEqual(Event(self.fixtureCgmOutOfRange).dalertidRaw, 14)
self.assertEqual(Event(self.fixtureCgmOutOfRange).dalertid,
eventtypes.LidCgmAlertClearedDex.DalertidEnum.CgmOutOfRange)
def test_sensortype_resolves_to_enum(self):
ev = Event(self.fixtureCgmLow)
self.assertEqual(ev.sensortypeRaw, 3)
self.assertEqual(ev.sensortype,
eventtypes.LidCgmAlertClearedDex.SensortypeEnum.CgmTypeDexcomG7)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureCgmOutOfRange)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 370)
self.assertEqual(d["name"], "LID_CGM_ALERT_CLEARED_DEX")
self.assertEqual(d["seqNum"], 409147)
self.assertEqual(d["dalertidRaw"], 14)
self.assertEqual(d["sensortypeRaw"], 3)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,169 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCgmDataG7(unittest.TestCase):
"""399 LID_CGM_DATA_G7 parsed from real captured pump-log events."""
maxDiff = None
def setUp(self):
# Rising: large positive rate, high glucose, FMR reading.
self.fixtureRising = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 399, "sequenceGroup": 0, "sequenceNumber": 464452,
"pumpDateTime": "2026-05-20T12:15:13",
"eventProperties": {
"glucoseValueStatus": 0, "cgmDataType": [0], "rate": 52,
"algorithmState": 32, "rssi": -87, "currentGlucoseDisplayValue": 287,
"egvTimeStamp": 580133707, "egvInfoBitmask": [0, 5, 6, 7, 8, 11, 12],
"interval": 0, "reservedD15": 0,
},
"estimatedDateTime": "2026-05-20T12:15:13Z",
}
# Falling: large negative rate.
self.fixtureFalling = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 399, "sequenceGroup": 0, "sequenceNumber": 416862,
"pumpDateTime": "2026-05-06T23:26:04",
"eventProperties": {
"glucoseValueStatus": 0, "cgmDataType": [0], "rate": -39,
"algorithmState": 32, "rssi": -54, "currentGlucoseDisplayValue": 195,
"egvTimeStamp": 578964361, "egvInfoBitmask": [0, 5, 6, 7, 8, 11, 12],
"interval": 0, "reservedD15": 0,
},
"estimatedDateTime": "2026-05-06T23:26:04Z",
}
# SpecialLow: glucoseValueStatus 2, very low display value.
self.fixtureSpecialLow = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 399, "sequenceGroup": 0, "sequenceNumber": 450303,
"pumpDateTime": "2026-05-16T15:44:53",
"eventProperties": {
"glucoseValueStatus": 2, "cgmDataType": [0], "rate": -5,
"algorithmState": 32, "rssi": -55, "currentGlucoseDisplayValue": 31,
"egvTimeStamp": 579800690, "egvInfoBitmask": [0, 5, 6, 7, 8, 11, 12],
"interval": 0, "reservedD15": 0,
},
"estimatedDateTime": "2026-05-16T15:44:53Z",
}
# Backfill: different cgmDataType/egvInfoBitmask, high glucose, zero rate.
self.fixtureBackfill = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 399, "sequenceGroup": 0, "sequenceNumber": 484027,
"pumpDateTime": "2026-05-25T22:34:18",
"eventProperties": {
"glucoseValueStatus": 0, "cgmDataType": [1], "rate": 0,
"algorithmState": 32, "rssi": -83, "currentGlucoseDisplayValue": 380,
"egvTimeStamp": 580602548, "egvInfoBitmask": [1, 5, 6, 7, 8, 11, 12],
"interval": 1, "reservedD15": 0,
},
"estimatedDateTime": "2026-05-25T22:34:18Z",
}
def test_dispatches_to_lidcgmdatag7(self):
for fx in (self.fixtureRising, self.fixtureFalling,
self.fixtureSpecialLow, self.fixtureBackfill):
ev = Event(fx)
self.assertIsInstance(ev, eventtypes.LidCgmDataG7)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureRising)
self.assertEqual(ev.eventId, 399)
self.assertEqual(ev.seqNum, 464452)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-20T12:15:13")
def test_display_value_and_rssi_round_trip(self):
ev = Event(self.fixtureRising)
self.assertEqual(ev.currentglucosedisplayvalue, 287)
self.assertEqual(ev.RSSI, -87)
low = Event(self.fixtureSpecialLow)
self.assertEqual(low.currentglucosedisplayvalue, 31)
self.assertEqual(low.RSSI, -55)
high = Event(self.fixtureBackfill)
self.assertEqual(high.currentglucosedisplayvalue, 380)
self.assertEqual(high.RSSI, -83)
def test_glucosevaluestatus_enum_resolves(self):
# 0 -> PreciseValue: the zero value must resolve, not be treated as missing.
ev = Event(self.fixtureRising)
self.assertEqual(ev.glucosevaluestatusRaw, 0)
self.assertEqual(ev.glucosevaluestatus,
eventtypes.LidCgmDataG7.GlucosevaluestatusEnum.PreciseValue)
# 2 -> SpecialLow
low = Event(self.fixtureSpecialLow)
self.assertEqual(low.glucosevaluestatusRaw, 2)
self.assertEqual(low.glucosevaluestatus,
eventtypes.LidCgmDataG7.GlucosevaluestatusEnum.SpecialLow)
def test_algorithmstate_enum_resolves(self):
# 32 -> ReportablePeriodValidEgv on every fixture.
for fx in (self.fixtureRising, self.fixtureFalling,
self.fixtureSpecialLow, self.fixtureBackfill):
ev = Event(fx)
self.assertEqual(ev.algorithmstateRaw, 32)
self.assertEqual(ev.algorithmstate,
eventtypes.LidCgmDataG7.AlgorithmstateEnum.ReportablePeriodValidEgv)
def test_cgm_datatype_bitmask_folds_to_raw_int(self):
# cgmDataType:[0] -> 1<<0 == 1 -> Fmr
ev = Event(self.fixtureRising)
self.assertEqual(ev.cgmDataTypeRaw, 1)
self.assertEqual(ev.cgmDataType,
eventtypes.LidCgmDataG7.CgmdatatypeBitmask.Fmr)
# cgmDataType:[1] -> 1<<1 == 2 -> Backfill
bf = Event(self.fixtureBackfill)
self.assertEqual(bf.cgmDataTypeRaw, 2)
self.assertEqual(bf.cgmDataType,
eventtypes.LidCgmDataG7.CgmdatatypeBitmask.Backfill)
def test_egvinfobitmask_folds_to_raw_int(self):
# [0,5,6,7,8,11,12] -> sum(1<<i) == 6625
ev = Event(self.fixtureRising)
self.assertEqual(ev.egvInfoBitmaskRaw,
sum(1 << i for i in [0, 5, 6, 7, 8, 11, 12]))
self.assertEqual(ev.egvInfoBitmaskRaw, 6625)
# [1,5,6,7,8,11,12] -> sum(1<<i) == 6626
bf = Event(self.fixtureBackfill)
self.assertEqual(bf.egvInfoBitmaskRaw,
sum(1 << i for i in [1, 5, 6, 7, 8, 11, 12]))
self.assertEqual(bf.egvInfoBitmaskRaw, 6626)
def test_rate_ratio_scales(self):
# rateRaw ×0.1 mg/dL/min
rising = Event(self.fixtureRising)
self.assertEqual(rising.rateRaw, 52)
self.assertAlmostEqual(rising.rate, 5.2)
falling = Event(self.fixtureFalling)
self.assertEqual(falling.rateRaw, -39)
self.assertAlmostEqual(falling.rate, -3.9)
flat = Event(self.fixtureBackfill)
self.assertEqual(flat.rateRaw, 0)
self.assertAlmostEqual(flat.rate, 0.0)
def test_egv_timestamp_is_raw_seconds(self):
# egvTimeStamp (camelCase) normalizes onto egvTimestamp, kept as raw seconds int.
self.assertEqual(Event(self.fixtureRising).egvTimestamp, 580133707)
self.assertEqual(Event(self.fixtureFalling).egvTimestamp, 578964361)
self.assertEqual(Event(self.fixtureSpecialLow).egvTimestamp, 579800690)
self.assertEqual(Event(self.fixtureBackfill).egvTimestamp, 580602548)
def test_todict_json_serializable(self):
for fx in (self.fixtureRising, self.fixtureFalling,
self.fixtureSpecialLow, self.fixtureBackfill):
json.dumps(Event(fx).todict()) # must not raise
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,91 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCgmJoinSessionG7(unittest.TestCase):
"""394 LID_CGM_JOIN_SESSION_G7: cgmTimestamp/sessionSignature are plain ints."""
maxDiff = None
def setUp(self):
# Real captured pump-log events (values copied verbatim).
self.fixtureLowCgmTs = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 394,
"sequenceGroup": 0,
"sequenceNumber": 413856,
"pumpDateTime": "2026-05-06T01:53:28",
"eventProperties": {"cgmTimestamp": 3042, "sessionSignature": 72},
"estimatedDateTime": "2026-05-06T01:53:28Z",
}
self.fixtureHighCgmTs = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 394,
"sequenceGroup": 0,
"sequenceNumber": 449825,
"pumpDateTime": "2026-05-16T12:38:39",
"eventProperties": {"cgmTimestamp": 51767, "sessionSignature": 117},
"estimatedDateTime": "2026-05-16T12:38:39Z",
}
self.fixtureSharedSignature = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 394,
"sequenceGroup": 0,
"sequenceNumber": 481763,
"pumpDateTime": "2026-05-25T09:39:19",
"eventProperties": {"cgmTimestamp": 248, "sessionSignature": 117},
"estimatedDateTime": "2026-05-25T09:39:19Z",
}
def test_dispatches_to_correct_class(self):
ev = Event(self.fixtureLowCgmTs)
self.assertIsInstance(ev, eventtypes.LidCgmJoinSessionG7)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureLowCgmTs)
self.assertEqual(ev.eventId, 394)
self.assertEqual(ev.seqNum, 413856)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureLowCgmTs)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-06T01:53:28")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureLowCgmTs)
self.assertEqual(ev.cgmtimestamp, 3042)
self.assertEqual(ev.sessionsignature, 72)
def test_high_cgm_timestamp_fixture(self):
ev = Event(self.fixtureHighCgmTs)
self.assertEqual(ev.eventId, 394)
self.assertEqual(ev.seqNum, 449825)
self.assertEqual(ev.cgmtimestamp, 51767)
self.assertEqual(ev.sessionsignature, 117)
def test_shared_signature_fixture(self):
# Same sessionSignature as the high-cgm fixture but a distinct cgmTimestamp.
ev = Event(self.fixtureSharedSignature)
self.assertEqual(ev.seqNum, 481763)
self.assertEqual(ev.cgmtimestamp, 248)
self.assertEqual(ev.sessionsignature, 117)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureLowCgmTs)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 394)
self.assertEqual(d["name"], "LID_CGM_JOIN_SESSION_G7")
self.assertEqual(d["seqNum"], 413856)
self.assertEqual(d["cgmtimestamp"], 3042)
self.assertEqual(d["sessionsignature"], 72)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,110 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidCgmStopSessionG7(unittest.TestCase):
"""447: LID_CGM_STOP_SESSION_G7. sessionStopReason is a plain int (no enum
in the generated class), so it round-trips as its captured value. Fixtures
are real captured events copied verbatim, one per distinct stop reason."""
maxDiff = None
def setUp(self):
# sessionStopReason=5 with a real stop time and stopSessionCode=1.
self.fixtureReason5 = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 447,
"sequenceGroup": 0,
"sequenceNumber": 413840,
"pumpDateTime": "2026-05-06T01:53:10",
"eventProperties": {
"currentTransmitterTime": 890598, "sessionStartTime": 73,
"sessionStopTime": 890591, "sessionDuration": 10,
"sessionStopReason": 5, "stopSessionCode": 1,
},
"estimatedDateTime": "2026-05-06T01:53:10Z",
}
# sessionStopReason=16 with zero stop time / code.
self.fixtureReason16 = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 447,
"sequenceGroup": 0,
"sequenceNumber": 449816,
"pumpDateTime": "2026-05-16T12:38:30",
"eventProperties": {
"currentTransmitterTime": 905595, "sessionStartTime": 72,
"sessionStopTime": 0, "sessionDuration": 10,
"sessionStopReason": 16, "stopSessionCode": 0,
},
"estimatedDateTime": "2026-05-16T12:38:30Z",
}
# sessionStopReason=15 with a UINT32-max sessionStartTime sentinel.
self.fixtureReason15 = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 447,
"sequenceGroup": 0,
"sequenceNumber": 481653,
"pumpDateTime": "2026-05-25T08:35:32",
"eventProperties": {
"currentTransmitterTime": 814740, "sessionStartTime": 4294967295,
"sessionStopTime": 0, "sessionDuration": 10,
"sessionStopReason": 15, "stopSessionCode": 0,
},
"estimatedDateTime": "2026-05-25T08:35:32Z",
}
def test_dispatches_to_lidcgmstopsessiong7(self):
ev = Event(self.fixtureReason5)
self.assertIsInstance(ev, eventtypes.LidCgmStopSessionG7)
self.assertIsNot(type(ev), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureReason5)
self.assertEqual(ev.eventId, 447)
self.assertEqual(ev.seqNum, 413840)
# eventTimestamp keeps pumpDateTime's wall-clock.
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-06T01:53:10")
def test_session_time_fields_round_trip(self):
ev = Event(self.fixtureReason5)
self.assertEqual(ev.currenttransmittertime, 890598)
self.assertEqual(ev.sessionstarttime, 73)
self.assertEqual(ev.sessionstoptime, 890591)
self.assertEqual(ev.sessionduration, 10)
self.assertEqual(ev.stopsessioncode, 1)
def test_session_start_time_sentinel_round_trips(self):
# UINT32-max sentinel must survive as-is.
ev = Event(self.fixtureReason15)
self.assertEqual(ev.sessionstarttime, 4294967295)
self.assertEqual(ev.sessionstoptime, 0)
self.assertEqual(ev.stopsessioncode, 0)
def test_session_stop_reason_is_raw_int(self):
# No enum is generated for sessionStopReason; it stays the captured int.
self.assertFalse(hasattr(eventtypes.LidCgmStopSessionG7,
"SessionstopreasonEnum"))
self.assertEqual(Event(self.fixtureReason5).sessionstopreason, 5)
self.assertEqual(Event(self.fixtureReason16).sessionstopreason, 16)
self.assertEqual(Event(self.fixtureReason15).sessionstopreason, 15)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureReason5, self.fixtureReason16,
self.fixtureReason15):
d = Event(fixture).todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 447)
self.assertEqual(d["name"], "LID_CGM_STOP_SESSION_G7")
self.assertEqual(
d["sessionstopreason"],
fixture["eventProperties"]["sessionStopReason"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,95 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidDateChanged(unittest.TestCase):
maxDiff = None
def setUp(self):
# Real captured code-14 (LID_DATE_CHANGED) events, copied verbatim from
# the clockChanges arrays of pump-log responses.
# datePrior == dateAfter: a resync that did not move the day.
self.fixtureEqual = {
"deviceAssignmentId": "73aeb403-1d22-4d12-a3fd-229e5b6641ee",
"eventCode": 14,
"sequenceGroup": 0,
"sequenceNumber": 364364,
"pumpDateTime": "2025-11-02T16:27:58",
"eventProperties": {"datePrior": 6515, "dateAfter": 6515, "rawRtcTime": 1625159578},
"estimatedDateTime": "2025-11-02T16:27:58Z",
}
# datePrior != dateAfter by one day: a small clock adjustment.
self.fixtureOffByOne = {
"deviceAssignmentId": "73aeb403-1d22-4d12-a3fd-229e5b6641ee",
"eventCode": 14,
"sequenceGroup": 0,
"sequenceNumber": 418774,
"pumpDateTime": "2025-11-18T22:55:04",
"eventProperties": {"datePrior": 6530, "dateAfter": 6531, "rawRtcTime": 2958809165},
"estimatedDateTime": "2025-11-18T22:55:04Z",
}
# Large jump (initial date set): datePrior far from dateAfter.
self.fixtureLargeJump = {
"deviceAssignmentId": "73aeb403-1d22-4d12-a3fd-229e5b6641ee",
"eventCode": 14,
"sequenceGroup": 0,
"sequenceNumber": 95,
"pumpDateTime": "2025-07-20T04:38:28",
"eventProperties": {"datePrior": 4394, "dateAfter": 6410, "rawRtcTime": 1089120495},
"estimatedDateTime": "2025-07-20T04:38:28Z",
}
def test_dispatches_to_liddatechanged(self):
for fx in (self.fixtureEqual, self.fixtureOffByOne, self.fixtureLargeJump):
ev = Event(fx)
self.assertIsInstance(ev, eventtypes.LidDateChanged)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureOffByOne)
self.assertEqual(ev.eventId, 14)
self.assertEqual(ev.seqNum, 418774)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2025-11-18T22:55:04")
def test_equal_dates_round_trip(self):
ev = Event(self.fixtureEqual)
self.assertEqual(ev.dateprior, 6515)
self.assertEqual(ev.dateafter, 6515)
self.assertEqual(ev.Rawrtctime, 1625159578)
def test_off_by_one_dates_round_trip(self):
ev = Event(self.fixtureOffByOne)
self.assertEqual(ev.dateprior, 6530)
self.assertEqual(ev.dateafter, 6531)
self.assertEqual(ev.Rawrtctime, 2958809165)
def test_large_jump_dates_round_trip(self):
ev = Event(self.fixtureLargeJump)
self.assertEqual(ev.dateprior, 4394)
self.assertEqual(ev.dateafter, 6410)
self.assertEqual(ev.Rawrtctime, 1089120495)
def test_todict_is_json_serializable(self):
for fx in (self.fixtureEqual, self.fixtureOffByOne, self.fixtureLargeJump):
ev = Event(fx)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 14)
self.assertEqual(d["name"], "LID_DATE_CHANGED")
self.assertEqual(d["seqNum"], fx["sequenceNumber"])
self.assertEqual(d["dateprior"], fx["eventProperties"]["datePrior"])
self.assertEqual(d["dateafter"], fx["eventProperties"]["dateAfter"])
self.assertEqual(d["Rawrtctime"], fx["eventProperties"]["rawRtcTime"])
if __name__ == "__main__":
unittest.main()
+117
View File
@@ -0,0 +1,117 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidNewDay(unittest.TestCase):
"""90: LID_NEW_DAY. Real captured pump-log events.
featuresBitmask / featureBitmaskIndex arrive as JSON arrays but the schema
declares them plain (no bitmask transform), so the parser stores the raw
value verbatim (the list as-is, e.g. []).
"""
maxDiff = None
def setUp(self):
# commandedBasalRate == 0
self.fixtureZeroBasal = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 90,
"sequenceGroup": 0,
"sequenceNumber": 393109,
"pumpDateTime": "2026-04-30T00:00:00",
"eventProperties": {
"commandedBasalRate": 0,
"featuresBitmask": [],
"featureBitmaskIndex": [],
},
"estimatedDateTime": "2026-04-30T00:00:00Z",
}
# fractional commandedBasalRate
self.fixtureFractionalBasal = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 90,
"sequenceGroup": 0,
"sequenceNumber": 396368,
"pumpDateTime": "2026-05-01T00:00:00",
"eventProperties": {
"commandedBasalRate": 0.763,
"featuresBitmask": [],
"featureBitmaskIndex": [],
},
"estimatedDateTime": "2026-05-01T00:00:00Z",
}
# integer-valued commandedBasalRate == 1
self.fixtureUnitBasal = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 90,
"sequenceGroup": 0,
"sequenceNumber": 409989,
"pumpDateTime": "2026-05-05T00:00:00",
"eventProperties": {
"commandedBasalRate": 1,
"featuresBitmask": [],
"featureBitmaskIndex": [],
},
"estimatedDateTime": "2026-05-05T00:00:00Z",
}
def test_dispatches_to_lidnewday(self):
ev = Event(self.fixtureZeroBasal)
self.assertIsInstance(ev, eventtypes.LidNewDay)
self.assertIsNot(type(ev), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureZeroBasal)
self.assertEqual(ev.eventId, 90)
self.assertEqual(ev.seqNum, 393109)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureZeroBasal)
self.assertEqual(
ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T00:00:00",
)
def test_commanded_basal_rate_zero(self):
ev = Event(self.fixtureZeroBasal)
self.assertEqual(ev.commandedbasalrate, 0)
def test_commanded_basal_rate_fractional(self):
ev = Event(self.fixtureFractionalBasal)
self.assertEqual(ev.commandedbasalrate, 0.763)
def test_commanded_basal_rate_unit(self):
ev = Event(self.fixtureUnitBasal)
self.assertEqual(ev.commandedbasalrate, 1)
def test_features_bitmask_holds_list_verbatim(self):
# No transform on the schema: the captured list is stored as-is.
ev = Event(self.fixtureZeroBasal)
self.assertEqual(ev.featuresbitmask, [])
self.assertIsInstance(ev.featuresbitmask, list)
def test_feature_bitmask_index_holds_list_verbatim(self):
ev = Event(self.fixtureZeroBasal)
self.assertEqual(ev.featurebitmaskindex, [])
self.assertIsInstance(ev.featurebitmaskindex, list)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureFractionalBasal)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 90)
self.assertEqual(d["name"], "LID_NEW_DAY")
self.assertEqual(d["seqNum"], 396368)
self.assertEqual(d["commandedbasalrate"], 0.763)
self.assertEqual(d["featuresbitmask"], [])
self.assertEqual(d["featurebitmaskindex"], [])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidPumpingResumed(unittest.TestCase):
"""12: LID_PUMPING_RESUMED. Real captures all share preResumeState:100 and
only differ in insulinAmount; two fixtures show that field varying."""
maxDiff = None
def setUp(self):
# Real captured pump-log events (verbatim).
self.fixtureA = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 12,
"sequenceGroup": 0,
"sequenceNumber": 394429,
"pumpDateTime": "2026-04-30T10:16:31",
"eventProperties": {"preResumeState": 100, "insulinAmount": 105},
"estimatedDateTime": "2026-04-30T10:16:31Z",
}
self.fixtureB = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 12,
"sequenceGroup": 0,
"sequenceNumber": 478545,
"pumpDateTime": "2026-05-24T11:41:19",
"eventProperties": {"preResumeState": 100, "insulinAmount": 13},
"estimatedDateTime": "2026-05-24T11:41:19Z",
}
def test_dispatches_to_correct_class(self):
self.assertIsInstance(Event(self.fixtureA), eventtypes.LidPumpingResumed)
self.assertNotIsInstance(Event(self.fixtureA), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureA)
self.assertEqual(ev.eventId, 12)
self.assertEqual(ev.seqNum, 394429)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureA)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T10:16:31")
def test_plain_fields(self):
ev = Event(self.fixtureA)
self.assertEqual(ev.preresumestate, 100)
self.assertEqual(ev.insulinamount, 105)
def test_insulin_amount_varies_across_captures(self):
self.assertEqual(Event(self.fixtureA).insulinamount, 105)
self.assertEqual(Event(self.fixtureB).insulinamount, 13)
# preResumeState is constant across real captures.
self.assertEqual(Event(self.fixtureB).preresumestate, 100)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureB)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 12)
self.assertEqual(d["name"], "LID_PUMPING_RESUMED")
self.assertEqual(d["seqNum"], 478545)
self.assertEqual(d["preresumestate"], 100)
self.assertEqual(d["insulinamount"], 13)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,88 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidPumpingSuspended(unittest.TestCase):
"""11: LID_PUMPING_SUSPENDED. All real captures share suspendReason:0
(UserAborted); only insulinAmount varies, so two captures suffice."""
maxDiff = None
def setUp(self):
# Real captured pump-log events (verbatim), eventCode 11.
self.fixtureA = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 11,
"sequenceGroup": 0,
"sequenceNumber": 394335,
"pumpDateTime": "2026-04-30T10:01:49",
"eventProperties": {
"preSuspendState": 106, "insulinAmount": 120,
"suspendReason": 0, "rpaTimeout": 15,
},
"estimatedDateTime": "2026-04-30T10:01:49Z",
}
self.fixtureB = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 11,
"sequenceGroup": 0,
"sequenceNumber": 401271,
"pumpDateTime": "2026-05-02T11:49:43",
"eventProperties": {
"preSuspendState": 106, "insulinAmount": 150,
"suspendReason": 0, "rpaTimeout": 15,
},
"estimatedDateTime": "2026-05-02T11:49:43Z",
}
def test_dispatches_to_correct_class(self):
self.assertIsInstance(Event(self.fixtureA), eventtypes.LidPumpingSuspended)
self.assertNotIsInstance(Event(self.fixtureA), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureA)
self.assertEqual(ev.eventId, 11)
self.assertEqual(ev.seqNum, 394335)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureA)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T10:01:49")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureA)
self.assertEqual(ev.presuspendstate, 106)
self.assertEqual(ev.insulinamount, 120)
self.assertEqual(ev.rpatimeout, 15)
def test_suspendreason_resolves_to_enum(self):
# suspendReason:0 -> UserAborted
ev = Event(self.fixtureA)
self.assertEqual(ev.suspendreasonRaw, 0)
self.assertEqual(ev.suspendreason,
eventtypes.LidPumpingSuspended.SuspendreasonEnum.UserAborted)
def test_second_capture_distinct_insulin_amount(self):
ev = Event(self.fixtureB)
self.assertEqual(ev.seqNum, 401271)
self.assertEqual(ev.insulinamount, 150)
self.assertEqual(ev.suspendreason,
eventtypes.LidPumpingSuspended.SuspendreasonEnum.UserAborted)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureA)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 11)
self.assertEqual(d["name"], "LID_PUMPING_SUSPENDED")
self.assertEqual(d["insulinamount"], 120)
self.assertEqual(d["suspendreasonRaw"], 0)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,100 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidTimeChanged(unittest.TestCase):
"""13: LID_TIME_CHANGED. Real captures from the clockChanges array.
timePrior/timeAfter are ms-of-day; compare them to see whether the clock
moved forward or backward."""
maxDiff = None
def setUp(self):
# Real capture: clock moved forward (timeAfter > timePrior).
self.fixtureForward = {
"deviceAssignmentId": "73aeb403-1d22-4d12-a3fd-229e5b6641ee",
"eventCode": 13,
"sequenceGroup": 0,
"sequenceNumber": 182932,
"pumpDateTime": "2025-09-11T10:28:11",
"eventProperties": {
"timePrior": 12513657, "timeAfter": 37691000,
"rawRtcTime": 1380540797,
},
"estimatedDateTime": "2025-09-11T10:28:11Z",
}
# Real capture: clock moved backward (timeAfter < timePrior).
self.fixtureBackward = {
"deviceAssignmentId": "73aeb403-1d22-4d12-a3fd-229e5b6641ee",
"eventCode": 13,
"sequenceGroup": 0,
"sequenceNumber": 364365,
"pumpDateTime": "2025-11-02T15:27:37",
"eventProperties": {
"timePrior": 59278139, "timeAfter": 55657000,
"rawRtcTime": 1625159580,
},
"estimatedDateTime": "2025-11-02T15:27:37Z",
}
def test_dispatches_to_lidtimechanged(self):
self.assertIsInstance(Event(self.fixtureForward), eventtypes.LidTimeChanged)
self.assertIsInstance(Event(self.fixtureBackward), eventtypes.LidTimeChanged)
self.assertNotIsInstance(Event(self.fixtureForward), RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureForward)
self.assertEqual(ev.eventId, 13)
self.assertEqual(ev.seqNum, 182932)
ev = Event(self.fixtureBackward)
self.assertEqual(ev.eventId, 13)
self.assertEqual(ev.seqNum, 364365)
def test_timestamp_preserves_wall_clock(self):
# eventTimestamp keeps pumpDateTime's wall-clock (tz forced to the
# configured TIMEZONE_NAME), so the naive portion round-trips exactly.
ev = Event(self.fixtureForward)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2025-09-11T10:28:11")
ev = Event(self.fixtureBackward)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2025-11-02T15:27:37")
def test_time_fields_round_trip(self):
ev = Event(self.fixtureForward)
self.assertEqual(ev.timeprior, 12513657)
self.assertEqual(ev.timeafter, 37691000)
self.assertEqual(ev.Rawrtctime, 1380540797)
ev = Event(self.fixtureBackward)
self.assertEqual(ev.timeprior, 59278139)
self.assertEqual(ev.timeafter, 55657000)
self.assertEqual(ev.Rawrtctime, 1625159580)
def test_forward_and_backward_direction(self):
# timeAfter > timePrior means the clock jumped forward, and vice versa.
fwd = Event(self.fixtureForward)
self.assertGreater(fwd.timeafter, fwd.timeprior)
bwd = Event(self.fixtureBackward)
self.assertLess(bwd.timeafter, bwd.timeprior)
def test_todict_is_json_serializable(self):
for fixture in (self.fixtureForward, self.fixtureBackward):
ev = Event(fixture)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 13)
self.assertEqual(d["name"], "LID_TIME_CHANGED")
self.assertEqual(d["timeprior"], fixture["eventProperties"]["timePrior"])
self.assertEqual(d["timeafter"], fixture["eventProperties"]["timeAfter"])
self.assertEqual(d["Rawrtctime"], fixture["eventProperties"]["rawRtcTime"])
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,96 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidTubingFilled(unittest.TestCase):
"""63: LID_TUBING_FILLED, parsed from real captured pump-log events.
Every captured code-63 event carries primeSize=-1 (a sentinel, not a real
fill volume) and completionStatus=3 (Completed); fixtures differ in
seqNum/position/wall-clock.
"""
maxDiff = None
def setUp(self):
# Real capture: primeSize sentinel (-1), completionStatus Completed (3).
self.fixtureNegativePrimeSize = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 63,
"sequenceGroup": 0,
"sequenceNumber": 394428,
"pumpDateTime": "2026-04-30T10:16:09",
"eventProperties": {"primeSize": -1, "completionStatus": 3, "position": 631224},
"estimatedDateTime": "2026-04-30T10:16:09Z",
}
# Second real capture with a different seqNum/position/timestamp.
self.fixtureCompleted = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 63,
"sequenceGroup": 0,
"sequenceNumber": 401423,
"pumpDateTime": "2026-05-02T12:27:17",
"eventProperties": {"primeSize": -1, "completionStatus": 3, "position": 594056},
"estimatedDateTime": "2026-05-02T12:27:17Z",
}
def test_dispatches_to_lidtubingfilled(self):
ev = Event(self.fixtureNegativePrimeSize)
self.assertIsInstance(ev, eventtypes.LidTubingFilled)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureNegativePrimeSize)
self.assertEqual(ev.eventId, 63)
self.assertEqual(ev.seqNum, 394428)
self.assertEqual(ev.NAME, "LID_TUBING_FILLED")
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureNegativePrimeSize)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T10:16:09")
def test_negative_primesize_sentinel_round_trips(self):
# -1 is a sentinel and must survive verbatim (not treated as missing).
ev = Event(self.fixtureNegativePrimeSize)
self.assertEqual(ev.primesize, -1)
def test_position_round_trips(self):
ev = Event(self.fixtureNegativePrimeSize)
self.assertEqual(ev.position, 631224)
def test_completionstatus_resolves_to_completed(self):
ev = Event(self.fixtureNegativePrimeSize)
self.assertEqual(ev.completionstatusRaw, 3)
self.assertEqual(ev.completionstatus,
eventtypes.LidTubingFilled.CompletionstatusEnum.Completed)
def test_second_capture_distinct_values(self):
ev = Event(self.fixtureCompleted)
self.assertEqual(ev.seqNum, 401423)
self.assertEqual(ev.position, 594056)
self.assertEqual(ev.primesize, -1)
self.assertEqual(ev.completionstatus,
eventtypes.LidTubingFilled.CompletionstatusEnum.Completed)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-05-02T12:27:17")
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureNegativePrimeSize)
td = ev.todict()
json.dumps(td) # must not raise
self.assertEqual(td["id"], 63)
self.assertEqual(td["name"], "LID_TUBING_FILLED")
self.assertEqual(td["seqNum"], 394428)
self.assertEqual(td["primesize"], -1)
self.assertEqual(td["completionstatusRaw"], 3)
self.assertEqual(td["position"], 631224)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
import json
import unittest
from tconnectsync.eventparser.generic import Event
from tconnectsync.eventparser import events as eventtypes
from tconnectsync.eventparser.raw_event import RawEvent
class TestLidVersionsA(unittest.TestCase):
"""307 / LID_VERSIONS_A: four plain uint32 version/part-number fields.
Only one distinct capture shape exists, so a single real fixture suffices."""
maxDiff = None
def setUp(self):
# Real captured LID_VERSIONS_A event, copied verbatim.
self.fixtureVersions = {
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
"eventCode": 307,
"sequenceGroup": 0,
"sequenceNumber": 393111,
"pumpDateTime": "2026-04-30T00:00:06",
"eventProperties": {
"armPartNumber": 1016587,
"armSwVersion": 1108201743,
"blePartNumber": 1016587,
"bleSwVersion": 1108201743,
},
"estimatedDateTime": "2026-04-30T00:00:06Z",
}
def test_dispatches_to_lidversionsa(self):
ev = Event(self.fixtureVersions)
self.assertIsInstance(ev, eventtypes.LidVersionsA)
self.assertNotIsInstance(ev, RawEvent)
def test_envelope_fields(self):
ev = Event(self.fixtureVersions)
self.assertEqual(ev.eventId, 307)
self.assertEqual(ev.seqNum, 393111)
def test_timestamp_preserves_wall_clock(self):
ev = Event(self.fixtureVersions)
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'),
"2026-04-30T00:00:06")
def test_plain_fields_round_trip(self):
ev = Event(self.fixtureVersions)
self.assertEqual(ev.armpartnumber, 1016587)
self.assertEqual(ev.armswversion, 1108201743)
self.assertEqual(ev.blepartnumber, 1016587)
self.assertEqual(ev.bleswversion, 1108201743)
def test_todict_is_json_serializable(self):
ev = Event(self.fixtureVersions)
d = ev.todict()
json.dumps(d) # must not raise
self.assertEqual(d["id"], 307)
self.assertEqual(d["name"], "LID_VERSIONS_A")
self.assertEqual(d["seqNum"], 393111)
self.assertEqual(d["armpartnumber"], 1016587)
self.assertEqual(d["armswversion"], 1108201743)
self.assertEqual(d["blepartnumber"], 1016587)
self.assertEqual(d["bleswversion"], 1108201743)
if __name__ == "__main__":
unittest.main()