dodil/erp-brick-hr

tests/scenario_bricks/hr_test_provider/brick.py

"""A throwaway PROVIDER brick for hr's scenarios (tests only): it provides the capabilities hr asks for, so the
harness can make them absent, failing, failed or connection_required (s.fault.capability), and so a wired
provider can be seen turning hr's manual tasks automatic with no playbook change (SC-HR-09).

It records what it was asked in memory (CALLS) — a stand-in for a vendor, not a model of one. drive.transfer_status
answers in_progress for the first IN_PROGRESS_POLLS asks of a transfer, then completed (SC-HR-12); contracts.status
answers SIGNED from the SIGN_AFTER-th ask of a contract.

What a contract does is chosen by the starter's legal name (the one value every draft carries, so it works live as
well as in process): "Approval" — it needs an approval, IN_APPROVAL for the first APPROVE_AFTER status asks, then
APPROVED (as contracts answers once legal approved it); "Regime" — a draft without client_size_regime is refused
422 "missing: client_size_regime — nothing was drafted" (contracts' own words; nothing stored); "Declines", "Recalled",
"Lapses" — once sent it ends DECLINED / RECALLED / EXPIRED_UNSIGNED; "Paper" — SIGNED_ELSEWHERE.
"""
import base64
import hashlib

from erp_platform.bricks import Brick

PROV = Brick("prov", version="0.1.0", instructions="A test provider of hr's capabilities.")
CALLS: list = []
POLLS: dict = {}
IN_PROGRESS_POLLS = 2
SIGN_AFTER = 1
APPROVE_AFTER = 1
CONTRACTS: dict = {}            # contract_id -> {"name": legal name, "sent": bool, "approval_polls": int}
ENDINGS = {"Declines": "DECLINED", "Recalled": "RECALLED", "Lapses": "EXPIRED_UNSIGNED", "Paper": "SIGNED_ELSEWHERE"}


def _ref(call) -> str:
    return "ref-" + hashlib.sha256(call.idempotency_key.encode()).hexdigest()[:12]


def _record(call):
    CALLS.append({"capability": call.capability.name, "key": call.idempotency_key, "input": dict(call.input)})


@PROV.capability("identity.provision_user")
def provision_user(call):
    _record(call)
    return _ref(call), {"account_id": _ref(call), "primary_email": call.input["primary_email"], "created": True}, "created"


@PROV.capability("identity.suspend_user")
def suspend_user(call):
    _record(call)
    return _ref(call), {"account_id": _ref(call), "suspended": True}, "suspended"


@PROV.capability("identity.delete_user")
def delete_user(call):
    _record(call)
    return _ref(call), {"account_id": _ref(call), "deleted": True}, "deleted"


@PROV.capability("identity.invite_app_user")
def invite_app_user(call):
    _record(call)
    sub = "sub-" + hashlib.sha256(call.input["email"].encode()).hexdigest()[:10]
    return sub, {"sub": sub, "email": call.input["email"], "created": True}, "invited"


def _member(call, key):
    _record(call)
    return _ref(call), {key: call.input.get("group") or call.input.get("space"),
                        ("added" if "add" in call.capability.name else "removed"): True}, "ok"


@PROV.capability("groups.add_member")
def add_member(call):
    return _member(call, "group_id")


@PROV.capability("groups.remove_member")
def remove_member(call):
    return _member(call, "group_id")


@PROV.capability("chat.add_to_space")
def add_to_space(call):
    return _member(call, "space_id")


@PROV.capability("chat.remove_from_space")
def remove_from_space(call):
    return _member(call, "space_id")


@PROV.capability("drive.transfer_ownership")
def transfer_ownership(call):
    _record(call)
    return _ref(call), {"transfer_ref": _ref(call), "state": "requested", "transferred_items": None}, "requested"


@PROV.capability("drive.transfer_status")
def transfer_status(call):
    _record(call)
    ref = call.input["transfer_ref"]
    POLLS[ref] = POLLS.get(ref, 0) + 1
    state = "in_progress" if POLLS[ref] <= IN_PROGRESS_POLLS else "completed"
    return ref, {"transfer_ref": ref, "state": state, "transferred_items": 3}, state


@PROV.capability("mail.send")
def send(call):
    _record(call)
    return _ref(call), {"message_id": _ref(call)}, "sent"


@PROV.capability("mail.set_autoreply")
def autoreply(call):
    _record(call)
    return _ref(call), {"enabled": True}, "on"


@PROV.capability("calendar.create_event")
def create_event(call):
    _record(call)
    return _ref(call), {"event_id": _ref(call), "url": "https://calendar.example/" + _ref(call)}, "created"


def _marked(name: str, word: str) -> bool:
    return word.lower() in (name or "").lower()


@PROV.capability("contracts.draft")
def draft(call):
    from erp_platform.bricks import Refused
    _record(call)
    variables = call.input.get("variables") or {}
    name = variables.get("legal_name") or ""
    if _marked(name, "Regime") and not variables.get("client_size_regime"):
        raise Refused(422, "missing: client_size_regime — nothing was drafted; send them and ask again")
    cid = "C-" + hashlib.sha256(call.idempotency_key.encode()).hexdigest()[:8]
    approval = _marked(name, "Approval")
    CONTRACTS.setdefault(cid, {"name": name, "sent": False, "approval_polls": 0})
    return cid, {"contract_id": cid, "state": "IN_APPROVAL" if approval else "DRAFT", "missing": [],
                 "deviations": [{"what": "a rewritten clause", "level": "1", "needs": "legal"}] if approval else [],
                 "approval_required": approval}, "drafted"


@PROV.capability("contracts.send")
def send_contract(call):
    from erp_platform.capabilities import Failed
    _record(call)
    c = CONTRACTS.get(call.input["contract_id"]) or {}
    if _marked(c.get("name"), "Approval") and c.get("approval_polls", 0) <= APPROVE_AFTER:
        raise Failed("an approval is open: it is sent once the approval is decided")
    if c:
        c["sent"] = True
    return call.input["contract_id"], {"contract_id": call.input["contract_id"], "state": "OUT_FOR_SIGNATURE",
                                       "version": 1, "sha256": "0" * 64, "deliveries": []}, "sent"


@PROV.capability("contracts.status")
def contract_status(call):
    _record(call)
    cid = call.input["contract_id"]
    c = CONTRACTS.get(cid)
    if c is not None and _marked(c["name"], "Approval") and not c["sent"]:
        c["approval_polls"] += 1
        state = "IN_APPROVAL" if c["approval_polls"] <= APPROVE_AFTER else "APPROVED"
        return cid, {"contract_id": cid, "state": state, "version": None, "signers": [], "executed": None,
                     "key_dates": [], "principal_statement_issued_at": None, "warnings": []}, state.lower()
    ending = next((st for word, st in ENDINGS.items() if c is not None and _marked(c["name"], word)), None)
    if ending and ending != "SIGNED_ELSEWHERE":
        return cid, {"contract_id": cid, "state": ending, "version": 1, "signers": [], "executed": None,
                     "key_dates": [], "principal_statement_issued_at": None, "warnings": []}, ending.lower()
    POLLS[cid] = POLLS.get(cid, 0) + 1
    signed = POLLS[cid] >= SIGN_AFTER
    final = ending or "SIGNED"
    return cid, {"contract_id": cid, "state": final if signed else "OUT_FOR_SIGNATURE", "version": 1,
                 "signers": [], "executed": ({"object_key": f"contracts/{cid}.pdf", "sha256": "0" * 64,
                                              "provenance": "test"} if signed else None),
                 "key_dates": [], "principal_statement_issued_at": None, "warnings": []}, "status"


@PROV.capability("contracts.fetch_executed")
def fetch_executed(call):
    _record(call)
    content = f"signed contract {call.input['contract_id']}".encode()
    return call.input["contract_id"], {"contract_id": call.input["contract_id"], "filename": "contract.pdf",
                                       "content_type": "application/pdf",
                                       "sha256": hashlib.sha256(content).hexdigest(), "size": len(content),
                                       "content_base64": base64.b64encode(content).decode()}, "fetched"