mirror of
https://github.com/kuhyx/diet-guard.git
synced 2026-11-10 19:07:15 +01:00
Gate: stand down inside declared holds
No lock and no owed slot while holds.covers("diet") is true; slots
inside a hold are excused; an open lock lifts within 30 s of a hold
starting. Fail closed (normal rules) if holds is missing or raises.
Pin holds-v0.1.0; bump coverage 7.17.1.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWVSoczW2NHQd887FUm
This commit is contained in:
parent
97af520fdd
commit
e3af1f571b
@ -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.132+1
|
||||
version: 1.0.133+1
|
||||
|
||||
environment:
|
||||
sdk: ^3.12.2
|
||||
|
||||
@ -15,11 +15,13 @@ prompts one slot at a time, with no separate weekday code path.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, time, timedelta
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import freedays
|
||||
import sickdays
|
||||
|
||||
from diet_guard._hold import hold_covers
|
||||
from diet_guard._meal_schedule_store import current_schedule
|
||||
from diet_guard._slots import missing_slots, slot_label
|
||||
from diet_guard._state import (
|
||||
@ -30,7 +32,7 @@ from diet_guard._state_today import (
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from datetime import datetime
|
||||
from datetime import date
|
||||
|
||||
|
||||
def due_slots(now: datetime | None = None) -> tuple[int, ...]:
|
||||
@ -50,16 +52,29 @@ def due_slots(now: datetime | None = None) -> tuple[int, ...]:
|
||||
read from a single local file and never the network, so this cannot hang
|
||||
the systemd-fired gate, and an unreadable pool reads as "not free" / "not
|
||||
sick" -- the normal rules apply rather than the gate switching itself off.
|
||||
|
||||
A declared hold (:mod:`diet_guard._hold`) is the same stand-down for a
|
||||
window: nothing is due while one is active, and a slot whose time falls
|
||||
inside a hold is *excused* -- never owed once the hold ends.
|
||||
"""
|
||||
reference = now if now is not None else now_local()
|
||||
day = reference.date()
|
||||
# A declared hold (``holds``): the PC stays usable, so nothing is due now.
|
||||
if hold_covers(reference):
|
||||
return ()
|
||||
if freedays.is_free_day(day) or sickdays.stands_down(sickdays.LOCKER_DIET, day):
|
||||
return ()
|
||||
# One schedule for both halves: snapping the log onto a different schedule
|
||||
# than the one deriving the due slots would leave a logged meal covering a
|
||||
# checkpoint that is not on the list -- and the real one nagging.
|
||||
schedule = current_schedule()
|
||||
return missing_slots(reference, logged_slots_today(schedule), schedule)
|
||||
missing = missing_slots(reference, logged_slots_today(schedule), schedule)
|
||||
return tuple(slot for slot in missing if not hold_covers(_slot_time(day, slot)))
|
||||
|
||||
|
||||
def _slot_time(day: date, slot: int) -> datetime:
|
||||
"""Return the naive wall-clock instant of ``slot`` minutes after midnight."""
|
||||
return datetime.combine(day, time()) + timedelta(minutes=slot)
|
||||
|
||||
|
||||
def gate_is_due(now: datetime | None = None) -> bool:
|
||||
|
||||
@ -63,6 +63,7 @@ from diet_guard._gatelock_ui import (
|
||||
GateCallbacks,
|
||||
make_vars,
|
||||
)
|
||||
from diet_guard._hold import watch_hold
|
||||
from diet_guard._meal_schedule_store import current_schedule
|
||||
from diet_guard._slots import current_slot, day_slots
|
||||
from diet_guard._state import now_local
|
||||
@ -104,10 +105,8 @@ def _pending_slots(*, demo_mode: bool) -> list[int]:
|
||||
return pending
|
||||
if demo_mode:
|
||||
schedule = current_schedule()
|
||||
# Spelled `is None` rather than `or` because slot 0 (midnight) is
|
||||
# falsy. The two happen to be equivalent -- if 0 is a slot it is
|
||||
# necessarily the *first* slot, so both arms return 0, verified
|
||||
# exhaustively -- but the intent should not depend on that coincidence.
|
||||
# `is None`, not `or`: slot 0 (midnight) is falsy. Both arms happen to
|
||||
# agree (0 is necessarily the first slot), but intent shouldn't hinge on it.
|
||||
elapsed = current_slot(now_local(), schedule)
|
||||
return [elapsed if elapsed is not None else day_slots(schedule)[0]]
|
||||
return []
|
||||
@ -220,7 +219,8 @@ class MealGate(_GateCalendar):
|
||||
self._lock.close()
|
||||
|
||||
def run(self) -> None:
|
||||
"""Run the Tk loop, restoring VT switching on every exit path."""
|
||||
"""Run the Tk loop (a hold starting mid-lock closes it), VT-safe."""
|
||||
watch_hold(self.root, self.close)
|
||||
self._lock.run()
|
||||
|
||||
def _wire_events(self) -> None:
|
||||
|
||||
76
diet_guard/_hold.py
Normal file
76
diet_guard/_hold.py
Normal file
@ -0,0 +1,76 @@
|
||||
"""Hold integration: no lock and no owed slot while a declared hold is active.
|
||||
|
||||
A *hold* (the ``holds`` package, contract holds-v0.1.0) is a ratcheted,
|
||||
time-boxed window in which the PC must stay fully usable (a proctored test).
|
||||
This module is the single seam between the gate and that package.
|
||||
|
||||
Fail mode, chosen out loud: **closed**. If ``holds`` is missing, corrupt or
|
||||
raises, :func:`hold_covers` returns False and the gate behaves exactly as it
|
||||
does without holds -- a broken holds package can never switch the gate off.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from importlib import import_module
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from diet_guard._state import now_local
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime
|
||||
import tkinter as tk
|
||||
|
||||
__all__ = ["HOLD_GATE", "HOLD_POLL_MS", "HoldWatch", "hold_covers", "watch_hold"]
|
||||
|
||||
HOLD_GATE = "diet"
|
||||
# An open lock re-reads the (local, tiny) holds file this often, so a hold that
|
||||
# starts under a live lock lifts it in <= 30 s -- inside the 60 s contract.
|
||||
HOLD_POLL_MS = 30_000
|
||||
|
||||
_LOG = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def hold_covers(at: datetime) -> bool:
|
||||
"""Return True if a hold covers the diet gate at local time ``at``.
|
||||
|
||||
Args:
|
||||
at: The instant to test (may be a future slot time); aware values are
|
||||
reduced to naive local wall-clock, which is what ``holds`` stores.
|
||||
|
||||
Returns:
|
||||
True inside a hold; False outside one *and* whenever ``holds`` cannot
|
||||
be imported or raises (fail closed: normal rules apply).
|
||||
"""
|
||||
try:
|
||||
holds = import_module("holds")
|
||||
return bool(holds.covers(HOLD_GATE, at.replace(tzinfo=None)))
|
||||
except ImportError, AttributeError, OSError, TypeError, ValueError:
|
||||
_LOG.exception("holds unavailable; treating the gate as not on hold")
|
||||
return False
|
||||
|
||||
|
||||
class HoldWatch:
|
||||
"""Close the open lock window as soon as a hold becomes active."""
|
||||
|
||||
def __init__(self, root: tk.Misc, close: Callable[[], None]) -> None:
|
||||
"""Bind the Tk root (for ``after``) and the window's close callback."""
|
||||
self._root = root
|
||||
self._close = close
|
||||
|
||||
def start(self) -> None:
|
||||
"""Schedule the first check."""
|
||||
self._root.after(HOLD_POLL_MS, self.tick)
|
||||
|
||||
def tick(self) -> None:
|
||||
"""Lift the lock if a hold is active now, else check again later."""
|
||||
if hold_covers(now_local()):
|
||||
self._close()
|
||||
return
|
||||
self._root.after(HOLD_POLL_MS, self.tick)
|
||||
|
||||
|
||||
def watch_hold(root: tk.Misc, close: Callable[[], None]) -> None:
|
||||
"""Start a :class:`HoldWatch` on ``root`` that calls ``close`` on a hold."""
|
||||
HoldWatch(root, close).start()
|
||||
@ -21,7 +21,9 @@ are functional in-memory stand-ins, shared by ``test_gatelock.py`` and
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import ExitStack
|
||||
import os
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
@ -106,6 +108,11 @@ def _isolate_state(tmp_path: Path) -> Iterator[None]:
|
||||
with ExitStack() as stack:
|
||||
for redirect in redirects:
|
||||
stack.enter_context(redirect)
|
||||
# The shared ``holds`` file lives under ~/.local/share: a real hold on
|
||||
# this machine must not stand the whole suite's gate down.
|
||||
stack.enter_context(
|
||||
patch.dict(os.environ, {"HOLDS_DATA": str(tmp_path / "holds.json")})
|
||||
)
|
||||
yield
|
||||
|
||||
|
||||
|
||||
@ -170,6 +170,15 @@ class TestLockDelegation:
|
||||
gate.run()
|
||||
mock_run.assert_called_once_with()
|
||||
|
||||
def test_run_arms_the_hold_watch(self, gate: MealGate) -> None:
|
||||
"""run() starts the poll that lifts the lock when a hold begins."""
|
||||
with (
|
||||
patch.object(gate._lock, "run"),
|
||||
patch("diet_guard._gatelock.watch_hold") as watch,
|
||||
):
|
||||
gate.run()
|
||||
watch.assert_called_once_with(gate.root, gate.close)
|
||||
|
||||
def test_close_delegates_to_lock(self, gate: MealGate) -> None:
|
||||
"""close() hands off to the owned LockWindow."""
|
||||
with patch.object(gate._lock, "close") as mock_close:
|
||||
|
||||
161
diet_guard/tests/test_hold.py
Normal file
161
diet_guard/tests/test_hold.py
Normal file
@ -0,0 +1,161 @@
|
||||
"""Hold contract (holds-v0.1.0): the gate never locks nor owes a slot in a hold.
|
||||
|
||||
``holds`` is faked through ``sys.modules`` so these tests pin diet_guard's side
|
||||
of the contract: ``holds.covers("diet", when)`` with a naive local datetime,
|
||||
failing closed (= normal rules) when the package is missing or misbehaves.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from diet_guard._gate import due_slots, gate_is_due
|
||||
from diet_guard._hold import HOLD_POLL_MS, HoldWatch, hold_covers, watch_hold
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
WED = datetime(2026, 11, 4, tzinfo=UTC)
|
||||
|
||||
|
||||
def _wed(hour: int, minute: int = 0) -> datetime:
|
||||
"""Return Wed 2026-11-04 at ``hour:minute`` (aware, like ``now_local``)."""
|
||||
return WED.replace(hour=hour, minute=minute)
|
||||
|
||||
|
||||
def _install_holds(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
covers: Callable[[str, datetime], bool],
|
||||
) -> MagicMock:
|
||||
"""Install a fake ``holds`` whose ``covers`` is ``covers``."""
|
||||
spy = MagicMock(side_effect=covers)
|
||||
monkeypatch.setitem(sys.modules, "holds", SimpleNamespace(covers=spy))
|
||||
return spy
|
||||
|
||||
|
||||
def _in_window(start: datetime, end: datetime) -> Callable[[str, datetime], bool]:
|
||||
"""A ``covers`` for one hold ``[start, end)`` on naive local time."""
|
||||
lo, hi = start.replace(tzinfo=None), end.replace(tzinfo=None)
|
||||
return lambda gate, when: gate == "diet" and lo <= when < hi
|
||||
|
||||
|
||||
def _logged(slots: set[int]) -> object:
|
||||
return patch("diet_guard._gate.logged_slots_today", return_value=slots)
|
||||
|
||||
|
||||
class TestHoldCovers:
|
||||
"""``hold_covers`` is a fail-closed wrapper over ``holds.covers``."""
|
||||
|
||||
def test_asks_the_diet_gate_with_a_naive_time(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
spy = _install_holds(monkeypatch, lambda gate, when: True)
|
||||
assert hold_covers(_wed(18)) is True
|
||||
spy.assert_called_once_with("diet", _wed(18).replace(tzinfo=None))
|
||||
|
||||
def test_missing_package_fails_closed(
|
||||
self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
monkeypatch.setitem(sys.modules, "holds", None)
|
||||
assert hold_covers(_wed(18)) is False
|
||||
assert "holds unavailable" in caplog.text
|
||||
|
||||
def test_a_raising_holds_fails_closed(
|
||||
self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
def boom(gate: str, when: datetime) -> bool:
|
||||
msg = "corrupt"
|
||||
raise ValueError(msg)
|
||||
|
||||
_install_holds(monkeypatch, boom)
|
||||
assert hold_covers(_wed(18)) is False
|
||||
assert "holds unavailable" in caplog.text
|
||||
|
||||
|
||||
class TestDueSlotsInHold:
|
||||
"""Wed 17:00-20:00 hold: no lock inside it, no debt for slots inside it."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _hold(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_install_holds(monkeypatch, _in_window(_wed(17), _wed(20)))
|
||||
|
||||
def test_nothing_due_while_the_hold_is_active(self) -> None:
|
||||
with _logged(set()):
|
||||
assert due_slots(_wed(18)) == ()
|
||||
assert not gate_is_due(_wed(18))
|
||||
|
||||
def test_slot_inside_the_hold_is_not_owed_afterwards(self) -> None:
|
||||
# 08:00/12:00 unlogged and elapsed; 17:30 and 19:45 fall in the hold.
|
||||
schedule = MagicMock()
|
||||
with (
|
||||
_logged({480, 720}),
|
||||
patch("diet_guard._gate.current_schedule", return_value=schedule),
|
||||
patch(
|
||||
"diet_guard._gate.missing_slots",
|
||||
return_value=(1050, 1185), # 17:30, 19:45
|
||||
),
|
||||
):
|
||||
assert due_slots(_wed(21)) == ()
|
||||
|
||||
def test_slots_outside_the_hold_stay_owed(self) -> None:
|
||||
with (
|
||||
_logged(set()),
|
||||
patch(
|
||||
"diet_guard._gate.missing_slots",
|
||||
return_value=(480, 1050, 1260), # 08:00, 17:30, 21:00
|
||||
),
|
||||
):
|
||||
assert due_slots(_wed(21, 30)) == (480, 1260)
|
||||
|
||||
def test_hold_ending_restores_the_normal_rules(self) -> None:
|
||||
with _logged(set()):
|
||||
assert gate_is_due(_wed(16))
|
||||
assert not gate_is_due(_wed(19))
|
||||
assert gate_is_due(_wed(20, 15))
|
||||
|
||||
def test_missing_holds_package_changes_nothing(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setitem(sys.modules, "holds", None)
|
||||
with _logged(set()):
|
||||
assert gate_is_due(_wed(18))
|
||||
|
||||
|
||||
class TestHoldWatch:
|
||||
"""An open lock lifts within one poll interval of a hold starting."""
|
||||
|
||||
def test_poll_interval_is_within_the_60_s_budget(self) -> None:
|
||||
assert 0 < HOLD_POLL_MS <= 30_000
|
||||
|
||||
def test_start_schedules_the_first_check(self) -> None:
|
||||
root, close = MagicMock(), MagicMock()
|
||||
HoldWatch(root, close).start()
|
||||
root.after.assert_called_once()
|
||||
assert root.after.call_args.args[0] == HOLD_POLL_MS
|
||||
|
||||
def test_watch_hold_starts_a_watch(self) -> None:
|
||||
root = MagicMock()
|
||||
watch_hold(root, MagicMock())
|
||||
root.after.assert_called_once()
|
||||
|
||||
def test_no_hold_reschedules_without_closing(self) -> None:
|
||||
root, close = MagicMock(), MagicMock()
|
||||
watch = HoldWatch(root, close)
|
||||
with patch("diet_guard._hold.hold_covers", return_value=False):
|
||||
watch.tick()
|
||||
close.assert_not_called()
|
||||
root.after.assert_called_once_with(HOLD_POLL_MS, watch.tick)
|
||||
|
||||
def test_hold_closes_the_lock_and_stops_polling(self) -> None:
|
||||
root, close = MagicMock(), MagicMock()
|
||||
watch = HoldWatch(root, close)
|
||||
with patch("diet_guard._hold.hold_covers", return_value=True):
|
||||
watch.tick()
|
||||
close.assert_called_once_with()
|
||||
root.after.assert_not_called()
|
||||
@ -7,6 +7,7 @@ dependencies = [
|
||||
"gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.11.0#subdirectory=gatelock",
|
||||
"crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync",
|
||||
"freedays @ git+https://github.com/kuhyx/utils@freedays-v0.4.0#subdirectory=freedays",
|
||||
"holds @ git+https://github.com/kuhyx/utils@holds-v0.1.0#subdirectory=holds",
|
||||
"sickdays @ git+https://github.com/kuhyx/utils@sickdays-v0.4.0#subdirectory=sickdays",
|
||||
"defusedxml==0.7.1",
|
||||
"requests==2.34.2",
|
||||
|
||||
@ -2,11 +2,12 @@
|
||||
# Runtime: tkinter/json/hmac/fcntl (stdlib) plus gatelock and requests below.
|
||||
bandit==1.9.4
|
||||
codespell==2.4.3
|
||||
coverage==7.17.0
|
||||
coverage==7.17.1
|
||||
crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync
|
||||
defusedxml==0.7.1
|
||||
freedays @ git+https://github.com/kuhyx/utils@freedays-v0.4.0#subdirectory=freedays
|
||||
gatelock @ git+https://github.com/kuhyx/utils@gatelock-v0.11.0#subdirectory=gatelock
|
||||
holds @ git+https://github.com/kuhyx/utils@holds-v0.1.0#subdirectory=holds
|
||||
mcp==2.3.0
|
||||
mypy==2.4.1
|
||||
pre-commit==4.7.0
|
||||
|
||||
Loading…
Reference in New Issue
Block a user