willemjan
9e11dd8d73
AndroidAPS records an automatic SMB micro-bolus as eventType "Correction Bolus" with type "SMB", not as its own eventType. bolus_stats() was grouping by eventType alone, so every SMB was silently counted as a manual correction bolus. Verified against production data: 100% of "Correction Bolus" treatments over 90 days were actually type=SMB. Also adds basal_stats(), computing delivered basal directly from Temp Basal treatments (rate x duration) without needing the pump profile. Closes gitea #20.
382 lines
14 KiB
Python
382 lines
14 KiB
Python
from datetime import date, datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
import nightscout_client as ns
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self._payload = payload
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
def test_mgdl_to_mmol_converts_correctly():
|
|
assert ns.mgdl_to_mmol(126.0) == pytest.approx(7.0, abs=0.05)
|
|
|
|
|
|
def test_base_url_requires_nightscout_url(monkeypatch):
|
|
monkeypatch.delenv("NIGHTSCOUT_URL", raising=False)
|
|
with pytest.raises(ns.NightscoutConfigError):
|
|
ns._base_url()
|
|
|
|
|
|
def test_base_url_strips_trailing_slash(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com/")
|
|
assert ns._base_url() == "https://example.com"
|
|
|
|
|
|
def test_auth_prefers_token_over_api_secret(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_TOKEN", "my-token")
|
|
monkeypatch.setenv("NIGHTSCOUT_API_SECRET", "my-secret")
|
|
headers, params = ns._auth()
|
|
assert params == {"token": "my-token"}
|
|
assert headers == {}
|
|
|
|
|
|
def test_auth_hashes_api_secret(monkeypatch):
|
|
monkeypatch.delenv("NIGHTSCOUT_TOKEN", raising=False)
|
|
monkeypatch.setenv("NIGHTSCOUT_API_SECRET", "my-secret")
|
|
headers, params = ns._auth()
|
|
assert params == {}
|
|
assert headers["API-SECRET"] != "my-secret" # nooit in plaintext
|
|
assert len(headers["API-SECRET"]) == 40 # sha1 hexdigest
|
|
|
|
|
|
def test_fetch_entries_converts_and_filters_by_time(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
monkeypatch.delenv("NIGHTSCOUT_TOKEN", raising=False)
|
|
monkeypatch.delenv("NIGHTSCOUT_API_SECRET", raising=False)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
recent = now - timedelta(hours=1)
|
|
too_old = now - timedelta(hours=48)
|
|
payload = [
|
|
{"date": int(recent.timestamp() * 1000), "sgv": 126},
|
|
{"date": int(too_old.timestamp() * 1000), "sgv": 90},
|
|
]
|
|
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse(payload))
|
|
|
|
result = ns.fetch_entries(hours=24)
|
|
assert len(result) == 1
|
|
assert result.iloc[0]["mmol"] == pytest.approx(7.0, abs=0.05)
|
|
|
|
|
|
def test_fetch_entries_handles_empty_response(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse([]))
|
|
|
|
result = ns.fetch_entries(hours=24)
|
|
assert result.empty
|
|
assert list(result.columns) == ["timestamp", "mmol"]
|
|
|
|
|
|
def test_fetch_treatments_fills_missing_columns(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
now = datetime.now(timezone.utc)
|
|
payload = [
|
|
{"created_at": now.isoformat(), "eventType": "Site Change"},
|
|
]
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse(payload))
|
|
|
|
result = ns.fetch_treatments(hours=24)
|
|
assert len(result) == 1
|
|
assert result.iloc[0]["eventType"] == "Site Change"
|
|
assert pd_isna(result.iloc[0]["insulin"])
|
|
assert pd_isna(result.iloc[0]["carbs"])
|
|
assert result.iloc[0]["type"] == ""
|
|
assert pd_isna(result.iloc[0]["absolute"])
|
|
assert pd_isna(result.iloc[0]["duration"])
|
|
|
|
|
|
def test_fetch_treatments_passes_through_type_and_temp_basal_fields(monkeypatch):
|
|
"""SMB-classificatie (reporting.bolus_stats) en basaalberekening
|
|
(reporting.basal_stats) hebben deze velden nodig — zie gitea #20."""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
now = datetime.now(timezone.utc)
|
|
payload = [
|
|
{
|
|
"created_at": now.isoformat(),
|
|
"eventType": "Correction Bolus",
|
|
"insulin": 0.5,
|
|
"type": "SMB",
|
|
},
|
|
{
|
|
"created_at": now.isoformat(),
|
|
"eventType": "Temp Basal",
|
|
"absolute": 1.2,
|
|
"duration": 30,
|
|
},
|
|
]
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse(payload))
|
|
|
|
result = ns.fetch_treatments(hours=24)
|
|
smb_row = result[result["eventType"] == "Correction Bolus"].iloc[0]
|
|
basal_row = result[result["eventType"] == "Temp Basal"].iloc[0]
|
|
assert smb_row["type"] == "SMB"
|
|
assert basal_row["absolute"] == pytest.approx(1.2)
|
|
assert basal_row["duration"] == pytest.approx(30)
|
|
|
|
|
|
def pd_isna(value):
|
|
import pandas as pd
|
|
|
|
return pd.isna(value)
|
|
|
|
|
|
def test_fetch_entries_range_filters_to_date_boundaries(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
monkeypatch.delenv("NIGHTSCOUT_TOKEN", raising=False)
|
|
monkeypatch.delenv("NIGHTSCOUT_API_SECRET", raising=False)
|
|
|
|
payload = [
|
|
{"date": int(datetime(2026, 1, 15, 12, 0, tzinfo=timezone.utc).timestamp() * 1000), "sgv": 126},
|
|
{"date": int(datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc).timestamp() * 1000), "sgv": 90},
|
|
{"date": int(datetime(2026, 2, 1, 12, 0, tzinfo=timezone.utc).timestamp() * 1000), "sgv": 100},
|
|
]
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse(payload))
|
|
|
|
result = ns.fetch_entries_range(date(2026, 1, 1), date(2026, 1, 31))
|
|
assert len(result) == 2 # de 1 feb-meting valt buiten het bereik
|
|
assert list(result.columns) == ["timestamp", "mmol"]
|
|
|
|
|
|
def test_fetch_entries_range_handles_empty_response(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse([]))
|
|
|
|
result = ns.fetch_entries_range(date(2026, 1, 1), date(2026, 1, 31))
|
|
assert result.empty
|
|
|
|
|
|
def test_fetch_treatments_range_filters_to_date_boundaries(monkeypatch):
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
payload = [
|
|
{"created_at": "2026-01-15T12:00:00Z", "eventType": "Meal Bolus", "insulin": 4.0, "carbs": 40},
|
|
{"created_at": "2026-02-01T12:00:00Z", "eventType": "Meal Bolus", "insulin": 3.0, "carbs": 30},
|
|
]
|
|
monkeypatch.setattr(ns.requests, "get", lambda *a, **k: FakeResponse(payload))
|
|
|
|
result = ns.fetch_treatments_range(date(2026, 1, 1), date(2026, 1, 31))
|
|
assert len(result) == 1
|
|
assert result.iloc[0]["insulin"] == pytest.approx(4.0)
|
|
|
|
|
|
def _entry(dt, sgv=100):
|
|
return {"date": int(dt.timestamp() * 1000), "sgv": sgv, "dateString": dt.isoformat()}
|
|
|
|
|
|
def test_paginated_get_stops_when_a_short_page_is_returned(monkeypatch):
|
|
"""Nightscout kan minder dan _PAGE_SIZE teruggeven omdat er simpelweg niet meer data is.
|
|
|
|
Dit was precies de bug: een enkel groot count-verzoek werd stilzwijgend
|
|
afgeknipt. Met paginering moet een pagina die kleiner is dan de gevraagde
|
|
paginagrootte de paginering laten stoppen — dat betekent "geen oudere
|
|
data beschikbaar", niet "er ging iets mis".
|
|
"""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
now = datetime.now(timezone.utc)
|
|
payload = [_entry(now - timedelta(minutes=5 * i)) for i in range(5)] # << _PAGE_SIZE
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
ns.requests,
|
|
"get",
|
|
lambda *a, **k: (calls.append(k), FakeResponse(payload))[1],
|
|
)
|
|
|
|
result = ns._paginated_get(
|
|
"/api/v1/entries/sgv.json", now - timedelta(days=1), None, ns.ENTRIES_CURSOR_FIELD
|
|
)
|
|
assert len(result) == 5
|
|
assert len(calls) == 1 # één pagina was genoeg, geen verdere paginering nodig
|
|
|
|
|
|
def test_paginated_get_walks_back_multiple_pages_until_since_is_reached(monkeypatch):
|
|
"""De kern-regressietest voor de bug: data die verder terugligt dan één
|
|
pagina (~1000 documenten) moet toch volledig worden opgehaald door
|
|
meerdere pagina's met find[date][$lte] te doorlopen."""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
now = datetime.now(timezone.utc)
|
|
|
|
# 3 volle pagina's (elk _PAGE_SIZE groot, 5 minuten uit elkaar, aflopend in tijd)
|
|
# + een laatste kleinere pagina die de paginering laat stoppen.
|
|
page_size = ns._PAGE_SIZE
|
|
pages = []
|
|
cursor = now
|
|
for _ in range(3):
|
|
page = [_entry(cursor - timedelta(minutes=5 * i)) for i in range(page_size)]
|
|
pages.append(page)
|
|
cursor = cursor - timedelta(minutes=5 * (page_size - 1)) - timedelta(minutes=5)
|
|
pages.append([_entry(cursor - timedelta(minutes=5 * i)) for i in range(10)])
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def fake_get(*_a, **_k):
|
|
response = FakeResponse(pages[call_count["n"]])
|
|
call_count["n"] += 1
|
|
return response
|
|
|
|
monkeypatch.setattr(ns.requests, "get", fake_get)
|
|
|
|
since = now - timedelta(days=365) # ruim voorbij alle test-pagina's
|
|
result = ns._paginated_get(
|
|
"/api/v1/entries/sgv.json", since, None, ns.ENTRIES_CURSOR_FIELD
|
|
)
|
|
|
|
assert call_count["n"] == 4 # 3 volle pagina's + 1 korte slotpagina
|
|
assert len(result) == 3 * page_size + 10
|
|
|
|
|
|
def test_paginated_get_stops_at_since_boundary_without_fetching_further(monkeypatch):
|
|
"""Zodra een pagina data ouder dan `since` bevat, hoeft er geen volgende
|
|
pagina meer opgehaald te worden, ook al was de pagina zelf vol."""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
now = datetime.now(timezone.utc)
|
|
page_size = ns._PAGE_SIZE
|
|
# deze ene pagina spant ruim meer dan een dag, dus bevat al data voorbij `since`
|
|
page = [_entry(now - timedelta(minutes=5 * i)) for i in range(page_size)]
|
|
|
|
calls = {"n": 0}
|
|
|
|
def fake_get(*_a, **_k):
|
|
calls["n"] += 1
|
|
return FakeResponse(page)
|
|
|
|
monkeypatch.setattr(ns.requests, "get", fake_get)
|
|
|
|
since = now - timedelta(hours=6) # ruim binnen de eerste pagina
|
|
result = ns._paginated_get(
|
|
"/api/v1/entries/sgv.json", since, None, ns.ENTRIES_CURSOR_FIELD
|
|
)
|
|
|
|
assert calls["n"] == 1
|
|
assert len(result) == page_size # filtering op since gebeurt pas in _entries_since, niet hier
|
|
|
|
|
|
def test_fetch_entries_range_retrieves_data_spanning_multiple_pages(monkeypatch):
|
|
"""End-to-end regressietest: een 90-dagenrapport moet echt 90 dagen data
|
|
krijgen, niet stilzwijgend worden afgeknipt tot de meest recente pagina."""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
monkeypatch.delenv("NIGHTSCOUT_TOKEN", raising=False)
|
|
monkeypatch.delenv("NIGHTSCOUT_API_SECRET", raising=False)
|
|
|
|
end_date = date(2026, 3, 31)
|
|
start_date = date(2026, 1, 1) # 90 dagen
|
|
|
|
# simuleer Nightscout: elke 5 minuten een meting over de volle periode,
|
|
# opgeknipt in door de server afgedwongen pagina's van _PAGE_SIZE
|
|
until = datetime.combine(end_date, datetime.max.time(), tzinfo=timezone.utc)
|
|
since = datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc)
|
|
all_entries = []
|
|
t = until
|
|
while t >= since:
|
|
all_entries.append(_entry(t))
|
|
t -= timedelta(minutes=5)
|
|
|
|
page_size = ns._PAGE_SIZE
|
|
|
|
def fake_get(*_args, **kwargs):
|
|
cursor_ms = kwargs["params"].get("find[date][$lte]")
|
|
pool = all_entries if cursor_ms is None else [e for e in all_entries if e["date"] <= cursor_ms]
|
|
return FakeResponse(pool[:page_size])
|
|
|
|
monkeypatch.setattr(ns.requests, "get", fake_get)
|
|
|
|
result = ns.fetch_entries_range(start_date, end_date)
|
|
assert len(result) == len(all_entries)
|
|
assert result["timestamp"].min().date() == start_date
|
|
assert result["timestamp"].max().date() == end_date
|
|
|
|
|
|
def _treatment(dt, event_type="Temp Basal"):
|
|
return {
|
|
"created_at": dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
|
|
"date": int(dt.timestamp() * 1000),
|
|
"eventType": event_type,
|
|
}
|
|
|
|
|
|
def test_fetch_treatments_range_retrieves_full_history_even_with_a_hidden_server_side_window(
|
|
monkeypatch,
|
|
):
|
|
"""Regressietest voor de echte productiebug (2026-07-31): treatments.json
|
|
filtert/sorteert intern op `created_at` (string), niet op `date`
|
|
(numeriek) zoals entries. Filteren op `date` — of helemaal geen filter
|
|
meegeven — leverde stilzwijgend hooguit de laatste ~4 dagen op, ook al
|
|
bevatten de documenten een geldig `date`-veld en stond de rest van de
|
|
geschiedenis (maanden) gewoon in de database (geverifieerd rechtstreeks
|
|
in MongoDB: 123k+ documenten, geen TTL-index, geen capped collection).
|
|
|
|
Deze test simuleert precies dat servergedrag: een find[date]-filter (of
|
|
het ontbreken van elk filter) geeft alleen de laatste 4 dagen terug;
|
|
alleen find[created_at] geeft de volledige, langere geschiedenis.
|
|
"""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
monkeypatch.delenv("NIGHTSCOUT_TOKEN", raising=False)
|
|
monkeypatch.delenv("NIGHTSCOUT_API_SECRET", raising=False)
|
|
|
|
end_date = date(2026, 7, 31)
|
|
start_date = date(2026, 5, 1) # 92 dagen, ruim voorbij het 4-dagen-venster
|
|
|
|
until = datetime.combine(end_date, datetime.max.time(), tzinfo=timezone.utc)
|
|
since = datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc)
|
|
all_treatments = []
|
|
t = until
|
|
while t >= since:
|
|
all_treatments.append(_treatment(t))
|
|
t -= timedelta(minutes=30)
|
|
|
|
hidden_window_start = until - timedelta(days=4)
|
|
page_size = ns._PAGE_SIZE
|
|
|
|
def fake_get(*_args, **kwargs):
|
|
params = kwargs["params"]
|
|
cursor = params.get("find[created_at][$lte]")
|
|
if cursor is None:
|
|
# geen (of het verkeerde) filter -> verborgen 4-dagenvenster
|
|
pool = [e for e in all_treatments if e["created_at"] >= hidden_window_start.strftime(
|
|
"%Y-%m-%dT%H:%M:%S.%f"
|
|
)[:-3] + "Z"]
|
|
else:
|
|
pool = [e for e in all_treatments if e["created_at"] <= cursor]
|
|
pool = sorted(pool, key=lambda e: e["created_at"], reverse=True)
|
|
return FakeResponse(pool[:page_size])
|
|
|
|
monkeypatch.setattr(ns.requests, "get", fake_get)
|
|
|
|
result = ns.fetch_treatments_range(start_date, end_date)
|
|
assert len(result) == len(all_treatments)
|
|
assert result["timestamp"].min().date() == start_date
|
|
assert result["timestamp"].max().date() == end_date
|
|
|
|
|
|
def test_paginated_get_uses_created_at_field_for_treatments_cursor(monkeypatch):
|
|
"""Losstaande, gerichte assertie op het daadwerkelijke queryparameter-veld
|
|
— moet `find[created_at][$lte]` zijn, nooit `find[date][$lte]`, voor het
|
|
treatments-endpoint."""
|
|
monkeypatch.setenv("NIGHTSCOUT_URL", "https://example.com")
|
|
now = datetime.now(timezone.utc)
|
|
payload = [_treatment(now - timedelta(minutes=30 * i)) for i in range(5)]
|
|
|
|
seen_params = []
|
|
|
|
def fake_get(*_a, **kwargs):
|
|
seen_params.append(kwargs["params"])
|
|
return FakeResponse(payload)
|
|
|
|
monkeypatch.setattr(ns.requests, "get", fake_get)
|
|
|
|
ns._paginated_get(
|
|
"/api/v1/treatments.json", now - timedelta(days=1), None, ns.TREATMENTS_CURSOR_FIELD
|
|
)
|
|
|
|
assert "find[created_at][$lte]" in seen_params[0]
|
|
assert "find[date][$lte]" not in seen_params[0]
|