dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

src/brick_ar/domain/migration_run.py

"""The MigrationRun aggregate: one load of another system's invoices, its reconciliation and the customer's signature.

loaded -> reconciled -> signed (or discarded). A re-load after a fix keeps the explanations of checks whose two sides did not
change. Signing is the owner's, is refused while any check is neither equal nor explained (readiness {gate, met, reason,
fix}), and records who, when and the run's checksums; a later re-run needs a new signature.
"""
import hashlib
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional

from brick_ar.domain import events
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.readiness import met, require, unmet
from brick_ar.domain.values import RunState


@dataclass
class MigrationCheck:
    run_id: str
    check_key: str
    slice: str = "all"
    odoo_value: str = ""
    ours_value: str = ""
    difference: str = ""
    equal: bool = False
    must: str = "equal"                 # equal | explained
    explanation: str = ""
    explained_by: str = ""

    @property
    def settled(self) -> bool:
        return self.equal or bool(self.explanation)


@dataclass
class MigrationRun:
    run_id: str
    source: str = "odoo"
    source_db: str = ""
    cutover_date: Optional[date] = None
    doors_json: str = "{}"
    status: str = RunState.LOADED.value
    loaded_at: Optional[datetime] = None
    checksums_json: str = ""
    signed_by: str = ""
    signed_at: Optional[datetime] = None
    counts_json: str = "{}"
    checks: list = field(default_factory=list)
    events: list = field(default_factory=list, repr=False, compare=False)

    def _assert_open(self) -> None:
        if self.status in (RunState.SIGNED.value, RunState.DISCARDED.value):
            raise RuleBroken(f"migration run {self.run_id} is {self.status}: a later load is a new run and needs a new signature")

    def loaded(self, *, at: datetime, counts: dict, changed: bool = True) -> None:
        """Documents were (re)loaded: back to `loaded` when something changed (the explanations survive until reconciled again).
        A reload that wrote nothing leaves a reconciled run reconciled: what was checked is what is there."""
        self._assert_open()
        if changed or self.status != RunState.RECONCILED.value:
            self.status = RunState.LOADED.value
        self.loaded_at, self.counts_json = at, json.dumps(counts, sort_keys=True)

    def reconciled(self, checks: list) -> None:
        if self.status not in (RunState.LOADED.value, RunState.RECONCILED.value):
            raise RuleBroken(f"migration run {self.run_id} is {self.status}")
        old = {(c.check_key, c.slice): c for c in self.checks}
        for c in checks:
            prev = old.get((c.check_key, c.slice))
            if prev and prev.explanation and (prev.odoo_value, prev.ours_value) == (c.odoo_value, c.ours_value):
                c.explanation, c.explained_by = prev.explanation, prev.explained_by
        self.checks = checks
        self.status = RunState.RECONCILED.value

    def explain(self, check_key: str, slice_: str, reason: str, *, by: str) -> MigrationCheck:
        if self.status != RunState.RECONCILED.value:
            raise RuleBroken("differences are explained once the run is reconciled")
        if not (reason or "").strip():
            raise RuleBroken("an explanation needs a reason")
        c = next((x for x in self.checks if x.check_key == check_key and x.slice == slice_), None)
        if c is None:
            raise RuleBroken(f"the run has no check {check_key} / {slice_}")
        if c.equal:
            raise RuleBroken(f"{check_key} / {slice_} is equal: there is nothing to explain")
        if c.must == "equal":
            raise RuleBroken(f"{check_key} / {slice_} must be equal and differs: fix the mapping and reload; it cannot be explained away")
        c.explanation, c.explained_by = reason.strip(), by
        return c

    def gates(self) -> list:
        """sign_reconciliation's gates as {gate, met, reason, fix}."""
        hard = [f"{c.check_key} / {c.slice}" for c in self.checks if c.must == "equal" and not c.equal]
        open_ = [f"{c.check_key} / {c.slice}" for c in self.checks if c.must != "equal" and not c.settled]
        return [
            met("must_equal") if not hard else unmet("must_equal", "a check that must be equal differs: " + ", ".join(hard[:8]),
                                                     "fix the mapping and reload; it cannot be explained away"),
            met("zero_unexplained") if not open_ else unmet("zero_unexplained", "checks neither equal nor explained: " + ", ".join(open_[:8]),
                                                            "explain each, or fix and reload"),
        ]

    def checksums(self) -> str:
        body = json.dumps([[c.check_key, c.slice, c.odoo_value, c.ours_value] for c in
                           sorted(self.checks, key=lambda c: (c.check_key, c.slice))])
        return hashlib.sha256(body.encode()).hexdigest()

    def sign(self, *, by: str, at: datetime) -> None:
        self._assert_open()
        if self.status != RunState.RECONCILED.value:
            raise RuleBroken("sign once the run is loaded and reconciled")
        require("sign the reconciliation", self.gates())
        self.status, self.signed_by, self.signed_at, self.checksums_json = RunState.SIGNED.value, by, at, self.checksums()
        self.events.append(events.RunSigned(by=by, run_id=self.run_id))

    def discard(self, *, reason: str, by: str) -> None:
        self._assert_open()
        if not (reason or "").strip():
            raise RuleBroken("discarding a run needs a reason")
        self.status = RunState.DISCARDED.value