"""The policy engine: check one action, or list what a principal can do on a packet. Order of checks, exactly: 1. removed packet: nothing passes, Admin included. 2. status: the rule's from statuses and the status conditions (RETURN_TO_CLERICAL needs wh_skipped, SET_VCLOSED needs a hand packet routed to the data book). An unknown status fails here. 3. change-order hold, Admin included. 4. system-only rules refuse people; the system principal passes only system-only rules. 5. admin bypass of who and custody, with an IT reason (SET_DATABOOK_AT needs none). 6. filed lock: only SET_DATABOOK_AT and ADMIN_REOPEN, Admin included. 7. who and custody. 8. preconditions in rule order, never bypassed. `check_scoped` decides the actions on a project or unit (Q.C. folder documents, the ITP) the same way without a packet: system, admin bypass, who, then the input preconditions. There is no status, CO hold, filed lock or custody. """ from __future__ import annotations from collections.abc import Callable, Mapping from dataclasses import dataclass from types import MappingProxyType from typing import Any from ..text import reason_ok from .actions import EDIT_WHAT, RULES, STATUS_LABELS, STATUSES, Action, Rule from .custody import allows, holder_name from .preconditions import INPUT_ONLY, PRECONDITIONS from .principal import PacketState, Principal EMPTY: Mapping[str, Any] = MappingProxyType({}) MSG_REMOVED = "This packet was removed." MSG_UNKNOWN_STATUS = "This packet has an unknown status." MSG_CO_HOLD = "A change order holds this step." MSG_SYSTEM_ONLY = "Only the VISUAL feed does this." MSG_SYSTEM_FORBIDDEN = "The VISUAL feed cannot do this." MSG_IT_REASON = "Give an IT reason of at least 8 characters." MSG_FILED = "This packet is filed and locked. IT can reopen it if a correction is needed." MSG_FORBIDDEN_STEP = "Your access doesn't allow this step." MSG_NOT_WH_SKIPPED = "This packet came through the warehouse. Send it back to the warehouse instead." MSG_NOT_DATABOOK = "Only packets routed to the data book are closed in VISUAL here." MSG_FEED_CLOSES = "The VISUAL feed marks this job closed when it sees every work order closed." MSG_FORBIDDEN_SCOPED = "Your access doesn't allow this." # The packet view check_scoped hands the who predicates. Only predicates that ignore the packet (CL, QC, QC_SUP, # CL_OR_QC and the like) can pass with it; HOLDER and the department predicates fail. _NO_PACKET = PacketState( id=0, status="", dept="", station_id=None, welder_id=None, wh_skipped=False, routed_to=None, vclosed=False, removed=False, source_kind="hand", weld_dept=False, co_hold=False, ) # Preconditions with their own refusal code (the API maps reason_required to 400 and ncr_required to 422). PRECONDITION_CODES: dict[str, str] = {"reason_given": "reason_required", "ncr_given": "ncr_required"} @dataclass(frozen=True) class Decision: ok: bool code: str = "" # removed | wrong_status | co_hold | filed | it_reason_required | forbidden | precondition # | reason_required | ncr_required | system_only why: str = "" # the message for the user admin_bypass: bool = False # an admin passed who and custody only through the bypass (an IT edit) signature: str | None = None # the rule's signature meaning, when one is required reasons: tuple[str, ...] = () # read mode only: every blocking reason (CO hold and state preconditions) # Status conditions beyond the from statuses: (predicate, message when it fails). STATUS_CONDITIONS: dict[Action, tuple[Callable[[PacketState], bool], str]] = { Action.RETURN_TO_CLERICAL: (lambda s: s.wh_skipped, MSG_NOT_WH_SKIPPED), Action.SET_VCLOSED: (lambda s: s.routed_to == "databook", MSG_NOT_DATABOOK), } def _status_condition(state: PacketState, action: Action) -> str | None: # C5: the feed alone closes a VISUAL packet in VISUAL; Clerical confirms a hand packet only. if action is Action.SET_VCLOSED and state.source_kind == "visual": return MSG_FEED_CLOSES cond = STATUS_CONDITIONS.get(action) if cond is not None and not cond[0](state): return cond[1] return None def _wrong_status(state: PacketState, action: Action, rule: Rule) -> str | None: if state.status not in STATUSES: return MSG_UNKNOWN_STATUS if rule.from_status and state.status not in rule.from_status: return f"This step isn't open while the packet is {STATUS_LABELS[state.status]}." return _status_condition(state, action) def _forbidden_why(action: Action, rule: Rule, state: PacketState, extra: Mapping[str, Any]) -> str: what = EDIT_WHAT.get(action) if rule.transition or what is None: return MSG_FORBIDDEN_STEP holder = extra.get("holder_name") if not isinstance(holder, str) or not holder: holder = holder_name(state) return f"Your access doesn't allow changing {what} on this packet. It's with {holder} right now." def _evaluate( principal: Principal, state: PacketState, action: Action, extra: Mapping[str, Any], it_reason: str | None, *, read: bool, assume_it_reason: bool, ) -> Decision: rule = RULES[action] # (1) removed if state.removed: return Decision(False, "removed", MSG_REMOVED) # (2) status why = _wrong_status(state, action, rule) if why is not None: return Decision(False, "wrong_status", why) # (3) change-order hold (deferred in read mode so the UI can show it as blocked after who passes) blockers: list[str] = [] if rule.co_hold and state.co_hold: hold_why = state.co_hold_why or MSG_CO_HOLD if not read: return Decision(False, "co_hold", hold_why) blockers.append(hold_why) # (4) system if rule.system_only and principal.system is None: return Decision(False, "system_only", MSG_SYSTEM_ONLY) if principal.system is not None and not rule.system_only: return Decision(False, "forbidden", MSG_SYSTEM_FORBIDDEN) # (5) admin bypass who_ok = allows(rule.who, principal, state, extra) bypass = False if not who_ok and principal.admin: if action is Action.SET_DATABOOK_AT: bypass = True elif (read and assume_it_reason) or reason_ok(it_reason) is not None: bypass = True else: return Decision(False, "it_reason_required", MSG_IT_REASON) # (6) filed lock if state.status == "filed" and not rule.filed_ok: return Decision(False, "filed", MSG_FILED) # (7) who and custody if not who_ok and not bypass: return Decision(False, "forbidden", _forbidden_why(action, rule, state, extra)) # (8) preconditions, never bypassed for name in rule.preconditions: if read and name in INPUT_ONLY: continue failed = PRECONDITIONS[name](state, extra) if failed is None: continue if read: blockers.append(failed) continue code = PRECONDITION_CODES.get(name, "precondition") if name == "reason_given" and not rule.reason_required: code = "precondition" return Decision(False, code, failed, admin_bypass=bypass) if blockers: code = "co_hold" if rule.co_hold and state.co_hold else "precondition" return Decision(False, code, blockers[0], admin_bypass=bypass, reasons=tuple(blockers)) return Decision(True, admin_bypass=bypass, signature=rule.signature) def check( principal: Principal, state: PacketState, action: Action, *, extra: Mapping[str, Any] = EMPTY, it_reason: str | None = None, ) -> Decision: """Decide one action with the request input in `extra`. Never raises for an unknown status.""" return _evaluate(principal, state, action, extra, it_reason, read=False, assume_it_reason=False) def evaluate(principal: Principal, state: PacketState, action: Action, *, assume_it_reason: bool = True) -> Decision: """Read-mode decision: input-only preconditions skipped, CO hold and state preconditions collected in reasons. For an admin the IT reason is assumed when `assume_it_reason` is true; admin_bypass says the UI will ask. """ return _evaluate(principal, state, action, EMPTY, None, read=True, assume_it_reason=assume_it_reason) def can( principal: Principal, state: PacketState, assume_it_reason: bool = True ) -> tuple[list[Action], dict[Action, list[str]]]: """What the principal can do now, and what is blocked with the reasons. System-only actions are never listed. `can` holds the actions that pass with an empty extra (input-only preconditions skipped). `blocked` holds the actions that pass who and custody but fail a state precondition or the CO hold. """ allowed: list[Action] = [] blocked: dict[Action, list[str]] = {} for action, rule in RULES.items(): if rule.system_only: continue d = evaluate(principal, state, action, assume_it_reason=assume_it_reason) if d.ok: allowed.append(action) elif d.reasons: blocked[action] = list(d.reasons) return allowed, blocked def check_scoped( principal: Principal, action: Action, *, extra: Mapping[str, Any] | None = None, it_reason: str | None = None, assume_it_reason: bool = False, ) -> Decision: """Decide an action on a project or unit (no packet): the steps of check() that do not need one. System principals fail `forbidden` (a system-only rule fails `system_only` for people). An admin the rule does not allow passes with an IT reason of at least 8 characters (`admin_bypass`), else `it_reason_required`. Then who (`forbidden`), then the input preconditions in rule order (`reason_given` fails `reason_required`). Custody is not checked: nobody holds a project or unit. `assume_it_reason` is the read mode for an admin (what the screen offers). """ rule = RULES[action] ex = extra or EMPTY if rule.system_only and principal.system is None: return Decision(False, "system_only", MSG_SYSTEM_ONLY) if principal.system is not None: return Decision(False, "forbidden", MSG_SYSTEM_FORBIDDEN) who_ok = allows(rule.who, principal, _NO_PACKET, ex) bypass = False if not who_ok and principal.admin: if assume_it_reason or reason_ok(it_reason) is not None: bypass = True else: return Decision(False, "it_reason_required", MSG_IT_REASON) if not who_ok and not bypass: return Decision(False, "forbidden", MSG_FORBIDDEN_SCOPED) for name in rule.preconditions: if name not in INPUT_ONLY: continue # a state precondition needs a packet; no scoped rule has one failed = PRECONDITIONS[name](_NO_PACKET, ex) if failed is None: continue code = "reason_required" if rule.reason_required and name == "reason_given" else "precondition" return Decision(False, code, failed, admin_bypass=bypass) return Decision(True, admin_bypass=bypass, signature=rule.signature)