feat: import Kuchnia Wikinga catering deliveries into the food bank

Pulls the day's delivered dishes and their real per-portion macros out of
panel.kuchniavikinga.pl, so a catering meal is one keypress instead of five
retyped numbers.

A delivered meal is not an eaten meal, so nothing is ever logged unattended:
dishes land in the *curated* bank, and entries are written only after an
explicit confirmation. Auto-logging would let the gate satisfy its own
checkpoint from a delivery note, and would record what the courier dropped off
rather than what was eaten.

Three surfaces:
  - `diet-guard kuchnia [--log] [--yes]`
  - the lock screen's "Today's delivery" button, which prefills the meal form
    dish by dish (the user still clicks Log & Continue)
  - a warm-up after each logged meal, on the existing background thread

All event-driven; no new systemd timer. The two automatic paths go through
`refresh_delivery_once`, guarded to one fetch per day.

The panel is an undocumented white-labelled Dietly API; the endpoint map was
recovered from its JS bundles and confirmed against live responses. Notes,
including the traps, are in docs/kuchnia-wikinga.md.

Load-bearing details:

* Macros are per portion and `weight` is that portion in grams -- proven, not
  assumed: 4P+4C+9F reproduces the stated calories within ~1% on every meal of
  the captured day. `_energy_is_consistent` re-runs that check per dish and
  drops any that fails; a per-100g mix-up is invisible in a payload and would
  silently skew every logged meal by a factor of several.
* `bank_dishes` compares nutritional fields before calling `add_manual_entry`,
  which restamps `t` unconditionally. Without the compare, every refresh
  republishes the whole curated bank to every peer. Tests assert on call count.
* Duplicate log suppression reads today's synced log, not a local marker:
  `log_meal` mints a fresh uuid, so a re-import after marker loss would merge
  as duplicates on every peer.
* The gate hook is in `on_focus_ready`, not `_cli_gate._should_lock`:
  `gate_is_due()` never reads the food bank, so a refresh there could not
  change the lock decision while adding a third-party round trip to the ~105ms
  `gate --check` fast path.
* `requests` is lazy-imported so the gate's not-due tick does not pay for it.
* Credentials and the cached session cookie live in ~/.config/diet_guard at
  mode 600, outside the synced tree. The credentials file is written by hand,
  never by the package -- the same contract as sync_token.

`_GateDashboard` is merged into `_GateMealFlow`, and both "pull it in from
elsewhere" flows share one `_PullFlows` mixin: pylint's max-parents counts
every MRO entry, so the gate's controller chain had no room for a new class.

Also adds a side-by-side row for the two pull buttons -- stacking them cost
42px and pushed the tab past the viewport at 1366x768.

989 tests, 100% branch coverage, pre-commit clean.
This commit is contained in:
Krzysztof Rudnicki 2026-08-22 23:04:22 +02:00
parent 9752142c28
commit 26f1770090
41 changed files with 3635 additions and 180 deletions

3
.gitignore vendored
View File

@ -17,3 +17,6 @@ htmlcov/
# ci_mirror.sh cached requirements.txt venv (rebuilt on demand)
.ci-mirror-venv/
# Kuchnia Wikinga probe captures: contain account data, never commit.
kuchnia_probe*.json

View File

@ -28,7 +28,7 @@ repos:
args: [--pytest-test-first]
# Shared fakes and fixtures, not test modules. Split out of conftest.py
# and test_sync.py only because those files passed the 250-line cap.
exclude: ^diet_guard/tests/_(tk_fakes(_containers|_canvas)?|gate_layout_probe|gate_focus_probe|gate_measurement|gate_keyboard_probe|gate_fixtures|sync_fakes)\.py$
exclude: ^diet_guard/tests/_(tk_fakes(_containers|_canvas)?|gate_layout_probe|gate_focus_probe|gate_measurement|gate_keyboard_probe|gate_fixtures|sync_fakes|kuchnia_fakes)\.py$
- id: check-ast
- id: mixed-line-ending
args: [--fix=lf]

View File

@ -225,6 +225,9 @@ phone show both.
- Don't strip the `components` field: nothing writes it, but historical
composite entries and the food bank read it, and it is part of the sync wire
format.
- Don't let the catering import write a log entry unattended: a delivered
meal is not an eaten meal, and the gate would satisfy its own checkpoint.
See [docs/kuchnia-wikinga.md](docs/kuchnia-wikinga.md).
- Don't add a dependency without the production install-path check above.
- Don't reintroduce a seal/`chattr +i` on the budget file.
- Don't re-add a Linux embedder target, and don't change the wrapper's port or

View File

@ -2,7 +2,7 @@ name: diet_guard_app
description: "Companion phone app for diet_guard: log meals on the go."
publish_to: 'none'
version: 1.0.52+1
version: 1.0.53+1
environment:
sdk: ^3.12.2

View File

@ -29,6 +29,7 @@ from diet_guard._cli_args import parse_args
from diet_guard._cli_averages import cmd_averages
from diet_guard._cli_gate import cmd_gate
from diet_guard._cli_init import cmd_init
from diet_guard._cli_kuchnia import cmd_kuchnia
from diet_guard._cli_log import ManualMacroArgs, Portion, cmd_ate
from diet_guard._cli_prune import cmd_prune_peers
from diet_guard._cli_sync import cmd_sync
@ -187,6 +188,11 @@ def _dispatch_gate(args: argparse.Namespace) -> int:
# A table rather than an if-chain: the chain grew a return per subcommand and
# tripped ruff's return-count limit at the seventh, and a table also has no
# unreachable trailing branch for the 100%-coverage gate to chase.
def _dispatch_kuchnia(args: argparse.Namespace) -> int:
"""Unpack the kuchnia flags for :func:`cmd_kuchnia`."""
return cmd_kuchnia(_emit, _ask, log=args.log, yes=args.yes)
def _dispatch_prune(args: argparse.Namespace) -> int:
"""Unpack the prune-peers flags for :func:`cmd_prune_peers`."""
return cmd_prune_peers(
@ -203,6 +209,7 @@ _COMMANDS: dict[str, Callable[[argparse.Namespace], int]] = {
"status": lambda _args: _cmd_status(),
"averages": lambda _args: cmd_averages(_emit),
"sync": lambda _args: cmd_sync(_emit),
"kuchnia": _dispatch_kuchnia,
"prune-peers": _dispatch_prune,
"gate": _dispatch_gate,
"undo": lambda _args: _cmd_undo(),

View File

@ -15,6 +15,7 @@ from __future__ import annotations
import argparse
from diet_guard._cli_averages import register_averages_subparser
from diet_guard._cli_kuchnia import register_kuchnia_subparser
from diet_guard._cli_prune import register_prune_subparser
from diet_guard._cli_sync import register_sync_subparser
@ -82,6 +83,7 @@ def parse_args(argv: list[str]) -> argparse.Namespace:
sub.add_parser("status", help="Show today's calories and budget band.")
sub.add_parser("undo", help="Remove today's most recent entry.")
register_averages_subparser(sub)
register_kuchnia_subparser(sub)
register_prune_subparser(sub)
register_sync_subparser(sub)

119
diet_guard/_cli_kuchnia.py Normal file
View File

@ -0,0 +1,119 @@
"""CLI handler for the ``kuchnia`` subcommand.
Fetches today's catering delivery, banks the dishes, and -- only with an
explicit confirmation -- logs them. Dry run by default: the whole point of
this feature is that a *delivered* meal is not an *eaten* meal, so writing
entries is always something the user asks for, never something that happens
because a command ran.
Structured like :mod:`diet_guard._cli_prune`: lazy attributes so registering
the subparser drags no HTTP stack into every other command, ``emit`` passed in
rather than imported, and an int exit code.
"""
from __future__ import annotations
from importlib import import_module
import sys
from typing import TYPE_CHECKING
from diet_guard._kuchnia_spread import assign_slots
from diet_guard._meal_schedule_store import current_schedule
from diet_guard._slots import day_slots
from diet_guard._state import now_local
if TYPE_CHECKING: # pragma: no cover - typing only
import argparse
from collections.abc import Callable
# ``_kuchnia_import`` reaches ``requests``; registering this subparser must not
# make every other command pay for an HTTP stack.
_LAZY_ATTRS = {
"refresh_delivery": ("diet_guard._kuchnia_import", "refresh_delivery"),
"log_dishes": ("diet_guard._kuchnia_log", "log_dishes"),
}
def __getattr__(name: str) -> object:
"""Resolve the deferred import helpers on first attribute access."""
target = _LAZY_ATTRS.get(name)
if target is None:
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)
module_name, attr = target
return getattr(import_module(module_name), attr)
def register_kuchnia_subparser(sub: argparse._SubParsersAction) -> None:
"""Register the ``kuchnia`` subcommand on ``sub``."""
kuchnia = sub.add_parser(
"kuchnia",
help="Import today's Kuchnia Wikinga delivery into the food bank.",
)
kuchnia.add_argument(
"--log",
action="store_true",
help="Also log the dishes as eaten (asks first unless --yes).",
)
kuchnia.add_argument(
"--yes",
action="store_true",
help="Skip the confirmation prompt for --log (for scripting).",
)
def cmd_kuchnia(
emit: Callable[[str], None],
ask: Callable[[str], str],
*,
log: bool,
yes: bool,
) -> int:
"""Import today's delivery; with ``--log``, offer to log it too.
Args:
emit: A one-line output sink.
ask: A prompt function (``_cli._ask``), used only for the confirmation.
log: Offer to write log entries as well as banking the dishes.
yes: Skip the confirmation prompt.
Returns:
0 on success, 1 when the catering panel could not be read.
"""
module = sys.modules[__name__]
# now_local() is the repo's tz-aware clock; its date is the same local
# calendar day the log is keyed by.
today = now_local().date()
dishes, reason = module.refresh_delivery(today)
if reason is not None:
emit(f"catering unavailable: {reason}")
return 1
if not dishes:
emit("no catering delivery today.")
return 0
slotted = assign_slots(dishes, day_slots(current_schedule()))
emit(f"{len(dishes)} dish(es) delivered today, banked:")
for item in slotted:
emit(
f" {item.slot:02d}:00 {item.dish.name} "
f"({item.dish.kcal:g} kcal, {item.dish.grams:g} g)",
)
total = sum(dish.kcal for dish in dishes)
emit(f" total: {total:g} kcal")
if not log:
emit("not logged -- pass --log to record these as eaten.")
return 0
if not yes and ask(f"log all {len(slotted)} as eaten? [y/N] ").strip().lower() != (
"y"
):
emit("nothing logged.")
return 0
written = module.log_dishes(slotted)
if not written:
emit("already logged today; nothing to add.")
return 0
emit(f"logged {len(written)} meal(s).")
return 0

View File

@ -14,9 +14,11 @@ rather than a periodic timer.
from __future__ import annotations
from dataclasses import dataclass
import threading
from typing import TYPE_CHECKING
from diet_guard._foodbank import remember_food
from diet_guard._kuchnia_import import refresh_delivery_once
from diet_guard._meal_schedule_store import current_schedule
from diet_guard._portions import DEFAULT_ITEM_GRAMS, estimate_unit_grams
from diet_guard._resolve import ManualMacros, resolve_nutrition
@ -166,5 +168,27 @@ def cmd_ate(
publish_after_log_detached(
lambda reason: emit(f"logged locally, not yet published ({reason})."),
)
# Warm the catering bank on the same background thread, so tomorrow's
# autocomplete already knows today's dishes. Guarded to one fetch per day
# (`refresh_delivery_once`), because this fires after *every* meal and each
# unguarded refresh is a login plus a three-request walk. Never inline: it
# would add up to the whole catering deadline to an interactive command.
_warm_catering_bank_detached()
print_summary()
return 0
def _warm_catering_bank_detached() -> None:
"""Refresh today's catering bank off-thread, ignoring any failure.
Non-daemon, matching ``publish_after_log_detached``: the process waits for
it at exit while the user already has their output. Failures are silent by
design -- the user asked to log a meal, not to hear about the caterer.
"""
def _warm() -> None:
# A module-global lookup, so the conftest patch on this module's
# attribute takes effect -- and so ruff can see the import is used.
refresh_delivery_once(now_local().date())
threading.Thread(target=_warm, daemon=False).start()

View File

@ -121,3 +121,38 @@ SYNC_TIMEOUT_SECONDS: float = 10.0
#: back to *15s per request* and one hung session refresh stalls the lock
#: window. The journal records five such refresh failures in five days.
INTERACTIVE_TIMEOUT_SECONDS: float = 2.0
# --- Kuchnia Wikinga catering import -------------------------------------
# The panel is a white-labelled Dietly SPA. Its API is undocumented; the
# endpoint map was recovered from the site's JS bundles and confirmed against
# live responses (see docs/kuchnia-wikinga.md).
KUCHNIA_API_BASE: str = "https://panel.kuchniavikinga.pl/api"
# Sent as the ``company-id`` header. The panel wants the company *name*, not a
# numeric id -- confirmed by its own postSideOrders call passing companyName.
KUCHNIA_COMPANY: str = "kuchniavikinga"
# The panel's own launcher tag for a browser panel session.
KUCHNIA_LAUNCHER_TYPE: str = "BROWSER_PANEL"
# Credentials the user writes by hand, mode 600: e-mail on line 1, password on
# line 2. Nothing in this package ever writes it, exactly as with
# SYNC_TOKEN_FILE. Under ~/.config, not DATA_DIR, so it is outside the synced
# tree -- a credential must never leave the machine it was entered on.
KUCHNIA_CREDENTIALS_FILE: Path = (
Path.home() / ".config" / "diet_guard" / "kuchnia_credentials"
)
# The cached SESSION cookie, so a refresh does not re-login every time. Also
# outside the synced tree, for the same reason.
KUCHNIA_SESSION_FILE: Path = (
Path.home() / ".config" / "diet_guard" / "kuchnia_session.json"
)
# One ISO date: the last day whose delivery was fetched. Without it the
# gate and the after-log hook would each pay a full auth + 3-request walk
# every time they fire. The curated bank cannot answer this -- it is keyed
# by dish name, and knowing today's names requires the fetch itself.
# Under ~/.config (not DATA_DIR): a per-device rate limit must not sync.
KUCHNIA_LAST_IMPORT_FILE: Path = (
Path.home() / ".config" / "diet_guard" / "kuchnia_last_import"
)
KUCHNIA_TIMEOUT_SECONDS: float = 8.0
# Whole-walk ceiling. The import is 3-4 sequential requests, so a per-request
# timeout alone would permit ~32s against a slow provider.
KUCHNIA_TOTAL_DEADLINE_SECONDS: float = 12.0

View File

@ -32,12 +32,11 @@ repo's 250-line limit: :mod:`._gatelock_core` provides the shared leaf
widget/field helpers and state (``_GateCore``, ``_GateState``);
:mod:`._gatelock_nutrition` provides the reference->total nutrition maths and
food lookup (``_GateNutrition``); :mod:`._gatelock_mealflow` provides the
submit/log flow, dashboard, and callback-error handling (``_GateMealFlow``);
and :mod:`._gatelock_calendar` provides the second ``ttk.Notebook`` tab --
the budget-adherence calendar, streaks, and YTD tally from
:mod:`._daystatus`, plus budget editing (``_GateCalendar``). ``MealGate``
wires these mixins together, owns the ``gatelock.LockWindow``, and handles
construction, layout, and event binding.
submit/log flow, dashboard, callback errors and the pull-it-in buttons
(``_GateMealFlow`` + ``_PullFlows``); :mod:`._gatelock_calendar` the tab --
the budget-adherence calendar, streaks, YTD tally (:mod:`._daystatus`) and
budget editing (``_GateCalendar``). ``MealGate`` wires these together, owns the
``gatelock.LockWindow``, and handles construction, layout and event binding.
"""
from __future__ import annotations
@ -167,6 +166,7 @@ class MealGate(_GateCalendar):
on_submit=self._on_submit,
on_close=self.close,
on_fetch_sync=self._on_fetch_sync,
on_load_delivery=self._on_load_delivery,
)
self._build()
@ -210,6 +210,7 @@ class MealGate(_GateCalendar):
if surface is None:
return
self._widgets.desc_text.focus_force()
self._autoload_delivery()
def on_close(self) -> None:
"""No hardware/state to release; meal-log writes already happened."""

View File

@ -1,75 +0,0 @@
"""The gate's running "how am I doing today" dashboard.
Split out of :mod:`._gatelock_mealflow` to keep every gate module under the
repo's 250-line limit. ``_GateDashboard`` sits between
:class:`~diet_guard._gatelock_nutrition._GateNutrition` and
:class:`~diet_guard._gatelock_mealflow._GateMealFlow` in the mixin chain and
owns only the *read* side: the prominent calorie headline and the detail panel
beneath it. Nothing here writes state, so it can be recomputed at any point
in the meal flow.
"""
from __future__ import annotations
from diet_guard._budget import BudgetError, daily_budget
from diet_guard._budget_derived import protein_target_g
from diet_guard._gatelock_nutrition import _GateNutrition
from diet_guard._state import (
entry_kcal,
)
from diet_guard._state_today import (
today_entries,
today_total_kcal,
today_total_macros,
)
# How many recent meals the dashboard lists.
_DASHBOARD_ROWS = 5
# ISO timestamp "YYYY-MM-DDTHH:MM:SS": HH:MM is characters 11..16.
_TIME_SLICE = slice(11, 16)
# Width a meal description is truncated to in the dashboard.
_DASH_DESC_WIDTH = 22
__all__ = ["_GateDashboard"]
class _GateDashboard(_GateNutrition):
"""The calorie headline and detail panel for the day so far."""
def _refresh_dashboard(self) -> None:
"""Recompute the prominent calorie headline and the detail panel."""
self._vars.cal_headline.set(self._cal_headline_text())
self._vars.dashboard.set(self._dashboard_text())
def _cal_headline_text(self) -> str:
"""Return the big calories-today line: consumed, target, and remaining."""
consumed = today_total_kcal()
try:
budget = daily_budget()
except (BudgetError, OSError):
return f"{consumed:g} kcal today"
return (
f"{consumed:g} / {budget:g} kcal · {round(budget - consumed, 1):g} left"
)
def _dashboard_text(self) -> str:
"""Build the detail panel: recent meals, then macros and protein."""
lines = ["── Today ───────────────────────────────"]
entries = today_entries()
if entries:
for entry in entries[-_DASHBOARD_ROWS:]:
clock = str(entry.get("time", ""))[_TIME_SLICE]
desc = str(entry.get("desc", "?"))[:_DASH_DESC_WIDTH]
lines.append(
f" {clock:>5} {desc:<{_DASH_DESC_WIDTH}} "
f"{entry_kcal(entry):>5.0f} kcal",
)
else:
lines.append(" (nothing logged yet today)")
protein, carbs, fat = today_total_macros()
lines.append(f" macros so far: P{protein:g} C{carbs:g} F{fat:g} g")
target = protein_target_g()
if target is not None:
left = round(target - protein, 1)
lines.append(f" protein {protein:g} / {target:g} g ({left:g} g left)")
return "\n".join(lines)

View File

@ -0,0 +1,240 @@
"""The lock screen's two "pull it in from elsewhere" flows.
Both let the user satisfy a slot without retyping a meal, and both run their
network call off the Tk thread: **Fetch from sync** for a meal a peer already
logged, and **Today's delivery** for the day's catering.
A mixin on :class:`~diet_guard._gatelock_mealflow._GateMealFlow` rather than a
link in the controller chain -- pylint caps that chain's depth and the gate is
at the limit -- and a separate module so both stay under the 250-line cap.
**Neither logs anything by itself.** The catering fetch fills the form and
stops; the user still clicks "Log & Continue". Auto-submitting would let the
gate satisfy its own checkpoint from a delivery note, which is the one thing
this feature is designed not to do.
"""
from __future__ import annotations
from importlib import import_module
import queue
import sys
from typing import TYPE_CHECKING
from diet_guard._gate import due_slots
from diet_guard._gatelock_fetch import FETCH_POLL_MS, start_fetch
from diet_guard._gatelock_kuchnia import start_delivery_fetch
from diet_guard._gatelock_nutrition import _GateNutrition
from diet_guard._kuchnia_spread import dish_field_values, dishes_in_slot_order
from diet_guard._state import now_local
from diet_guard._sync_refresh import pull_peer_logs
if TYPE_CHECKING: # pragma: no cover - typing only
from collections.abc import Callable
from diet_guard._gatelock_kuchnia import DeliveryResult
from diet_guard._kuchnia_parse import Dish
# ``_kuchnia_import`` reaches ``requests``. The gate's far more common not-due
# tick imports this module and must not pay ~78ms for an HTTP stack it never
# touches, so the name resolves on first access -- the same PEP 562 hook
# ``_cli_gate`` and ``_cli_prune`` use, which also keeps ``patch.object`` working.
_LAZY_ATTRS = ("refresh_delivery", "refresh_delivery_once")
def __getattr__(name: str) -> object:
"""Resolve the deferred catering import on first attribute access."""
if name not in _LAZY_ATTRS:
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)
return getattr(import_module("diet_guard._kuchnia_import"), name)
class _PullFlows(_GateNutrition):
"""The lock screen's two "pull it in from elsewhere" buttons.
Both let the user satisfy a slot without retyping a meal, and both run
their network call off the Tk thread. One mixin rather than two because
pylint caps the controller chain's depth and every MRO entry counts.
"""
#: Slots still to be filled. Declared on the controller this mixin is
#: mixed into; repeated here so mypy can type the reconcile below.
_pending: list[int]
#: The in-flight sync fetch's queue, or None when none is running.
#: Doubles as the poll's own guard: a stray poll after completion finds
#: None and stops rather than blocking on an empty queue forever.
_fetch_result: queue.Queue[str | None] | None = None
#: The in-flight catering fetch's queue, or None when none is running.
_delivery_result: queue.Queue[DeliveryResult] | None = None
#: Dishes fetched but not yet offered, in slot order. Popped one at a
#: time so each still passes through the normal submit path.
_delivery_pending: tuple[Dish, ...] = ()
#: True when the running fetch came from the button rather than the
#: window opening. An automatic fetch must not talk over the gate's own
#: prompt to report that the caterer had nothing today.
_delivery_asked: bool = False
#: The in-flight sync fetch's queue, or None when none is running.
#: Slots still to be filled; owned by the controller this mixes into.
# Supplied by ``_GateMealFlow``, which mixes this class in. Declared so the
# contract is explicit and pylint can see it: the dependency runs both ways
# (the flows drive the meal form, the form hosts the flows) and the
# ancestor cap leaves no room for another chain link.
_set_status: Callable[..., None]
_clear_inputs: Callable[[], None]
_refresh_dashboard: Callable[[], None]
_unlock: Callable[[str], None]
# -- catering delivery ------------------------------------------------------
def _on_load_delivery(self) -> None:
"""Fetch today's catering and offer it as prefilled entries.
Refuses in demo mode: a synthetic window must never satisfy a real
checkpoint, and there is no reason to send real credentials from one.
Fails closed -- any failure leaves the lock exactly as it was, and a
second click while a fetch is in flight is ignored.
"""
if self.demo_mode:
self._set_status("Loading a delivery is only available on the real lock.")
return
if self._delivery_result is not None:
return
self._delivery_asked = True
self._set_status("Loading today's delivery…")
# Through the module object so the lazy hook resolves it, and so a
# test's ``patch.object`` still wins.
self._delivery_result = start_delivery_fetch(
sys.modules[__name__].refresh_delivery,
now_local().date(),
)
self.root.after(FETCH_POLL_MS, self._poll_delivery)
def _poll_delivery(self) -> None:
"""On the Tk thread: pick up the catering worker's result."""
result = self._delivery_result
if result is None:
return
try:
outcome = result.get_nowait()
except queue.Empty:
self.root.after(FETCH_POLL_MS, self._poll_delivery)
return
self._delivery_result = None
if outcome.reason is not None:
if self._delivery_asked:
self._set_status(f"{outcome.reason} — still locked.", error=True)
return
if not outcome.dishes:
if self._delivery_asked:
self._set_status("No catering delivery today.")
return
self._delivery_pending = dishes_in_slot_order(outcome.dishes)
self._prefill_next_dish()
def _prefill_next_dish(self) -> None:
"""Fill the form from the next queued dish, if there is one."""
if not self._delivery_pending:
return
dish, *rest = self._delivery_pending
self._delivery_pending = tuple(rest)
self._clear_inputs()
self._set_desc(dish.name)
grams, macros = dish_field_values(dish)
self._widgets.amount_entry.insert(0, grams)
for entry, value in zip(self._macro_entries(), macros, strict=True):
entry.insert(0, value)
self._state.source = "kuchnia wikinga"
self._refresh_projection()
remaining = len(self._delivery_pending)
suffix = f" ({remaining} more to go)" if remaining else ""
self._set_status(f"Loaded: {dish.name}{suffix} — check, then Log & Continue.")
def _autoload_delivery(self) -> None:
"""Start a guarded catering fetch as the lock opens.
The user is stopped anyway, so the dishes are waiting by the time they
reach for them and the spinner sits *inside* the lock. Deliberately not
in ``_cli_gate._should_lock``: ``gate_is_due()`` never reads the food
bank, so a refresh there could not change the lock decision while
adding a third-party round trip to the ~105ms ``gate --check`` path.
Silent by design -- an outage must not greet the user the moment the
window appears, and the button is still there to retry.
"""
if self.demo_mode or self._delivery_result is not None:
return
self._delivery_asked = False
self._delivery_result = start_delivery_fetch(
sys.modules[__name__].refresh_delivery_once,
now_local().date(),
)
self.root.after(FETCH_POLL_MS, self._poll_delivery)
# -- manual sync ------------------------------------------------------------
def _on_fetch_sync(self) -> None:
"""Pull the shared log on demand and unlock any slots it now satisfies.
For a meal already logged on another device (typically the phone) that
has not propagated here yet: rather than re-entering it to unlock, the
user pulls it in.
The pull is the *narrow* one (:func:`pull_peer_logs`) and runs on a
worker thread -- inline it froze the fullscreen lock for ~18-27s. See
``docs/sync-latency.md``. Fails closed, and a second click while a
fetch is in flight is ignored: that is the reentrancy guard and the
write-race fix in one.
"""
if self.demo_mode:
self._set_status("Fetch from sync is only available on the real lock.")
return
if self._fetch_result is not None:
# Already fetching. Without this a double-click starts a second
# worker, orphans the first result and races two log writes.
return
self._set_status("Fetching from sync…")
self._fetch_result = start_fetch(pull_peer_logs)
self.root.after(FETCH_POLL_MS, self._poll_fetch)
def _poll_fetch(self) -> None:
"""On the Tk thread: pick up the worker's result once it lands."""
result = self._fetch_result
if result is None:
return
try:
reason = result.get_nowait()
except queue.Empty:
self.root.after(FETCH_POLL_MS, self._poll_fetch)
return
self._fetch_result = None
if reason is not None:
self._set_status(f"{reason} — still locked.", error=True)
return
self._reconcile_after_fetch()
def _reconcile_after_fetch(self) -> None:
"""Drop slots a pulled meal now covers; unlock when none remain."""
still_due = set(due_slots())
satisfied_slots = [slot for slot in self._pending if slot not in still_due]
self._refresh_dashboard()
if not satisfied_slots:
self._set_status("No new meals found in sync.")
return
self._pending = [slot for slot in self._pending if slot in still_due]
if not self._pending:
self._unlock("Synced from another device")
return
self._clear_inputs()
self._refresh_slot_header()
count = len(satisfied_slots)
meal_word = "meal" if count == 1 else "meals"
self._set_status(f"Pulled {count} {meal_word} — next meal, please.")
self._widgets.desc_text.focus_set()

View File

@ -0,0 +1,90 @@
"""Off-thread catering fetch for the lock screen's "Load delivery" button.
Its own module because :mod:`diet_guard._gatelock_mealflow` is already at the
repo's 250-line cap, and because the worker contract here is worth stating
once, plainly.
Modelled on :mod:`diet_guard._gatelock_fetch`, but it cannot reuse
:func:`~diet_guard._gatelock_fetch.start_fetch`: that is typed
``Callable[[], str | None] -> Queue[str | None]`` and this fetch returns
*dishes*. Squeezing them through by stashing results on the instance from the
worker would reintroduce exactly the cross-thread state that module's docstring
exists to forbid, so the pattern is copied rather than bent:
* **The worker touches no widget.** Tcl is not thread-safe, and that includes
``after`` -- the worker only ``put``s on a queue, and a poll scheduled on the
Tk thread picks it up. That also makes the completion path testable with no
thread at all: put a result, call the poll, assert.
* **The worker always feeds the queue**, even if the call raises. It has
nowhere to propagate to, so an escaping exception would leave the poll
waiting forever with the button disabled and the user stuck behind the lock.
* **Daemon thread**, so a hung request can never keep the process alive after
the window is gone -- the gate is a ``Type=oneshot`` unit and must exit.
Nothing here logs anything. The fetch only *offers* dishes; writing entries is
a separate, explicitly confirmed step (:mod:`diet_guard._kuchnia_log`).
"""
from __future__ import annotations
from collections.abc import Callable
from contextlib import suppress
import queue
import threading
from typing import TYPE_CHECKING, NamedTuple
if TYPE_CHECKING: # pragma: no cover - typing only
from collections.abc import Sequence
from datetime import date
from diet_guard._kuchnia_parse import Dish
#: What ``_kuchnia_import.refresh_delivery`` looks like to this module.
RefreshFn = Callable[["date"], tuple["Sequence[Dish]", "str | None"]]
class DeliveryResult(NamedTuple):
"""What the worker hands back: the dishes, or why there are none."""
dishes: tuple[Dish, ...]
reason: str | None
#: Reported when the worker itself fails unexpectedly. ``refresh_delivery`` is
#: already fail-closed, so reaching this means a bug rather than an outage.
UNEXPECTED = "catering fetch failed (unexpected error)"
def start_delivery_fetch(
refresh: RefreshFn,
day: date,
) -> queue.Queue[DeliveryResult]:
"""Run ``refresh(day)`` on a daemon thread; return its result queue.
Args:
refresh: A ``refresh_delivery``-shaped callable returning
``(dishes, reason)``. Passed in rather than imported so the gate
can be tested without an HTTP stack.
day: The delivery date to fetch.
Returns:
A queue that will receive exactly one :class:`DeliveryResult`.
"""
result: queue.Queue[DeliveryResult] = queue.Queue(maxsize=1)
def _worker() -> None:
"""Runs OFF the Tk thread. Touches no widget, by construction."""
outcome = DeliveryResult(dishes=(), reason=UNEXPECTED)
# ``refresh`` is already fail-closed, so an exception here is a bug --
# but it must not escape this thread. There is nowhere to propagate to,
# so it would surface only as an unraisable-thread warning while the
# poll waits forever and the user stays stuck behind the lock.
# ``suppress`` rather than a broad ``except``; the pre-seeded UNEXPECTED
# outcome is what reaches the window.
with suppress(Exception):
dishes, reason = refresh(day)
outcome = DeliveryResult(dishes=tuple(dishes), reason=reason)
result.put(outcome)
threading.Thread(target=_worker, daemon=True).start()
return result

View File

@ -116,15 +116,32 @@ def _build_actions(body: tk.Frame, callbacks: GateCallbacks) -> None:
callbacks.on_submit,
).pack(pady=(XS, XS))
# Manual pull for a meal already logged on another device (the phone) but
# not yet propagated to this machine -- saves re-typing it to unlock.
# Both "pull it in from elsewhere" actions share one row: stacking them
# cost 42px, which pushed the tab past the viewport on a 1366x768 screen
# (test_gate_fits_the_primary_screen). Side by side they fit.
pulls = tk.Frame(body, bg=BG)
pulls.pack(pady=(0, XS))
# A meal already logged on another device (the phone) but not yet
# propagated here -- saves re-typing it to unlock.
make_button(
body,
pulls,
_COLORS,
"⟳ Fetch from sync",
callbacks.on_fetch_sync,
ButtonStyle(variant="secondary", bold=False),
).pack(pady=(0, XS))
).pack(side="left", padx=(0, XS))
# Today's catering delivery, with its real macros. Loading only *offers*
# the dishes -- each still has to be checked and submitted, because a
# delivered meal is not an eaten meal.
make_button(
pulls,
_COLORS,
"🍱 Today's delivery",
callbacks.on_load_delivery,
ButtonStyle(variant="secondary", bold=False),
).pack(side="left")
def build_layout(

View File

@ -11,19 +11,22 @@ calorie/macro dashboard it calls into lives in :mod:`._gatelock_dashboard`.
from __future__ import annotations
import contextlib
import queue
import tkinter as tk
from typing import TYPE_CHECKING
from diet_guard._budget import BudgetError, daily_budget
from diet_guard._budget_derived import protein_target_g
from diet_guard._foodbank import remember_food
from diet_guard._gate import due_slots
from diet_guard._gatelock_dashboard import _GateDashboard
from diet_guard._gatelock_fetch import FETCH_POLL_MS, start_fetch
from diet_guard._gatelock_delivery import _PullFlows
from diet_guard._gatelock_ui import ERR, FG, UNIT_GRAMS
from diet_guard._resolve import lookup_candidates
from diet_guard._slots import slot_label
from diet_guard._state import log_meal
from diet_guard._sync_refresh import pull_peer_logs
from diet_guard._state import entry_kcal, log_meal
from diet_guard._state_today import (
today_entries,
today_total_kcal,
today_total_macros,
)
if TYPE_CHECKING:
from diet_guard._estimator import Nutrition
@ -31,15 +34,17 @@ if TYPE_CHECKING:
# How long the "unlocking..." confirmation lingers before the window tears down.
_UNLOCK_DELAY_MS = 1200
# How many recent meals the dashboard lists.
_DASHBOARD_ROWS = 5
# ISO timestamp "YYYY-MM-DDTHH:MM:SS": HH:MM is characters 11..16.
_TIME_SLICE = slice(11, 16)
# Width a meal description is truncated to in the dashboard.
_DASH_DESC_WIDTH = 22
class _GateMealFlow(_GateDashboard):
class _GateMealFlow(_PullFlows):
"""Submit/lookup/log flow for a logged food."""
#: The in-flight fetch's result queue, or None when no fetch is running.
#: Doubles as the poll's own guard: a stray poll after completion finds
#: None and stops rather than blocking on an empty queue forever.
_fetch_result: queue.Queue[str | None] | None = None
# -- slot walk ------------------------------------------------------------
def _clear_inputs(self) -> None:
@ -164,74 +169,43 @@ class _GateMealFlow(_GateDashboard):
self._set_status(f"{logged} — all meals logged, unlocking…")
self.root.after(_UNLOCK_DELAY_MS, self.close)
# -- manual sync ------------------------------------------------------------
def _refresh_dashboard(self) -> None:
"""Recompute the prominent calorie headline and the detail panel."""
self._vars.cal_headline.set(self._cal_headline_text())
self._vars.dashboard.set(self._dashboard_text())
def _on_fetch_sync(self) -> None:
"""Pull the shared log on demand and unlock any slots it now satisfies.
For a meal already logged on another device (typically the phone) that
has not yet propagated here: rather than re-entering it to unlock, the
user pulls it in.
The pull is the *narrow* one (:func:`pull_peer_logs`) and runs on a
worker thread, so the window stays live. It used to run the full tick
inline and froze the fullscreen lock for its whole duration (~18-27s
measured). The button's question is the same as the gate's -- "did a
peer log this slot?" -- and :meth:`_reconcile_after_fetch` reads only
``due_slots()``, so the budget and food banks it dropped were never
consulted here; the full tick still runs once the window closes.
Measured at ~92ms against the live remote.
Fails closed: a failure leaves the lock untouched. A second click while
a fetch is in flight is ignored, which is the reentrancy guard and the
write-race fix in one -- two workers would race two log writes and
orphan one result.
"""
if self.demo_mode:
self._set_status("Fetch from sync is only available on the real lock.")
return
if self._fetch_result is not None:
# Already fetching. Without this a double-click starts a second
# worker, orphans the first result and races two log writes.
return
self._set_status("Fetching from sync…")
self._fetch_result = start_fetch(pull_peer_logs)
self.root.after(FETCH_POLL_MS, self._poll_fetch)
def _poll_fetch(self) -> None:
"""On the Tk thread: pick up the worker's result once it lands."""
result = self._fetch_result
if result is None:
return
def _cal_headline_text(self) -> str:
"""Return the big calories-today line: consumed, target, and remaining."""
consumed = today_total_kcal()
try:
reason = result.get_nowait()
except queue.Empty:
self.root.after(FETCH_POLL_MS, self._poll_fetch)
return
self._fetch_result = None
if reason is not None:
self._set_status(f"{reason} — still locked.", error=True)
return
self._reconcile_after_fetch()
budget = daily_budget()
except (BudgetError, OSError):
return f"{consumed:g} kcal today"
return (
f"{consumed:g} / {budget:g} kcal · {round(budget - consumed, 1):g} left"
)
def _reconcile_after_fetch(self) -> None:
"""Drop pending slots a freshly pulled meal now covers; unlock if none."""
still_due = set(due_slots())
satisfied_slots = [slot for slot in self._pending if slot not in still_due]
self._refresh_dashboard()
if not satisfied_slots:
self._set_status("No new meals found in sync.")
return
self._pending = [slot for slot in self._pending if slot in still_due]
if not self._pending:
self._unlock("Synced from another device")
return
self._clear_inputs()
self._refresh_slot_header()
count = len(satisfied_slots)
meal_word = "meal" if count == 1 else "meals"
self._set_status(f"Pulled {count} {meal_word} — next meal, please.")
self._widgets.desc_text.focus_set()
def _dashboard_text(self) -> str:
"""Build the detail panel: recent meals, then macros and protein."""
lines = ["── Today ───────────────────────────────"]
entries = today_entries()
if entries:
for entry in entries[-_DASHBOARD_ROWS:]:
clock = str(entry.get("time", ""))[_TIME_SLICE]
desc = str(entry.get("desc", "?"))[:_DASH_DESC_WIDTH]
lines.append(
f" {clock:>5} {desc:<{_DASH_DESC_WIDTH}} "
f"{entry_kcal(entry):>5.0f} kcal",
)
else:
lines.append(" (nothing logged yet today)")
protein, carbs, fat = today_total_macros()
lines.append(f" macros so far: P{protein:g} C{carbs:g} F{fat:g} g")
target = protein_target_g()
if target is not None:
left = round(target - protein, 1)
lines.append(f" protein {protein:g} / {target:g} g ({left:g} g left)")
return "\n".join(lines)
def on_callback_error(self) -> None:
"""Surface an unexpected callback error without dropping the grab."""

View File

@ -84,3 +84,4 @@ class GateCallbacks:
on_submit: Callable[[], None]
on_close: Callable[[], None]
on_fetch_sync: Callable[[], None]
on_load_delivery: Callable[[], None]

View File

@ -0,0 +1,165 @@
"""HTTP access to the Kuchnia Wikinga panel.
Auth is a cookie session, not a bearer token: ``POST auth/login`` with a
**form-urlencoded** body sets a ``SESSION`` cookie that every later call
carries. The cookie is cached (:mod:`diet_guard._kuchnia_config`) so a refresh
normally skips the login round trip entirely, and an expired cookie costs one
extra request rather than a failed import.
``requests`` is imported lazily. The gate's not-due tick imports the CLI and
must not pay ~78ms for an HTTP stack it never touches -- the same reason
:mod:`diet_guard._estimator_off` and :mod:`diet_guard._cli_prune` defer theirs.
Call sites reach through ``sys.modules[__name__]`` so a test's ``patch.object``
still wins over the lazy hook.
"""
from __future__ import annotations
from importlib import import_module
import sys
import time
from typing import TYPE_CHECKING
from diet_guard._constants import (
KUCHNIA_API_BASE,
KUCHNIA_COMPANY,
KUCHNIA_LAUNCHER_TYPE,
KUCHNIA_TIMEOUT_SECONDS,
KUCHNIA_TOTAL_DEADLINE_SECONDS,
)
from diet_guard._kuchnia_config import (
SESSION_COOKIE,
clear_session_cookie,
load_session_cookie,
read_credentials,
save_session_cookie,
)
from diet_guard._kuchnia_errors import KuchniaError
if TYPE_CHECKING: # pragma: no cover - typing only
import requests
_LAZY_ATTRS = ("requests",)
_COOKIE_DOMAIN = "panel.kuchniavikinga.pl"
def __getattr__(name: str) -> object:
"""Resolve the deferred ``requests`` import on first attribute access."""
if name not in _LAZY_ATTRS:
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)
return import_module(name)
class PanelSession:
"""A logged-in panel session with a whole-walk deadline.
The deadline exists because the import is several sequential requests: a
per-request timeout alone would let a slow provider stack them into half a
minute, and this runs on paths a user is waiting on.
"""
def __init__(
self, deadline_seconds: float = KUCHNIA_TOTAL_DEADLINE_SECONDS
) -> None:
"""Build an unauthenticated session with the panel's required headers."""
module = sys.modules[__name__]
self._session = module.requests.Session()
# No session-level Content-Type: the login body is form-urlencoded and
# a sticky JSON default would mislabel it. ``requests`` sets it per call.
self._session.headers.update(
{
"company-id": KUCHNIA_COMPANY,
"X-Launcher-Type": KUCHNIA_LAUNCHER_TYPE,
"User-Agent": "diet_guard/1.0 (personal diet tracker)",
"Accept": "application/json",
},
)
self._expires_at = time.monotonic() + deadline_seconds
def _check_deadline(self) -> None:
"""Raise once the whole-walk budget is spent."""
if time.monotonic() >= self._expires_at:
msg = "catering panel too slow (deadline exceeded)"
raise KuchniaError(msg)
def _request(
self,
method: str,
path: str,
data: dict[str, str] | None = None,
) -> requests.Response:
"""Perform one request, translating transport errors into KuchniaError.
``data`` is the only body this client ever sends (the form-urlencoded
login), so it is spelled out rather than taken as ``**kwargs`` -- a
concrete signature types better and keeps the surface honest.
"""
self._check_deadline()
module = sys.modules[__name__]
try:
return self._session.request(
method,
f"{KUCHNIA_API_BASE}/{path}",
timeout=KUCHNIA_TIMEOUT_SECONDS,
data=data,
)
except module.requests.RequestException as exc:
msg = f"catering panel unreachable: {exc}"
raise KuchniaError(msg) from exc
def _login(self) -> None:
"""Authenticate with the stored credentials and cache the cookie."""
username, password = read_credentials()
response = self._request(
"POST",
"auth/login",
data={"username": username, "password": password},
)
if not response.ok:
msg = f"catering login rejected (HTTP {response.status_code})"
raise KuchniaError(msg)
cookie = self._session.cookies.get(SESSION_COOKIE)
if cookie is None:
msg = "catering login returned no session cookie"
raise KuchniaError(msg)
save_session_cookie(cookie)
def authenticate(self) -> None:
"""Attach the cached cookie, logging in only when there is not one."""
cached = load_session_cookie()
if cached is None:
self._login()
return
self._session.cookies.set(SESSION_COOKIE, cached, domain=_COOKIE_DOMAIN)
def get_json(self, path: str) -> object:
"""GET ``path`` and decode it, re-logging in once if the cookie expired.
A cached cookie that has expired is indistinguishable from a good one
until it is used, so an auth failure is retried exactly once with fresh
credentials. A second failure is real.
Args:
path: Path below the API base, without a leading slash.
Returns:
The decoded JSON body.
Raises:
KuchniaError: On a non-OK status or an undecodable body.
"""
response = self._request("GET", path)
if response.status_code in {401, 403}:
clear_session_cookie()
self._login()
response = self._request("GET", path)
if not response.ok:
msg = f"catering panel returned HTTP {response.status_code} for {path}"
raise KuchniaError(msg)
try:
return response.json()
except ValueError as exc:
msg = f"catering panel returned a non-JSON body for {path}"
raise KuchniaError(msg) from exc

View File

@ -0,0 +1,130 @@
"""Credentials and the cached session cookie for the catering import.
Sole reader of :data:`~diet_guard._constants.KUCHNIA_CREDENTIALS_FILE` and
:data:`~diet_guard._constants.KUCHNIA_SESSION_FILE`, so the test-suite redirect
in ``conftest._isolate_state`` has exactly one place to patch.
Two files, both mode 600, both under ``~/.config/diet_guard/`` -- deliberately
*outside* ``DATA_DIR``, which syncs. A password and a live session cookie must
not travel to another device.
The credentials file is **written by the user, never by this package**, which
is the same contract as ``sync_token``: nothing here can leak a password into a
backup or a log because nothing here has ever held one on disk.
"""
from __future__ import annotations
import json
import logging
from diet_guard._constants import (
KUCHNIA_CREDENTIALS_FILE,
KUCHNIA_LAST_IMPORT_FILE,
KUCHNIA_SESSION_FILE,
)
from diet_guard._kuchnia_errors import KuchniaError
_logger = logging.getLogger(__name__)
#: The only cookie the panel issues on login. It sets no ``XSRF-TOKEN``, so the
#: CSRF echo its own JavaScript performs is a no-op for this account -- the
#: session cookie alone authenticates.
SESSION_COOKIE = "SESSION"
_EXPECTED_LINES = 2
_SETUP_HINT = (
"create it with:\n"
" install -m 600 /dev/null {path}\n"
" printf '%s\\n%s\\n' 'you@example.com' 'your-password' > {path}"
)
def read_credentials() -> tuple[str, str]:
"""Return the panel ``(username, password)``.
Returns:
The credentials read from disk.
Raises:
KuchniaError: When the file is missing, empty, or not two lines. The
message carries the setup command, mirroring ``_read_token``.
"""
path = KUCHNIA_CREDENTIALS_FILE
if not path.exists():
msg = f"no catering credentials at {path} -- " + _SETUP_HINT.format(path=path)
raise KuchniaError(msg)
lines = [
line.strip()
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
if len(lines) < _EXPECTED_LINES:
msg = f"{path} needs two lines (e-mail, then password); found {len(lines)}"
raise KuchniaError(msg)
return lines[0], lines[1]
def load_session_cookie() -> str | None:
"""Return the cached session cookie, or None when there is no usable cache.
Never raises: a missing, unreadable or corrupt cache simply means "log in
again", which is always a valid next step.
"""
if not KUCHNIA_SESSION_FILE.exists():
return None
try:
cached = json.loads(KUCHNIA_SESSION_FILE.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
_logger.warning("Catering session cache %s is unreadable", KUCHNIA_SESSION_FILE)
return None
value = cached.get(SESSION_COOKIE) if isinstance(cached, dict) else None
return value if isinstance(value, str) and value else None
def save_session_cookie(value: str) -> None:
"""Cache ``value``, created mode 600 before it holds anything.
``touch(mode=0o600)`` first, then write, then atomically replace: writing
and chmod'ing afterwards would leave a live session cookie readable by
every local user for the duration of the write.
"""
KUCHNIA_SESSION_FILE.parent.mkdir(parents=True, exist_ok=True)
temp = KUCHNIA_SESSION_FILE.with_suffix(".json.tmp")
temp.touch(mode=0o600)
temp.write_text(json.dumps({SESSION_COOKIE: value}), encoding="utf-8")
temp.replace(KUCHNIA_SESSION_FILE)
def clear_session_cookie() -> None:
"""Drop the cached cookie so the next refresh logs in afresh."""
KUCHNIA_SESSION_FILE.unlink(missing_ok=True)
def last_import_day() -> str:
"""Return the ISO date whose delivery was last fetched, or "" if none.
Never raises: an unreadable marker just means "fetch again", which is
always safe.
"""
if not KUCHNIA_LAST_IMPORT_FILE.exists():
return ""
try:
return KUCHNIA_LAST_IMPORT_FILE.read_text(encoding="utf-8").strip()
except OSError:
return ""
def record_import_day(day: str) -> None:
"""Remember that ``day``'s delivery has been fetched.
Best-effort: this is a rate limit, not state anyone depends on. A failure
to write it costs one extra fetch, which is strictly better than failing an
import that already succeeded.
"""
try:
KUCHNIA_LAST_IMPORT_FILE.parent.mkdir(parents=True, exist_ok=True)
KUCHNIA_LAST_IMPORT_FILE.write_text(day, encoding="utf-8")
except OSError:
_logger.warning("Could not record the catering import day")

View File

@ -0,0 +1,18 @@
"""The one exception the Kuchnia Wikinga importer raises.
Its own module so :mod:`diet_guard._kuchnia_client`, which raises it, and the
callers that catch it do not have to import each other -- the same shape as
:mod:`diet_guard._sync_errors`.
Deliberately a single class rather than a hierarchy. Every failure mode here
(no credentials, login rejected, a non-200, an unparsable body, a blown
deadline) is handled identically by every caller: report the reason and carry
on without the catering data. A richer taxonomy would cost branch-coverage
tests for a distinction nothing acts on.
"""
from __future__ import annotations
class KuchniaError(Exception):
"""Raised when the catering panel cannot be read."""

View File

@ -0,0 +1,136 @@
"""Fetch the day's catering dishes and bank them.
Holds the module's single fail-closed boundary, :func:`refresh_delivery`: every
:class:`~diet_guard._kuchnia_errors.KuchniaError` is turned into a reason string
here, so no caller can be broken by a catering outage. That is the same idiom
as ``pull_peer_logs`` and ``publish_after_log``.
Imports go **only** to the curated bank
(:mod:`diet_guard._foodbank_manual`). The derived bank is rewritten from the
log on every meal, so anything written there would vanish at the next entry.
The idempotency guard is load-bearing and easy to get wrong.
``add_manual_entry`` restamps ``t`` unconditionally and rewrites the whole
file, and the sync merge derives each record's HLC from that ``t`` -- so
calling it for unchanged dishes would republish the *entire* curated bank to
every peer on every refresh. :func:`bank_dishes` therefore compares the
nutritional fields first and writes only what actually differs.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._foodbank_manual import add_manual_entry, read_manual_bank
from diet_guard._kuchnia_client import PanelSession
from diet_guard._kuchnia_config import last_import_day, record_import_day
from diet_guard._kuchnia_errors import KuchniaError
from diet_guard._kuchnia_orders import fetch_dishes
if TYPE_CHECKING: # pragma: no cover - typing only
from collections.abc import Sequence
from datetime import date
from diet_guard._kuchnia_parse import Dish
#: Fields compared to decide whether a bank record changed. ``t`` is excluded
#: deliberately -- it is the edit stamp, not part of the record's meaning, and
#: including it would make every comparison differ.
_COMPARED = ("desc", "kcal", "protein_g", "carbs_g", "fat_g", "grams")
def dish_to_record(dish: Dish) -> dict[str, object]:
"""Return the curated-bank record for ``dish``.
``count`` is 0: the bank's count ranks foods by how often they were
*eaten*, and a delivered dish has not been eaten yet. Logging it later
bumps the derived bank's own count normally.
"""
return {
"desc": dish.name,
"kcal": dish.kcal,
"protein_g": dish.protein_g,
"carbs_g": dish.carbs_g,
"fat_g": dish.fat_g,
"grams": dish.grams,
"count": 0,
}
def _matches(existing: dict[str, object], record: dict[str, object]) -> bool:
"""Return True when the banked record already says the same thing."""
return all(existing.get(key) == record.get(key) for key in _COMPARED)
def bank_dishes(dishes: Sequence[Dish]) -> int:
"""Add dishes to the curated bank, skipping ones already banked unchanged.
Args:
dishes: The dishes to bank.
Returns:
How many records were actually written.
"""
bank = read_manual_bank()
written = 0
for dish in dishes:
record = dish_to_record(dish)
existing = bank.get(dish.name.strip().casefold())
if isinstance(existing, dict) and _matches(existing, record):
continue
add_manual_entry(dish.name, record)
written += 1
return written
def refresh_delivery(day: date) -> tuple[list[Dish], str | None]:
"""Fetch ``day``'s dishes and bank them, never raising.
Args:
day: The delivery date to fetch.
Returns:
``(dishes, None)`` on success, or ``([], reason)`` when the catering
panel could not be read. Callers surface the reason and carry on.
"""
try:
session = PanelSession()
session.authenticate()
dishes = fetch_dishes(session, day)
except KuchniaError as exc:
return [], str(exc)
bank_dishes(dishes)
return dishes, None
def refresh_delivery_once(day: date) -> tuple[list[Dish], str | None]:
"""Fetch ``day``'s dishes unless today's have already been fetched.
The automatic triggers -- the gate opening, and the after-log publish --
can both fire many times a day, and each :func:`refresh_delivery` is a full
login plus a three-request walk against a third party. This wrapper makes
them cheap: the first call of the day pays, the rest are free.
The marker is a date on disk rather than a look in the curated bank. The
bank is keyed by dish name, so it can answer "is this dish known?" but
never "has today's delivery been fetched?" -- and knowing today's names
requires the very fetch being avoided.
The CLI deliberately does **not** use this: an explicit ``diet-guard
kuchnia`` should always go and look.
Args:
day: The delivery date to fetch.
Returns:
``([], None)`` when today was already fetched -- "nothing new", not an
error. Otherwise whatever :func:`refresh_delivery` returns.
"""
stamp = day.isoformat()
if last_import_day() == stamp:
return [], None
dishes, reason = refresh_delivery(day)
if reason is None:
# Only a clean fetch counts: an outage must not suppress the retry.
record_import_day(stamp)
return dishes, reason

View File

@ -0,0 +1,75 @@
"""Turn confirmed catering dishes into log entries.
Nothing here runs without an explicit confirmation from the user -- that is the
whole design of this feature. A delivered meal is not an eaten meal: if the
importer wrote entries unattended, the gate would auto-satisfy the exact
checkpoint it exists to enforce, and the log would record what the courier
dropped off rather than what was eaten.
Duplicate suppression works off **today's log**, not a local "already
imported" marker. ``log_meal`` mints a fresh uuid per entry, so a re-import
after a lost marker would merge as duplicates on every peer; comparing against
the synced log instead is robust to marker loss *and* correct across devices --
if the phone already logged the dish, the PC sees it and skips.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._estimator import Nutrition
from diet_guard._state import log_meal
from diet_guard._state_today import today_entries
if TYPE_CHECKING: # pragma: no cover - typing only
from collections.abc import Sequence
from diet_guard._kuchnia_spread import SlottedDish
#: Marks entries this importer created, so the provenance is visible in the log
#: and in the food bank's ``source`` column.
SOURCE = "kuchnia wikinga"
def _already_logged(entries: Sequence[dict[str, object]], name: str, slot: int) -> bool:
"""Return True when today's log already holds this dish in this slot."""
wanted = name.strip().casefold()
return any(
str(entry.get("desc", "")).strip().casefold() == wanted
and entry.get("slot") == slot
for entry in entries
)
def log_dishes(chosen: Sequence[SlottedDish]) -> list[str]:
"""Log each chosen dish, skipping any already logged today in that slot.
Args:
chosen: The dishes the user confirmed, already paired with slots.
Returns:
The descriptions actually written, in the order they were logged.
"""
entries = today_entries()
written: list[str] = []
for item in chosen:
dish = item.dish
if _already_logged(entries, dish.name, item.slot):
continue
log_meal(
dish.name,
Nutrition(
kcal=dish.kcal,
protein_g=dish.protein_g,
carbs_g=dish.carbs_g,
fat_g=dish.fat_g,
grams=dish.grams,
source=SOURCE,
),
item.slot,
)
# Reflect the write locally so two identical dishes in one batch do not
# both land: today_entries() was read once, before the loop.
entries = [*entries, {"desc": dish.name, "slot": item.slot}]
written.append(dish.name)
return written

View File

@ -0,0 +1,95 @@
"""Walk the panel's endpoints to today's delivered dishes.
Three calls, and the shape of the walk is not what the endpoint names suggest:
1. ``company/customer/order/active-ids`` -> ``[orderId]``.
2. ``company/customer/order/{orderId}`` -> the order, which **embeds every
delivery** with its date. No enumeration call is needed, but the embedded
``deliveryMeals`` carry ids only -- no dish names, no macros.
3. ``company/general/menus/delivery/{deliveryId}/new`` -> the day's actual
menu, with names and per-portion nutrition.
Step 3 is the important correction: the obvious-looking
``.../deliveries/{id}/details`` **404s** (and 400s when keyed by date), which
was confirmed against three separate delivery days. The menu endpoint is keyed
by the opaque ``deliveryId``, never by the date.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from diet_guard._kuchnia_errors import KuchniaError
from diet_guard._kuchnia_parse import parse_menu
if TYPE_CHECKING: # pragma: no cover - typing only
from datetime import date
from diet_guard._kuchnia_client import PanelSession
from diet_guard._kuchnia_parse import Dish
def _active_order_id(session: PanelSession) -> object | None:
"""Return the first active order id, or None when there is no active order."""
payload = session.get_json("company/customer/order/active-ids")
if not isinstance(payload, list) or not payload:
return None
return payload[0]
def _delivery_id_for(payload: object, wanted: str) -> object | None:
"""Return the delivery id whose date is ``wanted``.
Args:
payload: The decoded order.
wanted: An ISO ``YYYY-MM-DD`` date.
Returns:
The delivery id, or None when that day has no (undeleted) delivery.
"""
if not isinstance(payload, dict):
return None
deliveries = payload.get("deliveries")
if not isinstance(deliveries, list):
return None
for delivery in deliveries:
if not isinstance(delivery, dict):
continue
# ``deleted`` marks a cancelled day; it still appears in the list.
if delivery.get("deleted"):
continue
if str(delivery.get("date")) == wanted:
return delivery.get("deliveryId")
return None
def fetch_dishes(session: PanelSession, day: date) -> list[Dish]:
"""Return the dishes delivered on ``day``.
Args:
session: An authenticated session.
day: The delivery date to fetch.
Returns:
The day's dishes, empty when nothing is delivered that day.
Raises:
KuchniaError: When the panel cannot be read.
"""
order_id = _active_order_id(session)
if order_id is None:
return []
order = session.get_json(f"company/customer/order/{order_id}")
delivery_id = _delivery_id_for(order, day.isoformat())
if delivery_id is None:
return []
menu = session.get_json(f"company/general/menus/delivery/{delivery_id}/new")
dishes = parse_menu(menu)
if not dishes:
# A delivery exists but nothing survived parsing: either the menu is
# not published yet, or its units failed the consistency check. Both
# are "no data", not a crash -- but they are worth distinguishing from
# "no delivery" in the caller's message.
msg = f"catering menu for {day.isoformat()} has no usable dishes"
raise KuchniaError(msg)
return dishes

View File

@ -0,0 +1,155 @@
"""Turn the panel's menu payload into ``Dish`` records.
Written against a real capture (``scripts/probe_kuchnia.py``), not against the
field names guessed from the site's JavaScript. The shape that matters:
{"deliveryMenuMeal": [
{"mealName": "Śniadanie", # the slot label
"menuMealName": "Marchewkowe pancakes, ...", # the actual dish
"mealPriority": 1, # eating order, 1..N
"nutrition": {"weight": 270.0, "calories": 435.0, "protein": 25.86,
"carbohydrate": 54.64, "fat": 12.01, ...}},
...]}
**Macros are per portion and ``weight`` is that portion in grams.** Verified
arithmetically rather than assumed: 4·protein + 4·carbs + 9·fat reproduces the
stated ``calories`` within ~1% on every meal of the captured day, and the day
sums to 2055 kcal against the plan's declared 2000. Had they been per-100 g,
every imported meal would have been silently wrong by a factor of several --
which is why :func:`parse_menu` drops a dish whose numbers do not hold together
instead of importing a plausible-looking lie.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from diet_guard._coerce import as_float
#: Atwater factors. Used only as a sanity check on the *units*, never to
#: recompute a value the panel already states.
_KCAL_PER_G_PROTEIN = 4.0
_KCAL_PER_G_CARB = 4.0
_KCAL_PER_G_FAT = 9.0
#: How far the macro-derived energy may drift from the stated calories before
#: the dish is treated as unusable. Generous: the panel rounds, and fibre and
#: polyols carry energy these three factors ignore. A per-100 g mix-up would
#: miss by a factor of 2-5, nowhere near this band.
_ENERGY_TOLERANCE = 0.35
#: A portion outside this range is not a meal. Guards against a unit change
#: (kg for g) rather than against unusual food.
_MIN_GRAMS = 1.0
_MAX_GRAMS = 5000.0
@dataclass(frozen=True)
class Dish:
"""One delivered dish, with its portion macros.
Mirrors :class:`diet_guard._estimator.Nutrition`'s convention: every macro
is the total *for this portion*, not per 100 g.
"""
name: str
kcal: float
protein_g: float
carbs_g: float
fat_g: float
grams: float
priority: int
slot_label: str
def _energy_is_consistent(
kcal: float, protein: float, carbs: float, fat: float
) -> bool:
"""Return True when the macros roughly account for the stated calories.
This is a *unit* check, not a nutrition check: it catches a payload whose
macros and calories are quoted on different bases (per portion vs per
100 g), which is the one failure mode that would corrupt every import
without looking wrong.
"""
if kcal <= 0:
return False
derived = (
_KCAL_PER_G_PROTEIN * protein + _KCAL_PER_G_CARB * carbs + _KCAL_PER_G_FAT * fat
)
return abs(derived - kcal) <= _ENERGY_TOLERANCE * kcal
def _parse_meal(meal: object, fallback_priority: int) -> Dish | None:
"""Build one :class:`Dish`, or None when the entry is unusable.
Args:
meal: One ``deliveryMenuMeal`` element.
fallback_priority: Position to use when ``mealPriority`` is missing.
Returns:
The dish, or None when it lacks a name or fails the unit check.
"""
if not isinstance(meal, dict):
return None
nutrition = meal.get("nutrition")
if not isinstance(nutrition, dict):
return None
# ``menuMealName`` is the dish ("Kaszotto grzybowe..."); ``mealName`` is the
# slot label ("Kolacja"). The dish is what belongs in the food bank -- the
# label would collapse every Monday dinner onto one bank entry.
name = str(meal.get("menuMealName") or "").strip()
if not name:
return None
kcal = as_float(nutrition.get("calories"))
protein = as_float(nutrition.get("protein"))
carbs = as_float(nutrition.get("carbohydrate"))
fat = as_float(nutrition.get("fat"))
grams = as_float(nutrition.get("weight"))
if not _MIN_GRAMS <= grams <= _MAX_GRAMS:
return None
if not _energy_is_consistent(kcal, protein, carbs, fat):
return None
raw_priority = meal.get("mealPriority")
priority = (
raw_priority
if isinstance(raw_priority, int) and not isinstance(raw_priority, bool)
else fallback_priority
)
return Dish(
name=name,
kcal=kcal,
protein_g=protein,
carbs_g=carbs,
fat_g=fat,
grams=grams,
priority=priority,
slot_label=str(meal.get("mealName") or "").strip(),
)
def parse_menu(payload: object) -> list[Dish]:
"""Extract the day's dishes from a ``menus/delivery/{id}/new`` response.
Unusable entries are skipped rather than raising: a single malformed dish
must not cost the user the other four.
Args:
payload: The decoded response body.
Returns:
The dishes, in payload order (:mod:`diet_guard._kuchnia_spread` sorts
them by priority).
"""
if not isinstance(payload, dict):
return []
meals: Any = payload.get("deliveryMenuMeal")
if not isinstance(meals, list):
return []
parsed = (_parse_meal(meal, index + 1) for index, meal in enumerate(meals))
return [dish for dish in parsed if dish is not None]

View File

@ -0,0 +1,86 @@
"""Assign catering dishes to meal slots, in the provider's own order.
The panel returns each dish with a ``mealPriority`` (1..N: Śniadanie, II
śniadanie, Obiad, Podwieczorek, Kolacja) -- the order the catering intends them
to be eaten in. That beats inferring an order from the dish names, and it
beats spreading them arithmetically, so this module sorts by priority and then
maps position onto the user's configured slots.
Counts rarely match: a 5-meal plan against the default 4 slots means two dishes
share the first slot. The mapping is ``i * S // N``, which is monotonic
non-decreasing, keeps the first dish on the first slot and the last on the
last, and doubles up the earliest slots rather than dropping anything.
Integer ``//`` only, never ``round()``. That is repo convention for slot
arithmetic (``docs/meal-schedule.md``): Python's banker's rounding and Dart's
half-away-from-zero disagree on ``.5``, and a slot one device offers while the
other does not is a checkpoint that can never be satisfied. The parity risk is
theoretical *here* -- the phone has no importer and so no mirror of this code
(see the plan's §8) -- but the rule is cheap to keep and expensive to
rediscover.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, NamedTuple
from diet_guard._meal_schedule_store import current_schedule
from diet_guard._slots import day_slots
if TYPE_CHECKING:
from collections.abc import Sequence
from diet_guard._kuchnia_parse import Dish
class SlottedDish(NamedTuple):
"""A dish paired with the meal-slot hour it was assigned to."""
dish: Dish
slot: int
def assign_slots(dishes: Sequence[Dish], slots: Sequence[int]) -> list[SlottedDish]:
"""Pair each dish with a slot hour, following the provider's meal order.
Args:
dishes: The day's dishes, in any order (sorted here by ``priority``).
slots: The configured slot hours, ascending, e.g. ``(8, 12, 16, 20)``.
Returns:
One :class:`SlottedDish` per dish, ordered by the provider's priority.
Empty when either input is empty -- a day with no delivery and a
schedule with no slots are both "nothing to assign", not errors.
"""
if not dishes or not slots:
return []
# Ties (two dishes sharing a priority) fall back to name so the result is
# deterministic; a bank import that reshuffles between runs would look like
# a change and re-stamp every record.
ordered = sorted(dishes, key=lambda dish: (dish.priority, dish.name))
count = len(ordered)
span = len(slots)
return [
SlottedDish(dish=dish, slot=slots[index * span // count])
for index, dish in enumerate(ordered)
]
def dishes_in_slot_order(dishes: tuple[Dish, ...]) -> tuple[Dish, ...]:
"""Return ``dishes`` ordered by the provider's own meal priority.
So the dish offered for the slot being filled is the one the catering
actually intends for it, rather than whatever order the payload arrived in.
"""
slotted = assign_slots(dishes, day_slots(current_schedule()))
return tuple(item.dish for item in slotted)
def dish_field_values(dish: Dish) -> tuple[str, tuple[str, ...]]:
"""Return ``(portion_grams, (kcal, protein, carbs, fat))`` as form strings.
Formatted here rather than in the widget code so the numeric formatting is
covered by a test that needs no display.
"""
macros = (dish.kcal, dish.protein_g, dish.carbs_g, dish.fat_g)
return f"{dish.grams:g}", tuple(f"{value:g}" for value in macros)

View File

@ -92,6 +92,7 @@ def probe(screen_w: int = 1366, screen_h: int = 768) -> RingReport:
on_submit=_noop,
on_close=_noop,
on_fetch_sync=_noop,
on_load_delivery=_noop,
)
widgets = build_layout(
notebook, _make_vars(populated=True), callbacks, demo_mode=False

View File

@ -149,6 +149,7 @@ def measure(
on_submit=_noop,
on_close=_noop,
on_fetch_sync=_noop,
on_load_delivery=_noop,
)
widgets = build_layout(
notebook, _make_vars(populated=populated), callbacks, demo_mode=False

View File

@ -0,0 +1,145 @@
"""Shared fakes for the catering-import tests.
Not a ``test_`` module: these are fixtures-by-hand imported by the
``test_kuchnia_*`` files, and pytest must not collect them as tests. The repo
already excludes this naming pattern from the ``name-tests-test`` hook.
The fake stands in for ``requests`` at the module seam
(``patch.object(_kuchnia_client, "requests", ...)``), which is the same shape
``_estimator_off``'s tests use and the reason both modules resolve the name
through ``sys.modules`` rather than binding it at import.
"""
from __future__ import annotations
import json
from types import SimpleNamespace
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
class FakeResponse:
"""The slice of ``requests.Response`` the client actually touches."""
def __init__(
self, status: int = 200, payload: object = None, *, text: str = ""
) -> None:
"""Build a response with ``status`` and either JSON or raw text."""
self.status_code = status
self._payload = payload
self.text = text or (json.dumps(payload) if payload is not None else "")
@property
def ok(self) -> bool:
"""Mirror ``requests``: 2xx and 3xx are OK."""
return self.status_code < 400
def json(self) -> object:
"""Return the decoded payload, raising like ``requests`` on non-JSON."""
if self._payload is None:
msg = "no json"
raise ValueError(msg)
return self._payload
class FakeCookieJar(dict[str, str]):
"""A cookie jar with the two methods the client uses."""
def set(self, name: str, value: str, **_kwargs: object) -> None:
"""Store a cookie, ignoring domain/path keywords."""
self[name] = value
class FakeSession:
"""Records requests and replays queued responses in order."""
def __init__(self, responses: list[FakeResponse] | None = None) -> None:
"""Queue ``responses`` to be returned one per request."""
self.headers: dict[str, str] = {}
self.cookies = FakeCookieJar()
self.calls: list[tuple[str, str]] = []
self.bodies: list[dict[str, str]] = []
self._responses: Iterator[FakeResponse] = iter(responses or [])
self.login_cookie: str | None = "session-abc"
self.raise_on_request: Exception | None = None
def request(
self,
method: str,
url: str,
data: dict[str, str] | None = None,
timeout: float | None = None,
) -> FakeResponse:
"""Return the next queued response, recording the call and body."""
del timeout
if self.raise_on_request is not None:
raise self.raise_on_request
self.calls.append((method, url))
if data is not None:
self.bodies.append(data)
if method == "POST" and url.endswith("auth/login") and self.login_cookie:
self.cookies.set("SESSION", self.login_cookie)
try:
return next(self._responses)
except StopIteration:
return FakeResponse(200, {})
def fake_requests(session: FakeSession) -> SimpleNamespace:
"""Return a stand-in for the ``requests`` module itself.
A namespace rather than a class: ``requests.Session`` is capitalised, and
defining a method by that name would need an ``N802`` suppression for no
benefit. ``RequestException`` is included because the client catches it.
"""
return SimpleNamespace(
Session=lambda: session,
RequestException=FakeRequestError,
)
class FakeRequestError(Exception):
"""Stands in for ``requests.RequestException``."""
def write_credentials(path: Path, user: str = "me@example.com", pw: str = "pw") -> None:
"""Write a credentials file in the two-line format the reader expects."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(f"{user}\n{pw}\n", encoding="utf-8")
#: A menu payload shaped like the live one, macros per portion.
def menu_payload(count: int = 2) -> dict[str, object]:
"""Return a ``menus/delivery/{id}/new`` body with ``count`` valid dishes."""
return {
"deliveryMenuMeal": [
{
"menuMealName": f"Danie {index}",
"mealName": "Obiad",
"mealPriority": index,
"nutrition": {
"weight": 300.0,
"calories": 400.0,
"protein": 25.0,
"carbohydrate": 50.0,
"fat": 11.1,
},
}
for index in range(1, count + 1)
],
}
def order_payload(
delivery_id: int = 111, date: str = "2026-08-22"
) -> dict[str, object]:
"""Return an order body embedding one delivery on ``date``."""
return {
"orderId": 1,
"deliveries": [
{"deliveryId": delivery_id, "date": date, "deleted": False},
],
}

View File

@ -20,6 +20,7 @@ are functional in-memory stand-ins, shared by ``test_gatelock.py`` and
from __future__ import annotations
from contextlib import ExitStack
from typing import TYPE_CHECKING
from unittest.mock import patch
@ -95,8 +96,13 @@ if TYPE_CHECKING:
@pytest.fixture(autouse=True)
def _isolate_state(tmp_path: Path) -> Iterator[None]:
"""Redirect all on-disk diet_guard state into a temp dir."""
with (
"""Redirect all on-disk diet_guard state into a temp dir.
Built as a list fed through an ``ExitStack`` rather than one ``with``
tuple: the tuple form is a statically nested block per entry, and at
~20 redirects CPython refuses to compile it.
"""
redirects = [
patch(
"diet_guard._budget.BUDGET_FILE",
tmp_path / ".budget",
@ -159,7 +165,47 @@ def _isolate_state(tmp_path: Path) -> Iterator[None]:
# entry point (narrow peer-log pull, not the full tick). Without
# this every test that reaches `_should_lock` hits the real remote.
patch("diet_guard._cli_gate.pull_peer_logs", return_value=None),
):
# The catering credentials and its cached session cookie. Redirected
# for the same reason as `sync_token`: `_test_guard` raises on any
# write under the real ~/.config/diet_guard, and a test that logs in
# would otherwise clobber the live session. Deliberately NOT added to
# `test_state_redirect._REDIRECTED_CONSTANTS`, matching
# `SYNC_TOKEN_FILE` -- that check enforces a single naming module,
# which would forbid the CLI naming the path in its own setup text.
patch(
"diet_guard._kuchnia_config.KUCHNIA_CREDENTIALS_FILE",
tmp_path / "kuchnia_credentials",
),
patch(
"diet_guard._kuchnia_config.KUCHNIA_SESSION_FILE",
tmp_path / "kuchnia_session.json",
),
patch(
"diet_guard._kuchnia_config.KUCHNIA_LAST_IMPORT_FILE",
tmp_path / "kuchnia_last_import",
),
# The catering fetch is a third network entry point. Patched at each
# call site, so a test importing `refresh_delivery` directly still
# exercises the real helper.
patch("diet_guard._cli_kuchnia.refresh_delivery", return_value=([], None)),
# Logging a meal now warms the catering bank on a background
# thread. Without this every meal-logging test in the suite reaches
# the live panel.
patch(
"diet_guard._cli_log.refresh_delivery_once",
return_value=([], None),
),
# Logging a meal now warms the catering bank on a background
# thread. Without this every meal-logging test in the suite reaches
# the live panel.
patch(
"diet_guard._gatelock_delivery.refresh_delivery",
return_value=([], None),
),
]
with ExitStack() as stack:
for redirect in redirects:
stack.enter_context(redirect)
yield

View File

@ -12,7 +12,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, cast
from unittest.mock import MagicMock, patch
from diet_guard import _gatelock_mealflow
from diet_guard import _gatelock_delivery
from diet_guard._budget import write_budget
from diet_guard._state import log_meal
from diet_guard.tests.conftest import _nutrition
@ -118,7 +118,7 @@ class TestFetchFromSync:
def test_demo_mode_does_not_sync(self, gate: MealGate) -> None:
"""In demo the button is inert and never touches the network."""
with patch.object(_gatelock_mealflow, "pull_peer_logs") as pull:
with patch.object(_gatelock_delivery, "pull_peer_logs") as pull:
gate._on_fetch_sync()
pull.assert_not_called()
assert "only available on the real lock" in gate._vars.status.get()
@ -128,7 +128,7 @@ class TestFetchFromSync:
gate.demo_mode = False
gate._pending = [8, 12]
with patch.object(
_gatelock_mealflow,
_gatelock_delivery,
"pull_peer_logs",
return_value="sync unavailable (x)",
):
@ -142,8 +142,8 @@ class TestFetchFromSync:
gate.demo_mode = False
gate._pending = [8, 12]
with (
patch.object(_gatelock_mealflow, "pull_peer_logs", return_value=None),
patch.object(_gatelock_mealflow, "due_slots", return_value=(8, 12)),
patch.object(_gatelock_delivery, "pull_peer_logs", return_value=None),
patch.object(_gatelock_delivery, "due_slots", return_value=(8, 12)),
):
gate._on_fetch_sync()
_drive_fetch(gate)
@ -155,8 +155,8 @@ class TestFetchFromSync:
gate.demo_mode = False
gate._pending = [8, 12]
with (
patch.object(_gatelock_mealflow, "pull_peer_logs", return_value=None),
patch.object(_gatelock_mealflow, "due_slots", return_value=(12,)),
patch.object(_gatelock_delivery, "pull_peer_logs", return_value=None),
patch.object(_gatelock_delivery, "due_slots", return_value=(12,)),
):
gate._on_fetch_sync()
_drive_fetch(gate)
@ -168,8 +168,8 @@ class TestFetchFromSync:
gate.demo_mode = False
gate._pending = [8, 12, 16]
with (
patch.object(_gatelock_mealflow, "pull_peer_logs", return_value=None),
patch.object(_gatelock_mealflow, "due_slots", return_value=(16,)),
patch.object(_gatelock_delivery, "pull_peer_logs", return_value=None),
patch.object(_gatelock_delivery, "due_slots", return_value=(16,)),
):
gate._on_fetch_sync()
_drive_fetch(gate)
@ -181,8 +181,8 @@ class TestFetchFromSync:
gate.demo_mode = False
gate._pending = [8, 12]
with (
patch.object(_gatelock_mealflow, "pull_peer_logs", return_value=None),
patch.object(_gatelock_mealflow, "due_slots", return_value=()),
patch.object(_gatelock_delivery, "pull_peer_logs", return_value=None),
patch.object(_gatelock_delivery, "due_slots", return_value=()),
):
gate._on_fetch_sync()
_drive_fetch(gate)
@ -209,14 +209,14 @@ class TestFetchFromSync:
gate._poll_fetch()
after = cast("MagicMock", gate.root.after)
after.assert_called_with(_gatelock_mealflow.FETCH_POLL_MS, gate._poll_fetch)
after.assert_called_with(_gatelock_delivery.FETCH_POLL_MS, gate._poll_fetch)
assert gate._fetch_result is not None
def test_a_second_click_while_fetching_is_ignored(self, gate: MealGate) -> None:
"""Two workers would race two log writes and orphan one result."""
gate.demo_mode = False
with patch.object(
_gatelock_mealflow, "pull_peer_logs", return_value=None
_gatelock_delivery, "pull_peer_logs", return_value=None
) as pull:
gate._on_fetch_sync()
first = gate._fetch_result

View File

@ -0,0 +1,68 @@
"""Tests for the gate's automatic catering fetch (trigger b).
Split from ``test_kuchnia_gate.py`` for the repo's 250-line cap; the button,
worker and field-formatting tests live there.
The property that matters: an *automatic* fetch stays quiet about bad news. The
gate has just told the user which slots to fill, and a caterer problem they
never asked about must not talk over that.
"""
from __future__ import annotations
import queue
from unittest.mock import patch
from diet_guard import _gatelock_delivery
from diet_guard._gatelock_kuchnia import DeliveryResult
from diet_guard.tests.test_kuchnia_gate import _deliver, _dish, _Gate
class TestAutoload:
"""Trigger (b): the window opening fetches the day's delivery itself."""
def test_the_window_opening_starts_a_guarded_fetch(self) -> None:
gate = _Gate()
with patch.object(_gatelock_delivery, "start_delivery_fetch") as start:
start.return_value = queue.Queue(maxsize=1)
gate._autoload_delivery()
assert start.call_count == 1
# The guarded wrapper, not the raw refresh: several locks in one day
# must not each pay for a login plus a three-request walk.
assert start.call_args[0][0] is _gatelock_delivery.refresh_delivery_once
def test_demo_mode_does_not_autoload(self) -> None:
gate = _Gate()
gate.demo_mode = True
with patch.object(_gatelock_delivery, "start_delivery_fetch") as start:
gate._autoload_delivery()
assert start.call_count == 0
def test_it_does_not_stack_on_a_running_fetch(self) -> None:
gate = _Gate()
gate._delivery_result = queue.Queue(maxsize=1)
with patch.object(_gatelock_delivery, "start_delivery_fetch") as start:
gate._autoload_delivery()
assert start.call_count == 0
def test_an_automatic_outage_is_silent(self) -> None:
# The gate has just told the user which slots to fill; a caterer
# problem they never asked about must not talk over that.
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(), reason="panel down"), asked=False)
assert gate.statuses == []
def test_an_automatic_empty_day_is_silent(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(), reason=None), asked=False)
assert gate.statuses == []
def test_an_automatic_fetch_still_offers_the_dishes(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(_dish(),), reason=None), asked=False)
assert gate._descs == ["Kaszotto"]
def test_the_button_still_reports_both(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(), reason="panel down"), asked=True)
assert "panel down" in gate.last_status

View File

@ -0,0 +1,160 @@
"""Tests for the ``kuchnia`` subcommand and its dispatch wiring.
Split from ``test_kuchnia_import.py`` for the repo's 250-line cap; the banking
and logging behaviour those exercise lives there.
"""
from __future__ import annotations
from unittest.mock import patch
import pytest
from diet_guard import _cli, _cli_kuchnia
from diet_guard._kuchnia_parse import Dish
from diet_guard._state_today import today_entries
def _dish(name: str = "Kaszotto", kcal: float = 391.0, priority: int = 1) -> Dish:
return Dish(
name=name,
kcal=kcal,
protein_g=32.5,
carbs_g=35.6,
fat_g=13.0,
grams=318.0,
priority=priority,
slot_label="Kolacja",
)
class TestCli:
def _run(self, lines: list[str], **kwargs: bool) -> int:
answers = iter(["y"])
return _cli_kuchnia.cmd_kuchnia(
lines.append,
lambda _prompt: next(answers, "n"),
log=kwargs.get("log", False),
yes=kwargs.get("yes", False),
)
def test_dry_run_banks_but_does_not_log(self) -> None:
lines: list[str] = []
with patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([_dish()], None)
):
assert self._run(lines) == 0
assert today_entries() == []
assert any("not logged" in line for line in lines)
def test_reports_the_total(self) -> None:
lines: list[str] = []
dishes = [_dish("A", kcal=400.0), _dish("B", kcal=600.0, priority=2)]
with patch.object(
_cli_kuchnia, "refresh_delivery", return_value=(dishes, None)
):
self._run(lines)
assert any("total: 1000 kcal" in line for line in lines)
def test_an_outage_exits_nonzero_without_touching_the_log(self) -> None:
lines: list[str] = []
with patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([], "panel down")
):
assert self._run(lines) == 1
assert today_entries() == []
assert any("panel down" in line for line in lines)
def test_no_delivery_today_is_a_clean_exit(self) -> None:
lines: list[str] = []
with patch.object(_cli_kuchnia, "refresh_delivery", return_value=([], None)):
assert self._run(lines) == 0
assert any("no catering delivery" in line for line in lines)
def test_log_asks_first(self) -> None:
lines: list[str] = []
with (
patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([_dish()], None)
),
patch.object(_cli_kuchnia, "log_dishes", return_value=["A"]) as logger,
):
self._run(lines, log=True)
assert logger.call_count == 1
def test_declining_the_prompt_logs_nothing(self) -> None:
lines: list[str] = []
with (
patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([_dish()], None)
),
patch.object(_cli_kuchnia, "log_dishes") as logger,
):
_cli_kuchnia.cmd_kuchnia(
lines.append,
lambda _prompt: "n",
log=True,
yes=False,
)
assert logger.call_count == 0
assert any("nothing logged" in line for line in lines)
def test_yes_skips_the_prompt(self) -> None:
lines: list[str] = []
asked: list[str] = []
with (
patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([_dish()], None)
),
patch.object(_cli_kuchnia, "log_dishes", return_value=["A"]),
):
def _record(prompt: str) -> str:
asked.append(prompt)
return "y"
_cli_kuchnia.cmd_kuchnia(lines.append, _record, log=True, yes=True)
assert asked == []
def test_already_logged_says_so(self) -> None:
lines: list[str] = []
with (
patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([_dish()], None)
),
patch.object(_cli_kuchnia, "log_dishes", return_value=[]),
):
self._run(lines, log=True, yes=True)
assert any("already logged" in line for line in lines)
class TestDispatch:
"""The subcommand must actually be reachable through ``python -m``."""
def test_the_subcommand_is_wired_into_main(self) -> None:
with patch.object(
_cli_kuchnia,
"refresh_delivery",
return_value=([_dish()], None),
):
assert _cli.main(["kuchnia"]) == 0
def test_the_log_flags_reach_the_handler(self) -> None:
with (
patch.object(
_cli_kuchnia,
"refresh_delivery",
return_value=([_dish()], None),
),
patch.object(_cli_kuchnia, "log_dishes", return_value=["A"]) as logger,
):
assert _cli.main(["kuchnia", "--log", "--yes"]) == 0
assert logger.call_count == 1
def test_unknown_lazy_attribute_raises() -> None:
# The name lives in a variable so ruff cannot fold the lookup into a
# bare attribute access and then flag it as a useless expression.
missing = "not_a_real_helper"
with pytest.raises(AttributeError, match="no attribute"):
getattr(_cli_kuchnia, missing)

View File

@ -0,0 +1,235 @@
"""Tests for the catering panel client, its config files, and the walk.
The client never reaches the network here: ``requests`` is replaced at the
module seam, which is why :mod:`diet_guard._kuchnia_client` resolves that name
through ``sys.modules`` instead of binding it at import time.
"""
from __future__ import annotations
import datetime
import json
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from diet_guard import _kuchnia_client, _kuchnia_config
from diet_guard._kuchnia_client import PanelSession
from diet_guard._kuchnia_errors import KuchniaError
from diet_guard.tests._kuchnia_fakes import (
FakeRequestError,
FakeResponse,
FakeSession,
fake_requests,
write_credentials,
)
if TYPE_CHECKING:
from pathlib import Path
DAY = datetime.date(2026, 8, 22)
@pytest.fixture
def creds() -> None:
"""Write a credentials file at the (redirected) config path."""
write_credentials(_kuchnia_config.KUCHNIA_CREDENTIALS_FILE)
def _session_with(responses: list[FakeResponse]) -> FakeSession:
return FakeSession(responses)
class TestCredentials:
def test_missing_file_names_the_path_and_the_fix(self) -> None:
with pytest.raises(KuchniaError, match="no catering credentials"):
_kuchnia_config.read_credentials()
def test_a_one_line_file_is_rejected(self) -> None:
path = _kuchnia_config.KUCHNIA_CREDENTIALS_FILE
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("only-an-email\n", encoding="utf-8")
with pytest.raises(KuchniaError, match="two lines"):
_kuchnia_config.read_credentials()
def test_blank_lines_are_ignored(self, creds: None) -> None:
assert _kuchnia_config.read_credentials() == ("me@example.com", "pw")
class TestSessionCache:
def test_absent_cache_reads_as_none(self) -> None:
assert _kuchnia_config.load_session_cookie() is None
def test_round_trips_and_is_owner_only(self) -> None:
_kuchnia_config.save_session_cookie("abc")
assert _kuchnia_config.load_session_cookie() == "abc"
mode = _kuchnia_config.KUCHNIA_SESSION_FILE.stat().st_mode & 0o777
# 600 from the outset: a write-then-chmod would leave a live session
# cookie world-readable for the duration of the write.
assert mode == 0o600
def test_no_temp_file_is_left_behind(self) -> None:
_kuchnia_config.save_session_cookie("abc")
leftovers = list(_kuchnia_config.KUCHNIA_SESSION_FILE.parent.glob("*.tmp"))
assert leftovers == []
@pytest.mark.parametrize(
"content", ["{not json", '{"SESSION": 7}', "[]", '{"SESSION": ""}']
)
def test_an_unusable_cache_reads_as_none(self, content: str) -> None:
path = _kuchnia_config.KUCHNIA_SESSION_FILE
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
assert _kuchnia_config.load_session_cookie() is None
def test_clearing_is_idempotent(self) -> None:
_kuchnia_config.clear_session_cookie()
_kuchnia_config.save_session_cookie("abc")
_kuchnia_config.clear_session_cookie()
assert _kuchnia_config.load_session_cookie() is None
class TestAuthenticate:
def test_logs_in_and_caches_the_cookie(self, creds: None) -> None:
session = _session_with([FakeResponse(200, {})])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
PanelSession().authenticate()
assert _kuchnia_config.load_session_cookie() == "session-abc"
assert session.calls[0][0] == "POST"
def test_a_cached_cookie_skips_the_login(self, creds: None) -> None:
_kuchnia_config.save_session_cookie("cached")
session = _session_with([])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
PanelSession().authenticate()
assert session.calls == []
assert session.cookies["SESSION"] == "cached"
def test_a_rejected_login_raises(self, creds: None) -> None:
session = _session_with([FakeResponse(401, {})])
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="login rejected"),
):
PanelSession().authenticate()
def test_a_login_that_sets_no_cookie_raises(self, creds: None) -> None:
session = _session_with([FakeResponse(200, {})])
session.login_cookie = None
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="no session cookie"),
):
PanelSession().authenticate()
def test_login_body_is_form_encoded_not_json(self, creds: None) -> None:
# The panel rejects a JSON login body; this is the one call that must
# go out as application/x-www-form-urlencoded.
session = _session_with([FakeResponse(200, {})])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
PanelSession().authenticate()
assert session.bodies == [{"username": "me@example.com", "password": "pw"}]
class TestGetJson:
def test_returns_the_decoded_body(self, creds: None) -> None:
session = _session_with([FakeResponse(200, {"ok": True})])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
panel = PanelSession()
assert panel.get_json("x") == {"ok": True}
@pytest.mark.parametrize("status", [401, 403])
def test_an_expired_cookie_re_logs_in_once(self, creds: None, status: int) -> None:
_kuchnia_config.save_session_cookie("stale")
session = _session_with(
[
FakeResponse(status, {}),
FakeResponse(200, {}),
FakeResponse(200, {"ok": 1}),
],
)
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
panel = PanelSession()
panel.authenticate()
assert panel.get_json("x") == {"ok": 1}
methods = [call[0] for call in session.calls]
assert methods.count("POST") == 1, "exactly one re-login"
def test_a_second_auth_failure_is_real(self, creds: None) -> None:
session = _session_with(
[FakeResponse(401, {}), FakeResponse(200, {}), FakeResponse(401, {})],
)
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="HTTP 401"),
):
PanelSession().get_json("x")
def test_a_server_error_raises(self, creds: None) -> None:
session = _session_with([FakeResponse(500, {})])
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="HTTP 500"),
):
PanelSession().get_json("x")
def test_a_non_json_body_raises(self, creds: None) -> None:
session = _session_with([FakeResponse(200, None, text="<html>")])
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="non-JSON"),
):
PanelSession().get_json("x")
def test_a_transport_error_becomes_a_kuchnia_error(self, creds: None) -> None:
session = _session_with([])
session.raise_on_request = FakeRequestError("connection reset")
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="unreachable"),
):
PanelSession().get_json("x")
def test_the_whole_walk_has_a_deadline(self, creds: None) -> None:
# Per-request timeouts alone would let four sequential calls stack into
# half a minute in front of a user who is waiting.
session = _session_with([FakeResponse(200, {})])
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="deadline"),
):
PanelSession(deadline_seconds=-1.0).get_json("x")
def test_required_headers_are_set(self, creds: None) -> None:
session = _session_with([])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
PanelSession()
assert session.headers["company-id"] == "kuchniavikinga"
assert session.headers["X-Launcher-Type"] == "BROWSER_PANEL"
# No sticky Content-Type: the login body is form-encoded and a JSON
# default would mislabel it.
assert "Content-Type" not in session.headers
def test_lazy_import_keeps_requests_out_of_the_gate_path() -> None:
# The gate's not-due tick imports the CLI; paying ~78ms for an HTTP stack
# it never touches is exactly what the PEP 562 hook exists to prevent.
# The name lives in a variable so ruff cannot fold the lookup into a
# bare attribute access and then flag it as a useless expression.
missing = "definitely_not_a_real_name"
with pytest.raises(AttributeError, match="no attribute"):
getattr(_kuchnia_client, missing)
def test_the_session_file_is_json(creds: None) -> None:
_kuchnia_config.save_session_cookie("abc")
body = json.loads(_kuchnia_config.KUCHNIA_SESSION_FILE.read_text(encoding="utf-8"))
assert body == {"SESSION": "abc"}
def test_state_stays_out_of_the_real_config_dir(tmp_path: Path) -> None:
# The conftest redirect must actually redirect: a test that writes a
# session cookie into ~/.config/diet_guard would clobber the live one.
assert str(tmp_path) in str(_kuchnia_config.KUCHNIA_SESSION_FILE)
assert str(tmp_path) in str(_kuchnia_config.KUCHNIA_CREDENTIALS_FILE)

View File

@ -0,0 +1,221 @@
"""Tests for the lock screen's catering button and its off-thread worker.
Two properties matter more than the rest, because breaking either strands a
user behind a fullscreen lock:
* the worker **always** feeds its queue, even when the call raises, or the poll
waits forever with the button disabled;
* the flow **fails closed** -- an outage leaves the lock exactly as it was.
The completion path is driven directly (put a result, call the poll) rather
than through a thread, mirroring the doctrine in
:mod:`diet_guard._gatelock_fetch`: the worker touches no widget, so the Tk side
is testable with no thread at all.
"""
from __future__ import annotations
import datetime
import queue
from types import SimpleNamespace
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
import pytest
from diet_guard import _gatelock_delivery
from diet_guard._gatelock_delivery import _PullFlows, dish_field_values
from diet_guard._gatelock_kuchnia import (
UNEXPECTED,
DeliveryResult,
start_delivery_fetch,
)
from diet_guard._kuchnia_parse import Dish
if TYPE_CHECKING:
from collections.abc import Sequence
DAY = datetime.date(2026, 8, 22)
def _dish(name: str = "Kaszotto", priority: int = 1) -> Dish:
return Dish(
name=name,
kcal=391.0,
protein_g=32.5,
carbs_g=35.6,
fat_g=13.0,
grams=318.0,
priority=priority,
slot_label="Kolacja",
)
class TestWorker:
def test_hands_back_the_dishes(self) -> None:
result = start_delivery_fetch(lambda _day: ([_dish()], None), DAY)
outcome = result.get(timeout=5)
assert outcome.reason is None
assert outcome.dishes[0].name == "Kaszotto"
def test_hands_back_a_reason(self) -> None:
result = start_delivery_fetch(lambda _day: ([], "panel down"), DAY)
assert result.get(timeout=5).reason == "panel down"
def test_a_raising_call_still_feeds_the_queue(self) -> None:
# Without the finally, the poll waits forever with the button disabled
# and the user cannot get out from behind the lock.
def boom(_day: datetime.date) -> tuple[Sequence[Dish], str | None]:
msg = "kaboom"
raise RuntimeError(msg)
result = start_delivery_fetch(boom, DAY)
assert result.get(timeout=5).reason == UNEXPECTED
class _Gate(_PullFlows):
"""The slice of the controller the catering flow drives."""
def __init__(self) -> None:
self.root = MagicMock()
self.demo_mode = False
self.statuses: list[tuple[str, bool]] = []
self._widgets = SimpleNamespace(amount_entry=MagicMock())
self._state = SimpleNamespace(source="manual")
self._descs: list[str] = []
self._cleared = 0
self._projections = 0
self._entries = [MagicMock() for _ in range(4)]
def _set_status(self, text: str, *, error: bool = False) -> None:
self.statuses.append((text, error))
def _set_desc(self, value: str) -> None:
self._descs.append(value)
def _clear_inputs(self) -> None:
self._cleared += 1
def _refresh_projection(self) -> None:
self._projections += 1
def _macro_entries(self) -> tuple[MagicMock, ...]:
return tuple(self._entries)
@property
def last_status(self) -> str:
return self.statuses[-1][0]
def _deliver(gate: _Gate, outcome: DeliveryResult, *, asked: bool = True) -> None:
"""Hand the gate a completed result and run its poll, with no thread.
``asked`` mirrors whether the fetch came from the button (the default) or
from the window opening; an automatic one stays quiet about bad news.
"""
gate._delivery_asked = asked
result: queue.Queue[DeliveryResult] = queue.Queue(maxsize=1)
result.put(outcome)
gate._delivery_result = result
gate._poll_delivery()
class TestLoadButton:
def test_demo_mode_refuses(self) -> None:
# A synthetic window must never satisfy a real checkpoint, and there is
# no reason to send real credentials from one.
gate = _Gate()
gate.demo_mode = True
gate._on_load_delivery()
assert "only available on the real lock" in gate.last_status
assert gate._delivery_result is None
def test_a_second_click_while_in_flight_is_ignored(self) -> None:
gate = _Gate()
with patch.object(_gatelock_delivery, "start_delivery_fetch") as start:
start.return_value = queue.Queue(maxsize=1)
gate._on_load_delivery()
gate._on_load_delivery()
assert start.call_count == 1
def test_an_outage_leaves_the_lock_untouched(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(), reason="panel down"))
text, error = gate.statuses[-1]
assert "panel down" in text
assert "still locked" in text
assert error is True
assert gate._descs == []
def test_no_delivery_today_says_so(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(), reason=None))
assert "No catering delivery today" in gate.last_status
def test_a_dish_is_offered_but_never_logged(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(_dish(),), reason=None))
# The form is filled...
assert gate._descs == ["Kaszotto"]
gate._widgets.amount_entry.insert.assert_called_once_with(0, "318")
# ...and that is all. Submitting stays the user's explicit action.
assert "Log & Continue" in gate.last_status
def test_dishes_are_offered_one_at_a_time_in_meal_order(self) -> None:
gate = _Gate()
dishes = (_dish("Kolacja", priority=2), _dish("Sniadanie", priority=1))
_deliver(gate, DeliveryResult(dishes=dishes, reason=None))
assert gate._descs == ["Sniadanie"]
assert "1 more to go" in gate.last_status
gate._prefill_next_dish()
assert gate._descs == ["Sniadanie", "Kolacja"]
def test_the_last_dish_has_no_more_to_go_suffix(self) -> None:
gate = _Gate()
_deliver(gate, DeliveryResult(dishes=(_dish(),), reason=None))
assert "more to go" not in gate.last_status
def test_prefilling_past_the_end_is_a_no_op(self) -> None:
gate = _Gate()
gate._prefill_next_dish()
assert gate._descs == []
def test_the_poll_rearms_while_the_worker_runs(self) -> None:
gate = _Gate()
gate._delivery_result = queue.Queue(maxsize=1)
gate._poll_delivery()
gate.root.after.assert_called_with(
_gatelock_delivery.FETCH_POLL_MS,
gate._poll_delivery,
)
assert gate._delivery_result is not None
def test_a_stray_poll_after_completion_stops(self) -> None:
gate = _Gate()
gate._delivery_result = None
gate._poll_delivery()
assert gate.statuses == []
gate.root.after.assert_not_called()
class TestFieldFormatting:
def test_values_are_trimmed_not_padded(self) -> None:
grams, macros = dish_field_values(_dish())
assert grams == "318"
assert macros == ("391", "32.5", "35.6", "13")
def test_the_macro_order_matches_the_form(self) -> None:
# _macro_entries() is (kcal, protein, carbs, fat); a mismatch here
# would silently swap a dish's protein and carbohydrate.
_grams, macros = dish_field_values(
Dish("x", 100.0, 1.0, 2.0, 3.0, 50.0, 1, ""),
)
assert macros == ("100", "1", "2", "3")
def test_unknown_lazy_attribute_raises() -> None:
# The name lives in a variable so ruff cannot fold the lookup into a
# bare attribute access and then flag it as a useless expression.
missing = "not_a_real_helper"
with pytest.raises(AttributeError, match="no attribute"):
getattr(_gatelock_delivery, missing)

View File

@ -0,0 +1,225 @@
"""Tests for banking catering dishes, logging them, and the CLI around both.
The bank idempotency test is the one that earns its keep. ``add_manual_entry``
restamps ``t`` unconditionally and the sync merge derives each record's clock
from that stamp, so a refresh that re-banks unchanged dishes republishes the
*entire* curated bank to every peer. Asserting "the entry exists" passes while
that happens, so these assert on **call count** instead.
"""
from __future__ import annotations
import datetime
from unittest.mock import patch
from diet_guard import _kuchnia_config, _kuchnia_import
from diet_guard._foodbank_manual import read_manual_bank
from diet_guard._kuchnia_errors import KuchniaError
from diet_guard._kuchnia_import import bank_dishes, dish_to_record, refresh_delivery
from diet_guard._kuchnia_log import log_dishes
from diet_guard._kuchnia_parse import Dish
from diet_guard._kuchnia_spread import SlottedDish
from diet_guard._state_today import today_entries, today_total_kcal
DAY = datetime.date(2026, 8, 22)
def _dish(name: str = "Kaszotto", kcal: float = 391.0, priority: int = 1) -> Dish:
return Dish(
name=name,
kcal=kcal,
protein_g=32.5,
carbs_g=35.6,
fat_g=13.0,
grams=318.0,
priority=priority,
slot_label="Kolacja",
)
class TestBankDishes:
def test_banks_a_dish_with_its_portion_macros(self) -> None:
bank_dishes([_dish()])
bank = read_manual_bank()
record = bank["kaszotto"]
assert record["kcal"] == 391.0
assert record["grams"] == 318.0
# count is 0: the bank's count ranks foods by how often they were
# *eaten*, and a delivered dish has not been eaten yet.
assert record["count"] == 0
def test_polish_names_normalise_without_mangling(self) -> None:
bank_dishes([_dish(name="Kaszotto Grzybowe Z Kurczakiem")])
bank = read_manual_bank()
(key,) = bank
assert key == "kaszotto grzybowe z kurczakiem"
assert bank[key]["desc"] == "Kaszotto Grzybowe Z Kurczakiem"
def test_a_re_import_writes_nothing(self) -> None:
dishes = [_dish("A"), _dish("B")]
assert bank_dishes(dishes) == 2
with patch(
"diet_guard._kuchnia_import.add_manual_entry",
) as add:
assert bank_dishes(dishes) == 0
# Call count, not entry existence: the existence assertion passes even
# when every record is needlessly restamped and resynced.
assert add.call_count == 0
def test_a_changed_dish_is_rewritten(self) -> None:
bank_dishes([_dish("A", kcal=400.0)])
with patch(
"diet_guard._kuchnia_import.add_manual_entry",
) as add:
assert bank_dishes([_dish("A", kcal=450.0)]) == 1
assert add.call_count == 1
def test_callers_never_stamp_the_edit_time(self) -> None:
# ``t`` is set inside add_manual_entry; a caller-supplied one would
# skew the last-writer-wins merge.
assert "t" not in dish_to_record(_dish())
def test_banking_nothing_is_not_an_error(self) -> None:
assert bank_dishes([]) == 0
class TestRefreshDelivery:
def test_returns_dishes_and_banks_them(self) -> None:
dishes = [_dish()]
with (
patch.object(_kuchnia_import, "PanelSession"),
patch.object(_kuchnia_import, "fetch_dishes", return_value=dishes),
):
got, reason = refresh_delivery(DAY)
assert reason is None
assert got == dishes
assert read_manual_bank()
def test_an_outage_is_a_reason_not_an_exception(self) -> None:
# Fail-closed: no caller may be broken by the catering being down.
with patch.object(
_kuchnia_import,
"PanelSession",
side_effect=KuchniaError("panel down"),
):
got, reason = refresh_delivery(DAY)
assert got == []
assert reason == "panel down"
def test_a_failed_fetch_banks_nothing(self) -> None:
with patch.object(
_kuchnia_import,
"PanelSession",
side_effect=KuchniaError("nope"),
):
refresh_delivery(DAY)
assert read_manual_bank() == {}
class TestRefreshOnce:
"""The guard that makes the automatic triggers affordable."""
def test_the_first_call_of_the_day_fetches(self) -> None:
with patch.object(
_kuchnia_import,
"refresh_delivery",
return_value=([_dish()], None),
) as refresh:
dishes, reason = _kuchnia_import.refresh_delivery_once(DAY)
assert reason is None
assert len(dishes) == 1
assert refresh.call_count == 1
def test_the_second_call_does_not_touch_the_network(self) -> None:
# The gate and the after-log hook both fire many times a day; each
# unguarded refresh is a login plus a three-request walk.
with patch.object(
_kuchnia_import,
"refresh_delivery",
return_value=([_dish()], None),
) as refresh:
_kuchnia_import.refresh_delivery_once(DAY)
dishes, reason = _kuchnia_import.refresh_delivery_once(DAY)
assert refresh.call_count == 1
assert (dishes, reason) == ([], None), "nothing new is not an error"
def test_a_new_day_fetches_again(self) -> None:
with patch.object(
_kuchnia_import,
"refresh_delivery",
return_value=([_dish()], None),
) as refresh:
_kuchnia_import.refresh_delivery_once(DAY)
_kuchnia_import.refresh_delivery_once(DAY + datetime.timedelta(days=1))
assert refresh.call_count == 2
def test_an_outage_does_not_suppress_the_retry(self) -> None:
with patch.object(
_kuchnia_import,
"refresh_delivery",
return_value=([], "panel down"),
) as refresh:
_kuchnia_import.refresh_delivery_once(DAY)
_kuchnia_import.refresh_delivery_once(DAY)
assert refresh.call_count == 2, "a failed day must be retried"
def test_an_unreadable_marker_just_means_fetch_again(self) -> None:
marker = _kuchnia_config.KUCHNIA_LAST_IMPORT_FILE
marker.parent.mkdir(parents=True, exist_ok=True)
marker.mkdir() # a directory where a file belongs
with patch.object(
_kuchnia_import,
"refresh_delivery",
return_value=([_dish()], None),
) as refresh:
_kuchnia_import.refresh_delivery_once(DAY)
assert refresh.call_count == 1
def test_the_marker_never_reaches_real_state(self) -> None:
assert ".config/diet_guard" not in str(
_kuchnia_config.KUCHNIA_LAST_IMPORT_FILE,
)
class TestLogDishes:
def test_logs_each_dish_into_its_slot(self) -> None:
chosen = [
SlottedDish(dish=_dish("A", kcal=100.0), slot=8),
SlottedDish(dish=_dish("B", kcal=200.0), slot=12),
]
assert log_dishes(chosen) == ["A", "B"]
entries = today_entries()
assert [entry["slot"] for entry in entries] == [8, 12]
assert today_total_kcal() == 300.0
def test_records_the_provenance(self) -> None:
log_dishes([SlottedDish(dish=_dish("A"), slot=8)])
(entry,) = today_entries()
assert entry["source"] == "kuchnia wikinga"
def test_a_second_run_logs_nothing(self) -> None:
chosen = [SlottedDish(dish=_dish("A"), slot=8)]
log_dishes(chosen)
assert log_dishes(chosen) == []
assert len(today_entries()) == 1
def test_the_same_dish_in_a_different_slot_is_a_separate_meal(self) -> None:
log_dishes([SlottedDish(dish=_dish("A"), slot=8)])
assert log_dishes([SlottedDish(dish=_dish("A"), slot=12)]) == ["A"]
assert len(today_entries()) == 2
def test_duplicates_within_one_batch_land_once(self) -> None:
# today_entries() is read once, before the loop; without reflecting the
# writes locally both copies would be logged.
chosen = [
SlottedDish(dish=_dish("A"), slot=8),
SlottedDish(dish=_dish("A"), slot=8),
]
assert log_dishes(chosen) == ["A"]
def test_matching_ignores_case_and_padding(self) -> None:
log_dishes([SlottedDish(dish=_dish("Kaszotto"), slot=8)])
assert log_dishes([SlottedDish(dish=_dish(" kaszotto "), slot=8)]) == []
def test_logging_nothing_is_not_an_error(self) -> None:
assert log_dishes([]) == []

View File

@ -0,0 +1,119 @@
"""Tests for the panel's three-call walk to a day's menu.
Split from ``test_kuchnia_client.py`` for the repo's 250-line cap; the session,
credential and cookie-cache behaviour lives there.
The endpoint shape here is not the obvious one: ``.../deliveries/{id}/details``
404s despite its name, so the menu comes from
``company/general/menus/delivery/{deliveryId}/new``, keyed by the opaque id.
"""
from __future__ import annotations
import datetime
from unittest.mock import patch
import pytest
from diet_guard import _kuchnia_client, _kuchnia_config
from diet_guard._kuchnia_client import PanelSession
from diet_guard._kuchnia_errors import KuchniaError
from diet_guard._kuchnia_orders import fetch_dishes
from diet_guard.tests._kuchnia_fakes import (
FakeResponse,
FakeSession,
fake_requests,
menu_payload,
order_payload,
write_credentials,
)
DAY = datetime.date(2026, 8, 22)
@pytest.fixture
def creds() -> None:
"""Write a credentials file at the (redirected) config path."""
write_credentials(_kuchnia_config.KUCHNIA_CREDENTIALS_FILE)
def _session_with(responses: list[FakeResponse]) -> FakeSession:
return FakeSession(responses)
class TestFetchDishes:
def _panel(self, responses: list[FakeResponse]) -> tuple[PanelSession, FakeSession]:
session = _session_with(responses)
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
return PanelSession(), session
def test_walks_to_the_days_menu(self, creds: None) -> None:
panel, session = self._panel(
[
FakeResponse(200, [1]),
FakeResponse(200, order_payload(delivery_id=111)),
FakeResponse(200, menu_payload(2)),
],
)
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
dishes = fetch_dishes(panel, DAY)
assert len(dishes) == 2
# The menu endpoint is keyed by the opaque deliveryId, never the date:
# the date form 400s and the obvious `.../details` path 404s.
assert any("menus/delivery/111/new" in url for _, url in session.calls)
def test_no_active_order_is_not_an_error(self, creds: None) -> None:
panel, session = self._panel([FakeResponse(200, [])])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
assert fetch_dishes(panel, DAY) == []
def test_no_delivery_that_day_is_not_an_error(self, creds: None) -> None:
panel, session = self._panel(
[
FakeResponse(200, [1]),
FakeResponse(200, order_payload(date="2026-01-01")),
],
)
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
assert fetch_dishes(panel, DAY) == []
def test_a_cancelled_delivery_is_skipped(self, creds: None) -> None:
order = {
"orderId": 1,
"deliveries": [
{"deliveryId": 111, "date": "2026-08-22", "deleted": True},
],
}
panel, session = self._panel([FakeResponse(200, [1]), FakeResponse(200, order)])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
assert fetch_dishes(panel, DAY) == []
@pytest.mark.parametrize(
"order",
["nope", {}, {"deliveries": "x"}, {"deliveries": [None]}],
)
def test_a_malformed_order_yields_nothing(self, creds: None, order: object) -> None:
panel, session = self._panel([FakeResponse(200, [1]), FakeResponse(200, order)])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
assert fetch_dishes(panel, DAY) == []
def test_an_unpublished_menu_is_reported_not_silently_empty(
self, creds: None
) -> None:
panel, session = self._panel(
[
FakeResponse(200, [1]),
FakeResponse(200, order_payload()),
FakeResponse(200, {"deliveryMenuMeal": []}),
],
)
with (
patch.object(_kuchnia_client, "requests", fake_requests(session)),
pytest.raises(KuchniaError, match="no usable dishes"),
):
fetch_dishes(panel, DAY)
def test_active_ids_may_be_junk(self, creds: None) -> None:
panel, session = self._panel([FakeResponse(200, {"not": "a list"})])
with patch.object(_kuchnia_client, "requests", fake_requests(session)):
assert fetch_dishes(panel, DAY) == []

View File

@ -0,0 +1,186 @@
"""Tests for the catering menu parser and its slot assignment.
The unit check in :mod:`diet_guard._kuchnia_parse` is the reason this file
matters most: the panel could quote macros per portion or per 100 g, and the
difference is invisible in a payload but skews every logged meal by a factor of
several. Both directions of that mix-up are asserted below.
The fixture is a trimmed copy of a real capture (2026-08-22), so the field
names and value shapes are the ones the live panel actually sends.
"""
from __future__ import annotations
from typing import cast
import pytest
from diet_guard._kuchnia_parse import Dish, parse_menu
from diet_guard._kuchnia_spread import assign_slots
# One real meal from the captured payload, macros per portion.
_REAL_MEAL: dict[str, object] = {
"menuMealName": "Marchewkowe pancakes, twarożek cynamonowy, sos truskawkowy",
"mealName": "Śniadanie",
"mealPriority": 1,
"nutrition": {
"weight": 270.0,
"calories": 435.0,
"protein": 25.86,
"carbohydrate": 54.64,
"fat": 12.01,
"dietaryFiber": 2.03,
},
}
def _payload(
nutrition: dict[str, object] | None = None,
**overrides: object,
) -> dict[str, object]:
"""Return a one-meal payload with ``nutrition`` and top-level overrides."""
meal = dict(_REAL_MEAL)
nutrients = dict(cast("dict[str, object]", _REAL_MEAL["nutrition"]))
nutrients.update(nutrition or {})
meal["nutrition"] = nutrients
meal.update(overrides)
return {"deliveryMenuMeal": [meal]}
class TestParseMenu:
def test_reads_a_real_meal(self) -> None:
(dish,) = parse_menu(_payload())
assert dish.name.startswith("Marchewkowe pancakes")
assert dish.kcal == 435.0
assert dish.grams == 270.0
assert dish.protein_g == 25.86
assert dish.priority == 1
assert dish.slot_label == "Śniadanie"
def test_uses_the_dish_name_not_the_slot_label(self) -> None:
# ``mealName`` is "Śniadanie" for every breakfast ever delivered;
# banking under it would collapse the whole diet into five entries.
(dish,) = parse_menu(_payload())
assert dish.name != "Śniadanie"
def test_rejects_calories_quoted_per_100g(self) -> None:
# 435 kcal for 270 g is 161 per 100 g. If the panel ever quotes it that
# way, the macros no longer account for the energy and the dish must be
# dropped rather than imported at a third of its real value.
assert parse_menu(_payload(nutrition={"calories": 161.0})) == []
def test_rejects_macros_quoted_per_100g(self) -> None:
# The mirror image: calories per portion, macros per 100 g.
assert (
parse_menu(
_payload(
nutrition={"protein": 9.6, "carbohydrate": 20.2, "fat": 4.4},
),
)
== []
)
@pytest.mark.parametrize("grams", [0.0, 0.27, 6000.0])
def test_rejects_an_implausible_portion(self, grams: float) -> None:
assert parse_menu(_payload(nutrition={"weight": grams})) == []
def test_rejects_zero_calories(self) -> None:
assert parse_menu(_payload(nutrition={"calories": 0.0})) == []
def test_rejects_a_nameless_dish(self) -> None:
assert parse_menu(_payload(menuMealName=" ")) == []
def test_rejects_a_meal_with_no_nutrition_block(self) -> None:
assert parse_menu({"deliveryMenuMeal": [{"menuMealName": "X"}]}) == []
@pytest.mark.parametrize(
"payload",
["nope", {}, {"deliveryMenuMeal": "x"}, {"deliveryMenuMeal": [None]}, 7],
)
def test_survives_junk(self, payload: object) -> None:
assert parse_menu(payload) == []
def test_one_bad_dish_does_not_cost_the_others(self) -> None:
payload = {"deliveryMenuMeal": [_REAL_MEAL, {"bad": 1}]}
assert len(parse_menu(payload)) == 1
def test_a_boolean_priority_falls_back_to_position(self) -> None:
# bool is an int subclass; True must not become priority 1 by accident.
(dish,) = parse_menu(_payload(mealPriority=True))
assert dish.priority == 1
assert not isinstance(dish.priority, bool)
def test_a_missing_priority_falls_back_to_position(self) -> None:
meal = {k: v for k, v in _REAL_MEAL.items() if k != "mealPriority"}
(dish,) = parse_menu({"deliveryMenuMeal": [meal]})
assert dish.priority == 1
def test_a_missing_slot_label_is_empty_not_absent(self) -> None:
meal = {k: v for k, v in _REAL_MEAL.items() if k != "mealName"}
(dish,) = parse_menu({"deliveryMenuMeal": [meal]})
assert dish.slot_label == ""
def _dish(priority: int, name: str = "d") -> Dish:
return Dish(
name=f"{name}{priority}",
kcal=100.0,
protein_g=1.0,
carbs_g=2.0,
fat_g=3.0,
grams=50.0,
priority=priority,
slot_label="",
)
class TestAssignSlots:
def test_follows_the_providers_priority_not_payload_order(self) -> None:
dishes = [_dish(3), _dish(1), _dish(2)]
assigned = assign_slots(dishes, (8, 12, 16))
assert [item.dish.name for item in assigned] == ["d1", "d2", "d3"]
assert [item.slot for item in assigned] == [8, 12, 16]
def test_five_dishes_over_four_slots_double_up_the_earliest(self) -> None:
assigned = assign_slots([_dish(i) for i in range(1, 6)], (8, 12, 16, 20))
assert [item.slot for item in assigned] == [8, 8, 12, 16, 20]
def test_five_dishes_over_five_slots_map_one_to_one(self) -> None:
# kuhy's real schedule; the catering plan is also five meals.
assigned = assign_slots([_dish(i) for i in range(1, 6)], (8, 11, 14, 17, 20))
assert [item.slot for item in assigned] == [8, 11, 14, 17, 20]
@pytest.mark.parametrize("count", range(1, 9))
@pytest.mark.parametrize("span", range(2, 7))
def test_assignment_is_monotonic_and_in_range(self, count: int, span: int) -> None:
slots = tuple(range(8, 8 + span))
assigned = assign_slots([_dish(i) for i in range(1, count + 1)], slots)
got = [item.slot for item in assigned]
assert len(got) == count
assert got == sorted(got), "a later meal must never precede an earlier one"
assert set(got) <= set(slots)
assert got[0] == slots[0], "the first meal always starts the day"
if count >= span:
# Only a full (or over-full) day reaches the last slot. With fewer
# dishes than slots the delivery genuinely ends earlier -- three
# meals should not be stretched across breakfast-to-supper.
assert got[-1] == slots[-1]
def test_ties_break_deterministically(self) -> None:
# Two dishes sharing a priority must not reshuffle between runs: an
# unstable order looks like a change and re-stamps every bank record.
dishes = [_dish(1, "b"), _dish(1, "a")]
first = [item.dish.name for item in assign_slots(dishes, (8, 12))]
second = [item.dish.name for item in assign_slots(dishes[::-1], (8, 12))]
assert first == second
@pytest.mark.parametrize(
("dishes", "slots"),
[([], (8, 12)), ([_dish(1)], ())],
)
def test_nothing_to_assign_is_not_an_error(
self,
dishes: list[Dish],
slots: tuple[int, ...],
) -> None:
assert assign_slots(dishes, slots) == []

144
docs/kuchnia-wikinga.md Normal file
View File

@ -0,0 +1,144 @@
# Kuchnia Wikinga catering import
Pulls the day's delivered dishes — with their real per-portion macros — out of
`panel.kuchniavikinga.pl` and into diet_guard, so a catering meal is one
keypress instead of five retyped numbers.
## What it does, and deliberately does not
**A delivered meal is not an eaten meal.** Dishes land in the *curated* food
bank; entries are only ever written after an explicit confirmation. If the
importer logged automatically, the gate would satisfy its own checkpoint from a
delivery note, and the log would record what the courier dropped off rather than
what was eaten — skipped, shared and binned meals included.
So:
- **`diet-guard kuchnia`** banks the day's dishes and prints them. Nothing is
logged.
- **`diet-guard kuchnia --log`** asks first (`--yes` skips the prompt for
scripting).
- **The lock screen's "🍱 Today's delivery"** button *prefills* the meal form,
dish by dish, in the caterer's own meal order. The user still clicks
"Log & Continue" for each.
## The API (undocumented, recovered by inspection)
The panel is a white-labelled **Dietly** React SPA. There is no published API;
the endpoint map was recovered from its Vite bundles and confirmed against live
responses on 2026-08-22. Re-check if the bundle hashes change.
| | |
|---|---|
| Base | `https://panel.kuchniavikinga.pl/api` |
| Login | `POST auth/login`, **form-urlencoded** `username=…&password=…` |
| Session | a `SESSION` cookie — **not** a bearer token |
| Headers | `company-id: kuchniavikinga` (the company *name*), `X-Launcher-Type: BROWSER_PANEL` |
The walk is three calls:
1. `company/customer/order/active-ids` → `[orderId]`
2. `company/customer/order/{orderId}` → the order, which **embeds every
delivery** with its date. No enumeration call needed — but `deliveryMeals`
carries ids only, no names and no macros.
3. `company/general/menus/delivery/{deliveryId}/new` → the actual menu.
Three traps worth knowing before touching `_kuchnia_orders.py`:
- **`.../deliveries/{id}/details` does not work.** It 404s (400 keyed by date),
verified against three separate delivery days, despite being the obvious name
and the one the bundles pass a `deliveryId` to. The menu endpoint above is
the one that answers.
- **`deliveryId` is opaque, never a date.** The date form 400s.
- **No `XSRF-TOKEN` cookie is issued**, so the CSRF echo the panel's own
JavaScript performs is a no-op here. The client skips it when absent rather
than sending an empty header.
### Macros are per portion
`nutrition` = `{weight, calories, protein, carbohydrate, fat, dietaryFiber,
sugar, salt, saturatedFattyAcids}`. **`weight` is the portion in grams and the
macros are totals for that portion**, which maps straight onto diet_guard's
`grams`/`kcal`/`protein_g`/`carbs_g`/`fat_g`.
This was proven, not assumed: 4·protein + 4·carbs + 9·fat reproduces the stated
`calories` within ~1% on every meal of the captured day, and the day totals
2055 kcal against the plan's declared 2000.
`_kuchnia_parse._energy_is_consistent` re-runs that check on every dish and
**drops any that fails**. A per-100 g/per-portion mix-up is invisible in a
payload and would silently skew every logged meal by a factor of several, so
the parser refuses rather than importing a plausible-looking lie.
## Credentials
Two files under `~/.config/diet_guard/`, both mode 600, both **outside** the
synced tree — a password and a live session cookie must not travel to another
device.
```bash
install -m 600 /dev/null ~/.config/diet_guard/kuchnia_credentials
printf '%s\n%s\n' 'you@example.com' 'your-password' \
> ~/.config/diet_guard/kuchnia_credentials
```
`kuchnia_credentials` is **written by hand and never by the package**, the same
contract as `sync_token`. `kuchnia_session.json` caches the session cookie so a
refresh normally skips the login round trip; it is created `touch(mode=0o600)`
*before* it holds anything, then atomically replaced — writing and chmod'ing
afterwards would leave a live cookie world-readable for the duration.
Missing or unreadable credentials are not an error at the call site: the fetch
returns a reason string and the caller carries on.
## Triggers
Three, all event-driven. **No new systemd timer** — `diet-guard-sync.timer` was
deliberately deleted and `install.sh` actively uninstalls it.
| | Where | Guarded? |
|---|---|---|
| CLI | `diet-guard kuchnia` | no — an explicit ask always goes and looks |
| Gate | `MealGate.on_focus_ready` → `_autoload_delivery` | yes |
| After a meal | `_cli_log.cmd_ate`, on the existing background thread | yes |
The guarded pair go through `refresh_delivery_once`, which skips the walk when
today has already been fetched (`kuchnia_last_import`, one ISO date). Without
it the after-log hook would pay a login plus three requests after *every* meal.
Only a clean fetch records the date, so an outage is retried.
**Why the gate hook is in `on_focus_ready` and not `_cli_gate._should_lock`:**
`gate_is_due()` reads only the food log and the schedule, never the food bank,
so a catering refresh there could not change the lock decision — while adding a
third-party round trip to the ~105ms `gate --check` fast path, which opens no
window at all. Hooking it once the window is up puts the spinner *inside* the
lock instead of in front of it. An automatic fetch is also silent about bad
news: it must not talk over the prompt telling the user which slots to fill.
## Things that will bite
- **Re-banking unchanged dishes floods sync.** `add_manual_entry` restamps `t`
unconditionally and the merge derives each record's clock from it, so a
refresh that re-banks everything republishes the whole curated bank to every
peer. `bank_dishes` compares the nutritional fields first. Tests assert on
**call count** — "the entry exists" passes while this misbehaves.
- **Imports go to `food_bank_manual.json` only.** The derived bank is rewritten
from the log on every meal, so anything written there vanishes.
- **Duplicate log suppression reads today's log, not a marker.** `log_meal`
mints a fresh uuid per entry, so a re-import after a lost marker would merge
as duplicates on every peer. Matching `desc`+`slot` against the synced log is
robust to marker loss *and* correct across devices.
- **`requests` is lazy-imported.** The gate's not-due tick imports the CLI and
must not pay ~78ms for an HTTP stack it never touches. Call sites reach
through `sys.modules[__name__]` so `patch.object` still wins.
- **The phone needs no importer.** It receives the curated bank and the log
through existing sync, and `FoodBankRecord.fromJson`/`toJson` enumerate a
fixed key set — a provenance field written by the PC would be silently
dropped on every round-trip, producing an endless re-add/re-strip ping-pong.
## Re-probing the API
`scripts/probe_kuchnia.py` (login, order walk) and
`scripts/probe_kuchnia_details.py` (which URL form serves a day's meals) dump
redacted JSON captures. Run the first, then the second. They prompt once and
cache the session, and credential/cookie *values* never reach the dump.

211
scripts/probe_kuchnia.py Executable file
View File

@ -0,0 +1,211 @@
#!/usr/bin/env python3
"""Capture the Kuchnia Wikinga panel's API shape so the importer can be written.
The panel (``panel.kuchniavikinga.pl``) is a white-labelled *Dietly* React SPA
with no published API. Its endpoint map was recovered by grepping the Vite
bundles, which settled the URLs, the auth scheme and the header set -- but not
the *response shape*. This script fetches the real payloads so
``diet_guard._kuchnia_parse`` can be written against captured JSON instead of
against a guess.
It lives in ``scripts/`` rather than as a ``diet_guard`` subcommand on purpose:
the package carries a 100% branch-coverage gate, and a probe inside it would
need tests written against the very payload it exists to discover.
Three questions the capture must answer, because each changes the mapping code:
* **Are macros per portion or per 100 g?** The highest-risk unknown -- a
per-100 g figure stored as a portion total silently under-reports every meal.
Look for a weight/grams field next to ``calories``.
* Are the macro fields on the meal object, or nested under something like
``nutritionalValues``?
* Does the order object embed its deliveries, or is a separate call needed?
Nothing here writes to ``diet_guard`` state; the only output is a JSON dump.
Usage:
python3 scripts/probe_kuchnia.py
"""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import sys
from typing import Any
from probe_kuchnia_auth import prompt_credentials, remember_session
from probe_kuchnia_shape import describe, find_delivery_id
import requests
ORIGIN = "https://panel.kuchniavikinga.pl"
COMPANY = "kuchniavikinga"
#: Both candidate bases. The bundles define an RTK Query slice on ``/api`` and a
#: ``configSlice.baseUrl`` of ``/api/panel``, with code deriving one from the
#: other -- so which one ``auth/login`` answers on is decided here, not guessed.
LOGIN_BASES = (f"{ORIGIN}/api", f"{ORIGIN}/api/panel")
TIMEOUT = 20.0
def emit(text: str = "") -> None:
"""Write one progress line to stdout.
A thin wrapper over ``sys.stdout.write``, mirroring ``diet_guard._cli._emit``
so genuine operator output does not trip ruff's ``T201`` (no ``print``)
without resorting to a suppression. The progress lines are the point of an
interactive probe -- they are what tells the operator which base answered
and how far the walk got before something returned a non-200.
"""
sys.stdout.write(f"{text}\n")
def login(session: requests.Session, username: str, password: str) -> str | None:
"""Try each candidate base until one accepts the credentials.
The bundles show a form-urlencoded body, not JSON -- ``data=`` rather than
``json=`` -- and let ``requests`` set the content type.
Args:
session: The session that will carry the resulting cookies.
username: Panel login.
password: Panel password.
Returns:
The base URL that worked, or None when both were rejected.
"""
for base in LOGIN_BASES:
response = session.post(
f"{base}/auth/login",
data={"username": username, "password": password},
timeout=TIMEOUT,
)
emit(f" {base}/auth/login -> {response.status_code}")
if response.ok:
names = sorted({cookie.name for cookie in session.cookies})
emit(f" cookies received: {names or '(none)'}")
return base
return None
def walk(session: requests.Session, base: str) -> dict[str, Any]:
"""Follow active-ids -> order -> delivery details, capturing each payload.
Args:
session: A logged-in session.
base: The API base that accepted the login.
Returns:
The captured payloads, keyed by step.
"""
captured: dict[str, Any] = {}
active = session.get(f"{base}/company/customer/order/active-ids", timeout=TIMEOUT)
captured["active_ids"] = describe(active)
emit(f" active-ids -> {active.status_code}")
if not active.ok:
return captured
ids = active.json()
order_id = ids[0] if isinstance(ids, list) and ids else None
if order_id is None:
emit(" no active orders -- nothing further to capture.")
return captured
order = session.get(f"{base}/company/customer/order/{order_id}", timeout=TIMEOUT)
captured["order"] = describe(order)
emit(f" order/{order_id} -> {order.status_code}")
if not order.ok:
return captured
delivery_id = find_delivery_id(order.json())
captured["delivery_id_found"] = delivery_id
if delivery_id is None:
emit(" no deliveryId spotted in the order payload -- inspect it by hand.")
return captured
details = session.get(
f"{base}/company/customer/order/{order_id}/deliveries/{delivery_id}/details",
timeout=TIMEOUT,
)
captured["delivery_details"] = describe(details)
emit(f" deliveries/{delivery_id}/details -> {details.status_code}")
return captured
def build_session() -> requests.Session:
"""Return a session with the headers the panel requires on every call."""
session = requests.Session()
# No session-level Content-Type: login is form-urlencoded and a sticky JSON
# default would mislabel its body. ``requests`` sets it per request.
session.headers.update(
{
"company-id": COMPANY,
"X-Launcher-Type": "BROWSER_PANEL",
"User-Agent": "diet_guard/1.0 (personal diet tracker)",
"Accept": "application/json",
},
)
return session
def main() -> int:
"""Run the probe and write the capture.
Returns:
0 on success, 1 when login failed.
"""
parser = argparse.ArgumentParser(description="Probe the Kuchnia Wikinga panel.")
parser.add_argument(
"--out",
default="kuchnia_probe.json",
help="Where to write the captured payloads.",
)
args = parser.parse_args()
# Prompted, never an argv flag: a password in argv lands in shell history
# and in every ps listing on the machine.
username, password = prompt_credentials()
session = build_session()
emit("Trying login bases...")
base = login(session, username, password)
if base is None:
emit("Login rejected on both bases -- check the credentials.")
return 1
# Cache the session so the follow-up probes do not ask again.
remember_session(session)
xsrf = session.cookies.get("XSRF-TOKEN")
emit(f"XSRF-TOKEN cookie present: {xsrf is not None}")
if xsrf is not None:
session.headers["X-XSRF-TOKEN"] = xsrf
emit("Walking the order endpoints...")
captured = walk(session, base)
out = Path(args.out)
out.write_text(
json.dumps(
{
"captured_at": datetime.now(tz=timezone.utc).isoformat(),
"login_base": base,
"xsrf_cookie_present": xsrf is not None,
"steps": captured,
},
indent=2,
ensure_ascii=False,
),
encoding="utf-8",
)
emit(f"\nWrote {out.resolve()}")
emit("Check: are macros per portion or per 100 g? Is there a grams/weight field?")
return 0
if __name__ == "__main__":
sys.exit(main())

122
scripts/probe_kuchnia_auth.py Executable file
View File

@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""Log in once and reuse the session across probe runs.
Typing an e-mail and password on every probe invocation is not just tedious --
it is the thing the real importer must never do either, so the cache is built
here and carried over to ``diet_guard._kuchnia_config`` rather than invented
twice.
Two files, both mode 600, both under ``~/.config/diet_guard/`` (outside the
synced ``~/.local/share`` tree, because a credential must not sync):
* ``kuchnia_credentials`` -- optional, **written by the user by hand**, two
lines: e-mail then password. Nothing here ever writes it; that mirrors
``sync_token``, which the package also only ever reads.
* ``kuchnia_session.json`` -- the cached ``SESSION`` cookie, written by this
module.
The cache is created with ``touch(mode=0o600)`` *before* any content is
written, then atomically replaced -- writing first and chmod'ing after would
leave a live session cookie world-readable for the gap in between.
"""
from __future__ import annotations
import getpass
import json
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import requests
CONFIG_DIR = Path.home() / ".config" / "diet_guard"
CREDENTIALS_FILE = CONFIG_DIR / "kuchnia_credentials"
SESSION_FILE = CONFIG_DIR / "kuchnia_session.json"
#: The only cookie the panel issued on the first probe. No ``XSRF-TOKEN`` was
#: set, so the CSRF echo the bundles perform is a no-op for this account.
SESSION_COOKIE = "SESSION"
def read_credentials() -> tuple[str, str] | None:
"""Return ``(username, password)`` from the hand-written file, if present.
Returns:
The credentials, or None when the file is absent or malformed.
"""
if not CREDENTIALS_FILE.exists():
return None
lines = [
line.strip()
for line in CREDENTIALS_FILE.read_text(encoding="utf-8").splitlines()
if line.strip()
]
expected = 2
if len(lines) < expected:
return None
return lines[0], lines[1]
def load_session_cookie() -> str | None:
"""Return the cached session cookie, or None when there is no usable cache."""
if not SESSION_FILE.exists():
return None
try:
cached = json.loads(SESSION_FILE.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
value = cached.get(SESSION_COOKIE) if isinstance(cached, dict) else None
return value if isinstance(value, str) else None
def save_session_cookie(value: str) -> None:
"""Persist ``value`` as the cached session cookie, mode 600 from the outset."""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
temp = SESSION_FILE.with_suffix(".json.tmp")
# 600 before any content exists: write-then-chmod would publish a live
# session cookie to every local user for the duration of the write.
temp.touch(mode=0o600)
temp.write_text(json.dumps({SESSION_COOKIE: value}), encoding="utf-8")
temp.replace(SESSION_FILE)
def clear_session_cookie() -> None:
"""Drop the cached cookie so the next call logs in afresh."""
SESSION_FILE.unlink(missing_ok=True)
def prompt_credentials() -> tuple[str, str]:
"""Ask for credentials interactively, preferring the hand-written file.
Returns:
A ``(username, password)`` pair.
"""
stored = read_credentials()
if stored is not None:
return stored
username = input("Panel e-mail: ").strip()
return username, getpass.getpass("Panel password: ")
def attach_cached_session(session: requests.Session) -> bool:
"""Attach the cached cookie to ``session``.
Args:
session: The session to prime.
Returns:
True when a cached cookie was attached.
"""
cookie = load_session_cookie()
if cookie is None:
return False
session.cookies.set(SESSION_COOKIE, cookie, domain="panel.kuchniavikinga.pl")
return True
def remember_session(session: requests.Session) -> None:
"""Cache ``session``'s cookie if it carries one."""
value = session.cookies.get(SESSION_COOKIE)
if value is not None:
save_session_cookie(value)

174
scripts/probe_kuchnia_details.py Executable file
View File

@ -0,0 +1,174 @@
#!/usr/bin/env python3
"""Second probe pass: find which URL form returns a day's meals, and its shape.
The first pass (:mod:`probe_kuchnia`) settled login, the API base and the order
payload, but its ``deliveries/{id}/details`` call 404'd -- it had picked the
first delivery in document order, which was weeks in the future. It also found
no ``XSRF-TOKEN`` cookie, only ``SESSION``.
So two questions remain, and both are answered by trying rather than guessing:
* Does ``details`` key on the opaque ``deliveryId`` or on the ``YYYY-MM-DD``
date? The bundles pass a variable named ``deliveryId``, but the 404 means the
panel may want something else -- or that endpoint may only serve *past*
deliveries.
* Where do dish names and macros actually live? The order's ``deliveryMeals``
carry only ids (``dietCaloriesMealId``), so the names must come from
``details`` or from the menu endpoint.
Run this after ``probe_kuchnia.py``; it re-uses that capture to pick sensible
delivery ids instead of prompting for them.
Usage:
python3 scripts/probe_kuchnia_details.py [--capture kuchnia_probe.json]
"""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import json
from pathlib import Path
import sys
from typing import TYPE_CHECKING, Any
from probe_kuchnia import ORIGIN, TIMEOUT, build_session, emit, login
from probe_kuchnia_auth import (
attach_cached_session,
clear_session_cookie,
prompt_credentials,
remember_session,
)
from probe_kuchnia_shape import describe
if TYPE_CHECKING:
import requests
#: Candidate URL templates for a single day's meals. ``{order}`` is the order
#: id, ``{key}`` is whatever identifies the day -- id or date, which is exactly
#: what this probe is here to determine.
CANDIDATES = (
"company/customer/order/{order}/deliveries/{key}/details",
"company/general/menus/delivery/{key}/new",
"company/customer/order/{order}/deliveries/{key}/eta",
)
def pick_deliveries(capture: dict[str, Any]) -> list[dict[str, Any]]:
"""Return today's delivery and its neighbours from a first-pass capture.
A past, a current and a future delivery are all tried: if ``details`` only
serves days that have already been delivered, testing today alone would
report a false negative.
Args:
capture: The JSON written by ``probe_kuchnia.py``.
Returns:
Up to three delivery objects, oldest first.
"""
order = capture["steps"]["order"]["body"]
deliveries = sorted(order["deliveries"], key=lambda item: str(item["date"]))
today = datetime.now(tz=timezone.utc).astimezone().date().isoformat()
past = [item for item in deliveries if str(item["date"]) < today]
current = [item for item in deliveries if str(item["date"]) == today]
future = [item for item in deliveries if str(item["date"]) > today]
return past[-1:] + current + future[:1]
def probe_forms(
session: requests.Session,
base: str,
order_id: object,
delivery: dict[str, Any],
) -> dict[str, Any]:
"""Try every candidate URL for one delivery, keyed by id and by date.
Args:
session: A logged-in session.
base: The API base that accepted the login.
order_id: The active order's id.
delivery: One delivery object from the order payload.
Returns:
A record per attempted URL.
"""
results: dict[str, Any] = {}
keys = {"by_id": delivery["deliveryId"], "by_date": delivery["date"]}
for template in CANDIDATES:
for label, key in keys.items():
path = template.format(order=order_id, key=key)
response = session.get(f"{base}/{path}", timeout=TIMEOUT)
emit(f" {label:8} {path} -> {response.status_code}")
if response.ok:
results[f"{template}|{label}"] = describe(response)
return results
def _session_works(session: requests.Session, base: str, order_id: object) -> bool:
"""Return True when the attached cookie still authenticates.
A cached cookie that has expired looks identical to a good one until it is
used, so it is checked against a cheap authenticated endpoint before the
probe commits to it -- otherwise every subsequent call 401s and the run
reports a shape problem that is really an auth problem.
"""
probe = session.get(f"{base}/company/customer/order/{order_id}", timeout=TIMEOUT)
return probe.ok
def main() -> int:
"""Run the second probe pass and write its capture.
Returns:
0 on success, 1 when login failed.
"""
parser = argparse.ArgumentParser(description="Probe the delivery-details shape.")
parser.add_argument("--capture", default="kuchnia_probe.json")
parser.add_argument("--out", default="kuchnia_probe_details.json")
args = parser.parse_args()
capture = json.loads(Path(args.capture).read_text(encoding="utf-8"))
order_id = capture["steps"]["order"]["body"]["orderId"]
targets = pick_deliveries(capture)
emit(f"Order {order_id}; trying {len(targets)} deliveries.")
session = build_session()
base = f"{ORIGIN}/api"
if attach_cached_session(session) and _session_works(session, base, order_id):
emit("Reusing the cached session -- no login needed.")
else:
# Either no cache, or the cached cookie has expired. Drop it and log in
# once; the fresh cookie is cached so the next run is prompt-free again.
clear_session_cookie()
username, password = prompt_credentials()
resolved = login(session, username, password)
if resolved is None:
emit("Login rejected -- check the credentials.")
return 1
base = resolved
remember_session(session)
# No XSRF-TOKEN cookie was issued on the first pass, so the CSRF echo is
# simply skipped when absent rather than sent empty.
xsrf = session.cookies.get("XSRF-TOKEN")
if xsrf is not None:
session.headers["X-XSRF-TOKEN"] = xsrf
found: dict[str, Any] = {}
for delivery in targets:
emit(f" delivery {delivery['deliveryId']} ({delivery['date']}):")
found[str(delivery["date"])] = probe_forms(session, base, order_id, delivery)
Path(args.out).write_text(
json.dumps(
{"order_id": order_id, "results": found}, indent=2, ensure_ascii=False
),
encoding="utf-8",
)
emit(f"\nWrote {Path(args.out).resolve()}")
return 0
if __name__ == "__main__":
sys.exit(main())

96
scripts/probe_kuchnia_shape.py Executable file
View File

@ -0,0 +1,96 @@
#!/usr/bin/env python3
"""Payload shaping for :mod:`probe_kuchnia`: redaction and delivery-id search.
Split out of ``probe_kuchnia.py`` to keep both files under the repo's 250-line
cap. Nothing here performs I/O, which also makes the redaction rule -- the
part with a real security consequence -- readable on its own.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
import requests
#: Keys whose *values* never reach the dump. Names are kept: knowing a cookie is
#: called ``XSRF-TOKEN`` is the point; knowing its value is a liability, and the
#: capture is a plain file the operator may well paste into a chat window.
SECRET_KEYS = frozenset(
{"password", "token", "accesstoken", "refreshtoken", "authorization", "cookie"},
)
REDACTED = "<redacted>"
#: How much of a non-JSON body to keep -- enough to recognise an HTML error page
#: or a proxy banner without pasting an entire document into the capture.
BODY_EXCERPT_CHARS = 500
def redact(value: object) -> object:
"""Return ``value`` with any secret-looking field replaced by a placeholder.
Args:
value: Arbitrary decoded JSON.
Returns:
The same structure with secret values masked.
"""
if isinstance(value, dict):
return {
key: REDACTED if str(key).casefold() in SECRET_KEYS else redact(item)
for key, item in value.items()
}
if isinstance(value, list):
return [redact(item) for item in value]
return value
def describe(response: requests.Response) -> dict[str, Any]:
"""Summarise ``response`` for the dump, body included when it is JSON.
Args:
response: The response to describe.
Returns:
A JSON-serialisable record of status, cookie names and body.
"""
try:
body: object = redact(response.json())
except ValueError:
# Not JSON: an HTML login redirect or an error page. Keep an excerpt --
# which of those it is answers "did the session actually authenticate?".
body = response.text[:BODY_EXCERPT_CHARS]
return {
"url": response.url,
"status": response.status_code,
"set_cookie_names": sorted({cookie.name for cookie in response.cookies}),
"body": body,
}
def find_delivery_id(payload: object) -> object:
"""Return the first ``deliveryId`` found anywhere in ``payload``.
The bundles treat this id as opaque and never derive it from a date, so it
is located by searching the order payload rather than constructed from
today's date -- which is exactly the assumption that would break the walk.
Args:
payload: Decoded order JSON.
Returns:
The first delivery id encountered, or None.
"""
if isinstance(payload, dict):
for key, value in payload.items():
if key == "deliveryId" and value is not None:
return value
found = find_delivery_id(value)
if found is not None:
return found
elif isinstance(payload, list):
for item in payload:
found = find_delivery_id(item)
if found is not None:
return found
return None