gate + CLI: fill today's empty slots from catering in one confirmed step
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
file-length / file length <= 250 lines (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (mypy, mypy, ) (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (pylint, pylint, ) (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (rest, , mypy,pylint) (push) Waiting to run
Tests / test (push) Waiting to run
release-apk / release (APK) (push) Waiting to run
repo-contract / repo contract (push) Waiting to run

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-10-10 13:12:15 +02:00
parent 11173c7088
commit a7e3685963
18 changed files with 848 additions and 33 deletions

View File

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

View File

@ -24,13 +24,16 @@ from diet_guard._state import now_local
if TYPE_CHECKING: # pragma: no cover - typing only
import argparse
from collections.abc import Callable
from collections.abc import Callable, Sequence
from diet_guard._kuchnia_spread import SlottedDish
# ``_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"),
"fill_plan": ("diet_guard._kuchnia_log", "fill_plan"),
}
@ -53,7 +56,7 @@ def register_kuchnia_subparser(sub: argparse._SubParsersAction) -> None:
kuchnia.add_argument(
"--log",
action="store_true",
help="Also log the dishes as eaten (asks first unless --yes).",
help="Also log the dishes into today's empty slots (asks first unless --yes).",
)
kuchnia.add_argument(
"--yes",
@ -105,15 +108,31 @@ def cmd_kuchnia(
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).")
_log_into_empty_slots(module.fill_plan(slotted), emit, ask, yes=yes)
return 0
def _log_into_empty_slots(
plan: Sequence[SlottedDish],
emit: Callable[[str], None],
ask: Callable[[str], str],
*,
yes: bool,
) -> None:
"""Confirm, then log ``plan`` -- the same rule as the gate's "Fill all".
Args:
plan: The dishes whose slots are still empty today (``fill_plan``).
emit: A one-line output sink.
ask: A prompt function, used only for the confirmation.
yes: Skip the confirmation prompt.
"""
if not plan:
emit("every slot already logged today; nothing to add.")
return
question = f"log {len(plan)} into today's empty slots as eaten? [y/N] "
if not yes and ask(question).strip().lower() != "y":
emit("nothing logged.")
return
written = sys.modules[__name__].log_dishes(plan)
emit(f"logged {len(written)} meal(s).")

View File

@ -142,8 +142,7 @@ class MealGate(_GateCalendar):
app_name="diet_guard",
rank=RANK_DIET_GUARD,
)
# Published here, then owned by LockWindow, which releases it on
# close -- so deliberately not kept as an attribute.
# Published here, then owned (and released on close) by LockWindow.
arbiter = Arbiter(
"diet_guard",
RANK_DIET_GUARD,
@ -167,6 +166,7 @@ class MealGate(_GateCalendar):
on_close=self.close,
on_fetch_sync=self._on_fetch_sync,
on_load_delivery=self._on_load_delivery,
on_fill_all=self._on_fill_all,
)
self._build()

View File

@ -42,6 +42,9 @@ if TYPE_CHECKING: # pragma: no cover - typing only
# ``_cli_gate`` and ``_cli_prune`` use, which also keeps ``patch.object`` working.
_LAZY_ATTRS = ("refresh_delivery", "refresh_delivery_once")
#: ``(unlock reason, nothing-new status)`` for a sync pull; "Fill all" has its own.
_SYNC_LABELS = ("Synced from another device", "No new meals found in sync.")
def __getattr__(name: str) -> object:
"""Resolve the deferred catering import on first attribute access."""
@ -226,17 +229,18 @@ class _PullFlows(_GateNutrition):
return
self._reconcile_after_fetch()
def _reconcile_after_fetch(self) -> None:
"""Drop slots a pulled meal now covers; unlock when none remain."""
def _reconcile_after_fetch(self, labels: tuple[str, str] = _SYNC_LABELS) -> None:
"""Drop slots a pulled meal now covers; unlock if none remain."""
unlock, nothing_new = labels
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.")
self._set_status(nothing_new)
return
self._pending = [slot for slot in self._pending if slot in still_due]
if not self._pending:
self._unlock("Synced from another device")
self._unlock(unlock)
return
self._clear_inputs()
self._refresh_slot_header()

View File

@ -0,0 +1,169 @@
"""The lock screen's "Fill all" button: today's catering into every empty slot.
Two clicks, never one. The first fetches the day's delivery off the Tk thread
and shows exactly what would be written; the button relabels to "Confirm (N)".
Only the second click writes. A delivered meal is not an eaten meal, so the
user sees the dishes and the slots before anything lands -- and no modal
dialog asks instead, because a pop-up over a fullscreen grab risks stealing the
very keyboard grab the lock depends on.
This is *composed* into the gate rather than mixed in: pylint's
``max-parents`` already counts every class in the gate's MRO, so a new mixin
fails the hook, and module functions reaching into the gate trip ``SLF001``.
The gate hands this object the few callables it needs instead.
Which slots get filled -- every slot today, skipping any that already holds an
entry -- is :func:`diet_guard._kuchnia_log.fill_plan`, shared with
``diet-guard kuchnia --log`` so the CLI and the gate cannot disagree.
"""
from __future__ import annotations
from importlib import import_module
import queue
from typing import TYPE_CHECKING, NamedTuple
from diet_guard._gatelock_fetch import FETCH_POLL_MS
from diet_guard._gatelock_kuchnia import start_delivery_fetch
from diet_guard._kuchnia_log import fill_plan, log_dishes
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
from collections.abc import Callable, Sequence
from datetime import date
import tkinter as tk
from diet_guard._gatelock_kuchnia import DeliveryResult, RefreshFn
from diet_guard._kuchnia_parse import Dish
from diet_guard._kuchnia_spread import SlottedDish
#: The button's resting label; also what it returns to after each attempt.
FILL_LABEL = "🍱 Fill all"
#: ``(unlock reason, nothing-new status)`` for the gate's post-write reconcile:
#: a fill that covered only later slots must not report "found in sync".
RECONCILE_LABELS = (
"Filled from catering",
"Catering logged; this slot still needs a meal.",
)
def _refresh_delivery(day: date) -> tuple[Sequence[Dish], str | None]:
"""Fetch ``day``'s dishes, importing the HTTP stack only when clicked.
``_kuchnia_import`` reaches ``requests``; the gate's frequent not-due tick
imports this module and must not pay for it.
"""
refresh: RefreshFn = import_module("diet_guard._kuchnia_import").refresh_delivery
return refresh(day)
class FillHooks(NamedTuple):
"""What the flow calls back into the gate."""
#: The gate's status-line writer.
set_status: Callable[..., None]
#: Called after a write, so the gate drops the slots it now covers and
#: unlocks when none remain.
on_logged: Callable[[], None]
class FillAllFlow:
"""Drives one gate's "Fill all" button through fetch, confirm and log."""
def __init__(
self,
root: tk.Misc,
label: tk.StringVar,
hooks: FillHooks,
*,
demo_mode: bool,
) -> None:
"""Bind the flow to a gate.
Args:
root: Schedules the poll on the Tk thread.
label: The button's text, shared by its copy on every monitor.
hooks: The gate callbacks the flow drives.
demo_mode: A demo window refuses: it must never write real
entries from a delivery note.
"""
self._root = root
self._label = label
self._demo_mode = demo_mode
self._set_status = hooks.set_status
self._on_logged = hooks.on_logged
self._result: queue.Queue[DeliveryResult] | None = None
self._plan: list[SlottedDish] = []
self._plan_day: date | None = None
def click(self) -> None:
"""First click fetches and proposes; a second click confirms."""
if self._demo_mode:
self._set_status("Fill all is only available on the real lock.")
return
if self._result is not None:
return # a fetch is in flight; a double-click must not start two
today = now_local().date()
if self._plan and self._plan_day == today:
self._confirm()
return
# No plan, or one proposed before midnight: fetch (again). The day is
# pinned here, not when the result lands, so a fetch still in flight
# at midnight can never confirm yesterday's menu into today.
self._reset()
self._plan_day = today
self._set_status("Loading today's delivery…")
self._result = start_delivery_fetch(_refresh_delivery, today)
self._root.after(FETCH_POLL_MS, self._poll)
def _poll(self) -> None:
"""On the Tk thread: turn the fetched dishes into a proposal."""
result = self._result
if result is None:
return
try:
outcome = result.get_nowait()
except queue.Empty:
self._root.after(FETCH_POLL_MS, self._poll)
return
self._result = None
if outcome.reason is not None:
self._set_status(f"{outcome.reason} — still locked.", error=True)
return
if not outcome.dishes:
self._set_status("No catering delivery today.")
return
plan = fill_plan(assign_slots(outcome.dishes, day_slots(current_schedule())))
if not plan:
self._set_status("Every slot already has a meal — nothing to fill.")
return
self._plan = plan
self._label.set(f"✓ Confirm ({len(plan)})")
total = sum(item.dish.kcal for item in plan)
listing = ", ".join(f"{item.slot:02d}:00 {item.dish.name}" for item in plan)
self._set_status(
f"Will log {len(plan)} ({total:g} kcal): {listing} — click Confirm."
)
def _confirm(self) -> None:
"""Write the proposed dishes into the slots still empty right now."""
# Re-checked at write time: a meal typed in between the two clicks
# has taken its slot, and must keep it.
plan = fill_plan(self._plan)
self._reset()
if not plan:
self._set_status("Every slot already has a meal — nothing to fill.")
return
written = log_dishes(plan)
self._set_status(f"Logged {len(written)} catering dish(es).")
self._on_logged()
def _reset(self) -> None:
"""Forget any proposal and restore the button's label."""
self._plan = []
self._plan_day = None
self._label.set(FILL_LABEL)

View File

@ -107,8 +107,10 @@ def _build_readouts(body: tk.Frame, vars_: GateVars) -> None:
).pack(pady=(XS, XS))
def _build_actions(body: tk.Frame, callbacks: GateCallbacks) -> None:
"""Pack the submit button and the manual sync-fetch button."""
def _build_actions(
body: tk.Frame, callbacks: GateCallbacks, fill_label: tk.StringVar
) -> None:
"""Pack the submit button and the three "pull it in" buttons."""
make_button(
body,
_COLORS,
@ -141,7 +143,20 @@ def _build_actions(body: tk.Frame, callbacks: GateCallbacks) -> None:
"🍱 Today's delivery",
callbacks.on_load_delivery,
ButtonStyle(variant="secondary", bold=False),
).pack(side="left")
).pack(side="left", padx=(0, XS))
# Every empty slot from today's catering, in two clicks: the first shows
# what would be logged and relabels this to "Confirm (N)" -- through the
# shared variable, so the copy on every monitor relabels with it.
fill = make_button(
pulls,
_COLORS,
fill_label.get(),
callbacks.on_fill_all,
ButtonStyle(variant="secondary", bold=False),
)
fill.configure(textvariable=fill_label)
fill.pack(side="left")
def build_layout(
@ -183,7 +198,7 @@ def build_layout(
basis_prefix, per_entry, macros = build_macro_section(root, body, vars_)
_build_readouts(body, vars_)
_build_actions(body, callbacks)
_build_actions(body, callbacks, vars_.fill_label)
status_label = tk.Label(
body,

View File

@ -11,6 +11,7 @@ calorie/macro dashboard it calls into lives in :mod:`._gatelock_dashboard`.
from __future__ import annotations
import contextlib
from functools import partial
import tkinter as tk
from typing import TYPE_CHECKING
@ -18,6 +19,7 @@ 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._gatelock_delivery import _PullFlows
from diet_guard._gatelock_fillall import RECONCILE_LABELS, FillAllFlow, FillHooks
from diet_guard._gatelock_ui import ERR, FG, UNIT_GRAMS
from diet_guard._resolve import lookup_candidates
from diet_guard._slots import slot_label
@ -63,6 +65,18 @@ class _GateMealFlow(_PullFlows):
self._vars.preview.set("")
self._refresh_projection()
#: Built on the first "Fill all" click; it holds that button's state.
_fill_flow: FillAllFlow | None = None
def _on_fill_all(self) -> None:
"""Fill every empty slot from today's catering, on a confirming click."""
if self._fill_flow is None:
unlock = partial(self._reconcile_after_fetch, RECONCILE_LABELS)
hooks = FillHooks(self._set_status, unlock)
label, demo = self._vars.fill_label, self.demo_mode
self._fill_flow = FillAllFlow(self.root, label, hooks, demo_mode=demo)
self._fill_flow.click()
# -- behaviour ------------------------------------------------------------
def _set_status(self, text: str, *, error: bool = False) -> None:
@ -174,11 +188,13 @@ class _GateMealFlow(_PullFlows):
Teardown is scheduled *before* the budget is looked up, so a corrupt
budget file (which raises) can never re-trap the user at unlock time.
"""
# A 5-meal plan against 4 slots leaves a dish queued when the last slot
# is satisfied (``_kuchnia_spread`` doubles up the earliest slots), so
# say what was not offered rather than dropping it silently. Purely
# informational: the dishes are already banked, and the lock is over.
left = len(self._delivery_pending)
# 5 dishes on 4 slots leave one queued when the last slot is satisfied:
# name it rather than drop it. Already-logged dishes ("Fill all", a
# pull) do not count. Informational only; the lock is over.
eaten = {str(e.get("desc", "")).strip().casefold() for e in today_entries()}
left = sum(
d.name.strip().casefold() not in eaten for d in self._delivery_pending
)
if left:
noun = "dish" if left == 1 else "dishes"
logged = f"{logged} ({left} more {noun} delivered; log with 'ate')"

View File

@ -25,6 +25,7 @@ from gatelock import (
escape_text_tab_trap,
)
from diet_guard._gatelock_fillall import FILL_LABEL
from diet_guard._gatelock_spacing import SM, XS
from diet_guard._gatelock_typography import (
BODY,
@ -100,6 +101,7 @@ def make_vars(root: tk.Misc) -> GateVars:
carbs=tk.StringVar(master=root, value=""),
fat=tk.StringVar(master=root, value=""),
),
fill_label=tk.StringVar(master=root, value=FILL_LABEL),
)

View File

@ -55,6 +55,8 @@ class GateVars:
dashboard: tk.StringVar
unit: tk.StringVar
entries: GateEntryVars
#: "Fill all"'s label, shared so it relabels on every monitor at once.
fill_label: tk.StringVar
@dataclass
@ -85,3 +87,4 @@ class GateCallbacks:
on_close: Callable[[], None]
on_fetch_sync: Callable[[], None]
on_load_delivery: Callable[[], None]
on_fill_all: Callable[[], None]

View File

@ -42,6 +42,24 @@ def _already_logged(entries: Sequence[dict[str, object]], name: str, slot: int)
)
def fill_plan(slotted: Sequence[SlottedDish]) -> list[SlottedDish]:
"""Keep only the dishes whose slot has nothing logged in it today.
"Fill all with catering" fills *empty* slots: a slot the user already
logged something else into keeps that meal rather than gaining a second
one. Occupancy is read once, before the batch, so two dishes the spread
put in the same empty slot (5 dishes on 4 slots) both land.
Args:
slotted: Today's dishes, already paired with slots.
Returns:
The dishes to log, in the order given.
"""
taken = {entry.get("slot") for entry in today_entries()}
return [item for item in slotted if item.slot not in taken]
def dish_nutrition(dish: Dish) -> Nutrition:
"""Return a dish as the whole-portion :class:`Nutrition` it gets logged as.

View File

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

View File

@ -90,6 +90,8 @@ def _make_vars(*, populated: bool) -> GateVars:
dashboard=tk.StringVar(value=dashboard),
unit=tk.StringVar(value="grams"),
entries=entries,
# The wider of the button's two labels, so the row is measured at worst.
fill_label=tk.StringVar(value="✓ Confirm (5)" if populated else "🍱 Fill all"),
)
@ -150,6 +152,7 @@ def measure(
on_close=_noop,
on_fetch_sync=_noop,
on_load_delivery=_noop,
on_fill_all=_noop,
)
widgets = build_layout(
notebook, _make_vars(populated=populated), callbacks, demo_mode=False

View File

@ -136,6 +136,11 @@ def state_redirects(tmp_path: Path) -> list[_patch[object]]:
"diet_guard._gatelock_delivery.refresh_delivery",
return_value=([], None),
),
# The gate's "Fill all" button fetches through its own lazy helper.
patch(
"diet_guard._gatelock_fillall._refresh_delivery",
return_value=([], None),
),
# The Body tab's profile, weight log and published workouts.
patch("diet_guard._body_store.BODY_FILE", tmp_path / "body.json"),
# Workout sources live outside diet_guard entirely: RunnerUp's WebDAV

View File

@ -0,0 +1,250 @@
"""Tests for the lock screen's two-click "Fill all" catering flow.
The properties that matter, because the gate exists to stop unattended logging:
* the first click **never writes** -- it only proposes, and relabels the button;
* the second click writes only into slots that are **still empty right now**,
so a meal typed between the two clicks keeps its slot;
* a proposal does not survive midnight.
Driven without threads, as in :mod:`test_kuchnia_gate`: the fetch hands back a
pre-filled queue and the poll is called directly.
"""
from __future__ import annotations
from datetime import timedelta
import queue
from typing import TYPE_CHECKING, cast
from unittest.mock import MagicMock, patch
from diet_guard import _gatelock_fillall
from diet_guard._gatelock_fetch import FETCH_POLL_MS
from diet_guard._gatelock_fillall import FILL_LABEL, FillAllFlow, FillHooks
from diet_guard._gatelock_kuchnia import DeliveryResult
from diet_guard._kuchnia_parse import Dish
from diet_guard._meal_schedule_store import current_schedule
from diet_guard._slots import day_slots
from diet_guard._state import log_meal, now_local
from diet_guard._state_today import today_entries
from diet_guard.tests._tk_fakes import FakeVar
from diet_guard.tests.conftest import _nutrition
if TYPE_CHECKING:
from collections.abc import Iterator
from contextlib import AbstractContextManager
import tkinter as tk
def dish(name: str, priority: int, kcal: float = 400.0) -> Dish:
"""A catering dish with its own name, meal order and calories."""
return Dish(name, kcal, 30.0, 40.0, 10.0, 300.0, priority, "")
#: One dish per default slot, in the caterer's meal order.
FOUR = (
dish("Owsianka", 1, 350),
dish("Wrap", 2, 420.5),
dish("Obiad", 3, 600),
dish("Kolacja", 4, 194),
)
class Harness:
"""A flow plus everything it calls back into, recorded."""
def __init__(self, *, demo_mode: bool = False) -> None:
self.root = MagicMock()
self.label = FakeVar(value=FILL_LABEL)
self.statuses: list[tuple[str, bool]] = []
self.on_logged = MagicMock()
hooks = FillHooks(self._status, self.on_logged)
label = cast("tk.StringVar", self.label)
self.flow = FillAllFlow(self.root, label, hooks, demo_mode=demo_mode)
def _status(self, text: str, *, error: bool = False) -> None:
self.statuses.append((text, error))
@property
def status(self) -> str:
return self.statuses[-1][0]
def fetching(*outcomes: DeliveryResult) -> AbstractContextManager[MagicMock]:
"""Patch the fetch so each click gets one pre-delivered result, in order."""
def _results() -> Iterator[queue.Queue[DeliveryResult]]:
for outcome in outcomes:
result: queue.Queue[DeliveryResult] = queue.Queue(maxsize=1)
result.put(outcome)
yield result
return patch.object(
_gatelock_fillall, "start_delivery_fetch", side_effect=_results()
)
def delivered(*dishes: Dish) -> DeliveryResult:
return DeliveryResult(dishes=dishes, reason=None)
def logged() -> dict[int, list[str]]:
"""Today's on-disk log as slot -> descriptions."""
by_slot: dict[int, list[str]] = {}
for entry in today_entries():
by_slot.setdefault(int(entry["slot"]), []).append(str(entry["desc"]))
return by_slot
def propose(h: Harness) -> None:
"""The first click: fetch, then the poll that turns dishes into a plan."""
h.flow.click()
h.flow._poll()
def test_default_schedule_precondition() -> None:
# Every expectation below assumes the redirected (default) schedule.
assert day_slots(current_schedule()) == (8, 12, 16, 20)
class TestFirstClick:
def test_demo_mode_refuses_without_fetching(self) -> None:
h = Harness(demo_mode=True)
with fetching() as start:
h.flow.click()
start.assert_not_called()
assert "only available on the real lock" in h.status
assert h.label.get() == FILL_LABEL
def test_a_click_while_fetching_is_ignored(self) -> None:
h = Harness()
with patch.object(_gatelock_fillall, "start_delivery_fetch") as start:
start.return_value = queue.Queue(maxsize=1)
h.flow.click()
h.flow.click()
start.assert_called_once_with(
_gatelock_fillall._refresh_delivery, now_local().date()
)
assert h.statuses == [("Loading today's delivery…", False)]
h.root.after.assert_called_once_with(FETCH_POLL_MS, h.flow._poll)
def test_the_poll_rearms_while_the_worker_runs(self) -> None:
h = Harness()
with patch.object(_gatelock_fillall, "start_delivery_fetch") as start:
start.return_value = queue.Queue(maxsize=1)
h.flow.click()
h.flow._poll()
assert h.root.after.call_count == 2
h.root.after.assert_called_with(FETCH_POLL_MS, h.flow._poll)
def test_a_stray_poll_is_inert(self) -> None:
h = Harness()
h.flow._poll()
assert h.statuses == []
h.root.after.assert_not_called()
def test_an_outage_proposes_nothing(self) -> None:
h = Harness()
with fetching(DeliveryResult(dishes=(), reason="panel down")):
propose(h)
assert h.statuses[-1] == ("panel down — still locked.", True)
assert h.label.get() == FILL_LABEL
# Fetch done: the next click may fetch again rather than being ignored.
assert h.flow._result is None
def test_no_delivery_today(self) -> None:
h = Harness()
with fetching(delivered()):
propose(h)
assert h.status == "No catering delivery today."
assert h.label.get() == FILL_LABEL
def test_every_slot_already_logged(self) -> None:
for slot in (8, 12, 16, 20):
log_meal(f"mine {slot}", _nutrition(), slot)
h = Harness()
with fetching(delivered(*FOUR)):
propose(h)
assert h.status == "Every slot already has a meal — nothing to fill."
assert h.label.get() == FILL_LABEL
def test_the_proposal_writes_nothing(self) -> None:
log_meal("my sandwich", _nutrition(), 12)
h = Harness()
with fetching(delivered(*FOUR)):
propose(h)
assert h.label.get() == "✓ Confirm (3)"
assert h.status == (
"Will log 3 (1144 kcal): 08:00 Owsianka, 16:00 Obiad, "
"20:00 Kolacja — click Confirm."
)
assert logged() == {12: ["my sandwich"]}
h.on_logged.assert_not_called()
class TestConfirm:
def test_confirm_fills_only_the_empty_slots(self) -> None:
log_meal("my sandwich", _nutrition(), 12)
h = Harness()
with fetching(delivered(*FOUR)) as start:
propose(h)
h.flow.click()
assert start.call_count == 1 # the confirm did not refetch
assert logged() == {
8: ["Owsianka"],
12: ["my sandwich"],
16: ["Obiad"],
20: ["Kolacja"],
}
assert h.status == "Logged 3 catering dish(es)."
h.on_logged.assert_called_once_with()
assert h.label.get() == FILL_LABEL
def test_two_dishes_spread_into_one_empty_slot_both_land(self) -> None:
# 5 dishes on 4 slots: the spread doubles up 08:00.
h = Harness()
five = (*FOUR, dish("Przekaska", 5, 100))
with fetching(delivered(*five)):
propose(h)
h.flow.click()
assert logged()[8] == ["Owsianka", "Wrap"]
assert h.status == "Logged 5 catering dish(es)."
def test_a_meal_typed_between_the_clicks_keeps_its_slot(self) -> None:
h = Harness()
with fetching(delivered(*FOUR)):
propose(h)
assert h.label.get() == "✓ Confirm (4)"
log_meal("toast", _nutrition(), 8)
h.flow.click()
assert logged()[8] == ["toast"]
assert h.status == "Logged 3 catering dish(es)."
def test_every_slot_taken_between_the_clicks(self) -> None:
h = Harness()
with fetching(delivered(*FOUR)):
propose(h)
for slot in (8, 12, 16, 20):
log_meal(f"mine {slot}", _nutrition(), slot)
h.flow.click()
assert h.status == "Every slot already has a meal — nothing to fill."
h.on_logged.assert_not_called()
assert h.label.get() == FILL_LABEL
assert all(descs == [f"mine {s}"] for s, descs in logged().items())
def test_a_proposal_from_yesterday_is_refetched(self) -> None:
h = Harness()
tomorrow = now_local() + timedelta(days=1)
with fetching(delivered(*FOUR), delivered()) as start:
propose(h)
with patch.object(_gatelock_fillall, "now_local", return_value=tomorrow):
h.flow.click()
assert start.call_count == 2
assert start.call_args.args == (
_gatelock_fillall._refresh_delivery,
tomorrow.date(),
)
assert logged() == {}
h.on_logged.assert_not_called()
assert h.label.get() == FILL_LABEL
assert h.status == "Loading today's delivery…"

View File

@ -0,0 +1,226 @@
""" "Fill all" wired into the real gate: the button, the unlock, the leftovers.
:mod:`test_gatelock_fillall` drives the flow alone; this pins what the gate
adds around it -- one flow per gate, an unlock that names catering as its
reason, a button that relabels on every monitor, and an unlock line that does
not claim already-logged dishes are still waiting.
"""
from __future__ import annotations
from datetime import timedelta
import os
import queue
import tkinter as tk
from tkinter import ttk
from typing import TYPE_CHECKING
from unittest.mock import MagicMock, patch
import pytest
from diet_guard import _gatelock_delivery, _gatelock_fillall, _gatelock_layout
from diet_guard._gatelock import MealGate
from diet_guard._gatelock_fillall import FILL_LABEL, _refresh_delivery
from diet_guard._gatelock_layout import build_layout
from diet_guard._gatelock_mealflow import _UNLOCK_DELAY_MS
from diet_guard._gatelock_ui import GateCallbacks, make_vars
from diet_guard._state import log_meal, now_local
from diet_guard._state_today import today_entries
from diet_guard.tests._gate_fixtures import fake_tk
from diet_guard.tests.conftest import _nutrition
from diet_guard.tests.test_gatelock_fillall import (
FOUR,
Harness,
delivered,
dish,
fetching,
logged,
)
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
from gatelock import ButtonStyle, LockConfig
from pytest_xvfb import Xvfb
from diet_guard._gatelock_kuchnia import DeliveryResult
def _due_from_log(pending: list[int]) -> Callable[[], tuple[int, ...]]:
"""``due_slots`` as the log on disk says, so an unlock needs a real write."""
def due() -> tuple[int, ...]:
taken = {entry.get("slot") for entry in today_entries()}
return tuple(slot for slot in pending if slot not in taken)
return due
def _fill(gate: MealGate, outcome: DeliveryResult) -> None:
"""Click "Fill all", let the poll propose, then click again to confirm."""
result: queue.Queue[DeliveryResult] = queue.Queue(maxsize=1)
result.put(outcome)
with patch.object(_gatelock_fillall, "start_delivery_fetch", return_value=result):
gate._on_fill_all()
flow = gate._fill_flow
assert flow is not None
flow._poll()
gate._on_fill_all()
assert gate._fill_flow is flow # built once, then reused
def test_the_fetch_imports_the_http_stack_lazily() -> None:
day = now_local().date()
target = "diet_guard._kuchnia_import.refresh_delivery"
with patch(target, return_value=(FOUR, None)) as refresh:
assert _refresh_delivery(day) == (FOUR, None)
refresh.assert_called_once_with(day)
class TestTheGateDrivesTheFlow:
def test_a_demo_gate_refuses(self, gate: MealGate) -> None:
with patch.object(_gatelock_fillall, "start_delivery_fetch") as start:
gate._on_fill_all()
start.assert_not_called()
assert "only available on the real lock" in gate._vars.status.get()
def test_fill_and_confirm_unlocks_from_catering(self, gate: MealGate) -> None:
gate.demo_mode = False # read once, when the flow is first built
gate._pending = [8, 12]
# Fill all also banks the dishes for one-at-a-time offering; the
# unlock line must not then call them "still delivered".
gate._delivery_pending = FOUR
due = _due_from_log([8, 12])
with patch.object(_gatelock_delivery, "due_slots", side_effect=due):
_fill(gate, delivered(*FOUR))
status = gate._vars.status.get()
assert status == "Filled from catering — all meals logged, unlocking…"
assert gate._pending == []
gate.root.after.assert_called_with(_UNLOCK_DELAY_MS, gate.close)
assert sorted(logged()) == [8, 12, 16, 20]
assert gate._vars.fill_label.get() == FILL_LABEL
def test_a_partial_fill_advances_instead_of_unlocking(self, gate: MealGate) -> None:
gate.demo_mode = False
gate._pending = [8, 12]
due = _due_from_log([8, 12])
# Two dishes spread onto 08:00 and 16:00: 12:00 is still owed.
two = (dish("Owsianka", 1), dish("Obiad", 2))
with patch.object(_gatelock_delivery, "due_slots", side_effect=due):
_fill(gate, delivered(*two))
assert gate._pending == [12]
assert "Pulled 1 meal " in gate._vars.status.get()
assert "unlocking" not in gate._vars.status.get()
class TestLeftoverDishesAtUnlock:
"""The "(N more dishes delivered)" suffix counts only unlogged dishes."""
def test_an_already_logged_dish_is_not_leftover(self, gate: MealGate) -> None:
# Case and padding differ from the dish name on purpose.
log_meal(" KOLACJA ", _nutrition(), 20)
gate._delivery_pending = (dish("Kolacja", 4), dish("Obiad", 3))
gate._unlock("Logged 20:00")
assert gate._vars.status.get() == (
"Logged 20:00 (1 more dish delivered; log with 'ate') — "
"all meals logged, unlocking…"
)
def test_unlogged_dishes_are_all_counted(self, gate: MealGate) -> None:
gate._delivery_pending = (dish("Kolacja", 4), dish("Obiad", 3))
gate._unlock("Logged 20:00")
assert "(2 more dishes delivered;" in gate._vars.status.get()
def test_every_queued_dish_logged_means_no_suffix(self, gate: MealGate) -> None:
log_meal("Kolacja", _nutrition(), 20)
gate._delivery_pending = (dish("Kolacja", 4),)
gate._unlock("Logged 20:00")
assert "more dish" not in gate._vars.status.get()
class TestTheButton:
@pytest.mark.usefixtures("dual_output")
def test_every_monitor_shares_one_label(self) -> None:
built: list[tk.Button] = []
real = _gatelock_layout.make_button
def record(
parent: tk.Misc,
config: LockConfig,
text: str,
command: Callable[[], None],
style: ButtonStyle | None = None,
) -> tk.Button:
button = real(parent, config, text, command, style)
built.append(button)
return button
with (
fake_tk(),
patch.object(_gatelock_layout, "make_button", side_effect=record),
):
gate = MealGate(demo_mode=True)
fills = [
b
for b in built
if getattr(b, "configured", {}).get("textvariable") is gate._vars.fill_label
]
assert len(fills) == 2 # one per monitor, one shared variable
assert all(b.configured["command"] == gate._on_fill_all for b in fills)
def test_the_real_button_relabels_through_its_variable(
self, xvfb: Xvfb | None
) -> None:
# Real Tk, so only ever on pytest-xvfb's private display -- never the
# user's. No Xvfb means no test, not a window on :0.
if xvfb is None:
pytest.skip("needs pytest-xvfb's private display")
assert os.environ["DISPLAY"] == f":{xvfb.display}"
root = tk.Tk()
try:
root.withdraw()
on_fill_all = MagicMock()
callbacks = GateCallbacks(
*(MagicMock() for _ in range(5)), on_fill_all=on_fill_all
)
vars_ = make_vars(root)
notebook = ttk.Notebook(root)
widgets = build_layout(notebook, vars_, callbacks, demo_mode=False)
button = _find_fill_button(widgets.frame, str(vars_.fill_label))
assert button.cget("text") == FILL_LABEL
vars_.fill_label.set("✓ Confirm (3)")
assert button.cget("text") == "✓ Confirm (3)"
button.invoke()
on_fill_all.assert_called_once_with()
finally:
root.destroy()
def _walk(widget: tk.Misc) -> Iterator[tk.Misc]:
yield widget
for child in widget.winfo_children():
yield from _walk(child)
def _find_fill_button(frame: tk.Misc, var_name: str) -> tk.Button:
matches = [
w
for w in _walk(frame)
if isinstance(w, tk.Button) and str(w.cget("textvariable")) == var_name
]
assert len(matches) == 1
return matches[0]
def test_a_fetch_spanning_midnight_never_confirms_yesterdays_menu() -> None:
# Clicked on day D; the result lands, and Confirm comes on D+1. The plan
# belongs to the day that was *fetched*, so D+1 refetches and logs nothing.
h = Harness()
tomorrow = now_local() + timedelta(days=1)
with fetching(delivered(*FOUR), delivered()) as start:
h.flow.click()
with patch.object(_gatelock_fillall, "now_local", return_value=tomorrow):
h.flow._poll()
h.flow.click()
assert start.call_args.args[1] == tomorrow.date()
assert logged() == {}

View File

@ -116,16 +116,18 @@ class TestCli:
_cli_kuchnia.cmd_kuchnia(lines.append, _record, log=True, yes=True)
assert asked == []
def test_already_logged_says_so(self) -> None:
def test_every_slot_taken_says_so_and_logs_nothing(self) -> None:
lines: list[str] = []
with (
patch.object(
_cli_kuchnia, "refresh_delivery", return_value=([_dish()], None)
),
patch.object(_cli_kuchnia, "log_dishes", return_value=[]),
patch.object(_cli_kuchnia, "fill_plan", return_value=[]),
patch.object(_cli_kuchnia, "log_dishes") as logger,
):
self._run(lines, log=True, yes=True)
assert any("already logged" in line for line in lines)
assert logger.call_count == 0
assert any("every slot already logged" in line for line in lines)
class TestDispatch:

View File

@ -0,0 +1,78 @@
"""Tests for "fill all with catering": which slots a delivery may fill.
The rule is shared by ``diet-guard kuchnia --log`` and the gate's "Fill all"
button (:func:`diet_guard._kuchnia_log.fill_plan`): every slot today, but only
the ones with nothing logged yet. A slot the user already ate something in
keeps that meal rather than gaining a catering dish on top.
"""
from __future__ import annotations
from dataclasses import replace
from unittest.mock import patch
from diet_guard import _cli_kuchnia
from diet_guard._estimator import Nutrition
from diet_guard._kuchnia_log import fill_plan, log_dishes
from diet_guard._kuchnia_parse import Dish
from diet_guard._kuchnia_spread import SlottedDish
from diet_guard._state import log_meal
from diet_guard._state_today import today_entries
def _slotted(name: str, slot: int) -> SlottedDish:
dish = Dish(
name=name,
kcal=400.0,
protein_g=30.0,
carbs_g=40.0,
fat_g=12.0,
grams=300.0,
priority=1,
slot_label="Obiad",
)
return SlottedDish(dish, slot)
def _eat(desc: str, slot: int) -> None:
log_meal(desc, Nutrition(95.0, 0.5, 25.0, 0.3, 180.0, "manual"), slot)
class TestFillPlan:
def test_an_empty_day_fills_every_slot(self) -> None:
plan = [_slotted("A", 8), _slotted("B", 12), _slotted("C", 20)]
assert fill_plan(plan) == plan
def test_a_slot_with_a_meal_keeps_it(self) -> None:
_eat("apple", 12)
plan = fill_plan([_slotted("A", 8), _slotted("B", 12), _slotted("C", 20)])
assert [item.slot for item in plan] == [8, 20]
def test_two_dishes_sharing_an_empty_slot_both_land(self) -> None:
# 5 dishes on 4 slots double up the first; occupancy is read once,
# before the batch, so the second dish is not refused by the first.
plan = fill_plan([_slotted("A", 8), _slotted("B", 8)])
assert log_dishes(plan) == ["A", "B"]
def test_a_full_day_fills_nothing(self) -> None:
for slot in (8, 12):
_eat("apple", slot)
assert fill_plan([_slotted("A", 8), _slotted("B", 12)]) == []
class TestCliFill:
def test_log_skips_a_taken_slot(self) -> None:
# Two dishes on the default four slots spread to 08:00 and 16:00.
_eat("apple", 8)
dishes = [
replace(_slotted("Owsianka", 8).dish, priority=1),
replace(_slotted("Kaszotto", 16).dish, priority=2),
]
lines: list[str] = []
with patch.object(
_cli_kuchnia, "refresh_delivery", return_value=(dishes, None)
):
_cli_kuchnia.cmd_kuchnia(lines.append, lambda _p: "y", log=True, yes=False)
logged = {(entry["desc"], entry["slot"]) for entry in today_entries()}
assert logged == {("apple", 8), ("Kaszotto", 16)}
assert "logged 1 meal(s)." in lines

View File

@ -17,10 +17,14 @@ 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).
scripting), then fills today's *empty* slots — the same rule as below.
- **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.
- **"🍱 Fill all"** takes two clicks: the first lists what would be logged and
relabels it "✓ Confirm (N)", the second logs every dish into its slot —
all of today's, later ones included — skipping any slot that already holds
a meal (`_kuchnia_log.fill_plan`, re-checked at the second click).
The queue **survives each submit**: `_finish_slot` calls `_prefill_next_dish`
so the next dish is already in the form, carrying the "Logged HH:00 …"