mirror of
https://github.com/bckelley/tconnectsync.git
synced 2026-08-24 03:34:12 -05:00
386 lines
15 KiB
Python
386 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import logging
|
|
import time
|
|
import unittest
|
|
import datetime
|
|
import importlib
|
|
import contextlib
|
|
|
|
from unittest.mock import patch
|
|
|
|
from tconnectsync.autoupdate import Autoupdate, AutoupdateFailureError, AutoupdateFailureWarning, AutoupdateNoEventIndexesDetectedError, AutoupdateNoIndexChangeWarning
|
|
from tconnectsync.secret import AUTOUPDATE_FAILURE_MINUTES, AUTOUPDATE_NO_DATA_FAILURE_MINUTES, AUTOUPDATE_RESTART_ON_FAILURE, TCONNECT_PASSWORD
|
|
|
|
from .api.fake import TConnectApi
|
|
from .nightscout_fake import NightscoutApi
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def stub(*args, **kwargs):
|
|
pass
|
|
|
|
@contextlib.contextmanager
|
|
def build_mock_logger():
|
|
def fake_error(*args, **kwargs):
|
|
logger.error(*args, **kwargs)
|
|
|
|
def fake_warn(*args, **kwargs):
|
|
logger.warn(*args, **kwargs)
|
|
|
|
with patch("tconnectsync.autoupdate.logger.error") as mock_error, patch("tconnectsync.autoupdate.logger.warn") as mock_warn:
|
|
mock_error.side_effect = fake_error
|
|
mock_warn.side_effect = fake_warn
|
|
yield (mock_error, mock_warn)
|
|
|
|
def num_instances_of(cls, m):
|
|
return sum([isinstance(i[0][0], cls) for i in m.call_args_list])
|
|
|
|
class TestAutoupdate(unittest.TestCase):
|
|
maxDiff = None
|
|
|
|
# datetimes are unused
|
|
start = datetime.datetime(2021, 4, 20, 12, 0)
|
|
end = datetime.datetime(2021, 4, 21, 12, 0)
|
|
|
|
# each time the returned function is called, it returns the next argument
|
|
# until the end is reached, at which point it will repeat the last argument
|
|
def fake_last_event_uploaded(self, *indexes):
|
|
index = 0
|
|
def fake(*args, **kwargs):
|
|
nonlocal index
|
|
if index < len(indexes):
|
|
index += 1
|
|
data = {'maxPumpEventIndex': indexes[index-1], 'processingStatus': 1}
|
|
return data
|
|
|
|
return fake
|
|
|
|
# each time the returned function is called, it returns the next argument
|
|
# until the end is reached, at which point it will repeat the last argument
|
|
def fake_process_time_range(self, *returns):
|
|
index = 0
|
|
def fake(*args, **kwargs):
|
|
nonlocal index
|
|
if index < len(returns):
|
|
index += 1
|
|
return returns[index - 1]
|
|
|
|
return fake
|
|
|
|
def build_secrets(self, **kwargs):
|
|
secret = importlib.reload(importlib.import_module("tconnectsync.secret"))
|
|
class FakeSecret(object):
|
|
pass
|
|
|
|
fake = FakeSecret()
|
|
for k in dir(secret):
|
|
setattr(fake, k, getattr(secret, k))
|
|
|
|
for k, v in kwargs.items():
|
|
setattr(fake, k, v)
|
|
|
|
return fake
|
|
|
|
"""process_time_range should always be invoked the first time"""
|
|
def test_process_time_range_called_on_start(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range:
|
|
mock_process_time_range.return_value = 0
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 1)
|
|
self.assertEqual(u.autoupdate_invocations, 1)
|
|
self.assertEqual(u.last_event_index, 1)
|
|
|
|
|
|
"""process_time_range should never be called with pretend"""
|
|
def test_process_time_range_never_called_with_pretend(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range:
|
|
mock_process_time_range.return_value = 0
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=True)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 0)
|
|
self.assertEqual(u.autoupdate_invocations, 1)
|
|
self.assertEqual(u.last_event_index, 1)
|
|
|
|
|
|
"""
|
|
If the event index increases without process_time_range detecting new data,
|
|
AutoupdateFailureWarning should be raised.
|
|
"""
|
|
def test_autoupdate_failure_warning_on_index_process_time_range_discrepancy(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1, 2)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=2,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0,
|
|
TCONNECT_EMAIL="test@email.com",
|
|
TCONNECT_PASSWORD="testpassword"
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn):
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(0, 0)
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 2)
|
|
self.assertEqual(u.autoupdate_invocations, 2)
|
|
self.assertEqual(u.last_event_index, 2)
|
|
|
|
self.assertEqual(num_instances_of(AutoupdateFailureWarning, mock_warn), 1)
|
|
|
|
"""
|
|
On the first attempt, an AutoupdateFailureWarning should never be raised.
|
|
"""
|
|
def test_autoupdate_no_failure_warning_on_index_process_time_range_discrepancy_first_attempt(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn):
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(0, 0)
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 1)
|
|
self.assertEqual(u.autoupdate_invocations, 1)
|
|
self.assertEqual(u.last_event_index, 1)
|
|
|
|
self.assertEqual(num_instances_of(AutoupdateFailureWarning, mock_warn), 0)
|
|
|
|
|
|
"""
|
|
If the event index increases without process_time_range detecting new data
|
|
for AUTOUPDATE_FAILURE_MINUTES, an AutoupdateFailureError should be raised.
|
|
"""
|
|
def test_autoupdate_failure_error_on_index_process_time_range_discrepancy_for_failure_minutes(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1, 2)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=2,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0,
|
|
AUTOUPDATE_FAILURE_MINUTES=0,
|
|
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn):
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(0, 0)
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 2)
|
|
self.assertEqual(u.autoupdate_invocations, 2)
|
|
self.assertEqual(u.last_event_index, 2)
|
|
|
|
self.assertEqual(num_instances_of(AutoupdateFailureError, mock_error), 1)
|
|
|
|
"""
|
|
If the event index increases without process_time_range detecting new data
|
|
for AUTOUPDATE_FAILURE_MINUTES, and AUTOUPDATE_RESTART_ON_FAILURE is true,
|
|
then an AutoupdateFailureError should be raised AND 1 should be returned.
|
|
"""
|
|
def test_autoupdate_failure_error_on_index_process_time_range_discrepancy_for_failure_minutes_performs_restart(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1, 2)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=3,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0,
|
|
AUTOUPDATE_FAILURE_MINUTES=0,
|
|
AUTOUPDATE_RESTART_ON_FAILURE=True,
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn):
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(0, 0)
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 1)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 2)
|
|
self.assertEqual(u.autoupdate_invocations, 1) # exits before invocations is incremented
|
|
self.assertEqual(u.last_event_index, 1) # exits before changed to 2
|
|
|
|
self.assertEqual(num_instances_of(AutoupdateFailureError, mock_error), 1)
|
|
|
|
|
|
"""Validate state after first successful update"""
|
|
def test_state_after_first_successful_update(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range:
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(1)
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 1)
|
|
self.assertEqual(u.autoupdate_invocations, 1)
|
|
self.assertEqual(u.last_event_index, 1)
|
|
self.assertTrue(u.last_event_time == u.last_attempt_time == u.last_successful_process_time_range)
|
|
self.assertEqual(len(u.time_diffs_between_updates), 0)
|
|
|
|
"""Validate sleep occurs for the given fixed length when set"""
|
|
def test_sleep_for_fixed_length_when_set(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
sleep_length = 3 # sentinel
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=sleep_length
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, patch("tconnectsync.autoupdate.time.sleep") as mock_sleep:
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(1)
|
|
mock_sleep.side_effect = None
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(mock_process_time_range.call_count, 1)
|
|
self.assertEqual(u.autoupdate_invocations, 1)
|
|
self.assertEqual(u.last_event_index, 1)
|
|
self.assertTrue(mock_sleep.called)
|
|
self.assertEqual(mock_sleep.call_args[0], (sleep_length,))
|
|
|
|
|
|
"""
|
|
If there is no event index update for AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
|
|
then a AutoupdateNoEventIndexesDetectedError should be raised and
|
|
a restart should be triggered with AUTOUPDATE_RESTART_ON_FAILURE.
|
|
"""
|
|
def test_autoupdate_no_event_indexes_detected_error_on_no_index_change_less_than_three_iterations(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=100,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0.1,
|
|
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=0.2,
|
|
AUTOUPDATE_RESTART_ON_FAILURE=True,
|
|
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=1/600 # 0.1 second
|
|
)
|
|
|
|
# with more than 3 iterations, the sleep iteration will change
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, build_mock_logger() as (mock_error, mock_warn):
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(1, 0)
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 1)
|
|
|
|
self.assertEqual(num_instances_of(AutoupdateNoEventIndexesDetectedError, mock_error), 1)
|
|
|
|
|
|
"""
|
|
If there has been 3 failed attempts since the last time we found new data,
|
|
we should sleep for AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS and
|
|
a AutoupdateNoIndexChangeWarning should be logged.
|
|
"""
|
|
def test_autoupdate_no_index_change_warning_on_unexpected_no_index_sleep(self):
|
|
tconnect = TConnectApi()
|
|
tconnect._android.last_event_uploaded = self.fake_last_event_uploaded(1)
|
|
|
|
nightscout = NightscoutApi()
|
|
|
|
secret = self.build_secrets(
|
|
AUTOUPDATE_MAX_LOOP_INVOCATIONS=5,
|
|
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
|
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0.1,
|
|
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=0.2
|
|
)
|
|
|
|
with patch("tconnectsync.autoupdate.process_time_range") as mock_process_time_range, patch("tconnectsync.autoupdate.time.sleep") as mock_sleep, build_mock_logger() as (mock_error, mock_warn):
|
|
mock_process_time_range.side_effect = self.fake_process_time_range(0)
|
|
mock_sleep.side_effect = None
|
|
|
|
u = Autoupdate(secret)
|
|
ret = u.process(tconnect, nightscout, self.start, self.end, pretend=False)
|
|
self.assertEqual(ret, 0)
|
|
|
|
self.assertEqual(u.autoupdate_invocations, 5)
|
|
self.assertEqual(u.last_event_index, 1)
|
|
self.assertTrue(mock_sleep.called)
|
|
self.assertEqual(len(mock_sleep.call_args_list), 5)
|
|
self.assertEqual(mock_sleep.call_args_list[0][0], (0.1,))
|
|
self.assertEqual(mock_sleep.call_args_list[1][0], (0.1,))
|
|
self.assertEqual(mock_sleep.call_args_list[2][0], (0.1,))
|
|
self.assertEqual(mock_sleep.call_args_list[3][0], (0.2,))
|
|
self.assertEqual(mock_sleep.call_args_list[4][0], (0.2,))
|
|
|
|
self.assertEqual(num_instances_of(AutoupdateNoIndexChangeWarning, mock_warn), 2)
|
|
|