dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

src/brick_purchasing/domain/approval.py

"""Who may approve what: a second person is a different EMAIL and a different SUB; limits are dated rows; a release strategy
asks for up to three levels, each a different person (module.yaml PUR-02, PUR-38; known bug class: self-approval under a second
login)."""
import json
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Iterable, Optional

from brick_purchasing.domain.readiness import Gate, no, ok
from brick_purchasing.domain.values import EVERY, ZERO


def identities(email: Optional[str], sub: Optional[str]) -> frozenset:
    """Every identity a person has (email and sub, lower-case): the same person under either is the same person."""
    return frozenset(x.strip().lower() for x in (email, sub) if x and x.strip())


def same_person(a: Iterable[str], b: Iterable[str]) -> bool:
    return bool(set(a) & set(b))


@dataclass(frozen=True)
class Strategy:
    levels: int = 1
    thresholds: tuple = ()          # the minimum limit an approver of each level holds
    valid_from: Optional[date] = None


def thresholds_of(raw) -> tuple:
    try:
        return tuple(Decimal(str(x)) for x in (json.loads(raw) if isinstance(raw, str) else (raw or [])))
    except (ValueError, TypeError, ArithmeticError):
        return ()


def strategy_for(rows: list, category: str, total: Decimal, currency: str, on: date) -> Strategy:
    """The release strategy in force on `on` for this category and amount: of each (category, min_amount) the latest row
    valid_from <= on; the category's own rows win over the every-category rows; the highest min_amount <= total applies.
    None in force: one level."""
    best = {}
    for r in rows:
        if r["valid_from"] > on or r.get("currency", currency) != currency or r["category"] not in (category, EVERY):
            continue
        k = (r["category"], r["min_amount"])
        if k not in best or r["valid_from"] > best[k]["valid_from"]:
            best[k] = r
    for cat in (category, EVERY):
        mine = [r for (c, m), r in best.items() if c == cat and m <= total]
        if mine:
            r = max(mine, key=lambda x: x["min_amount"])
            return Strategy(int(r["levels"]), thresholds_of(r["level_thresholds"]), r["valid_from"])
    return Strategy()


def limit_in_force(rows: list, ids: frozenset, category: str, currency: str, on: date) -> Optional[Decimal]:
    """The approver's limit on `on`: the person's row for the category (else the every-category row) with the latest
    valid_from <= on. None: no limit in force (they may approve nothing)."""
    for cat in (category, EVERY):
        mine = [r for r in rows if r["category"] == cat and r["valid_from"] <= on and r.get("currency", currency) == currency
                and same_person(ids, identities(r["approver_email"], r.get("approver_sub")))]
        if mine:
            return max(mine, key=lambda r: r["valid_from"])["limit_amount"]
    return None


def level_gates(*, owner_ids: frozenset, earlier: list, approver_ids: frozenset, limit: Optional[Decimal], needed: Decimal,
                noun: str, final: bool, level: int = 1, levels: int = 1, threshold: Decimal = ZERO) -> list:
    """The readiness of one approval step. owner_ids: who asked (requester) or raised (buyer); earlier: the identities of the
    approvers of earlier levels. A final level's limit must cover the amount; an earlier level's, that level's threshold."""
    gates = []
    if not approver_ids or same_person(approver_ids, owner_ids) or any(same_person(approver_ids, e) for e in earlier):
        gates.append(no("second_person", f"you {'asked for' if noun == 'request' else 'raised'} this {noun}, or approved an earlier "
                        "level (same email or same sign-in)", "another approver"))
    else:
        gates.append(ok("second_person"))
    if final:
        need = max(needed, threshold)
        gates.append(ok("within_limit") if limit is not None and limit >= need else
                     no("within_limit", f"the amount {needed} is above your limit in force today"
                        + (f" ({limit})" if limit is not None else " (you have none)"), "an approver with a higher limit (Settings: approval limits)"))
    else:
        gates.append(ok("level_limit") if limit is not None and limit >= threshold else
                     no("level_limit", f"your limit is below level {level}'s threshold {threshold}", "an approver of this level (Settings: release strategy)"))
    return gates


def levels_waiting(levels: int, approved: int) -> int:
    return max(levels - approved, 0)


def stage_note(levels: int, approved: int) -> str:
    return f"level {approved + 1} of {levels} waiting" if approved < levels else "all levels approved"


def gate_list(gates: list) -> list:
    return [g for g in gates if isinstance(g, Gate)]


@dataclass
class Release:
    """One level of approval given on one revision of a request, order or agreement (pur_release)."""
    document_kind: str
    document_id: str
    revision: int
    level: int
    approved_by_email: str = ""
    approved_by_sub: str = ""
    approved_on: Optional[date] = None
    strategy_valid_from: Optional[date] = None

    @property
    def ids(self) -> frozenset:
        return identities(self.approved_by_email, self.approved_by_sub)


def step(doc, kind: str, doc_id: str, revision: int, *, owner_ids: frozenset, email: str, sub: str, limit: Optional[Decimal],
         needed: Decimal, today: date, noun: str, final: bool) -> Release:
    """One approval step on a document that carries `release_levels`, `release_thresholds` and `releases`. Adds the release
    row and returns it; raises GateUnmet naming every gate that fails (second_person, level_limit, within_limit)."""
    from brick_purchasing.domain.errors import RuleBroken
    from brick_purchasing.domain.readiness import require
    levels = int(doc.release_levels or 1)
    mine = [r for r in doc.releases if r.revision == revision]
    done = len(mine)
    if final and done != levels - 1:
        raise RuleBroken(f"{stage_note(levels, done)}: approve the earlier level first with approve_{noun}_level"
                         if done < levels - 1 else f"all {levels} levels of this {noun} are already approved")
    if not final and done >= levels - 1:
        raise RuleBroken(f"this is the last level of {levels}: approve the {noun} with approve_{noun}")
    level = done + 1
    thr = thresholds_of(doc.release_thresholds)
    threshold = thr[level - 1] if len(thr) >= level else ZERO
    require(level_gates(owner_ids=owner_ids, earlier=[r.ids for r in mine], approver_ids=identities(email, sub), limit=limit,
                        needed=needed, noun=noun, final=final, level=level, levels=levels, threshold=threshold))
    rel = Release(kind, doc_id, revision, level, (email or "").lower(), (sub or "").lower(), today, doc.strategy_valid_from)
    doc.releases.append(rel)
    return rel