diff --git a/tconnectsync/nightscout.py b/tconnectsync/nightscout.py index d24ce24..e57f293 100644 --- a/tconnectsync/nightscout.py +++ b/tconnectsync/nightscout.py @@ -19,12 +19,14 @@ DateLike = Union[str, datetime.datetime, arrow.Arrow] def format_datetime(date: DateLike) -> str: return arrow.get(date).isoformat() -def time_range(field_name: str, start_time: Optional[DateLike], end_time: Optional[DateLike], t_to_space: bool = False) -> str: +def time_range(field_name: str, start_time: Optional[DateLike], end_time: Optional[DateLike]) -> str: def fmt(date: DateLike) -> str: ret = format_datetime(date) - if t_to_space: - return ret.replace('T', ' ') - return ret + # URL-encode so the '+' in offsets like '+02:00' is not decoded + # to a space by the server, which would mangle the ISO-8601 value. + # Upstream instead retries with 'T' replaced by a space (t_to_space); + # encoding the value fixes the cause, so that fallback is not carried. + return urllib.parse.quote(ret, safe='') arg = '' if start_time: arg += '&find[%s][$gte]=%s' % (field_name, fmt(start_time)) @@ -70,37 +72,18 @@ class NightscoutApi: raise ApiException(r.status_code, "Nightscout put %s response: %s" % (r.status_code, r.text)) def last_uploaded_entry(self, eventType: str, time_start: Optional[DateLike] = None, time_end: Optional[DateLike] = None) -> Optional[dict]: - def internal(t_to_space: bool) -> Optional[dict]: - dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space) + dateFilter = time_range('created_at', time_start, time_end) + try: latest = requests.get(urljoin(self.url, 'api/v1/treatments?count=1&find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[eventType]=' + urllib.parse.quote(eventType) + dateFilter + '&ts=' + str(time.time())), headers={ 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() }, verify=self.verify) if latest.status_code != 200: - if 'as a valid ISO-8601 date' in latest.text: - logger.warning("Nightscout last_uploaded_entry %s could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (eventType, time_start, time_end, dateFilter)) - return None raise ApiException(latest.status_code, "Nightscout last_uploaded_entry %s response: %s" % (latest.status_code, latest.text)) j = latest.json() if j and len(j) > 0: return j[0] return None - try: - ret = None - try: - ret = internal(False) - except ApiException as e: - #logger.warning("last_uploaded_entry with no t_to_space: %s", e) - ret = None - if ret is None and (time_start or time_end): - try: - ret = internal(True) - except ApiException as e: - #logger.warning("last_uploaded_entry with t_to_space: %s", e) - ret = None - if ret is not None: - logger.warning("last_uploaded_entry with eventType=%s time_start=%s time_end=%s only returned data when timestamps contained a space" % (eventType, time_start, time_end)) - return ret except requests.exceptions.ConnectionError as e: if self.ignore_conn_errors: logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e) @@ -108,30 +91,18 @@ class NightscoutApi: raise e def last_uploaded_bg_entry(self, time_start: Optional[DateLike] = None, time_end: Optional[DateLike] = None) -> Optional[dict]: - def internal(t_to_space: bool) -> Optional[dict]: - dateFilter = time_range('dateString', time_start, time_end, t_to_space=t_to_space) + dateFilter = time_range('dateString', time_start, time_end) + try: latest = requests.get(urljoin(self.url, 'api/v1/entries.json?count=1&find[device]=' + urllib.parse.quote(ENTERED_BY) + dateFilter + '&ts=' + str(time.time())), headers={ 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() }, verify=self.verify) if latest.status_code != 200: - if 'as a valid ISO-8601 date' in latest.text: - logger.warning("Nightscout last_uploaded_bg_entry could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (time_start, time_end, dateFilter)) - return None raise ApiException(latest.status_code, "Nightscout last_uploaded_bg_entry %s response: %s" % (latest.status_code, latest.text)) j = latest.json() if j and len(j) > 0: return j[0] return None - - try: - ret = internal(False) - if ret is None and (time_start or time_end): - ret = internal(True) - if ret is not None: - logger.warning("last_uploaded_bg_entry with time_start=%s time_end=%s only returned data when timestamps contained a space" % (time_start, time_end)) - - return ret except requests.exceptions.ConnectionError as e: if self.ignore_conn_errors: logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e) @@ -139,29 +110,18 @@ class NightscoutApi: raise e def last_uploaded_activity(self, activityType: str, time_start: Optional[DateLike] = None, time_end: Optional[DateLike] = None) -> Optional[dict]: - def internal(t_to_space: bool) -> Optional[dict]: - dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space) + dateFilter = time_range('created_at', time_start, time_end) + try: latest = requests.get(urljoin(self.url, 'api/v1/activity?find[enteredBy]=' + urllib.parse.quote(ENTERED_BY) + '&find[activityType]=' + urllib.parse.quote(activityType) + dateFilter + '&ts=' + str(time.time())), headers={ 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() }, verify=self.verify) if latest.status_code != 200: - if 'as a valid ISO-8601 date' in latest.text: - logger.warning("Nightscout activity %s could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (activityType, time_start, time_end, dateFilter)) - return None raise ApiException(latest.status_code, "Nightscout activity %s response: %s" % (latest.status_code, latest.text)) j = latest.json() if j and len(j) > 0: return j[0] return None - - try: - ret = internal(False) - if ret is None and (time_start or time_end): - ret = internal(True) - if ret is not None: - logger.warning("last_uploaded_activity with activityType=%s time_start=%s time_end=%s only returned data when timestamps contained a space" % (activityType, time_start, time_end)) - return ret except requests.exceptions.ConnectionError as e: if self.ignore_conn_errors: logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e) @@ -169,29 +129,18 @@ class NightscoutApi: raise e def last_uploaded_devicestatus(self, time_start: Optional[DateLike] = None, time_end: Optional[DateLike] = None) -> Optional[dict]: - def internal(t_to_space: bool) -> Optional[dict]: - dateFilter = time_range('created_at', time_start, time_end, t_to_space=t_to_space) + dateFilter = time_range('created_at', time_start, time_end) + try: latest = requests.get(urljoin(self.url, 'api/v1/devicestatus?find[device]=' + urllib.parse.quote(ENTERED_BY) + dateFilter + '&ts=' + str(time.time())), headers={ 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() }, verify=self.verify) if latest.status_code != 200: - if 'as a valid ISO-8601 date' in latest.text: - logger.warning("Nightscout devicestatus could not process ISO-8601 date: start=%s end=%s dateFilter=%s" % (time_start, time_end, dateFilter)) - return None raise ApiException(latest.status_code, "Nightscout devicestatus %s response: %s" % (latest.status_code, latest.text)) j = latest.json() if j and len(j) > 0: return j[0] return None - - try: - ret = internal(False) - if ret is None and (time_start or time_end): - ret = internal(True) - if ret is not None: - logger.warning("devicestatus time_start=%s time_end=%s only returned data when timestamps contained a space" % (time_start, time_end)) - return ret except requests.exceptions.ConnectionError as e: if self.ignore_conn_errors: logger.warn('Ignoring ConnectionError because ignore_conn_errors=true', e) @@ -201,7 +150,7 @@ class NightscoutApi: """ Returns general status information about the Nightscout server. """ - def api_status(self) -> dict: + def api_status(self): status = requests.get(urljoin(self.url, 'api/v1/status.json'), headers={ 'api-secret': hashlib.sha1(self.secret.encode()).hexdigest() }, verify=self.verify) @@ -213,7 +162,7 @@ class NightscoutApi: Returns information on the currently configured Nightscout profile data store (contains all profiles in Nightscout under one mongo object). """ - def current_profile(self, time_start: Optional[DateLike] = None, time_end: Optional[DateLike] = None) -> dict: + def current_profile(self, time_start=None, time_end=None): r = requests.get(urljoin(self.url, 'api/v1/profile/current?api_secret=' + self.secret), json={}, headers={ 'Accept': 'application/json', 'Content-Type': 'application/json', diff --git a/tests/test_nightscout.py b/tests/test_nightscout.py new file mode 100644 index 0000000..c6f0920 --- /dev/null +++ b/tests/test_nightscout.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Tests for the Nightscout date-filter query building in time_range().""" + +import unittest +import urllib.parse + +import arrow + +from tconnectsync.nightscout import time_range + + +class TestTimeRangeEncoding(unittest.TestCase): + """A positive UTC offset ends an ISO-8601 timestamp with '+02:00'. Placed + raw into a query string, the '+' is a reserved character that servers decode + as a space, so Nightscout receives '2026-07-16T00:00:00 02:00' and rejects + it with "could not parse as a valid ISO-8601 date". Percent-encoding the + value keeps the offset intact.""" + + def test_positive_offset_is_percent_encoded(self): + start = arrow.get("2026-07-16T00:00:00+02:00") + + arg = time_range('created_at', start, None) + + self.assertNotIn( + '+', arg, + "A raw '+' in the query string is decoded to a space by the server, " + "mangling the timestamp. Got: %s" % arg, + ) + self.assertIn('%2B', arg, "Expected the offset '+' to be encoded as %%2B. Got: %s" % arg) + + def test_encoded_value_round_trips_to_the_original_timestamp(self): + """Decoding the query the way a server would must yield the timestamp + we meant to send.""" + start = arrow.get("2026-07-16T00:00:00+02:00") + + arg = time_range('created_at', start, None) + + value = arg.split('=', 1)[1] + self.assertEqual( + arrow.get(urllib.parse.unquote(value)), start, + "Round-tripping the encoded filter must reproduce the original instant", + ) + + def test_negative_offset_and_z_suffix_still_parse(self): + """US pumps report a '-04:00' offset and UTC values end in 'Z'; neither + is ambiguous in a query string, but both must survive encoding.""" + for iso in ("2026-07-16T00:00:00-04:00", "2026-07-16T00:00:00Z"): + with self.subTest(iso=iso): + expected = arrow.get(iso) + arg = time_range('created_at', expected, None) + value = arg.split('=', 1)[1] + self.assertEqual(arrow.get(urllib.parse.unquote(value)), expected) + + def test_both_bounds_are_emitted(self): + start = arrow.get("2026-07-16T00:00:00+02:00") + end = arrow.get("2026-07-17T00:00:00+02:00") + + arg = time_range('created_at', start, end) + + self.assertIn('find[created_at][$gte]=', arg) + self.assertIn('find[created_at][$lte]=', arg) + + def test_omitted_bounds_produce_no_filter(self): + self.assertEqual(time_range('created_at', None, None), '') + + +if __name__ == "__main__": + unittest.main()