Merge pull request 'Nightscout-koppeling + CI/CD-fixes' (#2) from dev into main
CI/CD / test (push) Failing after 49s
CI/CD / sonarqube (push) Skipped
CI/CD / build-and-scan (push) Skipped
CI/CD / deploy (push) Skipped

This commit was merged in pull request #2.
This commit is contained in:
2026-07-25 07:30:10 +00:00
14 changed files with 791 additions and 1 deletions
+8
View File
@@ -0,0 +1,8 @@
.env
.venv/
__pycache__/
*.pyc
.git/
.pytest_cache/
.ruff_cache/
tests/
+12
View File
@@ -0,0 +1,12 @@
# Kopieer dit bestand naar .env en vul je eigen Nightscout-gegevens in.
# .env wordt niet meegecommit (zie .gitignore).
NIGHTSCOUT_URL=https://jouw-nightscout.example.com
# Gebruik precies EEN van de twee onderstaande auth-opties.
# Optie 1: een Nightscout API-token (rol-gebaseerd, aan te raden — kan
# read-only worden ingesteld in Nightscout onder Admin Tools > Manage Tokens).
NIGHTSCOUT_TOKEN=
# Optie 2: het klassieke "API secret" van je Nightscout-instantie.
# NIGHTSCOUT_API_SECRET=
+84
View File
@@ -0,0 +1,84 @@
name: CI/CD
on:
push:
branches: [main]
jobs:
# ---------------------------------------------------------------------
# 1. Unit tests — moeten slagen voordat er iets gebouwd wordt
# ---------------------------------------------------------------------
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Installeer dependencies
run: pip install -r requirements.txt
- name: Lint (ruff)
run: ruff check .
- name: Unit tests (pytest)
run: pytest tests/ -v
# ---------------------------------------------------------------------
# 2. SonarQube — codekwaliteit-gate
# ---------------------------------------------------------------------
sonarqube:
runs-on: ubuntu-latest
needs: test
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0 # SonarQube wil de volledige git-historie zien
- name: SonarQube scan
uses: sonarsource/sonarqube-scan-action@v4
env:
SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
SONAR_HOST_URL: ${{ vars.SONAR_HOST_URL }}
# ---------------------------------------------------------------------
# 3. Image bouwen + Trivy security-scan + naar Gitea-registry pushen
# ---------------------------------------------------------------------
build-and-scan:
runs-on: ubuntu-latest
needs: [test, sonarqube]
steps:
- uses: actions/checkout@v4
- name: Build Docker image
run: docker build -t gitea.zb.local/willem/glucose-dashboard:${{ gitea.sha }} .
- name: Trivy scan
uses: aquasecurity/trivy-action@master
with:
image-ref: gitea.zb.local/willem/glucose-dashboard:${{ gitea.sha }}
severity: CRITICAL,HIGH
exit-code: "1" # laat de pipeline falen bij kritieke kwetsbaarheden
- name: Login bij Gitea-registry
run: echo "${{ secrets.GLUCOSE_TOKEN }}" | docker login gitea.zb.local -u willem --password-stdin
- name: Tag en push
run: |
docker tag gitea.zb.local/willem/glucose-dashboard:${{ gitea.sha }} gitea.zb.local/willem/glucose-dashboard:latest
docker push gitea.zb.local/willem/glucose-dashboard:${{ gitea.sha }}
docker push gitea.zb.local/willem/glucose-dashboard:latest
# ---------------------------------------------------------------------
# 4. Deploy — SSH naar de LXC en herstart met de nieuwe image
# ---------------------------------------------------------------------
deploy:
runs-on: ubuntu-latest
needs: build-and-scan
steps:
- name: Deploy via SSH
uses: appleboy/ssh-action@v1
with:
host: ${{ secrets.DEPLOY_HOST }}
username: ${{ secrets.DEPLOY_USER }}
key: ${{ secrets.DEPLOY_SSH_KEY }}
script: |
cd /opt/glucose-dashboard
docker compose pull
docker compose up -d
+7
View File
@@ -0,0 +1,7 @@
.venv/
__pycache__/
*.pyc
.env
.pytest_cache/
.ruff_cache/
homelab-cicd-werkinstructie.md
+18
View File
@@ -0,0 +1,18 @@
FROM python:3.12-slim
WORKDIR /app
# Systeem-dependency voor de HEALTHCHECK hieronder
RUN apt-get update && apt-get install -y --no-install-recommends curl \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8501
HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health || exit 1
ENTRYPOINT ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]
+36 -1
View File
@@ -1,3 +1,38 @@
# glucose-dashboard # glucose-dashboard
dashboard van glucose data Dashboard van glucosedata, live opgehaald uit een [Nightscout](https://nightscout.github.io/)-instantie.
Toont een rollend venster (standaard 24 uur) van je CGM-curve met insuline,
koolhydraten en pomp-/sensorwissels overlay, plus time-in-range-cijfers.
## Instellen
1. Zorg voor een draaiende Nightscout-instantie en een API-token (Admin Tools
> Manage Tokens in Nightscout — geef het token minimaal `readable`-rechten)
of het klassieke API secret.
2. Kopieer `.env.example` naar `.env` en vul `NIGHTSCOUT_URL` en één van de
twee auth-opties in. `app.py` laadt `.env` zelf in (via `python-dotenv`),
dus dit werkt hetzelfde onder PowerShell, cmd of bash — je hoeft de
variabelen niet zelf in je shell te zetten.
3. Lokaal draaien:
```powershell
uv venv
uv pip install -r requirements.txt --python .venv
.venv\Scripts\streamlit run app.py
```
4. Met Docker Compose:
```bash
docker compose up -d
```
`docker-compose.yml` leest `.env` via `env_file` — dat bestand wordt niet
meegecommit (zie `.gitignore`).
## Tests en lint
```bash
pytest tests/ -v
ruff check .
```
+241
View File
@@ -0,0 +1,241 @@
import plotly.graph_objects as go
import requests
import streamlit as st
from dotenv import load_dotenv
from data_processing import (
TARGET_HIGH,
TARGET_LOW,
classify_treatments,
summarize_time_in_range,
)
from nightscout_client import NightscoutConfigError, fetch_entries, fetch_treatments
load_dotenv() # laadt .env als die bestaat; bestaande env vars blijven leidend
st.set_page_config(page_title="Glucose dashboard", page_icon="📊", layout="wide")
st.title("Glucose dashboard")
HOURS_OPTIONS = {"6 uur": 6, "12 uur": 12, "24 uur": 24, "48 uur": 48}
# Kleuren per rol, licht/donker — zie de dataviz-conventies in dit project:
# categorale kleuren voor identiteit (glucose/insuline/carbs), status-kleuren
# (vast, nooit hergebruikt) voor te laag/te hoog.
PALETTE = {
"light": {
"glucose": "#2a78d6",
"insulin": "#1baf7a",
"carbs": "#eb6834",
"low": "#d03b3b",
"high": "#fab219",
"band": "rgba(12,163,12,0.08)",
"grid": "#e1e0d9",
"muted": "#898781",
"text": "#52514e",
},
"dark": {
"glucose": "#3987e5",
"insulin": "#199e70",
"carbs": "#d95926",
"low": "#d03b3b",
"high": "#fab219",
"band": "rgba(12,163,12,0.15)",
"grid": "#2c2c2a",
"muted": "#898781",
"text": "#c3c2b7",
},
}
@st.cache_data(ttl=60, show_spinner="Data ophalen van Nightscout...")
def load_nightscout_data(hours: int):
glucose = fetch_entries(hours=hours)
treatments = fetch_treatments(hours=hours)
return glucose, treatments
MMOL_HOVERTEMPLATE = "%{x|%d-%m %H:%M} · %{y:.1f} mmol/L<extra></extra>"
def build_chart(glucose, insulin, carbs, site_changes, sensor_changes, colors) -> go.Figure:
glucose_min = min(glucose["mmol"].min(), TARGET_LOW)
glucose_max = max(glucose["mmol"].max(), TARGET_HIGH)
y_top = glucose_max + 1.0
lane_base = glucose_min - 2.5
y_bottom = lane_base - 0.5
insulin_lane_y = lane_base + 1.5
carbs_lane_y = lane_base + 0.7
fig = go.Figure()
# Streefbereik als achtergrondband, geen losse legenda-entry nodig.
fig.add_hrect(
y0=TARGET_LOW,
y1=TARGET_HIGH,
fillcolor=colors["band"],
line_width=0,
layer="below",
)
fig.add_trace(
go.Scatter(
x=glucose["timestamp"],
y=glucose["mmol"],
mode="lines",
name="Glucose",
line={"color": colors["glucose"], "width": 2, "shape": "spline"},
hovertemplate=MMOL_HOVERTEMPLATE,
)
)
low_points = glucose[glucose["mmol"] < TARGET_LOW]
high_points = glucose[glucose["mmol"] > TARGET_HIGH]
if not low_points.empty:
fig.add_trace(
go.Scatter(
x=low_points["timestamp"],
y=low_points["mmol"],
mode="markers",
name=f"Laag (<{TARGET_LOW})",
marker={"color": colors["low"], "size": 6},
hovertemplate=MMOL_HOVERTEMPLATE,
)
)
if not high_points.empty:
fig.add_trace(
go.Scatter(
x=high_points["timestamp"],
y=high_points["mmol"],
mode="markers",
name=f"Hoog (>{TARGET_HIGH})",
marker={"color": colors["high"], "size": 6},
hovertemplate=MMOL_HOVERTEMPLATE,
)
)
if not insulin.empty:
fig.add_trace(
go.Scatter(
x=insulin["timestamp"],
y=[insulin_lane_y] * len(insulin),
mode="markers",
name="Insuline",
marker={"color": colors["insulin"], "size": 10, "symbol": "triangle-up"},
customdata=insulin["insulin"],
hovertemplate="%{x|%d-%m %H:%M} · %{customdata:.1f} E<extra></extra>",
)
)
if not carbs.empty:
fig.add_trace(
go.Scatter(
x=carbs["timestamp"],
y=[carbs_lane_y] * len(carbs),
mode="markers",
name="Koolhydraten",
marker={"color": colors["carbs"], "size": 10, "symbol": "circle"},
customdata=carbs["carbs"],
hovertemplate="%{x|%d-%m %H:%M} · %{customdata:.0f} g<extra></extra>",
)
)
# Let op: annotation_text/-position via add_vline() zelf triggert een bug
# in plotly's shapeannotation-helper met pandas Timestamp-x-waarden, dus
# de lijn en het label worden hier los toegevoegd.
for _, row in site_changes.iterrows():
fig.add_vline(x=row["timestamp"], line_dash="dot", line_color=colors["muted"], opacity=0.6)
fig.add_annotation(
x=row["timestamp"],
y=y_top,
yanchor="bottom",
text="Pomp",
showarrow=False,
font={"color": colors["muted"], "size": 10},
)
for _, row in sensor_changes.iterrows():
fig.add_vline(x=row["timestamp"], line_dash="dash", line_color=colors["muted"], opacity=0.6)
fig.add_annotation(
x=row["timestamp"],
y=y_top,
yanchor="bottom",
text="Sensor",
showarrow=False,
font={"color": colors["muted"], "size": 10},
)
fig.update_layout(
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(0,0,0,0)",
font_color=colors["text"],
legend={"orientation": "h", "yanchor": "bottom", "y": 1.02, "xanchor": "left", "x": 0},
margin={"l": 10, "r": 10, "t": 10, "b": 10},
height=480,
yaxis={
"title": "mmol/L",
"range": [y_bottom, y_top],
"gridcolor": colors["grid"],
"zeroline": False,
},
xaxis={"gridcolor": colors["grid"]},
)
return fig
with st.sidebar:
st.header("Instellingen")
hours_label = st.radio("Periode", list(HOURS_OPTIONS.keys()), index=2)
hours = HOURS_OPTIONS[hours_label]
if st.button("Vernieuwen"):
st.cache_data.clear()
try:
glucose, treatments = load_nightscout_data(hours)
except NightscoutConfigError as exc:
st.error(str(exc))
st.stop()
except (requests.RequestException, ValueError) as exc: # netwerk-/API-fouten van Nightscout
st.error(f"Kon geen data ophalen van Nightscout: {exc}")
st.stop()
if glucose.empty:
st.info("Geen glucosemetingen gevonden in deze periode.")
st.stop()
events = classify_treatments(treatments)
tir = summarize_time_in_range(glucose)
latest = glucose.iloc[-1]
previous = glucose.iloc[-2] if len(glucose) > 1 else None
delta = round(latest["mmol"] - previous["mmol"], 1) if previous is not None else None
col1, col2, col3, col4 = st.columns(4)
col1.metric("Actuele glucose", f"{latest['mmol']:.1f} mmol/L", delta=delta, delta_color="off")
col2.metric("Tijd in bereik", f"{tir['in_range']:.0f}%")
col3.metric("Tijd laag", f"{tir['low']:.0f}%")
col4.metric("Tijd hoog", f"{tir['high']:.0f}%")
st.caption(
f"Laatste meting: {latest['timestamp'].strftime('%d-%m-%Y %H:%M')} · "
f"streefbereik {TARGET_LOW}-{TARGET_HIGH} mmol/L"
)
theme_type = "light"
try:
theme_type = st.context.theme.type or "light"
except AttributeError:
pass
colors = PALETTE["dark" if theme_type == "dark" else "light"]
fig = build_chart(
glucose,
events["insulin"],
events["carbs"],
events["site_changes"],
events["sensor_changes"],
colors,
)
st.plotly_chart(fig, width="stretch")
st.caption(
"Driehoekjes = insuline · cirkels = koolhydraten · stippellijn = pompwissel · "
"streepjeslijn = sensorwissel"
)
+62
View File
@@ -0,0 +1,62 @@
"""
Data processing logic voor het glucose-dashboard.
Bewust gescheiden van app.py: Streamlit-UI-code kun je niet zomaar los
unit-testen (die heeft een draaiende Streamlit-context nodig). Door de
logica (classificeren, samenvatten) in een los module te zetten, kun je die
logica wel gewoon met pytest testen, en blijft app.py alleen de UI-laag.
"""
from __future__ import annotations
import pandas as pd
from nightscout_client import SENSOR_CHANGE_EVENT_TYPES, SITE_CHANGE_EVENT_TYPES
# Streefbereik in mmol/L — pas aan naar je eigen behandeldoelen.
TARGET_LOW = 3.9
TARGET_HIGH = 10.0
def classify_treatments(treatments: pd.DataFrame) -> dict[str, pd.DataFrame]:
"""Splits ruwe Nightscout-treatments op in aparte, tijd-gesorteerde subsets.
Een insuline- en koolhydraat-registratie kunnen dezelfde treatment zijn
(bijv. een "Meal Bolus" met zowel insuline als carbs) en verschijnen dan
in beide subsets.
"""
insulin = treatments[treatments["insulin"].fillna(0) > 0]
carbs = treatments[treatments["carbs"].fillna(0) > 0]
site_changes = treatments[treatments["eventType"].isin(SITE_CHANGE_EVENT_TYPES)]
sensor_changes = treatments[treatments["eventType"].isin(SENSOR_CHANGE_EVENT_TYPES)]
return {
"insulin": insulin.reset_index(drop=True),
"carbs": carbs.reset_index(drop=True),
"site_changes": site_changes.reset_index(drop=True),
"sensor_changes": sensor_changes.reset_index(drop=True),
}
def glucose_status(mmol: float, low: float = TARGET_LOW, high: float = TARGET_HIGH) -> str:
"""Classificeer een glucosewaarde t.o.v. het streefbereik."""
if mmol < low:
return "low"
if mmol > high:
return "high"
return "in_range"
def summarize_time_in_range(
glucose: pd.DataFrame, low: float = TARGET_LOW, high: float = TARGET_HIGH
) -> dict[str, float]:
"""Bereken het percentage metingen onder/in/boven het streefbereik."""
if glucose.empty:
return {"low": 0.0, "in_range": 0.0, "high": 0.0}
statuses = glucose["mmol"].apply(lambda v: glucose_status(v, low, high))
counts = statuses.value_counts(normalize=True) * 100
return {
"low": round(counts.get("low", 0.0), 1),
"in_range": round(counts.get("in_range", 0.0), 1),
"high": round(counts.get("high", 0.0), 1),
}
+9
View File
@@ -0,0 +1,9 @@
services:
glucose-dashboard:
image: gitea.zb.local/willem/glucose-dashboard:latest
container_name: glucose-dashboard
restart: unless-stopped
ports:
- "8501:8501"
env_file:
- .env # zie .env.example — NIGHTSCOUT_URL + NIGHTSCOUT_TOKEN of NIGHTSCOUT_API_SECRET
+129
View File
@@ -0,0 +1,129 @@
"""
Nightscout API client.
Praat met een bestaande Nightscout-instantie (CGM-data, insuline/koolhydraten
en pomp-/sensorwissels). Authenticatie loopt via environment variables:
- NIGHTSCOUT_URL: basis-URL van je Nightscout-instantie, bijv.
"https://jouw-nightscout.example.com" (verplicht).
- NIGHTSCOUT_TOKEN: een Nightscout API-token (rol-gebaseerd), OF
- NIGHTSCOUT_API_SECRET: het "API secret" van je instantie (wordt lokaal
ge-hashed, nooit in plaintext verstuurd).
Gebruik precies één van de twee auth-opties; als beide gezet zijn heeft het
token voorrang.
"""
from __future__ import annotations
import hashlib
import os
from datetime import datetime, timedelta, timezone
import pandas as pd
import requests
MGDL_PER_MMOL = 18.0182
INSULIN_EVENT_TYPES = {"Bolus", "Meal Bolus", "Correction Bolus", "Combo Bolus", "SMB"}
SITE_CHANGE_EVENT_TYPES = {"Site Change"}
SENSOR_CHANGE_EVENT_TYPES = {"Sensor Change", "Sensor Start"}
_REQUEST_TIMEOUT = 10
class NightscoutConfigError(RuntimeError):
"""NIGHTSCOUT_URL ontbreekt of is ongeldig."""
def mgdl_to_mmol(value: float) -> float:
"""Reken mg/dL (Nightscout's interne eenheid) om naar mmol/L."""
return round(value / MGDL_PER_MMOL, 1)
def _base_url() -> str:
url = os.environ.get("NIGHTSCOUT_URL", "").strip().rstrip("/")
if not url:
raise NightscoutConfigError(
"NIGHTSCOUT_URL is niet ingesteld. Zet deze environment variable, "
"bijv. https://jouw-nightscout.example.com"
)
return url
def _auth() -> tuple[dict, dict]:
"""Geeft (headers, params) terug voor authenticatie met Nightscout."""
token = os.environ.get("NIGHTSCOUT_TOKEN", "").strip()
api_secret = os.environ.get("NIGHTSCOUT_API_SECRET", "").strip()
if token:
return {}, {"token": token}
if api_secret:
hashed = hashlib.sha1(api_secret.encode("utf-8")).hexdigest()
return {"API-SECRET": hashed}, {}
return {}, {}
def _get(path: str, params: dict) -> list:
headers, auth_params = _auth()
response = requests.get(
f"{_base_url()}{path}",
headers=headers,
params={**params, **auth_params},
timeout=_REQUEST_TIMEOUT,
)
response.raise_for_status()
return response.json()
def fetch_entries(hours: int = 24) -> pd.DataFrame:
"""Haal glucosemetingen (SGV) op van de laatste `hours` uur, in mmol/L.
Geeft een DataFrame terug met kolommen 'timestamp' en 'mmol', oplopend
gesorteerd op tijd.
"""
since = datetime.now(timezone.utc) - timedelta(hours=hours)
# Nightscout geeft resultaten terug in aflopende volgorde; vraag ruim
# genoeg entries op (CGM meet doorgaans elke 5 min) en filter lokaal op
# tijd. Dit is robuuster dan server-side find[]-query's, die niet op elke
# Nightscout-versie identiek werken.
count = max(300, hours * 15)
data = _get("/api/v1/entries/sgv.json", {"count": count})
if not data:
return pd.DataFrame(columns=["timestamp", "mmol"])
df = pd.DataFrame(data)
df["timestamp"] = pd.to_datetime(df["date"], unit="ms", utc=True).dt.tz_localize(None)
df["mmol"] = df["sgv"].astype(float).apply(mgdl_to_mmol)
df = df[df["timestamp"] >= since.replace(tzinfo=None)]
return df[["timestamp", "mmol"]].sort_values("timestamp").reset_index(drop=True)
def fetch_treatments(hours: int = 24) -> pd.DataFrame:
"""Haal behandelingen op van de laatste `hours` uur: insuline, koolhydraten
en pomp-/sensorwissels.
Geeft een DataFrame terug met kolommen 'timestamp', 'eventType',
'insulin' en 'carbs' (de laatste twee NaN waar niet van toepassing).
"""
since = datetime.now(timezone.utc) - timedelta(hours=hours)
count = max(200, hours * 10)
data = _get("/api/v1/treatments.json", {"count": count})
if not data:
return pd.DataFrame(columns=["timestamp", "eventType", "insulin", "carbs"])
df = pd.DataFrame(data)
df["timestamp"] = pd.to_datetime(df["created_at"], utc=True).dt.tz_localize(None)
for col in ("insulin", "carbs"):
if col not in df.columns:
df[col] = pd.NA
df[col] = pd.to_numeric(df[col], errors="coerce")
if "eventType" not in df.columns:
df["eventType"] = ""
df = df[df["timestamp"] >= since.replace(tzinfo=None)]
return (
df[["timestamp", "eventType", "insulin", "carbs"]]
.sort_values("timestamp")
.reset_index(drop=True)
)
+7
View File
@@ -0,0 +1,7 @@
streamlit>=1.38,<2.0
pandas>=2.2,<3.0
requests>=2.32,<3.0
plotly>=5.24,<6.0
python-dotenv>=1.0,<2.0
pytest>=8.0,<9.0
ruff>=0.6,<1.0
+6
View File
@@ -0,0 +1,6 @@
sonar.projectKey=glucose-dashboard
sonar.projectName=Glucose Dashboard
sonar.sources=.
sonar.exclusions=data/**,tests/**
sonar.tests=tests
sonar.python.version=3.12
+74
View File
@@ -0,0 +1,74 @@
import pandas as pd
import pytest
from data_processing import classify_treatments, glucose_status, summarize_time_in_range
@pytest.fixture
def sample_glucose():
return pd.DataFrame(
{
"timestamp": pd.to_datetime(
["2026-07-25 08:00", "2026-07-25 09:00", "2026-07-25 10:00", "2026-07-25 11:00"]
),
"mmol": [3.5, 6.0, 11.0, 7.0], # laag, in bereik, hoog, in bereik
}
)
@pytest.fixture
def sample_treatments():
return pd.DataFrame(
{
"timestamp": pd.to_datetime(
[
"2026-07-25 08:00",
"2026-07-25 12:00",
"2026-07-24 09:00",
"2026-07-20 09:00",
]
),
"eventType": ["Meal Bolus", "Correction Bolus", "Site Change", "Sensor Change"],
"insulin": [4.0, 1.5, None, None],
"carbs": [45.0, None, None, None],
}
)
def test_glucose_status_classifies_correctly():
assert glucose_status(3.5) == "low"
assert glucose_status(6.0) == "in_range"
assert glucose_status(11.0) == "high"
def test_summarize_time_in_range(sample_glucose):
result = summarize_time_in_range(sample_glucose)
assert result == {"low": 25.0, "in_range": 50.0, "high": 25.0}
def test_summarize_time_in_range_handles_empty_input():
empty = pd.DataFrame(columns=["timestamp", "mmol"])
result = summarize_time_in_range(empty)
assert result == {"low": 0.0, "in_range": 0.0, "high": 0.0}
def test_classify_treatments_splits_into_expected_subsets(sample_treatments):
events = classify_treatments(sample_treatments)
assert len(events["insulin"]) == 2
assert set(events["insulin"]["insulin"]) == {4.0, 1.5}
assert len(events["carbs"]) == 1
assert events["carbs"].iloc[0]["carbs"] == 45.0
assert len(events["site_changes"]) == 1
assert events["site_changes"].iloc[0]["eventType"] == "Site Change"
assert len(events["sensor_changes"]) == 1
assert events["sensor_changes"].iloc[0]["eventType"] == "Sensor Change"
def test_classify_treatments_handles_empty_input():
empty = pd.DataFrame(columns=["timestamp", "eventType", "insulin", "carbs"])
events = classify_treatments(empty)
assert all(df.empty for df in events.values())
+98
View File
@@ -0,0 +1,98 @@
from datetime import 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"])
def pd_isna(value):
import pandas as pd
return pd.isna(value)