mirror of
https://github.com/jwoglom/tconnectsync.git
synced 2026-08-24 18:24:11 -05:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7c4b2f4ddb | ||
|
|
74576bbb51 | ||
|
|
272a329664 | ||
|
|
758204d4de | ||
|
|
ddeaa79ded | ||
|
|
c95fe424ab | ||
|
|
4bae860212 | ||
|
|
823bb785ba | ||
|
|
984487da44 | ||
|
|
4655e31561 | ||
|
|
7f88d88ea4 |
@@ -7,11 +7,11 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@master
|
||||
- name: Set up Python 3.9
|
||||
uses: actions/setup-python@v1
|
||||
- uses: actions/checkout@v4
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: 3.9
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Install pypa/build
|
||||
run: >-
|
||||
|
||||
@@ -27,7 +27,7 @@ jobs:
|
||||
run: |
|
||||
python -m venv .venv
|
||||
.venv/bin/python -m pip install --upgrade pip
|
||||
.venv/bin/python -m pip install -e . flake8 pytest coverage
|
||||
.venv/bin/python -m pip install -e . flake8 pytest coverage mypy
|
||||
# - name: Run pipenv check
|
||||
# run: |
|
||||
# # DDoS attacks in wheel and setuptools packages, not relevant
|
||||
@@ -53,6 +53,9 @@ jobs:
|
||||
.venv/bin/flake8 . --exclude=.venv --count --select=E9,F63,F7,F82 --ignore=F824 --show-source --statistics
|
||||
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
|
||||
.venv/bin/flake8 . --exclude=.venv --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
|
||||
- name: Type-check annotated modules with mypy
|
||||
run: |
|
||||
.venv/bin/mypy
|
||||
- name: Run tconnectsync --help
|
||||
run: |
|
||||
.venv/bin/tconnectsync --help
|
||||
|
||||
@@ -8,6 +8,7 @@ ptpython = "*"
|
||||
flake8 = "*"
|
||||
pytest = "*"
|
||||
coverage = "*"
|
||||
mypy = "*"
|
||||
|
||||
[packages]
|
||||
tconnectsync = {path = "."}
|
||||
@@ -17,3 +18,4 @@ tconnectsync = "python3 main.py"
|
||||
test = "python3 -m unittest discover -vv"
|
||||
build_events = "bash -c 'cd tconnectsync/eventparser && python3 build_events.py > events.py'"
|
||||
lint = "bash -c 'flake8 . --count --select=E9,F63,F7,F82 && flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 && echo PASS'"
|
||||
typecheck = "mypy"
|
||||
|
||||
@@ -381,6 +381,36 @@ An example `run.sh` if you built tconnectsync locally:
|
||||
docker run tconnectsync --auto-update
|
||||
```
|
||||
|
||||
#### Tuning Auto-Update
|
||||
|
||||
These optional environment variables control how `--auto-update` polls and how
|
||||
it behaves when things go wrong. The defaults are sensible; you generally only
|
||||
need these if you are seeing too many (or too few) restarts.
|
||||
|
||||
| Variable | Default | What it does |
|
||||
| --- | --- | --- |
|
||||
| `AUTOUPDATE_DEFAULT_SLEEP_SECONDS` | `300` | Poll interval when no better estimate is available. Also the ceiling for the retry backoff below. |
|
||||
| `AUTOUPDATE_MAX_SLEEP_SECONDS` | `1500` | Upper bound on the adaptive poll interval, regardless of how rarely new data appears. |
|
||||
| `AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS` | `60` | How long to wait when new data is overdue based on the pump's previous cadence. |
|
||||
| `AUTOUPDATE_USE_FIXED_SLEEP` | `false` | Set true to always sleep `AUTOUPDATE_DEFAULT_SLEEP_SECONDS` instead of adapting to the pump's observed upload cadence. |
|
||||
| `AUTOUPDATE_API_FAILURE_MINUTES` | `45` | Exit with a non-zero code after this many minutes of unbroken API/network failure, so your container platform restarts tconnectsync and can alert you. Set `0` to never exit. |
|
||||
| `AUTOUPDATE_NO_DATA_FAILURE_MINUTES` | `180` | Log an error if the pump has not reported new events for this long. Usually means the pump simply is not uploading. |
|
||||
| `AUTOUPDATE_FAILURE_MINUTES` | `75` | Log an error if events are appearing but no data has synced successfully for this long. |
|
||||
| `AUTOUPDATE_RESTART_ON_FAILURE` | `false` | Whether the two watchdogs above also exit non-zero. Independent of `AUTOUPDATE_API_FAILURE_MINUTES`. |
|
||||
| `AUTOUPDATE_MAX_LOOP_INVOCATIONS` | `-1` | Stop after this many poll cycles. `-1` means run forever; mainly useful for testing. |
|
||||
|
||||
**On failures and restarts.** Transient errors (DNS blips, timeouts, HTTP 404/502/503
|
||||
from Tandem) do not crash tconnectsync. It retries with a growing backoff — 30s,
|
||||
60s, 120s, 240s, then holding at `AUTOUPDATE_DEFAULT_SLEEP_SECONDS` — and resets
|
||||
as soon as a poll succeeds. Staying in-process matters: an exit discards the
|
||||
cached credentials, so a restart loop means a fresh login on every attempt,
|
||||
which risks tripping Tandem's rate limiting.
|
||||
|
||||
Only once the API has been failing continuously for `AUTOUPDATE_API_FAILURE_MINUTES`
|
||||
does tconnectsync give up and exit, so that a genuine outage surfaces (roughly one
|
||||
restart per hour) instead of disappearing into an endless quiet retry. Invalid
|
||||
credentials are never retried — they exit immediately, since retrying cannot help.
|
||||
|
||||
### Running with Cron
|
||||
|
||||
If you choose not to run tconnectsync with `--auto-update` continuously,
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[metadata]
|
||||
name = tconnectsync
|
||||
version = 3.0.0
|
||||
version = 3.0.1
|
||||
author = James Woglom
|
||||
author_email = j@wogloms.net
|
||||
description = Syncs Tandem Source (formerly t:connect) insulin pump data to Nightscout for the t:slim X2 and Tandem Mobi
|
||||
@@ -47,3 +47,12 @@ exclude =
|
||||
[options.entry_points]
|
||||
console_scripts =
|
||||
tconnectsync = tconnectsync:main
|
||||
|
||||
[mypy]
|
||||
files =
|
||||
tconnectsync
|
||||
follow_imports = silent
|
||||
ignore_missing_imports = True
|
||||
# Third-party deps such as requests ship no type stubs; treat them as untyped
|
||||
# instead of failing (older mypy does not silence this via ignore_missing_imports).
|
||||
disable_error_code = import-untyped
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import logging
|
||||
|
||||
from .tandemsource import TandemSourceApi
|
||||
from .. import secret
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -9,10 +10,13 @@ class TConnectApi:
|
||||
email = None
|
||||
password = None
|
||||
|
||||
def __init__(self, email, password, region='US'):
|
||||
def __init__(self, email, password, region=None):
|
||||
self.email = email
|
||||
self.password = password
|
||||
self.region = region
|
||||
# A caller which does not pass a region (e.g. tconnectsync-heroku)
|
||||
# must get the configured TCONNECT_REGION, not a hardcoded US
|
||||
# default which would send EU accounts to the US endpoints (#152).
|
||||
self.region = region or secret.TCONNECT_REGION
|
||||
self._tandemsource = None
|
||||
|
||||
@property
|
||||
|
||||
@@ -112,7 +112,7 @@ def days_between(start, end) -> int:
|
||||
return diff.days
|
||||
|
||||
# both inclusive
|
||||
def split_days_range(start_a, end_a, days: int = 5) -> List[Tuple[str, str]]:
|
||||
def split_days_range(start_a, end_a, days: int = 5) -> List[Tuple[arrow.Arrow, arrow.Arrow]]:
|
||||
ranges = []
|
||||
start = arrow.get(start_a)
|
||||
end = arrow.get(end_a)
|
||||
|
||||
@@ -20,10 +20,12 @@ from requests_oidc.plugins import OSCachedPlugin
|
||||
from requests_oidc.utils import ServerDetails
|
||||
from requests_oauthlib import OAuth2Session
|
||||
from jwt.algorithms import RSAAlgorithm
|
||||
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
|
||||
|
||||
|
||||
from ..util import timeago, cap_length
|
||||
from .common import parse_ymd_date, base_headers, base_session, ApiException, ApiLoginException
|
||||
from .. import secret
|
||||
from ..secret import CACHE_CREDENTIALS, CACHE_CREDENTIALS_PATH, TIMEZONE_NAME
|
||||
from ..eventparser.generic import Events
|
||||
|
||||
@@ -214,7 +216,13 @@ class TandemSourceApi:
|
||||
'AUTHORIZATION_ENDPOINT': 'https://tdcservices.eu.tandemdiabetes.com/accounts/api/connect/authorize'
|
||||
}
|
||||
|
||||
def __init__(self, email: str, password: str, region: str = 'US') -> None:
|
||||
def __init__(self, email: str, password: str, region: Optional[str] = None) -> None:
|
||||
# No region means "use the configured TCONNECT_REGION": a hardcoded
|
||||
# US default would send EU accounts to the US endpoints (#152).
|
||||
if not region:
|
||||
region = secret.TCONNECT_REGION
|
||||
if not region:
|
||||
raise ValueError("No region configured. Set TCONNECT_REGION to 'US' or 'EU'.")
|
||||
self.region = region.upper()
|
||||
if self.region not in ['US', 'EU']:
|
||||
raise ValueError(f"Invalid region '{region}'. Must be 'US' or 'EU'.")
|
||||
@@ -386,6 +394,10 @@ class TandemSourceApi:
|
||||
key = public_keys.get(kid)
|
||||
if not key:
|
||||
raise ApiException(0, 'Public key not found for JWT: %s' % kid)
|
||||
# A JWKS endpoint publishes public keys; from_jwk() is typed as possibly
|
||||
# returning a private key, so narrow it before passing to jwt.decode().
|
||||
if not isinstance(key, RSAPublicKey):
|
||||
raise ApiException(0, 'JWK is not an RSA public key for JWT: %s' % kid)
|
||||
|
||||
audience = self.TDC_OIDC_CLIENT_ID
|
||||
issuer = self.TDC_OIDC_ISSUER
|
||||
@@ -570,7 +582,7 @@ class TandemSourceApi:
|
||||
# Trigger automatic re-login, and try again once
|
||||
if e.status_code == 401:
|
||||
logger.info("Performing automatic re-login after HTTP 401 for TandemSourceApi")
|
||||
self.accessTokenExpiresAt = time.time()
|
||||
self.accessTokenExpiresAt = arrow.get()
|
||||
self.login(self._email, self._password)
|
||||
|
||||
return self.get(endpoint, query, tries=tries+1)
|
||||
|
||||
@@ -2,7 +2,7 @@ from enum import Enum
|
||||
|
||||
from ...eventparser import events
|
||||
|
||||
class EventClass(set, Enum):
|
||||
class EventClass(set, Enum): # type: ignore[misc] # set/Enum both define __hash__; the combination works at runtime
|
||||
# LidBasalDelivery = every 5min entry
|
||||
# LidBasalRateChange = only when basal rate changes
|
||||
BASAL = {events.LidBasalDelivery} # , LidBasalRateChange
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from dataclasses import dataclass
|
||||
from dataclasses_json import dataclass_json
|
||||
from dataclasses_json import dataclass_json, DataClassJsonMixin
|
||||
from typing import List
|
||||
|
||||
# These dataclasses model the `settings.details` blob from the Tandem Source
|
||||
@@ -52,6 +52,6 @@ class PumpCgmSettings:
|
||||
|
||||
@dataclass_json
|
||||
@dataclass
|
||||
class PumpSettings:
|
||||
class PumpSettings(DataClassJsonMixin):
|
||||
profiles: PumpProfiles
|
||||
cgmSettings: PumpCgmSettings
|
||||
|
||||
+20
-67
@@ -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,138 +72,89 @@ 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)
|
||||
return None
|
||||
else:
|
||||
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)
|
||||
return None
|
||||
else:
|
||||
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)
|
||||
return None
|
||||
else:
|
||||
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)
|
||||
return None
|
||||
else:
|
||||
raise e
|
||||
|
||||
"""
|
||||
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 +166,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',
|
||||
|
||||
@@ -215,32 +215,40 @@ class NightscoutEntry:
|
||||
return {
|
||||
# insulin duration in hours; Nightscout JS bug requires all top-level fields to be strings
|
||||
"dia": "%s" % (profile.insulinDuration / 60),
|
||||
"carbratio": list(sorted([
|
||||
# Sort by the typed segment.startTime (monotonic with timeAsSeconds)
|
||||
# so the sort key is a well-typed int rather than an untyped dict value.
|
||||
"carbratio": [
|
||||
{
|
||||
"time": minutes_to_ns_time(segment.startTime),
|
||||
"timeAsSeconds": segment.startTime * 60,
|
||||
"value": segment.carbRatio / 1000 # milliunits->units
|
||||
} for segment in profile.tDependentSegs if not segment.skip
|
||||
], key=lambda x: x["timeAsSeconds"])),
|
||||
} for segment in sorted(
|
||||
(s for s in profile.tDependentSegs if not s.skip),
|
||||
key=lambda s: s.startTime)
|
||||
],
|
||||
|
||||
"carbs_hr": NIGHTSCOUT_PROFILE_CARBS_HR_VALUE,
|
||||
"delay": NIGHTSCOUT_PROFILE_DELAY_VALUE,
|
||||
|
||||
"sens": list(sorted([ # Correction factor / isf
|
||||
"sens": [ # Correction factor / isf
|
||||
{
|
||||
"time": minutes_to_ns_time(segment.startTime),
|
||||
"timeAsSeconds": segment.startTime * 60,
|
||||
"value": segment.isf
|
||||
} for segment in profile.tDependentSegs if not segment.skip
|
||||
], key=lambda x: x["timeAsSeconds"])),
|
||||
} for segment in sorted(
|
||||
(s for s in profile.tDependentSegs if not s.skip),
|
||||
key=lambda s: s.startTime)
|
||||
],
|
||||
|
||||
"basal": list(sorted([
|
||||
"basal": [
|
||||
{
|
||||
"time": minutes_to_ns_time(segment.startTime),
|
||||
"timeAsSeconds": segment.startTime * 60,
|
||||
"value": segment.basalRate / 1000 # milliunits->units
|
||||
} for segment in profile.tDependentSegs
|
||||
], key=lambda x: x["timeAsSeconds"])),
|
||||
} for segment in sorted(
|
||||
profile.tDependentSegs,
|
||||
key=lambda s: s.startTime)
|
||||
],
|
||||
|
||||
"target_low": [
|
||||
{
|
||||
|
||||
@@ -73,6 +73,11 @@ AUTOUPDATE_USE_FIXED_SLEEP = get_bool('AUTOUPDATE_USE_FIXED_SLEEP', 'false')
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES = get_number('AUTOUPDATE_NO_DATA_FAILURE_MINUTES', '180') # 3 hours
|
||||
AUTOUPDATE_FAILURE_MINUTES = get_number('AUTOUPDATE_FAILURE_MINUTES', '75') # 75 minutes
|
||||
AUTOUPDATE_RESTART_ON_FAILURE = get_bool('AUTOUPDATE_RESTART_ON_FAILURE', 'false')
|
||||
# Give up and exit non-zero after this many minutes of unbroken API/network
|
||||
# failure, so the container platform notices (and, if configured, notifies).
|
||||
# Distinct from AUTOUPDATE_RESTART_ON_FAILURE, which covers the pump not
|
||||
# uploading -- a case where restarting achieves nothing. Set 0 to never exit.
|
||||
AUTOUPDATE_API_FAILURE_MINUTES = get_number('AUTOUPDATE_API_FAILURE_MINUTES', '45') # 45 minutes
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS = get_number('AUTOUPDATE_MAX_LOOP_INVOCATIONS', '-1')
|
||||
|
||||
NIGHTSCOUT_PROFILE_UPLOAD_MODE = get_one_of('NIGHTSCOUT_PROFILE_UPLOAD_MODE', 'add', ['add', 'replace'])
|
||||
|
||||
@@ -3,7 +3,9 @@ import logging
|
||||
import datetime
|
||||
import sys
|
||||
import arrow
|
||||
import requests
|
||||
|
||||
from ...api.common import ApiException, ApiLoginException
|
||||
from ...features import DEFAULT_FEATURES
|
||||
from ...api.tandemsource import naive_local_to_utc
|
||||
from .process import ProcessTimeRange
|
||||
@@ -11,11 +13,21 @@ from .choose_device import ChooseDevice
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Shortest wait after a failed poll. Doubles per consecutive failure, capped at
|
||||
# AUTOUPDATE_DEFAULT_SLEEP_SECONDS (5 min by default): 30, 60, 120, 240, 300...
|
||||
RETRY_INITIAL_SLEEP_SECONDS = 30
|
||||
|
||||
# Consecutive failures before the retry log line escalates from WARNING to
|
||||
# ERROR, so a sustained outage doesn't hide quietly inside the backoff.
|
||||
RETRY_ESCALATE_AFTER_FAILURES = 3
|
||||
|
||||
class TandemSourceAutoupdate:
|
||||
"""Wrap access to secrets for easier testing."""
|
||||
def __init__(self, secret):
|
||||
self.secret = secret
|
||||
self.autoupdate_invocations = 0
|
||||
self.consecutive_failures = 0
|
||||
self.first_failure_time = None
|
||||
self.last_max_date_with_events = None
|
||||
self.last_event_time = 0
|
||||
self.last_attempt_time = 0
|
||||
@@ -39,127 +51,227 @@ class TandemSourceAutoupdate:
|
||||
self.autoupdate_start = time.time()
|
||||
|
||||
while True:
|
||||
logger.debug("autoupdate loop")
|
||||
now = time.time()
|
||||
try:
|
||||
logger.debug("autoupdate loop")
|
||||
now = time.time()
|
||||
|
||||
time_end = datetime.datetime.now()
|
||||
time_start = time_end - datetime.timedelta(days=1)
|
||||
time_end = datetime.datetime.now()
|
||||
time_start = time_end - datetime.timedelta(days=1)
|
||||
|
||||
tconnectDevice = ChooseDevice(self.secret, tconnect).choose()
|
||||
tconnectDevice = ChooseDevice(self.secret, tconnect).choose()
|
||||
|
||||
event_seqnum = None
|
||||
cur_max_date_with_events = arrow.get(naive_local_to_utc(tconnectDevice['maxDateOfEvents'])).float_timestamp
|
||||
if not self.last_max_date_with_events or cur_max_date_with_events > self.last_max_date_with_events:
|
||||
logger.info('New reported tandemsource data. (cur_max_date: %s last_max_date: %s)' % (cur_max_date_with_events, self.last_max_date_with_events))
|
||||
event_seqnum = None
|
||||
cur_max_date_with_events = arrow.get(naive_local_to_utc(tconnectDevice['maxDateOfEvents'])).float_timestamp
|
||||
if not self.last_max_date_with_events or cur_max_date_with_events > self.last_max_date_with_events:
|
||||
logger.info('New reported tandemsource data. (cur_max_date: %s last_max_date: %s)' % (cur_max_date_with_events, self.last_max_date_with_events))
|
||||
|
||||
if pretend:
|
||||
logger.info('Would update now if not in pretend mode')
|
||||
if pretend:
|
||||
logger.info('Would update now if not in pretend mode')
|
||||
else:
|
||||
added, event_seqnum = ProcessTimeRange(tconnect, nightscout, tconnectDevice, pretend, self.secret, features=features).process(time_start, time_end)
|
||||
logger.info('Added %d items from ProcessTimeRange' % added)
|
||||
self.last_successful_process_time_range = now
|
||||
|
||||
# Track the time it took to find a new event between runs,
|
||||
# but skip this calculation the first process cycle (since
|
||||
# we don't know at what exact point the event index changed)
|
||||
if self.last_event_seqnum:
|
||||
# A negative diff means the pump's previously-reported maxDateWithEvents
|
||||
# was in the future of wall-clock `now` — almost always a timezone /
|
||||
# clock-skew issue (e.g. pump timestamps tagged as UTC but actually
|
||||
# local time). Recording it would poison the rolling average and
|
||||
# eventually produce a negative sleep_secs that crashes time.sleep().
|
||||
diff = now - self.last_max_date_with_events
|
||||
if diff >= 0:
|
||||
self.time_diffs_between_updates.append(diff)
|
||||
logger.debug('Updating tracking of time since last update: %s' % self.time_diffs_between_updates)
|
||||
else:
|
||||
logger.warning(
|
||||
'Skipping negative time diff (%0.1fs) — likely pump clock skew or timezone mismatch' % diff
|
||||
)
|
||||
|
||||
# Mark the last event index uploaded from the pump and timestamp
|
||||
if event_seqnum:
|
||||
self.last_event_seqnum = event_seqnum
|
||||
self.last_event_time = now
|
||||
self.last_max_date_with_events = cur_max_date_with_events
|
||||
self.last_attempt_time = now
|
||||
self.time_diffs_between_attempts = []
|
||||
else:
|
||||
added, event_seqnum = ProcessTimeRange(tconnect, nightscout, tconnectDevice, pretend, self.secret, features=features).process(time_start, time_end)
|
||||
logger.info('Added %d items from ProcessTimeRange' % added)
|
||||
self.last_successful_process_time_range = now
|
||||
logger.info('No new reported tandemsource data. cur_max_date: %s (%s) last_event_time: %s (%s)' % (
|
||||
arrow.get(cur_max_date_with_events) if cur_max_date_with_events else None,
|
||||
'%dm ago' % ((now - cur_max_date_with_events)//60) if cur_max_date_with_events else None,
|
||||
arrow.get(self.last_event_time) if self.last_event_time else None,
|
||||
'%dm ago' % ((now - self.last_event_time)//60) if self.last_event_time else None
|
||||
))
|
||||
|
||||
# Track the time it took to find a new event between runs,
|
||||
# but skip this calculation the first process cycle (since
|
||||
# we don't know at what exact point the event index changed)
|
||||
if self.last_event_seqnum:
|
||||
self.time_diffs_between_updates.append(now - self.last_max_date_with_events)
|
||||
logger.debug('Updating tracking of time since last update: %s' % self.time_diffs_between_updates)
|
||||
# If we haven't seen the pump event index update in AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
|
||||
# then trigger an error and potentially restart.
|
||||
# The most likely case here is that the pump isn't uploading right now.
|
||||
if self.last_event_time and (now - self.last_event_time) >= 60 * self.secret.AUTOUPDATE_NO_DATA_FAILURE_MINUTES:
|
||||
logger.error(AutoupdateNoEventIndexesDetectedError(
|
||||
"%s: No new data event indexes have been detected for %d minutes. " % (datetime.datetime.now(), (now - self.last_event_time)//60) +
|
||||
"New data might not be uploading."))
|
||||
|
||||
# Mark the last event index uploaded from the pump and timestamp
|
||||
if event_seqnum:
|
||||
self.last_event_seqnum = event_seqnum
|
||||
self.last_event_time = now
|
||||
self.last_max_date_with_events = cur_max_date_with_events
|
||||
self.last_attempt_time = now
|
||||
self.time_diffs_between_attempts = []
|
||||
else:
|
||||
logger.info('No new reported tandemsource data. cur_max_date: %s (%s) last_event_time: %s (%s)' % (
|
||||
arrow.get(cur_max_date_with_events) if cur_max_date_with_events else None,
|
||||
'%dm ago' % ((now - cur_max_date_with_events)//60) if cur_max_date_with_events else None,
|
||||
arrow.get(self.last_event_time) if self.last_event_time else None,
|
||||
'%dm ago' % ((now - self.last_event_time)//60) if self.last_event_time else None
|
||||
))
|
||||
# TODO: restarting doesn't really help anything here.
|
||||
# Should we notify the user?
|
||||
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
|
||||
logger.error("Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE")
|
||||
return 1
|
||||
|
||||
# If we haven't seen the pump event index update in AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
|
||||
# then trigger an error and potentially restart.
|
||||
# The most likely case here is that the pump isn't uploading right now.
|
||||
if self.last_event_time and (now - self.last_event_time) >= 60 * self.secret.AUTOUPDATE_NO_DATA_FAILURE_MINUTES:
|
||||
logger.error(AutoupdateNoEventIndexesDetectedError(
|
||||
"%s: No new data event indexes have been detected for %d minutes. " % (datetime.datetime.now(), (now - self.last_event_time)//60) +
|
||||
"New data might not be uploading."))
|
||||
# Similarly, if we HAVE seen pump event indexes update but have not successfully
|
||||
# found any associated data updates from the tconnect API for AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
|
||||
# trigger an error and potentially restart. This could either be a tconnectsync problem,
|
||||
# where we can see the indexes increasing, but it takes us until a period of no index
|
||||
# update to reach our AUTOUPDATE_FAILURE_MINUTES threshold; or, a side effect of the
|
||||
# above no indexes warning.
|
||||
elif self.last_successful_process_time_range and (now - self.last_successful_process_time_range) >= 60 * self.secret.AUTOUPDATE_FAILURE_MINUTES:
|
||||
logger.error(AutoupdateNoNewDataDetectedError(
|
||||
"%s: No new data has been detected via the API for %d minutes (last: %s). " % (datetime.datetime.now(), (now - self.last_successful_process_time_range)//60, self.last_successful_process_time_range) +
|
||||
"tconnectsync might not be functioning properly."))
|
||||
|
||||
# TODO: restarting doesn't really help anything here.
|
||||
# Should we notify the user?
|
||||
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
|
||||
logger.error("Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE")
|
||||
return 1
|
||||
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
|
||||
logger.error("%s: Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE" % datetime.datetime.now())
|
||||
return 1
|
||||
|
||||
# Similarly, if we HAVE seen pump event indexes update but have not successfully
|
||||
# found any associated data updates from the tconnect API for AUTOUPDATE_NO_DATA_FAILURE_MINUTES,
|
||||
# trigger an error and potentially restart. This could either be a tconnectsync problem,
|
||||
# where we can see the indexes increasing, but it takes us until a period of no index
|
||||
# update to reach our AUTOUPDATE_FAILURE_MINUTES threshold; or, a side effect of the
|
||||
# above no indexes warning.
|
||||
elif self.last_successful_process_time_range and (now - self.last_successful_process_time_range) >= 60 * self.secret.AUTOUPDATE_FAILURE_MINUTES:
|
||||
logger.error(AutoupdateNoNewDataDetectedError(
|
||||
"%s: No new data has been detected via the API for %d minutes (last: %s). " % (datetime.datetime.now(), (now - self.last_successful_process_time_range)//60, self.last_successful_process_time_range) +
|
||||
"tconnectsync might not be functioning properly."))
|
||||
# Track how long we've been retrying
|
||||
if self.last_attempt_time:
|
||||
self.time_diffs_between_attempts.append(now - self.last_attempt_time)
|
||||
|
||||
if self.secret.AUTOUPDATE_RESTART_ON_FAILURE:
|
||||
logger.error("%s: Exiting with error code due to AUTOUPDATE_RESTART_ON_FAILURE" % datetime.datetime.now())
|
||||
return 1
|
||||
self.last_attempt_time = now
|
||||
|
||||
# Track how long we've been retrying
|
||||
if self.last_attempt_time:
|
||||
self.time_diffs_between_attempts.append(now - self.last_attempt_time)
|
||||
# If it's been 3 loops since the last time we found new data,
|
||||
# then we're not in sync with the rate at which pump data is being
|
||||
# uploaded, so
|
||||
if len(self.time_diffs_between_attempts) >= 3:
|
||||
# The pump hasn't sent us data that, based on previous cadence, we were expecting
|
||||
logger.warning(AutoupdateNoIndexChangeWarning("Sleeping %d seconds after unexpected no index change based on previous cadence. (New data might be delayed.)" %
|
||||
int(self.secret.AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS)))
|
||||
|
||||
self.last_attempt_time = now
|
||||
logger.debug("Last event time: %s, time diffs between attempts: %s" % (self.last_event_time, self.time_diffs_between_attempts))
|
||||
|
||||
# If it's been 3 loops since the last time we found new data,
|
||||
# then we're not in sync with the rate at which pump data is being
|
||||
# uploaded, so
|
||||
if len(self.time_diffs_between_attempts) >= 3:
|
||||
# The pump hasn't sent us data that, based on previous cadence, we were expecting
|
||||
logger.warning(AutoupdateNoIndexChangeWarning("Sleeping %d seconds after unexpected no index change based on previous cadence. (New data might be delayed.)" %
|
||||
int(self.secret.AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS)))
|
||||
# The API answered, so any prior outage is over.
|
||||
self.consecutive_failures = 0
|
||||
self.first_failure_time = None
|
||||
|
||||
logger.debug("Last event time: %s, time diffs between attempts: %s" % (self.last_event_time, self.time_diffs_between_attempts))
|
||||
time.sleep(self.secret.AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS)
|
||||
|
||||
time.sleep(self.secret.AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS)
|
||||
# Since we bail early, update the invocations count and potentially exit after sleeping.
|
||||
self.autoupdate_invocations += 1
|
||||
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
|
||||
return 0
|
||||
|
||||
# Since we bail early, update the invocations count and potentially exit after sleeping.
|
||||
self.autoupdate_invocations += 1
|
||||
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
|
||||
return 0
|
||||
continue
|
||||
|
||||
continue
|
||||
# The API answered, so any prior outage is over.
|
||||
self.consecutive_failures = 0
|
||||
self.first_failure_time = None
|
||||
|
||||
sleep_secs = self.secret.AUTOUPDATE_DEFAULT_SLEEP_SECONDS
|
||||
sleep_secs = self.secret.AUTOUPDATE_DEFAULT_SLEEP_SECONDS
|
||||
|
||||
# Sleep for a rolling average of time between updates
|
||||
if self.secret.AUTOUPDATE_USE_FIXED_SLEEP != 1:
|
||||
logger.debug("Time diffs between updates: %s" % self.time_diffs_between_updates)
|
||||
# Sleep for a rolling average of time between updates
|
||||
if self.secret.AUTOUPDATE_USE_FIXED_SLEEP != 1:
|
||||
logger.debug("Time diffs between updates: %s" % self.time_diffs_between_updates)
|
||||
|
||||
# Only keep the 10 latest time diffs
|
||||
if len(self.time_diffs_between_updates) > 10:
|
||||
self.time_diffs_between_updates = self.time_diffs_between_updates[1:]
|
||||
# Only keep the 10 latest time diffs
|
||||
if len(self.time_diffs_between_updates) > 10:
|
||||
self.time_diffs_between_updates = self.time_diffs_between_updates[1:]
|
||||
|
||||
# If we have less than 3 data points,
|
||||
if len(self.time_diffs_between_updates) > 2:
|
||||
sleep_secs = sum(self.time_diffs_between_updates) / len(self.time_diffs_between_updates)
|
||||
# If we have less than 3 data points,
|
||||
if len(self.time_diffs_between_updates) > 2:
|
||||
sleep_secs = sum(self.time_diffs_between_updates) / len(self.time_diffs_between_updates)
|
||||
|
||||
# At minimum, update every AUTOUPDATE_MAX_SLEEP_SECONDS regardless
|
||||
# of how often we're seeing new data appear
|
||||
if sleep_secs > self.secret.AUTOUPDATE_MAX_SLEEP_SECONDS:
|
||||
sleep_secs = self.secret.AUTOUPDATE_MAX_SLEEP_SECONDS
|
||||
# At minimum, update every AUTOUPDATE_MAX_SLEEP_SECONDS regardless
|
||||
# of how often we're seeing new data appear
|
||||
if sleep_secs > self.secret.AUTOUPDATE_MAX_SLEEP_SECONDS:
|
||||
sleep_secs = self.secret.AUTOUPDATE_MAX_SLEEP_SECONDS
|
||||
|
||||
logger.info('Sleeping for %0.01f sec' % sleep_secs)
|
||||
time.sleep(sleep_secs)
|
||||
# Defensive: with the negative-diff filter above, sleep_secs should never be
|
||||
# negative, but legacy state from before the fix or other unexpected inputs
|
||||
# could still produce one. Clamp to AUTOUPDATE_DEFAULT_SLEEP_SECONDS so we
|
||||
# don't crash with ValueError nor tight-loop the API.
|
||||
if sleep_secs < 0:
|
||||
logger.warning(
|
||||
'Computed negative sleep duration (%0.1fs), falling back to default %ds' % (
|
||||
sleep_secs, self.secret.AUTOUPDATE_DEFAULT_SLEEP_SECONDS
|
||||
)
|
||||
)
|
||||
sleep_secs = self.secret.AUTOUPDATE_DEFAULT_SLEEP_SECONDS
|
||||
|
||||
self.autoupdate_invocations += 1
|
||||
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
|
||||
return 0
|
||||
logger.info('Sleeping for %0.01f sec' % sleep_secs)
|
||||
time.sleep(sleep_secs)
|
||||
|
||||
self.autoupdate_invocations += 1
|
||||
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
|
||||
return 0
|
||||
|
||||
except ApiLoginException:
|
||||
# A credentials failure is not transient: retrying it in-process
|
||||
# would hammer the login endpoint with attempts that cannot
|
||||
# succeed, which is the exact ban risk the backoff below exists
|
||||
# to prevent. Stay fatal so the user notices and fixes config.
|
||||
raise
|
||||
|
||||
except (
|
||||
ApiException,
|
||||
requests.exceptions.ConnectionError,
|
||||
requests.exceptions.Timeout,
|
||||
requests.exceptions.ChunkedEncodingError,
|
||||
requests.exceptions.RetryError,
|
||||
) as e:
|
||||
# Two failure families, one response. Transient network errors
|
||||
# (DNS, refused connections, timeouts, mid-stream disconnects,
|
||||
# urllib3 retry-budget exhaustion) and API errors that get()
|
||||
# does not retry itself (it only handles 401 and 500 — a 404,
|
||||
# 502 or 503 propagates) both used to exit the process and let
|
||||
# Docker restart the container.
|
||||
#
|
||||
# Restarting is the worst possible response: the credentials
|
||||
# cache dies with the process, so every restart performs a full
|
||||
# login. During the 2026-07-16 EU outage that meant a fresh
|
||||
# login every ~2 minutes for hours from a single IP. Staying in
|
||||
# the loop keeps the cache warm and the login endpoint untouched.
|
||||
self.consecutive_failures += 1
|
||||
if self.first_failure_time is None:
|
||||
self.first_failure_time = time.time()
|
||||
sleep_secs = self._retry_sleep_seconds()
|
||||
|
||||
log = logger.error if self.consecutive_failures >= RETRY_ESCALATE_AFTER_FAILURES else logger.warning
|
||||
log(
|
||||
'Error during autoupdate poll (%d consecutive): %s. Sleeping %ds before retry.' % (
|
||||
self.consecutive_failures, e, sleep_secs
|
||||
)
|
||||
)
|
||||
|
||||
time.sleep(sleep_secs)
|
||||
|
||||
# Staying alive forever would make a real outage silent on
|
||||
# deployments whose only alarm is the container dying. Once the
|
||||
# API has been unreachable for AUTOUPDATE_API_FAILURE_MINUTES,
|
||||
# exit so the platform can restart us and raise its own alert.
|
||||
failing_for = time.time() - self.first_failure_time
|
||||
if self.secret.AUTOUPDATE_API_FAILURE_MINUTES > 0 and failing_for >= 60 * self.secret.AUTOUPDATE_API_FAILURE_MINUTES:
|
||||
logger.error(
|
||||
AutoupdateFailureError(
|
||||
'%s: API has been failing for %d minutes (%d consecutive attempts). '
|
||||
'Exiting so the container platform restarts and reports it.' % (
|
||||
datetime.datetime.now(), failing_for // 60, self.consecutive_failures
|
||||
)
|
||||
)
|
||||
)
|
||||
return 1
|
||||
|
||||
self.autoupdate_invocations += 1
|
||||
if self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS > 0 and self.autoupdate_invocations >= self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS:
|
||||
return 0
|
||||
|
||||
def _retry_sleep_seconds(self):
|
||||
"""Exponential backoff for consecutive failed polls: 30, 60, 120, 240,
|
||||
then held at AUTOUPDATE_DEFAULT_SLEEP_SECONDS (300s default). The cap
|
||||
reuses the existing poll interval because a failing API should never be
|
||||
contacted more often than a healthy one."""
|
||||
backoff = RETRY_INITIAL_SLEEP_SECONDS * (2 ** (self.consecutive_failures - 1))
|
||||
return min(backoff, self.secret.AUTOUPDATE_DEFAULT_SLEEP_SECONDS)
|
||||
|
||||
|
||||
class AutoupdateError(RuntimeError):
|
||||
|
||||
@@ -10,7 +10,7 @@ import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def fetch_oneshot(username, password, time_start=None, time_end=None, region='US'):
|
||||
def fetch_oneshot(username, password, time_start=None, time_end=None, region=None):
|
||||
tconnect = TConnectApi(username, password, region)
|
||||
if not time_start and not time_end:
|
||||
time_end = datetime.datetime.now()
|
||||
|
||||
@@ -3,12 +3,19 @@ import collections
|
||||
import arrow
|
||||
|
||||
from types import ModuleType
|
||||
from typing import List, Optional, Tuple, TYPE_CHECKING
|
||||
from typing import Dict, Iterable, List, Optional, Protocol, Tuple, Type, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...api.tandemsource import BffPump
|
||||
|
||||
class EventProcessor(Protocol):
|
||||
"""Structural interface implemented by every Process* event handler."""
|
||||
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str]) -> None: ...
|
||||
def enabled(self) -> bool: ...
|
||||
def process(self, events: Iterable, time_start: arrow.Arrow, time_end: arrow.Arrow) -> List[dict]: ...
|
||||
def write(self, ns_entries: List[dict]) -> int: ...
|
||||
|
||||
from ...features import DEVICE_STATUS, DEFAULT_FEATURES
|
||||
from ...eventparser import events as eventtypes
|
||||
from ...domain.tandemsource.event_class import EventClass
|
||||
@@ -37,7 +44,7 @@ class ProcessTimeRange:
|
||||
self.secret = secret
|
||||
self.features = features
|
||||
|
||||
event_classes = {
|
||||
event_classes: Dict[str, Type[EventProcessor]] = {
|
||||
EventClass.BASAL.name: ProcessBasal,
|
||||
EventClass.BASAL_SUSPENSION.name: ProcessBasalSuspension,
|
||||
EventClass.BASAL_RESUME.name: ProcessBasalResume,
|
||||
@@ -93,7 +100,10 @@ class ProcessTimeRange:
|
||||
# Ensure time_end is timezone-aware for comparison
|
||||
time_end_aware = arrow.get(time_end)
|
||||
capped_time_end = min(events_last_time, time_end_aware) if events_last_time else time_end_aware
|
||||
ns_entries = c.process(events, events_first_time, capped_time_end)
|
||||
# events_first_time is populated whenever for_eventclass has entries
|
||||
# (i.e. at least one event was seen); fall back to time_start otherwise.
|
||||
time_start_for_events = events_first_time if events_first_time else time_start
|
||||
ns_entries = c.process(events, time_start_for_events, capped_time_end)
|
||||
w = c.write(ns_entries)
|
||||
if w:
|
||||
processed_count += w
|
||||
@@ -101,10 +111,10 @@ class ProcessTimeRange:
|
||||
logger.info("Skipping %s, is not enabled from features %s" % (clazz, self.features))
|
||||
|
||||
for updater_class in self.updater_classes:
|
||||
c = updater_class(self.tconnect, self.nightscout, self.tconnect_device_id, self.pretend, self.features)
|
||||
if c.enabled():
|
||||
updater = updater_class(self.tconnect, self.nightscout, self.tconnect_device_id, self.pretend, self.features)
|
||||
if updater.enabled():
|
||||
logger.info("%s is enabled from features %s" % (updater_class.__name__, self.features))
|
||||
done = c.update(self.pretend)
|
||||
done = updater.update(self.pretend)
|
||||
logger.info("%s completed with update required: %s" % (updater_class.__name__, done))
|
||||
else:
|
||||
logger.info("Skipping %s, is not enabled from features %s" % (updater_class.__name__, self.features))
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
import logging
|
||||
import arrow
|
||||
|
||||
from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
from typing import Iterable, List, Union, TYPE_CHECKING
|
||||
from typing_extensions import assert_never
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
from ...features import DEFAULT_FEATURES
|
||||
from ... import features
|
||||
@@ -20,6 +20,8 @@ from ...parser.nightscout import (
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
AlarmOrMalfunction = Union[eventtypes.LidAlarmActivated, eventtypes.LidMalfunctionActivated]
|
||||
|
||||
class ProcessAlarm:
|
||||
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
|
||||
self.tconnect = tconnect
|
||||
@@ -54,7 +56,10 @@ class ProcessAlarm:
|
||||
|
||||
return ns_entries
|
||||
|
||||
def skip_event(self, event: "BaseEvent") -> bool:
|
||||
def skip_event(self, event: AlarmOrMalfunction) -> bool:
|
||||
if not isinstance(event, eventtypes.LidAlarmActivated):
|
||||
return False
|
||||
|
||||
return event.alarmId in (
|
||||
eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm,
|
||||
eventtypes.LidAlarmActivated.AlarmidEnum.ResumePumpAlarm2
|
||||
@@ -73,16 +78,19 @@ class ProcessAlarm:
|
||||
return count
|
||||
|
||||
|
||||
def alarm_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
|
||||
if type(event) == eventtypes.LidAlarmActivated:
|
||||
def alarm_to_nsentry(self, event: AlarmOrMalfunction) -> dict:
|
||||
if isinstance(event, eventtypes.LidAlarmActivated):
|
||||
alarmId = event.alarmId
|
||||
reason = alarmId.name if alarmId is not None else "Alarm%s" % event.alarmIdRaw
|
||||
return NightscoutEntry.alarm(
|
||||
created_at = event.eventTimestamp.format(),
|
||||
reason = "%s" % event.alarmId.name,
|
||||
reason = reason,
|
||||
pump_event_id = "%s" % event.seqNum
|
||||
)
|
||||
elif type(event) == eventtypes.LidMalfunctionActivated:
|
||||
elif isinstance(event, eventtypes.LidMalfunctionActivated):
|
||||
return NightscoutEntry.alarm(
|
||||
created_at = event.eventTimestamp.format(),
|
||||
reason = "Malfunction",
|
||||
pump_event_id = "%s" % event.seqNum
|
||||
)
|
||||
assert_never(event)
|
||||
|
||||
@@ -15,14 +15,15 @@ from ...parser.nightscout import (
|
||||
NightscoutEntry
|
||||
)
|
||||
|
||||
from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
from typing import Iterable, List, Optional, Union, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
BasalEvent = Union[eventtypes.LidBasalRateChange, eventtypes.LidBasalDelivery]
|
||||
|
||||
class ProcessBasal:
|
||||
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
|
||||
self.tconnect = tconnect
|
||||
@@ -81,7 +82,7 @@ class ProcessBasal:
|
||||
return count
|
||||
|
||||
|
||||
def basal_to_nsentry(self, start: arrow.Arrow, duration: datetime.timedelta, event: "BaseEvent") -> Optional[dict]:
|
||||
def basal_to_nsentry(self, start: arrow.Arrow, duration: datetime.timedelta, event: BasalEvent) -> Optional[dict]:
|
||||
if type(event) == eventtypes.LidBasalRateChange:
|
||||
value = insulin_float_round(event.commandedBasalRate)
|
||||
if IGNORE_ZERO_UNIT_BASAL and value < 0.01:
|
||||
@@ -106,3 +107,5 @@ class ProcessBasal:
|
||||
reason = ', '.join(bitmask_to_list(event.commandedRateSource)),
|
||||
pump_event_id = "%s" % event.seqNum
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
@@ -5,7 +5,6 @@ from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
from ...features import DEFAULT_FEATURES
|
||||
from ... import features
|
||||
@@ -46,7 +45,9 @@ class ProcessBasalResume:
|
||||
logger.info("Skipping BasalResume event not after last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
|
||||
continue
|
||||
|
||||
ns_entries.append(self.resume_to_nsentry(event))
|
||||
ns = self.resume_to_nsentry(event)
|
||||
if ns:
|
||||
ns_entries.append(ns)
|
||||
|
||||
|
||||
return ns_entries
|
||||
@@ -64,9 +65,11 @@ class ProcessBasalResume:
|
||||
return count
|
||||
|
||||
|
||||
def resume_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
|
||||
def resume_to_nsentry(self, event: eventtypes.LidPumpingResumed) -> Optional[dict]:
|
||||
if type(event) == eventtypes.LidPumpingResumed:
|
||||
return NightscoutEntry.basalresume(
|
||||
created_at = event.eventTimestamp.format(),
|
||||
pump_event_id = "%s" % event.seqNum
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
@@ -5,7 +5,6 @@ from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
from ...features import DEFAULT_FEATURES
|
||||
from ... import features
|
||||
@@ -46,7 +45,9 @@ class ProcessBasalSuspension:
|
||||
logger.info("Skipping basalsuspension event not after last upload time: %s (time range: %s - %s)" % (event, time_start, time_end))
|
||||
continue
|
||||
|
||||
ns_entries.append(self.suspension_to_nsentry(event))
|
||||
ns = self.suspension_to_nsentry(event)
|
||||
if ns:
|
||||
ns_entries.append(ns)
|
||||
|
||||
|
||||
return ns_entries
|
||||
@@ -64,10 +65,12 @@ class ProcessBasalSuspension:
|
||||
return count
|
||||
|
||||
|
||||
def suspension_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
|
||||
def suspension_to_nsentry(self, event: eventtypes.LidPumpingSuspended) -> Optional[dict]:
|
||||
if type(event) == eventtypes.LidPumpingSuspended:
|
||||
return NightscoutEntry.basalsuspension(
|
||||
created_at = event.eventTimestamp.format(),
|
||||
reason = ', '.join(bitmask_to_list(event.suspendReason)),
|
||||
pump_event_id = "%s" % event.seqNum
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
@@ -17,7 +17,6 @@ from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -41,7 +40,7 @@ class ProcessBolus:
|
||||
logger.info("Last Nightscout bolus upload: %s" % last_upload_time)
|
||||
|
||||
# Correlate a bolus's request/completion messages by bolusid.
|
||||
bolusEventsForId = {}
|
||||
bolusEventsForId: dict = {}
|
||||
for event in sorted(events, key=lambda x: x.eventTimestamp):
|
||||
bolusEventsForId.setdefault(event.bolusId, {})[type(event)] = event
|
||||
|
||||
@@ -90,7 +89,7 @@ class ProcessBolus:
|
||||
return count
|
||||
|
||||
|
||||
def bolus_to_nsentry(self, bolusCompleted: "BaseEvent", bolusRequested1: "BaseEvent", bolusRequested2: "BaseEvent", bolusRequested3: "BaseEvent") -> Optional[dict]:
|
||||
def bolus_to_nsentry(self, bolusCompleted: eventtypes.LidBolusCompleted, bolusRequested1: Optional[eventtypes.LidBolusRequestedMsg1], bolusRequested2: Optional[eventtypes.LidBolusRequestedMsg2], bolusRequested3: Optional[eventtypes.LidBolusRequestedMsg3]) -> dict:
|
||||
suffixes = []
|
||||
if bolusRequested2 and bolusRequested2.userOverride == eventtypes.LidBolusRequestedMsg2.UseroverrideEnum.Yes:
|
||||
suffixes.append('(Override)')
|
||||
@@ -119,7 +118,7 @@ class ProcessBolus:
|
||||
pump_event_id = ",".join(seq_nums)
|
||||
)
|
||||
|
||||
def bolex_to_nsentry(self, bolexCompleted: "BaseEvent") -> Optional[dict]:
|
||||
def bolex_to_nsentry(self, bolexCompleted: eventtypes.LidBolexCompleted) -> dict:
|
||||
# The extended portion of a combo bolus, added as its own treatment at
|
||||
# the time it finished delivering. Insulin only; carbs/bg belong to the
|
||||
# initial LidBolusCompleted entry and must not be double-counted here.
|
||||
|
||||
@@ -12,11 +12,10 @@ from ...parser.nightscout import (
|
||||
NightscoutEntry
|
||||
)
|
||||
|
||||
from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
from typing import Iterable, List, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -84,7 +83,7 @@ class ProcessCartridge:
|
||||
|
||||
return count
|
||||
|
||||
def cart_to_nsentry(self, cartFilled: "BaseEvent") -> Optional[dict]:
|
||||
def cart_to_nsentry(self, cartFilled: eventtypes.LidCartridgeFilled) -> dict:
|
||||
# insulinVolume is populated on t:slim X2 / Mobi; v2Volume is a legacy fallback.
|
||||
volume = cartFilled.insulinVolume or cartFilled.v2Volume
|
||||
return NightscoutEntry.sitechange(
|
||||
@@ -93,7 +92,7 @@ class ProcessCartridge:
|
||||
pump_event_id = "%s" % cartFilled.seqNum
|
||||
)
|
||||
|
||||
def cannula_to_nsentry(self, cannulaFilled: "BaseEvent") -> Optional[dict]:
|
||||
def cannula_to_nsentry(self, cannulaFilled: eventtypes.LidCannulaFilled) -> dict:
|
||||
# primeSize is fractional (e.g. 0.3u); format with one decimal, not %d.
|
||||
primed = cannulaFilled.primeSize if cannulaFilled.primeSize and cannulaFilled.primeSize > 0 else None
|
||||
return NightscoutEntry.sitechange(
|
||||
@@ -102,7 +101,7 @@ class ProcessCartridge:
|
||||
pump_event_id = "%s" % cannulaFilled.seqNum
|
||||
)
|
||||
|
||||
def tubing_to_nsentry(self, tubingFilled: "BaseEvent") -> Optional[dict]:
|
||||
def tubing_to_nsentry(self, tubingFilled: eventtypes.LidTubingFilled) -> dict:
|
||||
# primeSize is -1 (sentinel, "not recorded") on real tubing fills; only show a real prime volume.
|
||||
primed = tubingFilled.primeSize if tubingFilled.primeSize and tubingFilled.primeSize > 0 else None
|
||||
return NightscoutEntry.sitechange(
|
||||
|
||||
@@ -12,14 +12,20 @@ from ...parser.nightscout import (
|
||||
NightscoutEntry
|
||||
)
|
||||
|
||||
from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
from typing import Iterable, List, Optional, Union, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# The three CGM alert event types all expose dalertId / dalertIdRaw / seqNum.
|
||||
CgmAlertEvent = Union[
|
||||
eventtypes.LidCgmAlertActivated,
|
||||
eventtypes.LidCgmAlertActivatedDex,
|
||||
eventtypes.LidCgmAlertActivatedFsl2,
|
||||
]
|
||||
|
||||
class ProcessCGMAlert:
|
||||
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
|
||||
self.tconnect = tconnect
|
||||
@@ -70,7 +76,7 @@ class ProcessCGMAlert:
|
||||
|
||||
return count
|
||||
|
||||
def alert_to_nsentry(self, alert: "BaseEvent") -> Optional[dict]:
|
||||
def alert_to_nsentry(self, alert: CgmAlertEvent) -> Optional[dict]:
|
||||
# FSL3 alert codes are defined in eventparser/static_dicts.py:CGM_ALERTS_DICT
|
||||
# Alert code meanings are documented in comments there.
|
||||
if not alert.dalertId:
|
||||
@@ -98,3 +104,5 @@ class ProcessCGMAlert:
|
||||
reason = ("Libre CGM Alert (%s)" % alert.dalertId.name) if alert.dalertId else "Libre CGM Alert (Unknown)",
|
||||
pump_event_id = "%s" % alert.seqNum
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
@@ -12,7 +12,6 @@ from typing import Iterable, List, Optional, Union, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
# The four CGM-reading event types share the glucoseValueStatus /
|
||||
# currentGlucoseDisplayValue fields determine_glucose_value() reads.
|
||||
@@ -52,22 +51,22 @@ def determine_glucose_value(event: CgmReadingEvent) -> int:
|
||||
status = event.glucoseValueStatus
|
||||
|
||||
if isinstance(event, eventtypes.LidCgmDataG7):
|
||||
e = eventtypes.LidCgmDataG7.GlucosevaluestatusEnum
|
||||
g7 = eventtypes.LidCgmDataG7.GlucosevaluestatusEnum
|
||||
return _resolve_glucose_value(display_value, status,
|
||||
precise=e.PreciseValue, high=e.SpecialHigh, low=e.SpecialLow)
|
||||
precise=g7.PreciseValue, high=g7.SpecialHigh, low=g7.SpecialLow)
|
||||
if isinstance(event, eventtypes.LidCgmDataGxb):
|
||||
e = eventtypes.LidCgmDataGxb.GlucosevaluestatusEnum
|
||||
gxb = eventtypes.LidCgmDataGxb.GlucosevaluestatusEnum
|
||||
return _resolve_glucose_value(display_value, status,
|
||||
precise=e.CurrentglucosedisplayvalueContainsTheGlucoseReading,
|
||||
high=e.TheGlucoseReadingIsHigh, low=e.TheGlucoseReadingIsLow)
|
||||
precise=gxb.CurrentglucosedisplayvalueContainsTheGlucoseReading,
|
||||
high=gxb.TheGlucoseReadingIsHigh, low=gxb.TheGlucoseReadingIsLow)
|
||||
if isinstance(event, eventtypes.LidCgmDataFsl3):
|
||||
e = eventtypes.LidCgmDataFsl3.GlucosevaluestatusEnum
|
||||
fsl3 = eventtypes.LidCgmDataFsl3.GlucosevaluestatusEnum
|
||||
return _resolve_glucose_value(display_value, status,
|
||||
precise=e.PreciseValue, high=e.SpecialHigh, low=e.SpecialLow)
|
||||
precise=fsl3.PreciseValue, high=fsl3.SpecialHigh, low=fsl3.SpecialLow)
|
||||
if isinstance(event, eventtypes.LidCgmDataFsl2):
|
||||
e = eventtypes.LidCgmDataFsl2.GlucosevaluestatusEnum
|
||||
fsl2 = eventtypes.LidCgmDataFsl2.GlucosevaluestatusEnum
|
||||
return _resolve_glucose_value(display_value, status,
|
||||
precise=e.PreciseValue, high=e.SpecialHigh, low=e.SpecialLow)
|
||||
precise=fsl2.PreciseValue, high=fsl2.SpecialHigh, low=fsl2.SpecialLow)
|
||||
|
||||
return display_value
|
||||
|
||||
@@ -120,12 +119,12 @@ class ProcessCGMReading:
|
||||
|
||||
return count
|
||||
|
||||
def timestamp_for(self, event: "BaseEvent") -> arrow.Arrow:
|
||||
def timestamp_for(self, event: CgmReadingEvent) -> arrow.Arrow:
|
||||
# For backfills the time the event was added to the pump's event store
|
||||
# might not be the time it actually occurred, so we use the egvTimestamp
|
||||
return arrow.get(TANDEM_EPOCH + event.egvTimeStamp, tzinfo='UTC').replace(tzinfo=self.timezone)
|
||||
|
||||
def to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
|
||||
def to_nsentry(self, event: CgmReadingEvent) -> dict:
|
||||
return NightscoutEntry.entry(
|
||||
sgv = determine_glucose_value(event),
|
||||
created_at = self.timestamp_for(event).format(),
|
||||
|
||||
@@ -3,6 +3,7 @@ import arrow
|
||||
|
||||
from ...features import DEFAULT_FEATURES
|
||||
from ... import features
|
||||
from ...eventparser import events as eventtypes
|
||||
from ...domain.tandemsource.event_class import EventClass
|
||||
from ...nightscout import format_datetime
|
||||
from ...parser.nightscout import (
|
||||
@@ -12,14 +13,28 @@ from ...parser.nightscout import (
|
||||
NightscoutEntry
|
||||
)
|
||||
|
||||
from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
from typing import Iterable, List, Optional, Union, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# The CGM session start/join/stop event types (see EventClass._CGM_START /
|
||||
# _CGM_JOIN / _CGM_STOP); all expose seqNum and eventTimestamp.
|
||||
CgmSessionEvent = Union[
|
||||
eventtypes.LidCgmStartSessionGx,
|
||||
eventtypes.LidCgmStartSessionFsl2,
|
||||
eventtypes.LidCgmJoinSessionGx,
|
||||
eventtypes.LidCgmJoinSessionG7,
|
||||
eventtypes.LidCgmJoinSessionFsl2,
|
||||
eventtypes.LidCgmJoinSessionFsl3,
|
||||
eventtypes.LidCgmStopSessionGx,
|
||||
eventtypes.LidCgmStopSessionG7,
|
||||
eventtypes.LidCgmStopSessionFsl2,
|
||||
eventtypes.LidCgmStopSessionFsl3,
|
||||
]
|
||||
|
||||
class ProcessCGMStartJoinStop:
|
||||
def __init__(self, tconnect: "TConnectApi", nightscout: "NightscoutApi", tconnect_device_id: str, pretend: bool, features: List[str] = DEFAULT_FEATURES) -> None:
|
||||
self.tconnect = tconnect
|
||||
@@ -63,7 +78,9 @@ class ProcessCGMStartJoinStop:
|
||||
|
||||
ns_entries = []
|
||||
for event in allEvents:
|
||||
ns_entries.append(self.to_nsentry(event))
|
||||
ns = self.to_nsentry(event)
|
||||
if ns:
|
||||
ns_entries.append(ns)
|
||||
|
||||
return ns_entries
|
||||
|
||||
@@ -80,7 +97,7 @@ class ProcessCGMStartJoinStop:
|
||||
return count
|
||||
|
||||
|
||||
def to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
|
||||
def to_nsentry(self, event: CgmSessionEvent) -> Optional[dict]:
|
||||
if type(event) in EventClass._CGM_START:
|
||||
return NightscoutEntry.cgm_start(
|
||||
created_at = format_datetime(event.eventTimestamp),
|
||||
@@ -99,3 +116,5 @@ class ProcessCGMStartJoinStop:
|
||||
reason = "CGM Session Stopped",
|
||||
pump_event_id = "%s" % event.seqNum
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
@@ -17,7 +17,6 @@ from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -63,7 +62,7 @@ class ProcessDeviceStatus:
|
||||
return []
|
||||
return [entry]
|
||||
|
||||
def daily_basal_to_nsentry(self, event: "BaseEvent") -> Optional[dict]:
|
||||
def daily_basal_to_nsentry(self, event: eventtypes.LidDailyBasal) -> Optional[dict]:
|
||||
# NOTE: the pump-logs endpoint does not emit event 81 (LID_DAILY_BASAL)
|
||||
# for either t:slim X2 or Mobi (verified against live accounts), and no
|
||||
# other returned event carries battery data. DEVICE_STATUS therefore
|
||||
|
||||
@@ -5,7 +5,6 @@ from typing import Iterable, List, Optional, TYPE_CHECKING
|
||||
if TYPE_CHECKING:
|
||||
from ...api import TConnectApi
|
||||
from ...nightscout import NightscoutApi
|
||||
from ...eventparser.raw_event import BaseEvent
|
||||
|
||||
from ...features import DEFAULT_FEATURES
|
||||
from ... import features
|
||||
@@ -90,7 +89,7 @@ class ProcessUserMode:
|
||||
processed_sleep.append((start_sleep, event))
|
||||
start_sleep = None
|
||||
else:
|
||||
if sleep_not_ended:
|
||||
if sleep_not_ended and sleep_last_upload:
|
||||
logger.info("ProcessUserMode: Found StopSleep without StartSleep, with incomplete sleep event in nightscout: %s NS: %s" % (event, sleep_last_upload))
|
||||
ns_entries.append(self.process_unended_sleep_stop(event, sleep_last_upload))
|
||||
else:
|
||||
@@ -102,7 +101,7 @@ class ProcessUserMode:
|
||||
processed_exercise.append((start_exercise, event))
|
||||
start_exercise = None
|
||||
else:
|
||||
if exercise_not_ended:
|
||||
if exercise_not_ended and exercise_last_upload:
|
||||
logger.info("ProcessUserMode: Found StopExercise without StartExercise, with incomplete exercise event in nightscout: %s NS: %s" % (event, exercise_last_upload))
|
||||
ns_entries.append(self.process_unended_exercise_stop(event, exercise_last_upload))
|
||||
else:
|
||||
@@ -118,10 +117,14 @@ class ProcessUserMode:
|
||||
logger.info("ProcessUserMode: exercise is active")
|
||||
|
||||
for items in processed_sleep:
|
||||
ns_entries.append(self.sleep_to_nsentry(start=items[0], stop=items[1], time_end=time_end))
|
||||
ns = self.sleep_to_nsentry(start=items[0], stop=items[1], time_end=time_end)
|
||||
if ns:
|
||||
ns_entries.append(ns)
|
||||
|
||||
for items in processed_exercise:
|
||||
ns_entries.append(self.exercise_to_nsentry(start=items[0], stop=items[1], time_end=time_end))
|
||||
ns = self.exercise_to_nsentry(start=items[0], stop=items[1], time_end=time_end)
|
||||
if ns:
|
||||
ns_entries.append(ns)
|
||||
|
||||
return ns_entries
|
||||
|
||||
@@ -137,19 +140,19 @@ class ProcessUserMode:
|
||||
|
||||
return count
|
||||
|
||||
def is_start_sleep(self, event: "BaseEvent") -> bool:
|
||||
def is_start_sleep(self, event: eventtypes.LidAaUserModeChange) -> bool:
|
||||
return event.requestedAction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StartSleep
|
||||
def is_stop_sleep(self, event: "BaseEvent") -> bool:
|
||||
def is_stop_sleep(self, event: eventtypes.LidAaUserModeChange) -> bool:
|
||||
return event.requestedAction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopSleep or \
|
||||
event.requestedAction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopAll
|
||||
def is_start_exercise(self, event: "BaseEvent") -> bool:
|
||||
def is_start_exercise(self, event: eventtypes.LidAaUserModeChange) -> bool:
|
||||
return event.requestedAction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StartExercise
|
||||
def is_stop_exercise(self, event: "BaseEvent") -> bool:
|
||||
def is_stop_exercise(self, event: eventtypes.LidAaUserModeChange) -> bool:
|
||||
return event.requestedAction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopExercise or \
|
||||
event.requestedAction == eventtypes.LidAaUserModeChange.RequestedactionEnum.StopAll
|
||||
|
||||
|
||||
def sleep_to_nsentry(self, start: "BaseEvent", stop: Optional["BaseEvent"] = None, time_end: Optional[arrow.Arrow] = None) -> Optional[dict]:
|
||||
def sleep_to_nsentry(self, start: eventtypes.LidAaUserModeChange, stop: Optional[eventtypes.LidAaUserModeChange] = None, time_end: Optional[arrow.Arrow] = None) -> Optional[dict]:
|
||||
if start and stop:
|
||||
reason = None
|
||||
if start.sleepStartedByGui == eventtypes.LidAaUserModeChange.SleepstartedbyguiEnum.TrueVal:
|
||||
@@ -181,8 +184,10 @@ class ProcessUserMode:
|
||||
pump_event_id = "%s" % start.seqNum
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
def exercise_to_nsentry(self, start: "BaseEvent", stop: Optional["BaseEvent"] = None, time_end: Optional[arrow.Arrow] = None) -> Optional[dict]:
|
||||
|
||||
def exercise_to_nsentry(self, start: eventtypes.LidAaUserModeChange, stop: Optional[eventtypes.LidAaUserModeChange] = None, time_end: Optional[arrow.Arrow] = None) -> Optional[dict]:
|
||||
if start and stop:
|
||||
reason = "Exercise"
|
||||
if start.exerciseChoice == eventtypes.LidAaUserModeChange.ExercisechoiceEnum.Timed:
|
||||
@@ -213,7 +218,9 @@ class ProcessUserMode:
|
||||
pump_event_id = "%s" % start.seqNum
|
||||
)
|
||||
|
||||
def process_unended_sleep_stop(self, event: "BaseEvent", sleep_last_upload: dict) -> dict:
|
||||
return None
|
||||
|
||||
def process_unended_sleep_stop(self, event: eventtypes.LidAaUserModeChange, sleep_last_upload: dict) -> dict:
|
||||
logger.info("ProcessUserMode: Deleting old sleep event treatment before pushing update (delete treatments/%s)" % sleep_last_upload["_id"])
|
||||
if self.pretend:
|
||||
logger.info("ProcessUserMode: Skipping delete in pretend mode")
|
||||
@@ -229,7 +236,7 @@ class ProcessUserMode:
|
||||
pump_event_id="%s,%s" % (sleep_last_upload.get("pump_event_id",""), event.seqNum)
|
||||
)
|
||||
|
||||
def process_unended_exercise_stop(self, event: "BaseEvent", exercise_last_upload: dict) -> dict:
|
||||
def process_unended_exercise_stop(self, event: eventtypes.LidAaUserModeChange, exercise_last_upload: dict) -> dict:
|
||||
logger.info("ProcessUserMode: Deleting old exercise event treatment before pushing update (delete treatments/%s)" % exercise_last_upload["_id"])
|
||||
if self.pretend:
|
||||
logger.info("ProcessUserMode: Skipping delete in pretend mode")
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
#!/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 TestLidMalfunctionActivated(unittest.TestCase):
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.fixture = {
|
||||
"deviceAssignmentId": "4ff6bebc-d4d6-4423-b123-eecfcf5a4238",
|
||||
"eventCode": 6,
|
||||
"sequenceGroup": 0,
|
||||
"sequenceNumber": 500123,
|
||||
"pumpDateTime": "2026-05-16T00:07:00",
|
||||
"eventProperties": {"malfId": 7, "faultLocatorData": 8311, "param1": 42, "param2": 0},
|
||||
"estimatedDateTime": "2026-05-16T00:07:00Z",
|
||||
}
|
||||
|
||||
def test_dispatches_to_correct_class(self):
|
||||
self.assertIsInstance(Event(self.fixture), eventtypes.LidMalfunctionActivated)
|
||||
self.assertNotIsInstance(Event(self.fixture), RawEvent)
|
||||
|
||||
def test_has_no_alarmid_attribute(self):
|
||||
ev = Event(self.fixture)
|
||||
self.assertFalse(hasattr(ev, 'alarmId'))
|
||||
self.assertEqual(ev.malfIdRaw, 7)
|
||||
|
||||
def test_envelope_fields(self):
|
||||
ev = Event(self.fixture)
|
||||
self.assertEqual(ev.eventId, 6)
|
||||
self.assertEqual(ev.seqNum, 500123)
|
||||
|
||||
def test_timestamp_preserves_wall_clock(self):
|
||||
ev = Event(self.fixture)
|
||||
self.assertEqual(ev.eventTimestamp.format('YYYY-MM-DDTHH:mm:ss'), "2026-05-16T00:07:00")
|
||||
|
||||
def test_plain_fields(self):
|
||||
ev = Event(self.fixture)
|
||||
self.assertEqual(ev.faultLocatorData, 8311)
|
||||
self.assertEqual(ev.param1, 42)
|
||||
self.assertEqual(ev.param2, 0)
|
||||
|
||||
def test_todict_is_json_serializable(self):
|
||||
ev = Event(self.fixture)
|
||||
d = ev.todict()
|
||||
json.dumps(d) # must not raise
|
||||
self.assertEqual(d["id"], 6)
|
||||
self.assertEqual(d["name"], "LID_MALFUNCTION_ACTIVATED")
|
||||
self.assertEqual(d["malfIdRaw"], 7)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,745 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Regression tests for negative-sleep crash in TandemSourceAutoupdate.
|
||||
|
||||
When the pump's reported maxDateWithEvents is interpreted as being in the
|
||||
future (e.g. timezone mismatch where arrow tags a local-time string as UTC),
|
||||
`now - last_max_date_with_events` produced a negative value that landed in
|
||||
the rolling-average list, which in turn fed `time.sleep()` and crashed the
|
||||
process with `ValueError: sleep length must be non-negative`.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from unittest.mock import patch
|
||||
|
||||
import arrow
|
||||
import requests
|
||||
|
||||
from tconnectsync.api.common import ApiException, ApiLoginException
|
||||
from tconnectsync.sync.tandemsource.autoupdate import TandemSourceAutoupdate
|
||||
|
||||
from ...secrets import build_secrets
|
||||
|
||||
|
||||
class _FakeTConnect:
|
||||
pass
|
||||
|
||||
|
||||
class _FakeNightscout:
|
||||
pass
|
||||
|
||||
|
||||
class TestAutoupdateNegativeSleep(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.secret = build_secrets(
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=300,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=1500,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=60,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=0,
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=180,
|
||||
AUTOUPDATE_FAILURE_MINUTES=75,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
)
|
||||
|
||||
def _run_one_iteration(self, autoupdate, future_offset_seconds=None, max_date_iso=None):
|
||||
"""Drive one autoupdate loop iteration. Either pass `future_offset_seconds`
|
||||
(produces a UTC-tagged ISO string `future_offset_seconds` ahead of now) or
|
||||
pass `max_date_iso` directly (used by tests that need a specific format,
|
||||
e.g. naive local-time strings to exercise the TIMEZONE_NAME parsing fix)."""
|
||||
if max_date_iso is None:
|
||||
assert future_offset_seconds is not None
|
||||
max_date_iso = arrow.utcnow().shift(seconds=future_offset_seconds).isoformat()
|
||||
future_iso = max_date_iso
|
||||
|
||||
sleep_calls = []
|
||||
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=lambda s: sleep_calls.append(s),
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
) as mock_process:
|
||||
mock_choose.return_value.choose.return_value = {
|
||||
"assignmentId": "test-device-1",
|
||||
"maxDateOfEvents": future_iso,
|
||||
}
|
||||
mock_process.return_value.process.return_value = (1, 999)
|
||||
|
||||
autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
return sleep_calls
|
||||
|
||||
def test_time_sleep_never_called_with_negative_value(self):
|
||||
"""Defensive clamp: even with negative rolling-avg entries, time.sleep
|
||||
must receive a non-negative argument."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
# Simulate state after prior iterations where pump timestamps were
|
||||
# consistently ~2h in the future (TZ skew).
|
||||
autoupdate.time_diffs_between_updates = [-7200.0, -7200.0, -7200.0]
|
||||
autoupdate.last_max_date_with_events = (
|
||||
arrow.utcnow().float_timestamp + 7200
|
||||
)
|
||||
autoupdate.last_event_seqnum = 12345
|
||||
|
||||
sleep_calls = self._run_one_iteration(autoupdate, future_offset_seconds=7260)
|
||||
|
||||
self.assertTrue(sleep_calls, "Expected at least one time.sleep call")
|
||||
for call_arg in sleep_calls:
|
||||
self.assertGreaterEqual(
|
||||
call_arg, 0,
|
||||
"time.sleep was called with negative value %r" % call_arg,
|
||||
)
|
||||
|
||||
def test_negative_diff_not_recorded_in_rolling_average(self):
|
||||
"""Root cause: a negative `now - last_max_date_with_events` indicates
|
||||
clock skew and must not be appended to the rolling-average list."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
# Previous max-date is 2h in the future, so `now - past_future = negative`.
|
||||
autoupdate.last_max_date_with_events = (
|
||||
arrow.utcnow().float_timestamp + 7200
|
||||
)
|
||||
autoupdate.last_event_seqnum = 12345
|
||||
|
||||
self._run_one_iteration(autoupdate, future_offset_seconds=7260)
|
||||
|
||||
for diff in autoupdate.time_diffs_between_updates:
|
||||
self.assertGreaterEqual(
|
||||
diff, 0,
|
||||
"Negative diff %r leaked into time_diffs_between_updates" % diff,
|
||||
)
|
||||
|
||||
def test_positive_diff_is_still_recorded(self):
|
||||
"""Sanity check: the happy path (pump timestamp in the past) still
|
||||
feeds the rolling average."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
# Previous max-date is 5min in the PAST — normal case.
|
||||
autoupdate.last_max_date_with_events = (
|
||||
arrow.utcnow().float_timestamp - 300
|
||||
)
|
||||
autoupdate.last_event_seqnum = 12345
|
||||
|
||||
self._run_one_iteration(autoupdate, future_offset_seconds=60)
|
||||
|
||||
self.assertEqual(
|
||||
len(autoupdate.time_diffs_between_updates), 1,
|
||||
"Expected exactly one positive diff to be recorded",
|
||||
)
|
||||
self.assertGreater(autoupdate.time_diffs_between_updates[0], 0)
|
||||
|
||||
|
||||
class TestAutoupdateNaiveTimestampParsing(unittest.TestCase):
|
||||
"""Root cause regression: Tandem Source EU returns maxDateOfEvents as a
|
||||
naive ISO string in the pump's local timezone (no offset marker). Before
|
||||
the fix, arrow.get() defaulted naive strings to UTC, shifting the timestamp
|
||||
into the future of `now` by the local UTC offset and producing chronic
|
||||
negative time diffs (every cycle in production logs from 2026-05-19/20).
|
||||
|
||||
Parsing now routes through the API layer's naive_local_to_utc(), which
|
||||
applies tzinfo=TIMEZONE_NAME only when the string carries no offset marker.
|
||||
Strings with an embedded offset (Z, +HH, +HHMM, +HH:MM) are honored as-is.
|
||||
|
||||
Note that naive_local_to_utc() reads the module-level TIMEZONE_NAME rather
|
||||
than the secret object passed to TandemSourceAutoupdate, so these tests
|
||||
patch the constant where the function looks it up. Both resolve to the same
|
||||
env var in production."""
|
||||
|
||||
def test_naive_local_time_string_parsed_in_configured_tz(self):
|
||||
secret = build_secrets(
|
||||
TIMEZONE_NAME="Europe/Berlin",
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=300,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=1500,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=60,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=0,
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=180,
|
||||
AUTOUPDATE_FAILURE_MINUTES=75,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
)
|
||||
autoupdate = TandemSourceAutoupdate(secret)
|
||||
|
||||
# Simulate the production scenario: pump reports its local wall-clock
|
||||
# time as a naive ISO string with no offset marker.
|
||||
now_berlin = arrow.now("Europe/Berlin")
|
||||
naive_local_iso = now_berlin.format("YYYY-MM-DDTHH:mm:ss")
|
||||
self.assertNotIn("+", naive_local_iso, "fixture must be naive (no TZ)")
|
||||
self.assertNotIn("Z", naive_local_iso, "fixture must be naive (no TZ)")
|
||||
|
||||
sleep_calls = []
|
||||
with patch(
|
||||
"tconnectsync.api.tandemsource.TIMEZONE_NAME", "Europe/Berlin"
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=lambda s: sleep_calls.append(s),
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
) as mock_process:
|
||||
mock_choose.return_value.choose.return_value = {
|
||||
"assignmentId": "test-device-1",
|
||||
"maxDateOfEvents": naive_local_iso,
|
||||
}
|
||||
mock_process.return_value.process.return_value = (1, 999)
|
||||
|
||||
autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
# After the fix, the parsed epoch should match wall-clock now (give or
|
||||
# take a second for test execution), NOT now + UTC_offset.
|
||||
recorded_epoch = autoupdate.last_max_date_with_events
|
||||
wall_clock_epoch = arrow.utcnow().float_timestamp
|
||||
delta = abs(recorded_epoch - wall_clock_epoch)
|
||||
self.assertLess(
|
||||
delta, 10,
|
||||
"Naive local-time string was misinterpreted as UTC (delta=%0.1fs). "
|
||||
"Expected parser to honor TIMEZONE_NAME=Europe/Berlin." % delta,
|
||||
)
|
||||
|
||||
def test_embedded_tz_marker_still_honored(self):
|
||||
"""A maxDateWithEvents that DOES carry an offset (e.g. US fixtures,
|
||||
future format changes) must still parse correctly even with a
|
||||
mismatching TIMEZONE_NAME, because the helper short-circuits to
|
||||
plain arrow.get() when an offset is present."""
|
||||
secret = build_secrets(
|
||||
TIMEZONE_NAME="Europe/Berlin", # deliberately wrong for the fixture
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=300,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=1500,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=60,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=0,
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=1,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=180,
|
||||
AUTOUPDATE_FAILURE_MINUTES=75,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
)
|
||||
autoupdate = TandemSourceAutoupdate(secret)
|
||||
|
||||
# Pump in US Eastern reports with explicit -05:00 / -04:00 offset,
|
||||
# like the existing test_process.py fixture.
|
||||
now_eastern = arrow.now("America/New_York")
|
||||
tz_tagged_iso = now_eastern.isoformat()
|
||||
self.assertIn(
|
||||
":", tz_tagged_iso[-6:],
|
||||
"fixture must include an explicit TZ offset",
|
||||
)
|
||||
|
||||
sleep_calls = []
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=lambda s: sleep_calls.append(s),
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
) as mock_process:
|
||||
mock_choose.return_value.choose.return_value = {
|
||||
"assignmentId": "test-device-1",
|
||||
"maxDateOfEvents": tz_tagged_iso,
|
||||
}
|
||||
mock_process.return_value.process.return_value = (1, 999)
|
||||
|
||||
autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
recorded_epoch = autoupdate.last_max_date_with_events
|
||||
wall_clock_epoch = arrow.utcnow().float_timestamp
|
||||
delta = abs(recorded_epoch - wall_clock_epoch)
|
||||
self.assertLess(
|
||||
delta, 10,
|
||||
"Embedded TZ offset was overridden by TIMEZONE_NAME (delta=%0.1fs). "
|
||||
"Helper should short-circuit to arrow.get() when offset present." % delta,
|
||||
)
|
||||
|
||||
|
||||
class TestAutoupdateTransientNetworkError(unittest.TestCase):
|
||||
"""Regression: DNS failures and connection resets used to propagate up
|
||||
from ChooseDevice / ProcessTimeRange and exit the process, leading
|
||||
Docker/Synology to restart the container hourly and email the user.
|
||||
|
||||
The fix wraps the loop body in a try/except for requests' ConnectionError,
|
||||
Timeout, ChunkedEncodingError, and RetryError; logs a warning; sleeps;
|
||||
and continues. Sustained outages still trigger the NO_DATA_FAILURE_MINUTES
|
||||
safety net (covered by other paths).
|
||||
|
||||
Network errors share the incremental backoff of TestAutoupdateApiErrorBackoff
|
||||
(30s, doubling, capped at DEFAULT_SLEEP_SECONDS) rather than the flat
|
||||
DEFAULT_SLEEP_SECONDS they originally used: a 2-second DNS blip should not
|
||||
cost a 5-minute sync gap, while a real outage still settles at 5 minutes."""
|
||||
|
||||
def setUp(self):
|
||||
self.secret = build_secrets(
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=300,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=1500,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=60,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=0,
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=2,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=180,
|
||||
AUTOUPDATE_FAILURE_MINUTES=75,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
)
|
||||
|
||||
def _drive(self, autoupdate, choose_side_effect):
|
||||
"""Drive autoupdate.process() with patched ChooseDevice and ProcessTimeRange.
|
||||
Returns (sleep_calls, result)."""
|
||||
sleep_calls = []
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=lambda s: sleep_calls.append(s),
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
) as mock_process:
|
||||
mock_choose.return_value.choose.side_effect = choose_side_effect
|
||||
mock_process.return_value.process.return_value = (1, 999)
|
||||
result = autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
return sleep_calls, result, future_iso
|
||||
|
||||
def test_connection_error_does_not_crash_loop(self):
|
||||
"""A DNS failure on the first iteration must not exit the process;
|
||||
the loop should sleep and try again."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
sleep_calls, result, _ = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
requests.exceptions.ConnectionError(
|
||||
"HTTPSConnectionPool(host='source.eu.tandemdiabetes.com', port=443): "
|
||||
"Max retries exceeded with url: /api/... "
|
||||
"(Caused by NameResolutionError(...Temporary failure in name resolution))"
|
||||
),
|
||||
{"assignmentId": "test-device-1", "maxDateOfEvents": future_iso},
|
||||
],
|
||||
)
|
||||
|
||||
self.assertIn(result, (0, None))
|
||||
self.assertEqual(autoupdate.autoupdate_invocations, 2)
|
||||
self.assertGreaterEqual(len(sleep_calls), 2)
|
||||
self.assertEqual(
|
||||
sleep_calls[0], 30,
|
||||
"First retry after a network blip should be the short backoff, "
|
||||
"not a flat 5-minute wait",
|
||||
)
|
||||
|
||||
def test_timeout_does_not_crash_loop(self):
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
sleep_calls, _, _ = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
requests.exceptions.Timeout("Read timed out"),
|
||||
{"assignmentId": "x", "maxDateOfEvents": future_iso},
|
||||
],
|
||||
)
|
||||
self.assertEqual(autoupdate.autoupdate_invocations, 2)
|
||||
self.assertGreaterEqual(len(sleep_calls), 2)
|
||||
|
||||
def test_chunked_encoding_error_does_not_crash_loop(self):
|
||||
"""A mid-stream disconnect during pump_events download surfaces as
|
||||
ChunkedEncodingError (subclass of RequestException, NOT ConnectionError),
|
||||
so it must be in the catch tuple explicitly."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
_, _, _ = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
requests.exceptions.ChunkedEncodingError("Connection broken"),
|
||||
{"assignmentId": "x", "maxDateOfEvents": future_iso},
|
||||
],
|
||||
)
|
||||
self.assertEqual(autoupdate.autoupdate_invocations, 2)
|
||||
|
||||
def test_retry_error_does_not_crash_loop(self):
|
||||
"""urllib3 retry-budget exhaustion bubbles up as requests.RetryError,
|
||||
which is RequestException but not ConnectionError."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
_, _, _ = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
requests.exceptions.RetryError("Max retries exceeded"),
|
||||
{"assignmentId": "x", "maxDateOfEvents": future_iso},
|
||||
],
|
||||
)
|
||||
self.assertEqual(autoupdate.autoupdate_invocations, 2)
|
||||
|
||||
def test_non_network_exception_still_propagates(self):
|
||||
"""Programming bugs (e.g. KeyError) must NOT be swallowed by the
|
||||
network-error handler — they should still crash so they get noticed."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
):
|
||||
mock_choose.return_value.choose.side_effect = KeyError("simulated bug")
|
||||
with self.assertRaises(KeyError):
|
||||
autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
def test_max_loop_invocations_respected_on_persistent_failure(self):
|
||||
"""If the network never recovers, the loop must still terminate at
|
||||
MAX_LOOP_INVOCATIONS rather than spinning forever."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
sleep_calls = []
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=lambda s: sleep_calls.append(s),
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
):
|
||||
mock_choose.return_value.choose.side_effect = (
|
||||
requests.exceptions.ConnectionError("dns fail")
|
||||
)
|
||||
result = autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
self.assertIn(result, (0, None))
|
||||
self.assertEqual(
|
||||
autoupdate.autoupdate_invocations,
|
||||
self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS,
|
||||
)
|
||||
|
||||
|
||||
class TestAutoupdateApiErrorBackoff(unittest.TestCase):
|
||||
"""Regression: on 2026-07-16 Tandem retired the reportsfacade endpoints in
|
||||
the EU region, so pump_event_metadata() began returning HTTP 404. get()
|
||||
only retries 401 and 500, so the ApiException propagated out of the loop
|
||||
and exited the process. Docker restarted the container roughly every two
|
||||
minutes, and because the credentials cache is lost on restart, EVERY
|
||||
restart performed a fresh login against sso.tandemdiabetes.com — hundreds
|
||||
of logins per hour from one IP, which risks a WAF ban.
|
||||
|
||||
The fix keeps API errors inside the loop and backs off incrementally
|
||||
(30s, 60s, 120s, ... capped at AUTOUPDATE_DEFAULT_SLEEP_SECONDS) so the
|
||||
process stays alive, the credentials cache stays warm, and a sustained
|
||||
outage settles into one quiet poll every 5 minutes."""
|
||||
|
||||
def setUp(self):
|
||||
self.secret = build_secrets(
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=300,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=1500,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=60,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=0,
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=6,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=180,
|
||||
AUTOUPDATE_FAILURE_MINUTES=75,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
)
|
||||
|
||||
def _drive(self, autoupdate, choose_side_effect):
|
||||
sleep_calls = []
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=lambda s: sleep_calls.append(s),
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
) as mock_process:
|
||||
mock_choose.return_value.choose.side_effect = choose_side_effect
|
||||
mock_process.return_value.process.return_value = (1, 999)
|
||||
result = autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
return sleep_calls, result
|
||||
|
||||
def test_api_exception_does_not_crash_loop(self):
|
||||
"""The production symptom: HTTP 404 from pumpeventmetadata must be
|
||||
survivable, not fatal."""
|
||||
# One failure + one success, so stop the loop after two invocations
|
||||
# rather than running past the fixtures.
|
||||
self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS = 2
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
|
||||
sleep_calls, result = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
ApiException(404, "TandemSourceApi HTTP 404 response: "),
|
||||
{"assignmentId": "test-device-1", "maxDateOfEvents": future_iso},
|
||||
],
|
||||
)
|
||||
|
||||
self.assertIn(result, (0, None))
|
||||
self.assertGreaterEqual(len(sleep_calls), 2)
|
||||
|
||||
def test_backoff_grows_incrementally_and_caps_at_default_sleep(self):
|
||||
"""A persistent outage must not poll at a fixed fast rate. Waits grow
|
||||
30 -> 60 -> 120 -> 240 and then hold at AUTOUPDATE_DEFAULT_SLEEP_SECONDS
|
||||
(300s = 5 minutes), never above it."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
|
||||
sleep_calls, _ = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=ApiException(404, "TandemSourceApi HTTP 404 response: "),
|
||||
)
|
||||
|
||||
self.assertEqual(sleep_calls, [30, 60, 120, 240, 300, 300])
|
||||
|
||||
def test_backoff_resets_after_successful_iteration(self):
|
||||
"""A single blip must not permanently penalize the poll rate: once a
|
||||
poll succeeds, the next failure starts again at the shortest wait."""
|
||||
# Four fixtures below, so stop after four invocations.
|
||||
self.secret.AUTOUPDATE_MAX_LOOP_INVOCATIONS = 4
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
device = {"assignmentId": "test-device-1", "maxDateOfEvents": future_iso}
|
||||
|
||||
sleep_calls, _ = self._drive(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
ApiException(502, "TandemSourceApi HTTP 502 response: "),
|
||||
ApiException(502, "TandemSourceApi HTTP 502 response: "),
|
||||
device,
|
||||
ApiException(502, "TandemSourceApi HTTP 502 response: "),
|
||||
],
|
||||
)
|
||||
|
||||
# Expected: 30 and 60 for the two failures, then the normal poll
|
||||
# interval for the successful iteration, then back to 30 — not 120 —
|
||||
# because the success reset the counter.
|
||||
self.assertEqual(
|
||||
sleep_calls[:2], [30, 60],
|
||||
"Expected the first outage to back off 30 then 60, got %r" % sleep_calls,
|
||||
)
|
||||
self.assertEqual(
|
||||
sleep_calls[-1], 30,
|
||||
"Backoff must reset to 30s after the successful poll in between, "
|
||||
"got %r (full sequence: %r)" % (sleep_calls[-1], sleep_calls),
|
||||
)
|
||||
|
||||
def test_login_exception_still_propagates(self):
|
||||
"""Guard: a credentials failure is NOT transient. Retrying it in-process
|
||||
would hammer the login endpoint with doomed attempts, which is exactly
|
||||
the ban risk this backoff exists to avoid. It must stay fatal so the
|
||||
user notices and fixes their config."""
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
):
|
||||
mock_choose.return_value.choose.side_effect = ApiLoginException(
|
||||
401, "Invalid credentials"
|
||||
)
|
||||
with self.assertRaises(ApiLoginException):
|
||||
autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
|
||||
class TestAutoupdateSustainedFailureExit(unittest.TestCase):
|
||||
"""Staying alive through an outage costs the only alarm this deployment
|
||||
has: Synology's Container Manager mails on container exit, and nothing
|
||||
watches the log stream. With the backoff swallowing API errors forever, a
|
||||
real outage (like the 2026-07-16 EU cutover) would now be silent.
|
||||
|
||||
So a sustained failure escalates one final step: after
|
||||
AUTOUPDATE_API_FAILURE_MINUTES of unbroken failure, exit non-zero. Docker
|
||||
restarts, Synology sends exactly one mail per outage-hour instead of one
|
||||
per two minutes. Short blips stay silent, which is the whole point.
|
||||
|
||||
This is deliberately NOT gated on AUTOUPDATE_RESTART_ON_FAILURE: that flag
|
||||
covers the pump-not-uploading watchdog, where restarting fixes nothing.
|
||||
A dead API is a different failure and deserves its own knob."""
|
||||
|
||||
def _secret(self, **overrides):
|
||||
base = dict(
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=300,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=1500,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=60,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=0,
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=50,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=180,
|
||||
AUTOUPDATE_FAILURE_MINUTES=75,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
AUTOUPDATE_API_FAILURE_MINUTES=45,
|
||||
)
|
||||
base.update(overrides)
|
||||
return build_secrets(**base)
|
||||
|
||||
def _drive_with_clock(self, autoupdate, choose_side_effect):
|
||||
"""Drive the loop with a fake clock that advances by each sleep, so
|
||||
simulated wall-clock time passes without the test actually waiting."""
|
||||
clock = [10_000.0]
|
||||
sleeps = []
|
||||
|
||||
def fake_sleep(secs):
|
||||
sleeps.append(secs)
|
||||
clock[0] += secs
|
||||
|
||||
with patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.sleep",
|
||||
side_effect=fake_sleep,
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.time.time",
|
||||
side_effect=lambda: clock[0],
|
||||
), patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ChooseDevice"
|
||||
) as mock_choose, patch(
|
||||
"tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange"
|
||||
) as mock_process:
|
||||
mock_choose.return_value.choose.side_effect = choose_side_effect
|
||||
mock_process.return_value.process.return_value = (1, 999)
|
||||
result = autoupdate.process(_FakeTConnect(), _FakeNightscout(), pretend=False)
|
||||
|
||||
return result, sleeps, clock[0] - 10_000.0
|
||||
|
||||
def test_exits_nonzero_after_sustained_api_failure(self):
|
||||
"""The production scenario: a dead endpoint. After 45 simulated minutes
|
||||
of unbroken 404s the process must exit non-zero so the platform mails."""
|
||||
autoupdate = TandemSourceAutoupdate(self._secret())
|
||||
|
||||
result, sleeps, elapsed = self._drive_with_clock(
|
||||
autoupdate,
|
||||
choose_side_effect=ApiException(404, "TandemSourceApi HTTP 404 response: "),
|
||||
)
|
||||
|
||||
self.assertEqual(result, 1, "Expected a non-zero exit after a sustained outage")
|
||||
self.assertGreaterEqual(
|
||||
elapsed, 45 * 60,
|
||||
"Exited after only %0.0fs; must persist a full AUTOUPDATE_API_FAILURE_MINUTES "
|
||||
"before giving up" % elapsed,
|
||||
)
|
||||
self.assertLess(
|
||||
elapsed, 75 * 60,
|
||||
"Took %0.0fs to give up; backoff should reach the threshold promptly "
|
||||
"once capped" % elapsed,
|
||||
)
|
||||
|
||||
def test_recovery_before_threshold_does_not_exit(self):
|
||||
"""A 10-minute outage that recovers must not trigger a mail."""
|
||||
autoupdate = TandemSourceAutoupdate(self._secret(AUTOUPDATE_MAX_LOOP_INVOCATIONS=6))
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
device = {"assignmentId": "x", "maxDateOfEvents": future_iso}
|
||||
|
||||
result, _, _ = self._drive_with_clock(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
ApiException(503, "down"),
|
||||
ApiException(503, "down"),
|
||||
ApiException(503, "down"),
|
||||
device,
|
||||
device,
|
||||
device,
|
||||
],
|
||||
)
|
||||
|
||||
self.assertIn(result, (0, None), "A recovered outage must not exit non-zero")
|
||||
|
||||
def test_failure_clock_resets_on_success(self):
|
||||
"""Two separate short outages must not add up to an exit: the failure
|
||||
clock restarts from the successful poll between them."""
|
||||
autoupdate = TandemSourceAutoupdate(self._secret(AUTOUPDATE_MAX_LOOP_INVOCATIONS=12))
|
||||
future_iso = arrow.utcnow().shift(seconds=60).isoformat()
|
||||
device = {"assignmentId": "x", "maxDateOfEvents": future_iso}
|
||||
|
||||
result, _, _ = self._drive_with_clock(
|
||||
autoupdate,
|
||||
choose_side_effect=[
|
||||
ApiException(503, "down"), ApiException(503, "down"),
|
||||
ApiException(503, "down"), ApiException(503, "down"),
|
||||
ApiException(503, "down"),
|
||||
device,
|
||||
ApiException(503, "down"), ApiException(503, "down"),
|
||||
ApiException(503, "down"), ApiException(503, "down"),
|
||||
ApiException(503, "down"), device,
|
||||
],
|
||||
)
|
||||
|
||||
self.assertIn(
|
||||
result, (0, None),
|
||||
"Two short outages separated by a success must not accumulate into an exit",
|
||||
)
|
||||
|
||||
def test_zero_minutes_disables_the_exit(self):
|
||||
"""Opt-out: 0 means never give up, for users who would rather have a
|
||||
silent process than a restarting one."""
|
||||
autoupdate = TandemSourceAutoupdate(
|
||||
self._secret(AUTOUPDATE_API_FAILURE_MINUTES=0, AUTOUPDATE_MAX_LOOP_INVOCATIONS=30)
|
||||
)
|
||||
|
||||
result, _, elapsed = self._drive_with_clock(
|
||||
autoupdate,
|
||||
choose_side_effect=ApiException(404, "gone"),
|
||||
)
|
||||
|
||||
self.assertIn(result, (0, None), "0 must disable the sustained-failure exit")
|
||||
self.assertGreater(
|
||||
elapsed, 45 * 60,
|
||||
"Test must simulate past the default threshold to prove it is ignored",
|
||||
)
|
||||
|
||||
|
||||
class FakeChooseDevice:
|
||||
def __init__(self, secret, tconnect):
|
||||
self.secret = secret
|
||||
self.tconnect = tconnect
|
||||
|
||||
def choose(self):
|
||||
return {
|
||||
'tconnectDeviceId': 'test-device-123',
|
||||
'maxDateOfEvents': '2025-11-18T13:00:00-05:00',
|
||||
}
|
||||
|
||||
|
||||
class FakeProcessTimeRange:
|
||||
def __init__(self, tconnect, nightscout, tconnectDevice, pretend, secret, features=None):
|
||||
self.tconnect = tconnect
|
||||
self.nightscout = nightscout
|
||||
self.tconnectDevice = tconnectDevice
|
||||
self.pretend = pretend
|
||||
self.secret = secret
|
||||
self.features = features
|
||||
|
||||
def process(self, time_start, time_end):
|
||||
return 0, None
|
||||
|
||||
|
||||
class TestTandemSourceAutoupdate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.secret = build_secrets(
|
||||
AUTOUPDATE_MAX_LOOP_INVOCATIONS=2,
|
||||
AUTOUPDATE_DEFAULT_SLEEP_SECONDS=0,
|
||||
AUTOUPDATE_USE_FIXED_SLEEP=True,
|
||||
AUTOUPDATE_MAX_SLEEP_SECONDS=0,
|
||||
AUTOUPDATE_NO_DATA_FAILURE_MINUTES=9999,
|
||||
AUTOUPDATE_FAILURE_MINUTES=9999,
|
||||
AUTOUPDATE_UNEXPECTED_NO_INDEX_SLEEP_SECONDS=0,
|
||||
AUTOUPDATE_RESTART_ON_FAILURE=False,
|
||||
)
|
||||
|
||||
def test_process_does_not_crash_when_no_events_are_found(self):
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
|
||||
with mock.patch('tconnectsync.sync.tandemsource.autoupdate.ChooseDevice', FakeChooseDevice), \
|
||||
mock.patch('tconnectsync.sync.tandemsource.autoupdate.ProcessTimeRange', FakeProcessTimeRange), \
|
||||
mock.patch('tconnectsync.sync.tandemsource.autoupdate.time.time', return_value=1000), \
|
||||
mock.patch('tconnectsync.sync.tandemsource.autoupdate.time.sleep', return_value=None), \
|
||||
self.assertLogs('tconnectsync.sync.tandemsource.autoupdate', level='INFO') as logs:
|
||||
result = autoupdate.process(object(), object(), pretend=False)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertTrue(any('No new reported tandemsource data.' in message for message in logs.output))
|
||||
|
||||
def test_process_does_not_crash_in_pretend_mode_without_successful_update_time(self):
|
||||
autoupdate = TandemSourceAutoupdate(self.secret)
|
||||
|
||||
with mock.patch('tconnectsync.sync.tandemsource.autoupdate.ChooseDevice', FakeChooseDevice), \
|
||||
mock.patch('tconnectsync.sync.tandemsource.autoupdate.time.time', return_value=2000), \
|
||||
mock.patch('tconnectsync.sync.tandemsource.autoupdate.time.sleep', return_value=None), \
|
||||
self.assertLogs('tconnectsync.sync.tandemsource.autoupdate', level='INFO') as logs:
|
||||
result = autoupdate.process(object(), object(), pretend=True)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertIsNone(autoupdate.last_successful_process_time_range)
|
||||
self.assertTrue(any('No new reported tandemsource data.' in message for message in logs.output))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -92,6 +92,16 @@ ALARM_RESUME = {
|
||||
"eventProperties": {"alarmId": 18, "faultLocatorData": 8311, "param1": 5228339, "param2": 0},
|
||||
}
|
||||
|
||||
MALFUNCTION = {
|
||||
"deviceAssignmentId": "00000000-0000-0000-0000-000000000000",
|
||||
"eventCode": 6,
|
||||
"sequenceGroup": 0,
|
||||
"sequenceNumber": 500123,
|
||||
"pumpDateTime": "2026-05-16T00:07:00",
|
||||
"estimatedDateTime": "2026-05-16T00:07:00Z",
|
||||
"eventProperties": {"malfId": 7, "faultLocatorData": 8311, "param1": 42, "param2": 0},
|
||||
}
|
||||
|
||||
|
||||
class TestProcessAlarmJson(unittest.TestCase):
|
||||
maxDiff = None
|
||||
@@ -127,6 +137,41 @@ class TestProcessAlarmJson(unittest.TestCase):
|
||||
p = self.process.process(list(Events([dict(ALARM_RESUME)])), None, None)
|
||||
self.assertEqual(p, [])
|
||||
|
||||
def test_malfunction_alarm_uploaded(self):
|
||||
event = Event(dict(MALFUNCTION))
|
||||
self.assertEqual(type(event), eventtypes.LidMalfunctionActivated)
|
||||
self.assertFalse(hasattr(event, 'alarmId'))
|
||||
|
||||
p = self.process.process([event], None, None)
|
||||
|
||||
self.assertEqual(len(p), 1)
|
||||
self.assertDictEqual(p[0], {
|
||||
'eventType': 'Alarm',
|
||||
'reason': 'Malfunction',
|
||||
'notes': 'Malfunction',
|
||||
'created_at': '2026-05-16 00:07:00-04:00',
|
||||
'enteredBy': 'Pump (tconnectsync)',
|
||||
'pump_event_id': '500123'
|
||||
})
|
||||
|
||||
def test_alarm_and_malfunction_mixed(self):
|
||||
# A batch mixing both ALARM-class event types must not crash and must
|
||||
# emit an entry for each.
|
||||
p = self.process.process(list(Events([dict(ALARM_PUMP_RESET), dict(MALFUNCTION)])), None, None)
|
||||
|
||||
self.assertEqual(len(p), 2)
|
||||
reasons = {entry['reason'] for entry in p}
|
||||
self.assertEqual(reasons, {'PumpResetAlarm', 'Malfunction'})
|
||||
|
||||
|
||||
class TestAlarmOrMalfunctionUnion(unittest.TestCase):
|
||||
def test_union_matches_eventclass(self):
|
||||
from typing import get_args
|
||||
from tconnectsync.sync.tandemsource.process_alarm import AlarmOrMalfunction
|
||||
from tconnectsync.domain.tandemsource.event_class import EventClass
|
||||
|
||||
self.assertEqual(set(get_args(AlarmOrMalfunction)), set(EventClass.ALARM))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -0,0 +1,320 @@
|
||||
#!/usr/bin/env python3
|
||||
"""EU region integration tests (issue #152).
|
||||
|
||||
Configures the EU region via each supported mechanism (TCONNECT_REGION
|
||||
environment variable, .env file, --region CLI flag) and runs the real
|
||||
downstream code against a mocked HTTP layer on which only the EU
|
||||
endpoints are registered, so any request to a US endpoint fails.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
import urllib.parse
|
||||
|
||||
import jwt
|
||||
import requests_mock
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from jwt.algorithms import RSAAlgorithm
|
||||
|
||||
TEST_KID = 'eu-integration-test-key'
|
||||
PUMPER_ID = 'aaaaaaaa-1111-2222-3333-444444444444'
|
||||
ACCOUNT_ID = 'bbbbbbbb-5555-6666-7777-888888888888'
|
||||
DEVICE_ID = '1b493210-9336-4901-a329-a352775738c5'
|
||||
|
||||
EU_API = 'https://tdcservices.eu.tandemdiabetes.com/accounts/api'
|
||||
EU_SOURCE = 'https://source.eu.tandemdiabetes.com/'
|
||||
EU_CLIENT_ID = '1519e414-eeec-492e-8c5e-97bea4815a10'
|
||||
|
||||
EU_LOGIN_URL = EU_API + '/login'
|
||||
EU_TOKEN_URL = EU_API + '/connect/token'
|
||||
EU_AUTHORIZE_URL = EU_API + '/connect/authorize'
|
||||
EU_JWKS_URL = EU_API + '/.well-known/openid-configuration/jwks'
|
||||
EU_CALLBACK_URL = EU_SOURCE + 'authorize/callback'
|
||||
EU_PUMPER_URL = EU_SOURCE + 'api/reports/bff/pumper/' + PUMPER_ID
|
||||
EU_PUMP_LOGS_URL = EU_SOURCE + 'api/reports/bff/pump-logs/' + DEVICE_ID
|
||||
|
||||
NS_URL = 'http://nightscout.example.com/'
|
||||
|
||||
US_HOSTS = {'tdcservices.tandemdiabetes.com', 'source.tandemdiabetes.com'}
|
||||
|
||||
# sso.tandemdiabetes.com hosts the login page for both regions.
|
||||
ALLOWED_HOSTS = {
|
||||
'sso.tandemdiabetes.com',
|
||||
'tdcservices.eu.tandemdiabetes.com',
|
||||
'source.eu.tandemdiabetes.com',
|
||||
'nightscout.example.com',
|
||||
}
|
||||
|
||||
TEST_EMAIL = 'eu-user@example.com'
|
||||
TEST_PASSWORD = 'eu-password'
|
||||
|
||||
_PRIVATE_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
_PRIVATE_PEM = _PRIVATE_KEY.private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.PKCS8,
|
||||
serialization.NoEncryption(),
|
||||
)
|
||||
|
||||
|
||||
def make_id_token():
|
||||
"""RS256-signed id_token with the EU issuer/audience, so extract_jwt()
|
||||
performs full verification against the mocked EU jwks endpoint."""
|
||||
now = int(time.time())
|
||||
claims = {
|
||||
'iss': EU_API,
|
||||
'aud': EU_CLIENT_ID,
|
||||
'iat': now,
|
||||
'nbf': now,
|
||||
'exp': now + 3600,
|
||||
'sub': 'eu-test-subject',
|
||||
'pumperId': PUMPER_ID,
|
||||
'accountId': ACCOUNT_ID,
|
||||
}
|
||||
return jwt.encode(claims, _PRIVATE_PEM, algorithm='RS256', headers={'kid': TEST_KID})
|
||||
|
||||
|
||||
def make_jwks():
|
||||
jwk = json.loads(RSAAlgorithm.to_jwk(_PRIVATE_KEY.public_key()))
|
||||
jwk.update({'kid': TEST_KID, 'use': 'sig', 'alg': 'RS256'})
|
||||
return {'keys': [jwk]}
|
||||
|
||||
|
||||
BFF_PUMPER = {
|
||||
'firstName': 'Eu',
|
||||
'lastName': 'User',
|
||||
'name': 'Eu User',
|
||||
'country': 'DE',
|
||||
'pumps': [
|
||||
{
|
||||
'algorithm': 'Control-IQ',
|
||||
'availableDataRange': {'start': '2026-01-01T00:00:00', 'end': '2026-07-16T10:00:00'},
|
||||
'assignmentId': DEVICE_ID,
|
||||
'lastUploadDate': '2026-07-16T10:00:00Z',
|
||||
'maxDateOfEvents': '2026-07-16T10:00:00',
|
||||
'modelNumber': '1000354',
|
||||
'modelName': 't:slim X2™ Insulin Pump',
|
||||
'partNumber': '1011979',
|
||||
'serialNumber': '90556643',
|
||||
'softwareVersion': '7.8.0.0',
|
||||
'lastUploadClientType': 'mobile_tconnect',
|
||||
'settings': None,
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
PUMP_LOGS = {'events': [], 'clockChanges': []}
|
||||
|
||||
|
||||
def register_eu_endpoints(m):
|
||||
"""Register only the EU (and region-shared) endpoints; any US request
|
||||
raises requests_mock.NoMockAddress."""
|
||||
m.get('https://sso.tandemdiabetes.com/', text='')
|
||||
m.post(EU_LOGIN_URL, json={'redirectUrl': '/', 'status': 'SUCCESS'})
|
||||
m.get(EU_AUTHORIZE_URL, status_code=302,
|
||||
headers={'Location': EU_CALLBACK_URL + '?code=eu-test-code'})
|
||||
m.get(EU_CALLBACK_URL, text='')
|
||||
m.post(EU_TOKEN_URL, json={
|
||||
'access_token': 'eu-access-token',
|
||||
'id_token': make_id_token(),
|
||||
'expires_in': 3600,
|
||||
})
|
||||
m.get(EU_JWKS_URL, json=make_jwks())
|
||||
m.get(EU_PUMPER_URL, json=BFF_PUMPER)
|
||||
m.get(EU_PUMP_LOGS_URL, json=PUMP_LOGS)
|
||||
|
||||
|
||||
def register_nightscout_endpoints(m):
|
||||
m.get(NS_URL + 'api/v1/status.json', json={'status': 'ok', 'version': '15.0.3'})
|
||||
m.get(NS_URL + 'api/v1/treatments', json=[])
|
||||
|
||||
|
||||
class EuRegionTestBase(unittest.TestCase):
|
||||
"""Runs each test in a scratch cwd with a controlled environment, and
|
||||
re-imports tconnectsync so secret.py is loaded from that environment."""
|
||||
maxDiff = None
|
||||
|
||||
ENV_KEYS = [
|
||||
'TCONNECT_EMAIL', 'TCONNECT_PASSWORD', 'TCONNECT_REGION',
|
||||
'CACHE_CREDENTIALS', 'NS_URL', 'NS_SECRET', 'API_SECRET',
|
||||
'TIMEZONE_NAME', 'TZ', 'PUMP_SERIAL_NUMBER', 'REQUESTS_PROXY',
|
||||
]
|
||||
|
||||
BASE_ENV = {
|
||||
'TCONNECT_EMAIL': TEST_EMAIL,
|
||||
'TCONNECT_PASSWORD': TEST_PASSWORD,
|
||||
'CACHE_CREDENTIALS': 'false',
|
||||
'NS_URL': NS_URL,
|
||||
'NS_SECRET': 'ns-secret',
|
||||
'TIMEZONE_NAME': 'Europe/Berlin',
|
||||
}
|
||||
|
||||
def setUp(self):
|
||||
self._saved_env = {k: os.environ.get(k) for k in self.ENV_KEYS}
|
||||
for k in self.ENV_KEYS:
|
||||
os.environ.pop(k, None)
|
||||
self._old_cwd = os.getcwd()
|
||||
self._tmpdir = tempfile.mkdtemp(prefix='tconnectsync-eu-test-')
|
||||
os.chdir(self._tmpdir)
|
||||
self._purge_modules()
|
||||
|
||||
def tearDown(self):
|
||||
os.chdir(self._old_cwd)
|
||||
shutil.rmtree(self._tmpdir, ignore_errors=True)
|
||||
for k, v in self._saved_env.items():
|
||||
if v is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = v
|
||||
self._purge_modules()
|
||||
|
||||
@staticmethod
|
||||
def _purge_modules():
|
||||
for name in list(sys.modules):
|
||||
if name == 'tconnectsync' or name.startswith('tconnectsync.'):
|
||||
del sys.modules[name]
|
||||
|
||||
def import_tconnectsync(self, extra_env=None):
|
||||
env = dict(self.BASE_ENV)
|
||||
env.update(extra_env or {})
|
||||
os.environ.update(env)
|
||||
self._purge_modules()
|
||||
import tconnectsync
|
||||
return tconnectsync
|
||||
|
||||
def called(self, m):
|
||||
"""(method, url-without-query) pairs for every mocked request."""
|
||||
return [(r.method, r.url.split('?')[0]) for r in m.request_history]
|
||||
|
||||
def assert_only_eu_hosts(self, m):
|
||||
hosts = {urllib.parse.urlparse(r.url).netloc.lower() for r in m.request_history}
|
||||
self.assertTrue(hosts, 'expected at least one HTTP request')
|
||||
self.assertFalse(hosts & US_HOSTS,
|
||||
'US endpoints were contacted with EU region configured: %s' % (hosts & US_HOSTS))
|
||||
self.assertLessEqual(hosts, ALLOWED_HOSTS,
|
||||
'unexpected hosts contacted: %s' % (hosts - ALLOWED_HOSTS))
|
||||
|
||||
def assert_eu_login_flow(self, m):
|
||||
calls = self.called(m)
|
||||
self.assertIn(('POST', EU_LOGIN_URL), calls)
|
||||
self.assertIn(('GET', EU_AUTHORIZE_URL), calls)
|
||||
self.assertIn(('POST', EU_TOKEN_URL), calls)
|
||||
self.assertIn(('GET', EU_JWKS_URL), calls)
|
||||
self.assert_only_eu_hosts(m)
|
||||
|
||||
|
||||
class TestEuRegionFromEnvironmentVariable(EuRegionTestBase):
|
||||
"""TCONNECT_REGION=EU set as an environment variable."""
|
||||
|
||||
def test_tconnect_api_without_region_argument_uses_eu(self):
|
||||
# The downstream pattern that regressed in #152: TConnectApi built
|
||||
# without a region argument.
|
||||
self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
|
||||
from tconnectsync.api import TConnectApi
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
register_eu_endpoints(m)
|
||||
|
||||
api = TConnectApi(TEST_EMAIL, TEST_PASSWORD)
|
||||
self.assertEqual(api.region, 'EU')
|
||||
|
||||
tandemsource = api.tandemsource
|
||||
self.assertEqual(tandemsource.region, 'EU')
|
||||
self.assertEqual(tandemsource.LOGIN_API_URL, EU_LOGIN_URL)
|
||||
self.assertEqual(tandemsource.SOURCE_URL, EU_SOURCE)
|
||||
self.assertEqual(tandemsource.pumperId, PUMPER_ID)
|
||||
self.assertEqual(tandemsource.accountId, ACCOUNT_ID)
|
||||
|
||||
pumper = tandemsource.get_pumper()
|
||||
self.assertEqual(pumper['pumps'][0]['assignmentId'], DEVICE_ID)
|
||||
|
||||
self.assert_eu_login_flow(m)
|
||||
self.assertIn(('GET', EU_PUMPER_URL), self.called(m))
|
||||
|
||||
def test_tandem_source_api_without_region_argument_uses_eu(self):
|
||||
self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
|
||||
from tconnectsync.api.tandemsource import TandemSourceApi
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
register_eu_endpoints(m)
|
||||
|
||||
api = TandemSourceApi(TEST_EMAIL, TEST_PASSWORD)
|
||||
self.assertEqual(api.region, 'EU')
|
||||
self.assertEqual(api.pumperId, PUMPER_ID)
|
||||
self.assert_eu_login_flow(m)
|
||||
|
||||
def test_secret_exposes_eu_region(self):
|
||||
tconnectsync = self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
|
||||
self.assertEqual(tconnectsync.secret.TCONNECT_REGION, 'EU')
|
||||
|
||||
|
||||
class TestEuRegionFromDotEnvFile(EuRegionTestBase):
|
||||
"""TCONNECT_REGION=EU set through a .env file in the working directory."""
|
||||
|
||||
def test_tconnect_api_without_region_argument_uses_eu(self):
|
||||
# secret.py reads $CWD/.env; setUp chdir'd into a scratch directory.
|
||||
with open(os.path.join(self._tmpdir, '.env'), 'w') as f:
|
||||
for k, v in dict(self.BASE_ENV, TCONNECT_REGION='EU').items():
|
||||
f.write('%s=%s\n' % (k, v))
|
||||
|
||||
os.environ.update({'CACHE_CREDENTIALS': 'false'})
|
||||
self._purge_modules()
|
||||
import tconnectsync # noqa: F401
|
||||
from tconnectsync import secret
|
||||
from tconnectsync.api import TConnectApi
|
||||
|
||||
self.assertEqual(secret.TCONNECT_REGION, 'EU')
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
register_eu_endpoints(m)
|
||||
|
||||
api = TConnectApi(secret.TCONNECT_EMAIL, secret.TCONNECT_PASSWORD)
|
||||
self.assertEqual(api.region, 'EU')
|
||||
self.assertEqual(api.tandemsource.LOGIN_API_URL, EU_LOGIN_URL)
|
||||
self.assert_eu_login_flow(m)
|
||||
|
||||
|
||||
class TestEuRegionThroughMainEntrypoint(EuRegionTestBase):
|
||||
"""Full `tconnectsync --check-login` runs through main()."""
|
||||
|
||||
def run_check_login(self, tconnectsync, argv):
|
||||
with requests_mock.Mocker() as m:
|
||||
register_eu_endpoints(m)
|
||||
register_nightscout_endpoints(m)
|
||||
stdout = io.StringIO()
|
||||
with contextlib.redirect_stdout(stdout):
|
||||
tconnectsync.main(argv)
|
||||
return m, stdout.getvalue()
|
||||
|
||||
def assert_full_eu_check_login(self, m, output):
|
||||
self.assert_eu_login_flow(m)
|
||||
calls = self.called(m)
|
||||
self.assertIn(('GET', EU_PUMPER_URL), calls)
|
||||
self.assertIn(('GET', EU_PUMP_LOGS_URL), calls)
|
||||
self.assertIn('No API errors returned!', output)
|
||||
self.assertNotIn('API errors occurred', output)
|
||||
|
||||
def test_check_login_with_region_from_environment_variable(self):
|
||||
tconnectsync = self.import_tconnectsync({'TCONNECT_REGION': 'EU'})
|
||||
m, output = self.run_check_login(tconnectsync, ['--check-login'])
|
||||
self.assertIn("TCONNECT_REGION='EU'", output)
|
||||
self.assert_full_eu_check_login(m, output)
|
||||
|
||||
def test_check_login_with_region_from_cli_flag(self):
|
||||
# No TCONNECT_REGION configured: --region EU alone must route
|
||||
# everything to the EU endpoints.
|
||||
tconnectsync = self.import_tconnectsync()
|
||||
self.assertEqual(tconnectsync.secret.TCONNECT_REGION, 'US')
|
||||
m, output = self.run_check_login(tconnectsync, ['--check-login', '--region', 'EU'])
|
||||
self.assert_full_eu_check_login(m, output)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user