"""Custody: who holds a packet right now, and the Who predicates. A direct port of guard.js holds().""" from __future__ import annotations from collections.abc import Mapping from typing import Any from .actions import ACTIVE, SUPPORT_CODES, Who from .principal import PacketState, Principal WORKER, LEAD, SUPERVISOR = 1, 2, 3 def works_on(principal: Principal, state: PacketState) -> bool: """A member of the packet's department whose station limits allow the packet's station. Leads and supervisors always pass. A worker passes when they have no station limits or the limits hold the packet's station. This is the old worksOn(); START and SUBMIT add the assigned-welder check. """ level = principal.level(state.dept) if level == 0: return False if level >= LEAD: return True return not principal.station_limits or ( state.station_id is not None and state.station_id in principal.station_limits ) def holds(principal: Principal, state: PacketState) -> bool: """True when the principal holds the packet in its current status. Admin is not a holder; see the bypass.""" st = state.status if st == "draft": return "CL" in principal.support if st == "warehouse": return "WH" in principal.support if st in ("ready", "review"): return principal.level(state.dept) >= LEAD if st in ACTIVE: return principal.level(state.dept) >= LEAD or works_on(principal, state) if st == "qc": return "QC" in principal.support if st == "complete": return "CL" in principal.support return False def holder_name(state: PacketState, dept_name: str | None = None, station_name: str | None = None) -> str: """Who has the packet, for messages. The department name falls back to its code.""" st = state.status if st == "draft": return "Clerical" if st == "warehouse": return "Warehouse" if st == "qc": return "Q.C." if st == "complete": return "Clerical (data book)" if st == "filed": return "the data book (filed)" name = dept_name or state.dept if st in ACTIVE and station_name: return f"{name} ยท {station_name}" return name def _works_on_assigned(principal: Principal, state: PacketState) -> bool: """START and SUBMIT: a lead or above in the department, or the assigned welder within their station limits.""" level = principal.level(state.dept) if level >= LEAD: return True return ( level == WORKER and principal.user_id is not None and state.welder_id is not None and str(state.welder_id) == str(principal.user_id) and works_on(principal, state) ) def _co_verifier(principal: Principal, extra: Mapping[str, Any]) -> bool: """A supervisor of the department named in the request (extra["dept"]).""" dept = extra.get("dept") if not isinstance(dept, str) or not dept: return False if dept in ("WH", "QC"): return dept in principal.support_sup if dept in SUPPORT_CODES: return False # Clerical issues change orders; it never verifies them. PM has no supervisor. return principal.level(dept) >= SUPERVISOR def allows(who: Who, principal: Principal, state: PacketState, extra: Mapping[str, Any]) -> bool: """Evaluate one Who predicate. Admin is never allowed here except by Who.ADMIN; the bypass is a rule.""" if who is Who.CL: return "CL" in principal.support if who is Who.CL_SUP: return "CL" in principal.support_sup if who is Who.WH: return "WH" in principal.support if who is Who.WH_SUP: return "WH" in principal.support_sup if who is Who.QC: return "QC" in principal.support if who is Who.QC_SUP: return "QC" in principal.support_sup if who is Who.CL_OR_QC: return "CL" in principal.support or "QC" in principal.support if who is Who.HOLDER: return holds(principal, state) if who is Who.DEPT_LEAD: return principal.level(state.dept) >= LEAD if who is Who.DEPT_SUP: return principal.level(state.dept) >= SUPERVISOR if who is Who.WORKS_ON: return _works_on_assigned(principal, state) if who is Who.CO_VERIFIER: return _co_verifier(principal, extra) if who is Who.ADMIN: return principal.admin if who is Who.SYSTEM: return principal.system is not None return False