669 lines
22 KiB
Python
669 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
import datetime
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
import pytest
|
|
|
|
from datastore import DatastoreEngine
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_datastore():
|
|
"""Creates a temporary datastore directory with schemas for testing."""
|
|
temp_dir = Path(tempfile.mkdtemp())
|
|
data_dir = temp_dir / "data"
|
|
data_dir.mkdir(parents=True)
|
|
|
|
# Copy schemas from knowledge-wiki/data/_schema if available, or create them
|
|
schema_src = Path("/Users/moritz/src/infrapuzzle/knowledge-wiki/data/_schema")
|
|
if schema_src.is_dir():
|
|
shutil.copytree(schema_src, data_dir / "_schema")
|
|
|
|
(data_dir / "_meta.json").write_text("{}")
|
|
engine = DatastoreEngine(data_dir=data_dir)
|
|
yield engine
|
|
shutil.rmtree(temp_dir)
|
|
|
|
|
|
def test_nutrition_explicit_logging(temp_datastore: DatastoreEngine):
|
|
"""Verifies logging food with explicit macro values and daily aggregation."""
|
|
res1 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-06T08:30:00+02:00",
|
|
meal="Frühstück",
|
|
payload={
|
|
"item": "Protein Porridge",
|
|
"qty": 300,
|
|
"unit": "g",
|
|
"protein_g": 35.5,
|
|
"carbs_g": 45.0,
|
|
"sugar_g": 5.0,
|
|
"fat_g": 8.0,
|
|
"fiber_g": 6.0,
|
|
"salt_g": 0.5,
|
|
"kcal": 420.0,
|
|
},
|
|
)
|
|
assert res1["status"] == "success"
|
|
assert res1["event"]["id"] is not None
|
|
|
|
res2 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-06T12:30:00+02:00",
|
|
meal="Mittagessen",
|
|
payload={
|
|
"item": "Hähnchen mit Reis",
|
|
"qty": 400,
|
|
"unit": "g",
|
|
"protein_g": 50.0,
|
|
"carbs_g": 60.0,
|
|
"sugar_g": 2.0,
|
|
"fat_g": 10.0,
|
|
"fiber_g": 4.0,
|
|
"salt_g": 1.2,
|
|
"kcal": 550.0,
|
|
},
|
|
)
|
|
assert res2["status"] == "success"
|
|
|
|
summary = temp_datastore.get_summary(subject="moritz", date_str="2026-10-06")
|
|
nut = summary["nutrition"]
|
|
assert nut["item_count"] == 2
|
|
assert nut["total_protein_g"] == 85.5
|
|
assert nut["total_carbs_g"] == 105.0
|
|
assert nut["total_kcal"] == 970.0
|
|
|
|
|
|
def test_nutrition_from_food_reference(temp_datastore: DatastoreEngine):
|
|
"""Verifies that linking a food_id computes macros automatically from foods.per_100."""
|
|
temp_datastore.put_entity(
|
|
collection="foods",
|
|
record={
|
|
"id": "oatly-haferdrink",
|
|
"name": "Oatly Haferdrink",
|
|
"per_100": {
|
|
"kcal": 61.0,
|
|
"protein_g": 1.1,
|
|
"carbs_g": 7.1,
|
|
"sugar_g": 3.4,
|
|
"fat_g": 3.0,
|
|
"fiber_g": 0.8,
|
|
"salt_g": 0.1,
|
|
},
|
|
},
|
|
)
|
|
|
|
# Log 200 ml Oatly Haferdrink
|
|
res = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-06T09:00:00+02:00",
|
|
payload={
|
|
"item": "Hafermilch im Kaffee",
|
|
"food_id": "oatly-haferdrink",
|
|
"qty": 200,
|
|
"unit": "ml",
|
|
},
|
|
)
|
|
assert res["status"] == "success"
|
|
|
|
# Query v_nutrition
|
|
rows = temp_datastore.query("SELECT protein_g, carbs_g, kcal FROM v_nutrition WHERE food_id = 'oatly-haferdrink'")
|
|
assert len(rows) == 1
|
|
assert round(rows[0]["protein_g"], 2) == 2.2
|
|
assert round(rows[0]["carbs_g"], 2) == 14.2
|
|
assert round(rows[0]["kcal"], 0) == 122.0
|
|
|
|
|
|
def test_baby_sleep_open_and_close(temp_datastore: DatastoreEngine):
|
|
"""Verifies open-ended interval sleep logging and stop_event duration calculations."""
|
|
start_time = "2026-10-06T13:00:00+02:00"
|
|
end_time = "2026-10-06T14:30:00+02:00"
|
|
|
|
# 1. Start open sleep
|
|
res_start = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start=start_time,
|
|
payload={"type": "nap", "location": "bett"},
|
|
)
|
|
assert res_start["status"] == "success"
|
|
assert res_start["event"]["end"] is None
|
|
|
|
# Verify open status
|
|
summary = temp_datastore.get_summary(subject="mila", date_str="2026-10-06")
|
|
assert "open_sleep" in summary
|
|
assert summary["open_sleep"]["sleep_type"] == "nap"
|
|
|
|
# 2. Prevent overlapping second open sleep
|
|
with pytest.raises(ValueError, match="already has an open sleep session"):
|
|
temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-06T13:30:00+02:00",
|
|
payload={"type": "nap"},
|
|
)
|
|
|
|
# 3. Stop sleep
|
|
res_stop = temp_datastore.stop_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
end_time=end_time,
|
|
)
|
|
assert res_stop["status"] == "success"
|
|
assert res_stop["duration_minutes"] == 90
|
|
assert res_stop["duration_formatted"] == "1 h 30 min"
|
|
|
|
# 4. Verify aggregated summary
|
|
summary2 = temp_datastore.get_summary(subject="mila", date_str="2026-10-06")
|
|
slp = summary2["sleep"]
|
|
assert slp["nap_count"] == 1
|
|
assert slp["nap_sleep_minutes"] == 90
|
|
assert slp["total_sleep_hours"] == 1.5
|
|
|
|
|
|
def test_soft_delete_event(temp_datastore: DatastoreEngine):
|
|
"""Verifies that soft-deleted events are excluded from views."""
|
|
res = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-06T07:00:00+02:00",
|
|
payload={"item": "Snack", "qty": 50, "unit": "g", "protein_g": 5.0, "kcal": 100.0},
|
|
)
|
|
event_id = res["event"]["id"]
|
|
|
|
# Verify present in summary
|
|
sum1 = temp_datastore.get_summary(subject="moritz", date_str="2026-10-06")
|
|
assert sum1["nutrition"]["total_kcal"] == 100.0
|
|
|
|
# Delete
|
|
del_res = temp_datastore.delete_event(event_id)
|
|
assert del_res["status"] == "success"
|
|
|
|
# Verify removed from view
|
|
sum2 = temp_datastore.get_summary(subject="moritz", date_str="2026-10-06")
|
|
assert sum2["nutrition"]["total_kcal"] == 0.0
|
|
|
|
|
|
def test_places_spatial_radius_search_both_engines(temp_datastore: DatastoreEngine):
|
|
"""Verifies distance calculation using both DuckDB spatial extension and pure-SQL fallback macro."""
|
|
temp_datastore.put_entity(
|
|
collection="places",
|
|
record={
|
|
"id": "spitalgarten-regensburg",
|
|
"name": "Spitalgarten Regensburg",
|
|
"category": "biergarten",
|
|
"lat": 49.0223,
|
|
"lon": 12.0978,
|
|
"address": "St.-Katharinen-Platz 1, 93059 Regensburg",
|
|
"google_maps_url": "https://maps.app.goo.gl/spitalgarten",
|
|
"tags": ["biergarten", "spielplatz"],
|
|
},
|
|
)
|
|
temp_datastore.put_entity(
|
|
collection="places",
|
|
record={
|
|
"id": "walhalla-donaustauf",
|
|
"name": "Walhalla",
|
|
"category": "sightseeing",
|
|
"lat": 49.0315,
|
|
"lon": 12.2245,
|
|
"address": "Walhallastraße 48, 93093 Donaustauf",
|
|
"google_maps_url": "https://maps.app.goo.gl/walhalla",
|
|
"tags": ["ausflug", "kultur"],
|
|
},
|
|
)
|
|
|
|
# 1. Test with spatial extension enabled
|
|
places_1k = temp_datastore.find_places_near(lat=49.0205, lon=12.0975, radius_m=1000.0)
|
|
assert len(places_1k) == 1
|
|
assert places_1k[0]["id"] == "spitalgarten-regensburg"
|
|
assert 150.0 < places_1k[0]["distance_meters"] < 300.0
|
|
|
|
# 2. Test fallback macro by forcing has_spatial_extension = False
|
|
temp_datastore.has_spatial_extension = False
|
|
places_fallback_1k = temp_datastore.find_places_near(lat=49.0205, lon=12.0975, radius_m=1000.0)
|
|
assert len(places_fallback_1k) == 1
|
|
assert places_fallback_1k[0]["id"] == "spitalgarten-regensburg"
|
|
assert 150.0 < places_fallback_1k[0]["distance_meters"] < 300.0
|
|
# Differences between Spheroid (WGS84 ellipsoid) and Haversine (sphere) are < 3m
|
|
assert abs(places_1k[0]["distance_meters"] - places_fallback_1k[0]["distance_meters"]) < 5.0
|
|
|
|
|
|
def test_garmin_daily_and_daily_balance(temp_datastore: DatastoreEngine):
|
|
"""Verifies that Garmin daily summaries join with nutrition to calculate caloric deficit/surplus deterministically."""
|
|
# 1. Log nutrition intake on 2026-10-01
|
|
temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-01T12:30:00+02:00",
|
|
payload={"item": "Mittagessen", "qty": 400, "unit": "g", "protein_g": 60.0, "kcal": 800.0},
|
|
)
|
|
temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-01T19:30:00+02:00",
|
|
payload={"item": "Abendessen", "qty": 500, "unit": "g", "protein_g": 80.0, "kcal": 1346.0},
|
|
)
|
|
|
|
# 2. Add Garmin daily summary for 2026-10-01
|
|
garmin_dir = temp_datastore.data_dir / "garmin" / "daily"
|
|
garmin_dir.mkdir(parents=True, exist_ok=True)
|
|
garmin_file = garmin_dir / "2026-10.jsonl"
|
|
with open(garmin_file, "a", encoding="utf-8") as f:
|
|
f.write(
|
|
json.dumps(
|
|
{
|
|
"calendarDate": "2026-10-01",
|
|
"totalKilocalories": 2890,
|
|
"activeKilocalories": 650,
|
|
"bmrKilocalories": 2240,
|
|
"totalSteps": 10500,
|
|
"restingHeartRate": 51,
|
|
}
|
|
)
|
|
+ "\n"
|
|
)
|
|
|
|
# 3. Reload data into datastore
|
|
temp_datastore._load_all_data()
|
|
|
|
# 4. Query v_daily_balance
|
|
res = temp_datastore.query("SELECT * FROM v_daily_balance WHERE log_date = '2026-10-01'")
|
|
assert len(res) == 1
|
|
row = res[0]
|
|
assert row["subject"] == "moritz"
|
|
assert row["intake_kcal"] == 2146.0 # 800 + 1346
|
|
assert row["protein_g"] == 140.0 # 60 + 80
|
|
assert row["expenditure_kcal"] == 2890
|
|
assert row["active_kcal"] == 650
|
|
assert row["rmr_kcal"] == 2240
|
|
assert row["steps"] == 10500
|
|
assert row["balance_kcal"] == -744.0 # 2146 - 2890
|
|
assert row["balance_status"] == "Defizit"
|
|
|
|
|
|
def test_add_event_with_since(temp_datastore: DatastoreEngine):
|
|
"""Verifies that relative offsets like '15m' calculate start time deterministically in Berlin timezone."""
|
|
from zoneinfo import ZoneInfo
|
|
berlin_tz = ZoneInfo("Europe/Berlin")
|
|
|
|
before = datetime.datetime.now(berlin_tz)
|
|
res = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
since="15m",
|
|
payload={"type": "nap", "location": "bett"},
|
|
)
|
|
after = datetime.datetime.now(berlin_tz)
|
|
|
|
assert res["status"] == "success"
|
|
event = res["event"]
|
|
assert event["end"] is None
|
|
|
|
start_dt = datetime.datetime.fromisoformat(event["start"])
|
|
assert start_dt.tzinfo is not None
|
|
# Verify Berlin timezone offset (+02:00 in CEST or +01:00 in CET)
|
|
assert start_dt.utcoffset() == before.utcoffset()
|
|
|
|
# The logged start must be ~15 minutes before the test run
|
|
expected_approx = before - datetime.timedelta(minutes=15)
|
|
assert abs((start_dt - expected_approx).total_seconds()) < 5
|
|
|
|
# Test error if both start and since are provided
|
|
with pytest.raises(ValueError, match="Cannot specify both 'since' and 'start'"):
|
|
temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-06T12:00:00+02:00",
|
|
since="10m",
|
|
payload={"item": "Snack", "qty": 10},
|
|
)
|
|
|
|
# Test invalid since format
|
|
with pytest.raises(ValueError, match="Unrecognized duration format"):
|
|
temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
since="invalid_format",
|
|
payload={"item": "Snack", "qty": 10},
|
|
)
|
|
|
|
|
|
def test_sleep_over_midnight_attribution(temp_datastore: DatastoreEngine):
|
|
"""Verifies that overnight sleep (Wed evening -> Thu morning) stays attributed to Wednesday."""
|
|
# Sleep starts Wednesday 19:15
|
|
res_start = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-07T19:15:00+02:00",
|
|
payload={"type": "night", "location": "bett"},
|
|
)
|
|
assert res_start["status"] == "success"
|
|
|
|
# Sleep ends Thursday 07:00
|
|
res_stop = temp_datastore.stop_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
end_time="2026-10-08T07:00:00+02:00",
|
|
)
|
|
assert res_stop["status"] == "success"
|
|
# Duration: 11 hours 45 min = 705 minutes
|
|
assert res_stop["duration_minutes"] == 705
|
|
|
|
# Check v_sleep view: log_date must be the day it STARTED (Wednesday 2026-10-07)
|
|
rows = temp_datastore.query("SELECT log_date, duration_minutes, sleep_type FROM v_sleep WHERE subject = 'mila'")
|
|
assert len(rows) == 1
|
|
assert rows[0]["log_date"] == "2026-10-07"
|
|
assert rows[0]["sleep_type"] == "night"
|
|
assert rows[0]["duration_minutes"] == 705
|
|
|
|
# Check daily summary for Wednesday: must contain the full 11.75 hours night sleep
|
|
wed_summary = temp_datastore.get_summary(subject="mila", date_str="2026-10-07")
|
|
assert "sleep" in wed_summary
|
|
assert wed_summary["sleep"]["night_sleep_minutes"] == 705
|
|
assert wed_summary["sleep"]["total_sleep_hours"] == 11.75
|
|
|
|
# Check daily summary for Thursday: must NOT contain Wednesday night's sleep
|
|
thu_summary = temp_datastore.get_summary(subject="mila", date_str="2026-10-08")
|
|
assert "sleep" not in thu_summary or thu_summary["sleep"] is None
|
|
|
|
|
|
def test_duplicate_suppression_nutrition(temp_datastore: DatastoreEngine):
|
|
"""Verifies that duplicate intake events within 15 minutes are suppressed idempotently."""
|
|
start_time = "2026-10-09T08:00:00+02:00"
|
|
payload = {
|
|
"item": "Protein Porridge",
|
|
"qty": 300,
|
|
"unit": "g",
|
|
"protein_g": 35.0,
|
|
"kcal": 400.0,
|
|
}
|
|
|
|
# 1. First logging
|
|
res1 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start=start_time,
|
|
payload=payload,
|
|
)
|
|
assert res1["status"] == "success"
|
|
assert res1.get("duplicate") is not True
|
|
first_id = res1["event"]["id"]
|
|
|
|
# 2. Duplicate logging 5 minutes later
|
|
res2 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-09T08:05:00+02:00",
|
|
payload=payload,
|
|
)
|
|
assert res2["status"] == "success"
|
|
assert res2.get("duplicate") is True
|
|
assert res2["event"]["id"] == first_id
|
|
|
|
# Verify DuckDB summary: only 1 item counted
|
|
sum1 = temp_datastore.get_summary(subject="moritz", date_str="2026-10-09")
|
|
assert sum1["nutrition"]["item_count"] == 1
|
|
assert sum1["nutrition"]["total_kcal"] == 400.0
|
|
|
|
# 3. Explicit bypass with allow_duplicate=True
|
|
res3 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-09T08:06:00+02:00",
|
|
payload=payload,
|
|
allow_duplicate=True,
|
|
)
|
|
assert res3["status"] == "success"
|
|
assert res3.get("duplicate") is not True
|
|
assert res3["event"]["id"] != first_id
|
|
|
|
# Verify DuckDB summary: 2 items counted
|
|
sum2 = temp_datastore.get_summary(subject="moritz", date_str="2026-10-09")
|
|
assert sum2["nutrition"]["item_count"] == 2
|
|
assert sum2["nutrition"]["total_kcal"] == 800.0
|
|
|
|
# 4. Different item within 15 minutes is NOT suppressed
|
|
res4 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-09T08:07:00+02:00",
|
|
payload={"item": "Apfel", "qty": 150, "kcal": 80.0},
|
|
)
|
|
assert res4.get("duplicate") is not True
|
|
|
|
# 5. Same item outside 15 minutes window is NOT suppressed
|
|
res5 = temp_datastore.add_event(
|
|
kind="nutrition.intake",
|
|
subject="moritz",
|
|
actor="agent:koch",
|
|
start="2026-10-09T09:30:00+02:00",
|
|
payload=payload,
|
|
)
|
|
assert res5.get("duplicate") is not True
|
|
|
|
|
|
def test_duplicate_suppression_sleep(temp_datastore: DatastoreEngine):
|
|
"""Verifies that starting the same sleep session within 15 minutes returns existing open session."""
|
|
res1 = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-09T13:00:00+02:00",
|
|
payload={"type": "nap", "location": "bett"},
|
|
)
|
|
assert res1["status"] == "success"
|
|
first_id = res1["event"]["id"]
|
|
|
|
# Repeat 3 minutes later: should return existing open session without throwing ValueError
|
|
res2 = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-09T13:03:00+02:00",
|
|
payload={"type": "nap", "location": "bett"},
|
|
)
|
|
assert res2["status"] == "success"
|
|
assert res2.get("duplicate") is True
|
|
assert res2["event"]["id"] == first_id
|
|
|
|
# Stop session
|
|
temp_datastore.stop_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
end_time="2026-10-09T14:00:00+02:00",
|
|
)
|
|
|
|
|
|
def test_cli_kind_flag_and_positional(temp_datastore: DatastoreEngine):
|
|
"""Verifies CLI accepts both positional and --kind flag for add and stop commands."""
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
script_path = Path(__file__).resolve().parent.parent / "scripts" / "datastore.py"
|
|
data_dir = str(temp_datastore.data_dir)
|
|
|
|
# 1. Add using --kind flag
|
|
proc_add = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script_path),
|
|
"add",
|
|
"--kind",
|
|
"nutrition.intake",
|
|
"--subject",
|
|
"moritz",
|
|
"--payload",
|
|
'{"item": "CLI Snack", "qty": 50, "kcal": 100}',
|
|
"--start",
|
|
"2026-10-09T15:00:00+02:00",
|
|
],
|
|
env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert proc_add.returncode == 0
|
|
data = json.loads(proc_add.stdout)
|
|
assert data["status"] == "success"
|
|
assert data["event"]["kind"] == "nutrition.intake"
|
|
|
|
# 2. Add open sleep using positional kind
|
|
proc_sleep_start = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script_path),
|
|
"add",
|
|
"sleep.session",
|
|
"--subject",
|
|
"mila",
|
|
"--payload",
|
|
'{"type": "nap"}',
|
|
"--start",
|
|
"2026-10-09T15:00:00+02:00",
|
|
],
|
|
env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert proc_sleep_start.returncode == 0
|
|
|
|
# 3. Stop sleep using --kind flag
|
|
proc_sleep_stop = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script_path),
|
|
"stop",
|
|
"--kind",
|
|
"sleep.session",
|
|
"--subject",
|
|
"mila",
|
|
"--end",
|
|
"2026-10-09T16:00:00+02:00",
|
|
],
|
|
env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert proc_sleep_stop.returncode == 0
|
|
stop_data = json.loads(proc_sleep_stop.stdout)
|
|
assert stop_data["status"] == "success"
|
|
assert stop_data["duration_minutes"] == 60
|
|
|
|
|
|
def test_auto_close_stale_sleep_session(temp_datastore: DatastoreEngine):
|
|
"""Verifies that an open sleep session from yesterday is auto-closed when a new sleep session is logged."""
|
|
# 1. Open night sleep session yesterday at 19:15
|
|
res_night = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-08T19:15:00+02:00",
|
|
payload={"type": "night", "location": "bett"},
|
|
)
|
|
assert res_night["status"] == "success"
|
|
night_id = res_night["event"]["id"]
|
|
|
|
# 2. Add midday nap today at 12:35 without closing the night session first
|
|
res_nap = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-09T12:35:00+02:00",
|
|
payload={"type": "nap", "location": "bett"},
|
|
)
|
|
assert res_nap["status"] == "success"
|
|
nap_id = res_nap["event"]["id"]
|
|
assert nap_id != night_id
|
|
|
|
# 3. Query the night session to verify it was auto-closed with estimated=True and end=07:00
|
|
rows = temp_datastore.query(f"SELECT id, start, \"end\", estimated, note FROM events WHERE id = '{night_id}'")
|
|
assert len(rows) == 1
|
|
night_row = rows[0]
|
|
assert night_row["end"] is not None
|
|
assert "07:00:00" in str(night_row["end"])
|
|
assert night_row["estimated"] is True
|
|
assert "Auto-beendet" in str(night_row["note"])
|
|
|
|
|
|
def test_auto_close_stale_sleep_session_on_bounded_nap(temp_datastore: DatastoreEngine):
|
|
"""Verifies that logging a bounded interval today auto-closes a forgotten night session from yesterday."""
|
|
res_night = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-08T19:15:00+02:00",
|
|
payload={"type": "night", "location": "bett"},
|
|
)
|
|
night_id = res_night["event"]["id"]
|
|
|
|
# Add bounded nap from 15:20 to 15:35
|
|
res_nap = temp_datastore.add_event(
|
|
kind="sleep.session",
|
|
subject="mila",
|
|
actor="agent:hebamme",
|
|
start="2026-10-09T15:20:00+02:00",
|
|
end="2026-10-09T15:35:00+02:00",
|
|
payload={"type": "nap", "location": "bett"},
|
|
)
|
|
assert res_nap["status"] == "success"
|
|
|
|
# Verify night session was closed
|
|
rows = temp_datastore.query(f"SELECT id, \"end\", estimated FROM events WHERE id = '{night_id}'")
|
|
assert rows[0]["end"] is not None
|
|
assert rows[0]["estimated"] is True
|
|
|
|
|
|
def test_cli_rejects_invalid_kind(temp_datastore: DatastoreEngine):
|
|
"""Verifies CLI rejects boolean words or strings without dots as kind."""
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
script_path = Path(__file__).resolve().parent.parent / "scripts" / "datastore.py"
|
|
data_dir = str(temp_datastore.data_dir)
|
|
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(script_path),
|
|
"add",
|
|
"true",
|
|
"--subject",
|
|
"moritz",
|
|
"--payload",
|
|
'{"item": "Snack", "qty": 100}',
|
|
],
|
|
env=dict(os.environ, OPENCLAW_DATA_DIR=data_dir),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
assert proc.returncode != 0
|
|
assert "Invalid kind 'true'" in proc.stderr
|
|
|
|
|
|
|
|
|