From 26f17700901e76ea4403b0a53cad8342766590f7 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 22 Aug 2026 23:04:22 +0200 Subject: [PATCH] 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. --- .gitignore | 3 + .pre-commit-config.yaml | 2 +- AGENTS.md | 3 + app/pubspec.yaml | 2 +- diet_guard/_cli.py | 7 + diet_guard/_cli_args.py | 2 + diet_guard/_cli_kuchnia.py | 119 ++++++++++ diet_guard/_cli_log.py | 24 ++ diet_guard/_constants.py | 35 +++ diet_guard/_gatelock.py | 13 +- diet_guard/_gatelock_dashboard.py | 75 ------ diet_guard/_gatelock_delivery.py | 240 ++++++++++++++++++++ diet_guard/_gatelock_kuchnia.py | 90 ++++++++ diet_guard/_gatelock_layout.py | 25 +- diet_guard/_gatelock_mealflow.py | 128 +++++------ diet_guard/_gatelock_ui_types.py | 1 + diet_guard/_kuchnia_client.py | 165 ++++++++++++++ diet_guard/_kuchnia_config.py | 130 +++++++++++ diet_guard/_kuchnia_errors.py | 18 ++ diet_guard/_kuchnia_import.py | 136 +++++++++++ diet_guard/_kuchnia_log.py | 75 ++++++ diet_guard/_kuchnia_orders.py | 95 ++++++++ diet_guard/_kuchnia_parse.py | 155 +++++++++++++ diet_guard/_kuchnia_spread.py | 86 +++++++ diet_guard/tests/_gate_keyboard_probe.py | 1 + diet_guard/tests/_gate_layout_probe.py | 1 + diet_guard/tests/_kuchnia_fakes.py | 145 ++++++++++++ diet_guard/tests/conftest.py | 52 ++++- diet_guard/tests/test_gatelock_dashboard.py | 26 +-- diet_guard/tests/test_kuchnia_autoload.py | 68 ++++++ diet_guard/tests/test_kuchnia_cli.py | 160 +++++++++++++ diet_guard/tests/test_kuchnia_client.py | 235 +++++++++++++++++++ diet_guard/tests/test_kuchnia_gate.py | 221 ++++++++++++++++++ diet_guard/tests/test_kuchnia_import.py | 225 ++++++++++++++++++ diet_guard/tests/test_kuchnia_orders.py | 119 ++++++++++ diet_guard/tests/test_kuchnia_parse.py | 186 +++++++++++++++ docs/kuchnia-wikinga.md | 144 ++++++++++++ scripts/probe_kuchnia.py | 211 +++++++++++++++++ scripts/probe_kuchnia_auth.py | 122 ++++++++++ scripts/probe_kuchnia_details.py | 174 ++++++++++++++ scripts/probe_kuchnia_shape.py | 96 ++++++++ 41 files changed, 3635 insertions(+), 180 deletions(-) create mode 100644 diet_guard/_cli_kuchnia.py delete mode 100644 diet_guard/_gatelock_dashboard.py create mode 100644 diet_guard/_gatelock_delivery.py create mode 100644 diet_guard/_gatelock_kuchnia.py create mode 100644 diet_guard/_kuchnia_client.py create mode 100644 diet_guard/_kuchnia_config.py create mode 100644 diet_guard/_kuchnia_errors.py create mode 100644 diet_guard/_kuchnia_import.py create mode 100644 diet_guard/_kuchnia_log.py create mode 100644 diet_guard/_kuchnia_orders.py create mode 100644 diet_guard/_kuchnia_parse.py create mode 100644 diet_guard/_kuchnia_spread.py create mode 100644 diet_guard/tests/_kuchnia_fakes.py create mode 100644 diet_guard/tests/test_kuchnia_autoload.py create mode 100644 diet_guard/tests/test_kuchnia_cli.py create mode 100644 diet_guard/tests/test_kuchnia_client.py create mode 100644 diet_guard/tests/test_kuchnia_gate.py create mode 100644 diet_guard/tests/test_kuchnia_import.py create mode 100644 diet_guard/tests/test_kuchnia_orders.py create mode 100644 diet_guard/tests/test_kuchnia_parse.py create mode 100644 docs/kuchnia-wikinga.md create mode 100755 scripts/probe_kuchnia.py create mode 100755 scripts/probe_kuchnia_auth.py create mode 100755 scripts/probe_kuchnia_details.py create mode 100755 scripts/probe_kuchnia_shape.py diff --git a/.gitignore b/.gitignore index 8c80821..e010ef1 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 0dfd775..5046b16 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -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] diff --git a/AGENTS.md b/AGENTS.md index 00019d7..5de3d19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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 diff --git a/app/pubspec.yaml b/app/pubspec.yaml index a922bb8..c20d0ae 100644 --- a/app/pubspec.yaml +++ b/app/pubspec.yaml @@ -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 diff --git a/diet_guard/_cli.py b/diet_guard/_cli.py index 534c14e..26fc122 100644 --- a/diet_guard/_cli.py +++ b/diet_guard/_cli.py @@ -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(), diff --git a/diet_guard/_cli_args.py b/diet_guard/_cli_args.py index 6df4b1f..bcbad82 100644 --- a/diet_guard/_cli_args.py +++ b/diet_guard/_cli_args.py @@ -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) diff --git a/diet_guard/_cli_kuchnia.py b/diet_guard/_cli_kuchnia.py new file mode 100644 index 0000000..e6c9e77 --- /dev/null +++ b/diet_guard/_cli_kuchnia.py @@ -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 diff --git a/diet_guard/_cli_log.py b/diet_guard/_cli_log.py index 079a2c7..b47fd98 100644 --- a/diet_guard/_cli_log.py +++ b/diet_guard/_cli_log.py @@ -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() diff --git a/diet_guard/_constants.py b/diet_guard/_constants.py index ac60698..4fd4797 100644 --- a/diet_guard/_constants.py +++ b/diet_guard/_constants.py @@ -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 diff --git a/diet_guard/_gatelock.py b/diet_guard/_gatelock.py index 2b6cfd8..262f332 100644 --- a/diet_guard/_gatelock.py +++ b/diet_guard/_gatelock.py @@ -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.""" diff --git a/diet_guard/_gatelock_dashboard.py b/diet_guard/_gatelock_dashboard.py deleted file mode 100644 index ff91485..0000000 --- a/diet_guard/_gatelock_dashboard.py +++ /dev/null @@ -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) diff --git a/diet_guard/_gatelock_delivery.py b/diet_guard/_gatelock_delivery.py new file mode 100644 index 0000000..a36ba47 --- /dev/null +++ b/diet_guard/_gatelock_delivery.py @@ -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() diff --git a/diet_guard/_gatelock_kuchnia.py b/diet_guard/_gatelock_kuchnia.py new file mode 100644 index 0000000..41e0f6f --- /dev/null +++ b/diet_guard/_gatelock_kuchnia.py @@ -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 diff --git a/diet_guard/_gatelock_layout.py b/diet_guard/_gatelock_layout.py index f3737da..0f5ff33 100644 --- a/diet_guard/_gatelock_layout.py +++ b/diet_guard/_gatelock_layout.py @@ -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( diff --git a/diet_guard/_gatelock_mealflow.py b/diet_guard/_gatelock_mealflow.py index 113a2c3..8b939e1 100644 --- a/diet_guard/_gatelock_mealflow.py +++ b/diet_guard/_gatelock_mealflow.py @@ -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.""" diff --git a/diet_guard/_gatelock_ui_types.py b/diet_guard/_gatelock_ui_types.py index bbe3cdf..117fb8e 100644 --- a/diet_guard/_gatelock_ui_types.py +++ b/diet_guard/_gatelock_ui_types.py @@ -84,3 +84,4 @@ class GateCallbacks: on_submit: Callable[[], None] on_close: Callable[[], None] on_fetch_sync: Callable[[], None] + on_load_delivery: Callable[[], None] diff --git a/diet_guard/_kuchnia_client.py b/diet_guard/_kuchnia_client.py new file mode 100644 index 0000000..0d04e7c --- /dev/null +++ b/diet_guard/_kuchnia_client.py @@ -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 diff --git a/diet_guard/_kuchnia_config.py b/diet_guard/_kuchnia_config.py new file mode 100644 index 0000000..d1ce85d --- /dev/null +++ b/diet_guard/_kuchnia_config.py @@ -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") diff --git a/diet_guard/_kuchnia_errors.py b/diet_guard/_kuchnia_errors.py new file mode 100644 index 0000000..bd52090 --- /dev/null +++ b/diet_guard/_kuchnia_errors.py @@ -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.""" diff --git a/diet_guard/_kuchnia_import.py b/diet_guard/_kuchnia_import.py new file mode 100644 index 0000000..745e735 --- /dev/null +++ b/diet_guard/_kuchnia_import.py @@ -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 diff --git a/diet_guard/_kuchnia_log.py b/diet_guard/_kuchnia_log.py new file mode 100644 index 0000000..04748db --- /dev/null +++ b/diet_guard/_kuchnia_log.py @@ -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 diff --git a/diet_guard/_kuchnia_orders.py b/diet_guard/_kuchnia_orders.py new file mode 100644 index 0000000..fc19436 --- /dev/null +++ b/diet_guard/_kuchnia_orders.py @@ -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 diff --git a/diet_guard/_kuchnia_parse.py b/diet_guard/_kuchnia_parse.py new file mode 100644 index 0000000..6ebeaa5 --- /dev/null +++ b/diet_guard/_kuchnia_parse.py @@ -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] diff --git a/diet_guard/_kuchnia_spread.py b/diet_guard/_kuchnia_spread.py new file mode 100644 index 0000000..25f40df --- /dev/null +++ b/diet_guard/_kuchnia_spread.py @@ -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) diff --git a/diet_guard/tests/_gate_keyboard_probe.py b/diet_guard/tests/_gate_keyboard_probe.py index 02df314..8094eb2 100644 --- a/diet_guard/tests/_gate_keyboard_probe.py +++ b/diet_guard/tests/_gate_keyboard_probe.py @@ -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 diff --git a/diet_guard/tests/_gate_layout_probe.py b/diet_guard/tests/_gate_layout_probe.py index fac4923..5b6fb0b 100644 --- a/diet_guard/tests/_gate_layout_probe.py +++ b/diet_guard/tests/_gate_layout_probe.py @@ -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 diff --git a/diet_guard/tests/_kuchnia_fakes.py b/diet_guard/tests/_kuchnia_fakes.py new file mode 100644 index 0000000..ec3c340 --- /dev/null +++ b/diet_guard/tests/_kuchnia_fakes.py @@ -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}, + ], + } diff --git a/diet_guard/tests/conftest.py b/diet_guard/tests/conftest.py index 2bb730b..214e6ac 100644 --- a/diet_guard/tests/conftest.py +++ b/diet_guard/tests/conftest.py @@ -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 diff --git a/diet_guard/tests/test_gatelock_dashboard.py b/diet_guard/tests/test_gatelock_dashboard.py index c896cb9..a21e263 100644 --- a/diet_guard/tests/test_gatelock_dashboard.py +++ b/diet_guard/tests/test_gatelock_dashboard.py @@ -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 diff --git a/diet_guard/tests/test_kuchnia_autoload.py b/diet_guard/tests/test_kuchnia_autoload.py new file mode 100644 index 0000000..43301c6 --- /dev/null +++ b/diet_guard/tests/test_kuchnia_autoload.py @@ -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 diff --git a/diet_guard/tests/test_kuchnia_cli.py b/diet_guard/tests/test_kuchnia_cli.py new file mode 100644 index 0000000..c19fdba --- /dev/null +++ b/diet_guard/tests/test_kuchnia_cli.py @@ -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) diff --git a/diet_guard/tests/test_kuchnia_client.py b/diet_guard/tests/test_kuchnia_client.py new file mode 100644 index 0000000..91156c1 --- /dev/null +++ b/diet_guard/tests/test_kuchnia_client.py @@ -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="")]) + 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) diff --git a/diet_guard/tests/test_kuchnia_gate.py b/diet_guard/tests/test_kuchnia_gate.py new file mode 100644 index 0000000..cbdeacc --- /dev/null +++ b/diet_guard/tests/test_kuchnia_gate.py @@ -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) diff --git a/diet_guard/tests/test_kuchnia_import.py b/diet_guard/tests/test_kuchnia_import.py new file mode 100644 index 0000000..a28b257 --- /dev/null +++ b/diet_guard/tests/test_kuchnia_import.py @@ -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([]) == [] diff --git a/diet_guard/tests/test_kuchnia_orders.py b/diet_guard/tests/test_kuchnia_orders.py new file mode 100644 index 0000000..782515f --- /dev/null +++ b/diet_guard/tests/test_kuchnia_orders.py @@ -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) == [] diff --git a/diet_guard/tests/test_kuchnia_parse.py b/diet_guard/tests/test_kuchnia_parse.py new file mode 100644 index 0000000..518b909 --- /dev/null +++ b/diet_guard/tests/test_kuchnia_parse.py @@ -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) == [] diff --git a/docs/kuchnia-wikinga.md b/docs/kuchnia-wikinga.md new file mode 100644 index 0000000..2f42b8c --- /dev/null +++ b/docs/kuchnia-wikinga.md @@ -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. diff --git a/scripts/probe_kuchnia.py b/scripts/probe_kuchnia.py new file mode 100755 index 0000000..a5e9822 --- /dev/null +++ b/scripts/probe_kuchnia.py @@ -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()) diff --git a/scripts/probe_kuchnia_auth.py b/scripts/probe_kuchnia_auth.py new file mode 100755 index 0000000..2ab06fa --- /dev/null +++ b/scripts/probe_kuchnia_auth.py @@ -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) diff --git a/scripts/probe_kuchnia_details.py b/scripts/probe_kuchnia_details.py new file mode 100755 index 0000000..5434cdd --- /dev/null +++ b/scripts/probe_kuchnia_details.py @@ -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()) diff --git a/scripts/probe_kuchnia_shape.py b/scripts/probe_kuchnia_shape.py new file mode 100755 index 0000000..3e1897b --- /dev/null +++ b/scripts/probe_kuchnia_shape.py @@ -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 = "" + +#: 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