#!/usr/bin/env python3 """Classify authored read observations. No token, HTTP, BC access or grant changes.""" from dataclasses import dataclass, replace from typing import Optional @dataclass(frozen=True) class Case: name: str company: str resource: str record: str marker: str allowed: bool @dataclass(frozen=True) class Observation: case: str context: tuple[str, str, str] # fictional tenant, environment, client identity company: str resource: str record: str run: str observed_at: int status: int method: str = "GET" body_id: Optional[str] = None body_marker: Optional[str] = None # Normalized category supplied by an adapter, not a parsed native BC payload. error_category: Optional[str] = None def inspect(cases, observations, *, context=("tenant-fixture", "sandbox", "client-export"), run="review-42", now=1000, max_age=90): if not cases or max_age <= 0: raise ValueError("invalid_review") if any(not c.name or not c.company or not c.resource or not c.record or not c.marker or type(c.allowed) is not bool for c in cases): raise ValueError("incomplete_case") expected = {c.name: c for c in cases} if len(expected) != len(cases): raise ValueError("duplicate_case") seen = {} for o in observations: if o.case not in expected or o.case in seen: raise ValueError("unknown_or_duplicate_observation") c = expected[o.case] if (o.context, o.company, o.resource, o.record, o.run) != (context, c.company, c.resource, c.record, run): raise ValueError("context_mismatch") if o.method != "GET" or type(o.status) is not int or not 100 <= o.status <= 599: raise ValueError("unsupported_observation") if type(o.observed_at) is not int or not 0 <= now-o.observed_at <= max_age: raise ValueError("observation_outside_review_window") if o.status == 200: if o.error_category is not None or (o.body_id, o.body_marker) != (c.record, c.marker): raise ValueError("success_not_matched_to_fixture") verdict = "allowed_read_observed" if c.allowed else "excess_read_observed" elif o.status == 403 and o.error_category == "Authorization": verdict = "required_read_denied" if c.allowed else "authorization_refusal_observed" elif o.status == 401: verdict = "authentication_unresolved" elif o.status == 404: verdict = "not_found_not_authorization_proof" else: verdict = "indeterminate_response" seen[o.case] = verdict return {c.name: seen.get(c.name, "missing_observation") for c in cases} def run(): count = 0 def check(condition, label): nonlocal count if not condition: raise AssertionError(label) count += 1 def rejects(call, label): try: call() except ValueError: check(True, label) else: raise AssertionError(label) a = Case("customers-A", "A", "customers", "client-A-42", "fixture-A", True) b = Case("customers-B", "B", "customers", "client-B-42", "fixture-B", False) cases = (a, b) good = Observation(a.name, ("tenant-fixture", "sandbox", "client-export"), "A", "customers", a.record, "review-42", 990, 200, body_id=a.record, body_marker=a.marker) denied = Observation(b.name, good.context, "B", "customers", b.record, good.run, 991, 403, error_category="Authorization") review = lambda observations=(good, denied), **kw: inspect(cases, observations, **kw) check(review()=={a.name:"allowed_read_observed",b.name:"authorization_refusal_observed"},"both planned observations qualified") check(review((denied,good))==review(),"observation order irrelevant") excess=replace(denied,status=200,body_id=b.record,body_marker=b.marker,error_category=None) check(review((good,excess))[b.name]=="excess_read_observed","forbidden fixture actually readable") check(review((replace(good,status=403,body_id=None,body_marker=None,error_category="Authorization"),denied))[a.name]=="required_read_denied","required read refused") check(review((good,))[b.name]=="missing_observation","untested company remains untested") check(review(())[a.name]=="missing_observation","no evidence is not success") for status, category, verdict in [(404,None,"not_found_not_authorization_proof"),(401,"Authentication","authentication_unresolved"),(403,None,"indeterminate_response"),(500,"Internal","indeterminate_response"),(429,None,"indeterminate_response"),(204,None,"indeterminate_response")]: check(review((good,replace(denied,status=status,error_category=category)))[b.name]==verdict,"HTTP status alone does not certify exclusion") for change, label in [({"context":("other","sandbox","client-export")},"wrong tenant"),({"context":("tenant-fixture","production","client-export")},"wrong environment"),({"context":("tenant-fixture","sandbox","admin")},"admin identity instead of connector"),({"company":"B"},"wrong company"),({"resource":"vendors"},"wrong resource"),({"record":"client-A-43"},"wrong requested record"),({"run":"review-43"},"wrong review"),({"method":"POST"},"mutation not allowed in read exercise"),({"status":True},"boolean is not HTTP status"),({"status":600},"invalid HTTP status"),({"observed_at":909},"stale observation"),({"observed_at":1001},"future observation"),({"observed_at":True},"boolean is not observation time"),({"body_id":"client-A-43"},"wrong returned record"),({"body_marker":"another-fixture"},"wrong fixture marker"),({"body_id":None,"body_marker":None},"empty response not access denial"),({"error_category":"Authorization"},"contradictory success/error observation")]: rejects(lambda change=change:review((replace(good,**change),denied)),label) rejects(lambda:review((good,good,denied)),"duplicate observation") rejects(lambda:review((replace(good,case="unknown"),denied)),"unplanned case") rejects(lambda:inspect((a,a),(good,)),"duplicate plan") rejects(lambda:inspect((replace(a,allowed=1),b),(good,denied)),"explicit boolean policy required") rejects(lambda:inspect((replace(a,record=""),b),(good,denied)),"known fixture identity required") rejects(lambda:inspect((),()),"empty review not closure") rejects(lambda:review(max_age=0),"explicit positive observation window") check(review((replace(good,observed_at=910),denied))[a.name]=="allowed_read_observed","configured local window boundary") print(f"{count} local checks passed. Native Business Central reads: 0; HTTP calls: 0; permission changes: 0.") if __name__ == "__main__": run()