"""democlock — the demo clock as a route a brick can mount, so a DEMO deployment can be walked through time.
A 30-day onboarding cannot be demonstrated, or checked live by the scenario harness, in real time. In process
the harness moves `work.now()` directly; against a deployment it can only ask, through a door. This is that
door, declared as two ordinary use cases on the brick that mounts it (so both doors serve it, with the
permission checked the one way):
from erp_platform import democlock
democlock.mount(HR) # GET /demo-clock and POST /demo-clock, permission hr:clock:set
GET /<brick>/demo-clock what the business clock reads, and whether it was moved (signed-in callers)
POST /<brick>/demo-clock {"at": "2026-11-02T09:00 Europe/London"} | {"advance": "+3d"} | {"reset": true},
with "freeze": true (the default) to stop it there. Needs `<brick>:clock:set`.
REFUSED IN PRODUCTION, twice over. Under DODIL_ENV=prod|live both routes answer 404 — a live deployment has no
demo clock to find, and saying "forbidden" would confirm there is one — and `work.set_clock` itself raises, so
a brick that called it some other way still cannot move a live clock. Publish `<brick>:clock:set` in the module
spec, and give it to nobody in a production pool.
The clock is PROCESS state: a demo deployment that moves it runs one replica (`--max-replicas 1`), or each
replica keeps its own time. Mount it on ONE brick per composition — it is one clock for the whole process.
"""
from datetime import timedelta
from typing import Optional
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
PATH = "/demo-clock"
def _offset(text: str) -> timedelta:
m = work._OFFSET.fullmatch((text or "").strip())
if not m or not any(m.groups()[1:]):
raise Refused(422, f"advance is an offset like +3d, +2h30m or -15m, not {text!r}")
sign = -1 if m.group(1) == "-" else 1
d, h, mins = (int(g or 0) for g in m.groups()[1:])
return sign * timedelta(days=d, hours=h, minutes=mins)
def _guard() -> None:
if work._live_env():
raise Refused(404, "not found")
def mount(brick: Brick, *, permission: Optional[str] = None) -> tuple:
"""Declare the clock's two use cases on `brick`. Returns their names (read, set)."""
perm = permission or f"{brick.name}:clock:set"
if not permission and not auth.is_concrete(perm):
# 0.9.10: a brick named by its module id (sales-order, business-partner) gave `sales-order:clock:set`, which is
# not a concrete permission (a hyphen), so NO ONE could move the clock: the spec's namespace is used instead.
perm = f"{(getattr(brick, 'namespace', None) or brick.name).replace('-', '_')}:clock:set"
read_name, set_name = f"{brick.name}_demo_clock", f"set_{brick.name}_demo_clock"
def read(caller: dict) -> dict:
_guard()
return work.clock_state()
def set_(caller: dict, at: Optional[str] = None, advance: Optional[str] = None, freeze: bool = True,
reset: bool = False) -> dict:
_guard()
if sum(x is not None and x is not False for x in (at, advance, reset or None)) != 1:
raise Refused(422, "give exactly one of at, advance or reset")
try:
if reset:
work.reset_clock()
elif at is not None:
work.set_clock(at=work.parse_instant(at), freeze=freeze)
else:
delta = _offset(advance)
if freeze and not work._SET:
work.set_clock(offset=delta, freeze=True)
else:
work.advance_clock(delta)
except work.DemoClockRefused:
raise Refused(404, "not found") from None
except ValueError as e:
raise Refused(422, str(e)) from None
return work.clock_state()
read.__name__ = read.__qualname__ = read_name
read.__doc__ = (f"What the business clock reads in this {brick.name} deployment, and whether a demo moved it. "
"Not found on a live deployment.")
set_.__name__ = set_.__qualname__ = set_name
set_.__doc__ = ("Move the business clock of this DEMO deployment: to a time ('2026-11-02T09:00 Europe/London'), "
"by an offset ('+3d'), or back to real time (reset). Never available on a live deployment.")
# The probe generators (erp_platform.scenarios.probes) leave the clock alone: it is the harness's own lever.
read.__erp_probe_exempt__ = set_.__erp_probe_exempt__ = "the demo clock"
brick.use_case(kind="query", route=f"GET {PATH}")(read)
brick.use_case(kind="command", route=f"POST {PATH}", permission=perm, idempotent=True)(set_)
return read_name, set_name