"""Offline inventory for one synthetic Business Central collection notification. Python 3.9+ standard library. No HTTP, credentials, native connector or ERP writes. FixturePage/Row are internal fixtures, NOT native Business Central payloads. This is an acceptance experiment, not a webhook server or pagination adapter. """ from dataclasses import dataclass, replace from typing import Dict, Mapping, Optional, Tuple import copy class IncompleteInventory(ValueError): pass @dataclass(frozen=True) class Scope: company: str subscription: str client_state: str approved_resource: str @dataclass(frozen=True) class Row: company: str record_id: str revision: str @dataclass(frozen=True) class FixturePage: rows: Tuple[Row, ...] continuation: Optional[str] = None read_ok: bool = True def collect(scope: Scope, notification: dict, pages: Mapping[str, FixturePage], expected: Tuple[Tuple[str, str], ...]) -> Dict[str, str]: """Return a full matched inventory, or raise without publishing a partial one. The caller supplies an already qualified resource and trusted scope. Equality with client_state is only a fixture guard, not origin authentication. Page tokens are local fixture keys, never URLs to follow over the network. Expected revisions are synthetic business projections, not native ETags. """ if not all(isinstance(v, str) and v for v in vars(scope).values()): raise IncompleteInventory("invalid fixture scope") if not isinstance(notification, dict) or notification.get("changeType") != "collection": raise IncompleteInventory("not a collection notification") for key, value in [("subscriptionId", scope.subscription), ("clientState", scope.client_state), ("resource", scope.approved_resource)]: if notification.get(key) != value: raise IncompleteInventory("notification outside approved fixture scope") wanted = {} for record_id, revision in expected: if not all(isinstance(v, str) and v for v in (record_id, revision)): raise IncompleteInventory("invalid expected row") if record_id in wanted: raise IncompleteInventory("duplicate expected identity") wanted[record_id] = revision if not wanted: raise IncompleteInventory("this experiment requires a known nonempty cohort") token, visited, observed = "chunk-0", set(), {} while token is not None: if not isinstance(token, str) or not token or token in visited: raise IncompleteInventory("invalid or repeated local continuation") visited.add(token) page = pages.get(token) if not isinstance(page, FixturePage) or page.read_ok is not True: raise IncompleteInventory("page missing or unreadable") for row in page.rows: if not isinstance(row, Row) or row.company != scope.company: raise IncompleteInventory("row outside fixture company") if not all(isinstance(v, str) and v for v in (row.record_id, row.revision)): raise IncompleteInventory("unidentified row or revision") if row.record_id in observed: raise IncompleteInventory("repeated record; qualify the scan before acceptance") observed[row.record_id] = row.revision token = page.continuation if observed != wanted: raise IncompleteInventory("missing, unexpected or changed business projection") return observed def main() -> None: scope = Scope("company-A", "subscription-A", "dummy-state", "/api/v2.0/companies(company-A)/customers?qualified-fixture-filter") # Symbolic company and filter are intentionally not native request examples. note = dict(subscriptionId=scope.subscription, clientState=scope.client_state, resource=scope.approved_resource, changeType="collection") rows = tuple(Row(scope.company, "customer-%04d" % n, "terms-reviewed") for n in range(1203)) expected = tuple((r.record_id, r.revision) for r in rows) pages = {"chunk-0": FixturePage(rows[:500], "chunk-1"), "chunk-1": FixturePage(rows[500:1000], "chunk-2"), "chunk-2": FixturePage(rows[1000:])} checks = 0 def check(ok): nonlocal checks if not ok: raise AssertionError("acceptance check failed") checks += 1 def rejected(s=scope, n=None, p=None, e=expected): try: collect(s, note if n is None else n, pages if p is None else p, e) except IncompleteInventory: check(True) else: raise AssertionError("an incomplete inventory was accepted") snapshot = copy.deepcopy(pages) result = collect(scope, note, pages, expected) check(len(result) == 1203) check(result == dict(expected)) check(result["customer-1202"] == "terms-reviewed") check(pages == snapshot) check(collect(scope, note, pages, expected) == result) check(collect(scope, note, {"chunk-0": FixturePage(rows)}, expected) == result) check(collect(scope, note, {"chunk-0": FixturePage((), "all"), "all": FixturePage(rows)}, expected) == result) for change in [dict(changeType="updated"), dict(changeType="deleted"), dict(changeType="created"), dict(subscriptionId="other"), dict(clientState="other"), dict(resource="other-company"), dict(resource=scope.approved_resource + "-rewritten")]: rejected(n={**note, **change}) for key in ["changeType", "subscriptionId", "clientState", "resource"]: n = dict(note) del n[key] rejected(n=n) rejected(s=replace(scope, company="")) rejected(s=replace(scope, company="company-B")) rejected(p={"chunk-0": pages["chunk-0"]}) rejected(p={**pages, "chunk-0": replace(pages["chunk-0"], continuation=None)}) rejected(p={**pages, "chunk-2": FixturePage((), "chunk-0")}) rejected(p={**pages, "chunk-2": FixturePage((), "missing-page")}) rejected(p={**pages, "chunk-2": replace(pages["chunk-2"], read_ok=False)}) rejected(p={**pages, "chunk-2": replace(pages["chunk-2"], read_ok=1)}) rejected(p={**pages, "chunk-2": replace(pages["chunk-2"], continuation="")}) rejected(p={**pages, "chunk-2": replace(pages["chunk-2"], continuation=True)}) rejected(p={"chunk-0": FixturePage(rows[:-1])}) rejected(p={"chunk-0": FixturePage(rows + (rows[0],))}) rejected(p={"chunk-0": FixturePage((replace(rows[0], revision="old-terms"),) + rows[1:])}) rejected(p={"chunk-0": FixturePage(rows[1:] + (replace(rows[0], record_id="unexpected"),))}) rejected(p={"chunk-0": FixturePage((replace(rows[0], company="company-B"),) + rows[1:])}) rejected(p={"chunk-0": FixturePage((replace(rows[0], record_id=""),) + rows[1:])}) rejected(p={"chunk-0": FixturePage((replace(rows[0], revision=""),) + rows[1:])}) rejected(e=expected + (expected[0],)) rejected(e=()) rejected(e=(("", "terms-reviewed"),)) rejected(e=(("customer-0000", ""),)) print(f"{checks} local checks passed. Matched fixture customers: {len(result)}.") print("Native Business Central reads: 0; Power Automate runs: 0; HTTP calls: 0; writes: 0.") if __name__ == "__main__": main()