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

URL-encode Nightscout date filters instead of retrying with a mangled timestamp

Browse Source 
arrow's isoformat() ends a timestamp with its UTC offset, e.g.
'2026-07-16T00:00:00+02:00'. Placed raw into a query string, '+' is a
reserved character that servers decode as a space, so Nightscout receives
'2026-07-16T00:00:00 02:00' and answers 'could not parse as a valid ISO-8601
date'. Any user on a positive UTC offset hits this on every date-filtered
query.

The current workaround retries the request with 'T' replaced by a space
(t_to_space) and treats a failure as 'no previous entry'. That gets a
response, but it works around the symptom: the value is still mangled, every
affected query costs two round-trips, and a genuinely empty result is
indistinguishable from a rejected one.

Percent-encoding the value fixes the cause: the offset survives, the first
request succeeds, and the t_to_space fallback and its retry wrapper are no
longer needed and are removed.

Adds tests for time_range(): that a positive offset is encoded rather than
emitted raw, that the encoded value round-trips back to the original instant,
that negative offsets and 'Z' still parse, and the bounds/no-bounds cases.
There was no coverage of time_range() before. The encoding test fails on
master with '+' present in the query and passes with this change.

Running with this in my EU deployment since May.
This commit is contained in:
xannasavin
2026-07-20 20:50:13 -04:00
committed by James Woglom
parent ddeaa79ded
commit 758204d4de
2 changed files with 84 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tconnectsync/nightscout.py
+16 -67
View File
@@ -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',
tests/test_nightscout.py
+68
View File
@@ -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()