#!/usr/bin/env python3 """Read synthetic multipart responses locally. No HTTP or SAP operations.""" from dataclasses import dataclass, replace from email import policy from email.parser import BytesParser import json import re @dataclass(frozen=True) class Unit: name: str # Empty tuple means a read; otherwise Content-ID and expected final status. operations: tuple[tuple[str, int], ...] = () def parse_part(part): if len(part.get_all("Content-ID", [])) > 1 or len(part.get_all("Content-Type", [])) != 1: raise ValueError("ambiguous_mime_headers") if part.defects: raise ValueError("malformed_mime") if part.get_content_type() == "multipart/mixed": children = [parse_part(child) for child in part.iter_parts()] if not children or any(child["kind"] != "http" for child in children): raise ValueError("unexpected_nested_group") return {"kind": "changeset", "children": children} if part.get_content_type() != "application/http": raise ValueError("unexpected_part_type") payload = part.get_payload(decode=True) if not isinstance(payload, bytes): raise ValueError("missing_http_payload") first, separator, rest = payload.partition(b"\r\n") match = re.fullmatch(rb"HTTP/1\.[01] ([1-5][0-9]{2}) [^\r\n]+", first) if not separator or not match or b"\r\n\r\n" not in rest: raise ValueError("invalid_http_response") raw_body = rest.split(b"\r\n\r\n", 1)[1] native_error = False if raw_body: try: body = json.loads(raw_body) except (ValueError, UnicodeDecodeError): body = None native_error = isinstance(body, dict) and isinstance(body.get("error"), dict) and isinstance(body["error"].get("code"), (int, str)) and not isinstance(body["error"].get("code"), bool) and str(body["error"].get("code")) != "" return {"kind": "http", "status": int(match[1]), "cid": part.get("Content-ID"), "native_error_shape": native_error} def inspect_batch(plan, content_type, raw_body, *, outer_status=200, expected_context=("sap-fixture", "A", "run-42"), observed_context=("sap-fixture", "A", "run-42")): """Strict local fixture policy for synchronous OData V4 multipart only.""" if observed_context != expected_context: raise ValueError("wrong_context") if outer_status != 200: raise ValueError("unexpected_outer_status") if any(not cid or status not in {200,201,204} for unit in plan for cid,status in unit.operations): raise ValueError("invalid_plan_operation") ids = [cid for unit in plan for cid, status in unit.operations] if len(ids) != len(set(ids)) or len({u.name for u in plan}) != len(plan): raise ValueError("ambiguous_plan") if "\r" in content_type or "\n" in content_type: raise ValueError("invalid_content_type") message = BytesParser(policy=policy.default).parsebytes(("Content-Type: "+content_type+"\r\nMIME-Version: 1.0\r\n\r\n").encode()+raw_body) if message.get_content_type() != "multipart/mixed" or not message.is_multipart() or message.defects: raise ValueError("incomplete_or_invalid_batch") parts = [parse_part(p) for p in message.iter_parts()] if len(parts) > len(plan): raise ValueError("unexpected_unit") result = {} stopped = False for index, unit in enumerate(plan): if index >= len(parts): result[unit.name] = "not_reported" continue if stopped: raise ValueError("unexpected_response_after_stop") part = parts[index] if not unit.operations: if part["kind"] != "http" or part["status"] != 200: raise ValueError("unexpected_read_result") result[unit.name] = "read_success" elif part["kind"] == "http": if not (400 <= part["status"] <= 599 and part["native_error_shape"]): raise ValueError("unqualified_changeset_error") if part["cid"] is not None and part["cid"] not in dict(unit.operations): raise ValueError("unknown_error_content_id") result[unit.name] = "changeset_failed" stopped = True else: seen = [child["cid"] for child in part["children"]] expected = dict(unit.operations) if len(seen) != len(set(seen)) or set(seen) != set(expected): raise ValueError("missing_or_duplicate_content_id") if any(child["status"] != expected[child["cid"]] for child in part["children"]): raise ValueError("unexpected_operation_status") result[unit.name] = "changeset_success" return result # All payloads below are authored synthetic fixtures, not captured SAP traffic. def http(status=200, cid=None, body=None): headers = b"Content-Type: application/http\r\nContent-Transfer-Encoding: binary\r\n" if cid is not None: headers += ("Content-ID: "+cid+"\r\n").encode() data = b"" if body is None else json.dumps(body).encode() return headers+b"\r\n"+(f"HTTP/1.1 {status} Fixture\r\nContent-Type: application/json\r\n\r\n").encode()+data def multipart(parts, boundary): return b"".join(("--"+boundary+"\r\n").encode()+p+b"\r\n" for p in parts)+("--"+boundary+"--\r\n").encode() def group(parts, boundary="cs_fixture"): return ("Content-Type: multipart/mixed; boundary="+boundary+"\r\n\r\n").encode()+multipart(parts,boundary) def run(): count = 0 def check(value, label): nonlocal count if not value: raise AssertionError(label) count += 1 def rejects(call, label): try: call() except ValueError: check(True,label) else: raise AssertionError(label) plan = (Unit("read"), Unit("catalogue", (("1",201),("2",204))), Unit("next", (("3",201),))) ctype = "multipart/mixed; boundary=b_fixture" success = group([http(201,"1",{"ItemCode":"FICTIF-A"}),http(204,"2")]) following = group([http(201,"3",{"ItemCode":"FICTIF-B"})],"cs_next") error = http(400,body={"error":{"code":-10,"message":{"value":"Synthetic rejected item"}}}) payload = multipart([http(),success,following],"b_fixture") read = lambda body=payload, p=plan, ct=ctype, **kw:inspect_batch(p,ct,body,**kw) check(read()=={"read":"read_success","catalogue":"changeset_success","next":"changeset_success"},"complete success") failed = read(multipart([http(),error],"b_fixture")) check(failed["catalogue"]=="changeset_failed","one native-shaped error applies to whole expected changeset") check(failed["next"]=="not_reported","later unit is not certified") check(failed["read"]=="read_success","prior read remains observed") prior = read(multipart([http(),success,error],"b_fixture")) check(prior["catalogue"]=="changeset_success","earlier independent changeset stays successful") check(prior["next"]=="changeset_failed","later changeset fails separately") reverse = group([http(204,"2"),http(201,"1")]) check(read(multipart([http(),reverse,following],"b_fixture"))["catalogue"]=="changeset_success","Content-ID matches reordered responses") check(read(multipart([http()],"b_fixture"))["catalogue"]=="not_reported","complete MIME is not complete business evidence") check(read(multipart([http()],"b_fixture"))["next"]=="not_reported","no fabricated tail success") rejects(lambda:read(observed_context=("sap-fixture","B","run-42")),"wrong company") rejects(lambda:read(observed_context=("other-fixture","A","run-42")),"wrong installation") rejects(lambda:read(observed_context=("sap-fixture","A","run-43")),"wrong source run") rejects(lambda:read(outer_status=202),"asynchronous/interim outer status unsupported") rejects(lambda:read(outer_status=500),"outer error not enough to classify changesets") rejects(lambda:read(payload.rsplit(b"--b_fixture--",1)[0]),"truncated batch closing boundary") rejects(lambda:read(ct="application/json"),"wrong top media type") rejects(lambda:read(ct="multipart/mixed"),"missing batch boundary") rejects(lambda:read(ct=ctype+"\r\nX-Test: bad"),"header injection refused") rejects(lambda:read(multipart([http(),success,following,http()],"b_fixture")),"extra unexpected unit") rejects(lambda:read(p=(plan[0],plan[1],Unit("next",(("1",201),)))),"Content-ID unique across batch plan") rejects(lambda:read(p=(plan[0],Unit("bad",(("",201),)))),"blank content ID in plan") rejects(lambda:read(p=(plan[0],Unit("bad",(("1",202),)))),"interim status cannot be a final plan result") rejects(lambda:read(p=(plan[0],replace(plan[1],name="read"),plan[2])),"unit names unambiguous") for bad, label in [(group([http(201,"1"),http(204,"1")]),"duplicate response ID"),(group([http(201,"1")]),"missing response ID"),(group([http(201,"1"),http(204,"9")]),"unknown response ID"),(group([http(201),http(204,"2")]),"absent response ID"),(group([http(201,"1"),http(202,"2")]),"interim operation not final success"),(group([http(201,"1"),http(400,"2",{"error":{"code":-10}})]),"unexpected mixed error shape"),(http(200),"bare HTTP success is not changeset success"),(http(400,body={"message":"proxy refusal"}),"unqualified error origin"),(http(400,cid="9",body={"error":{"code":-10}}),"unknown error ID"),(group([group([http(201,"1")],"cs_inner")]),"nested changeset"),(b"Content-Type: text/plain\r\n\r\nHTTP/1.1 200 Fixture\r\n\r\n", "wrong MIME child type")]: rejects(lambda part=bad:read(multipart([http(),part],"b_fixture")),label) check(read(multipart([http(),http(400,cid="2",body={"error":{"code":-10}})],"b_fixture"))["catalogue"]=="changeset_failed","error content ID does not split atomic group") duplicate_id = http(201,"1").replace(b"Content-ID: 1\r\n", b"Content-ID: 1\r\nContent-ID: 9\r\n") rejects(lambda:read(multipart([http(),group([duplicate_id,http(204,"2")])],"b_fixture")),"contradictory duplicate MIME content ID header") duplicate_type = http(201,"1").replace(b"Content-Type: application/http\r\n", b"Content-Type: application/http\r\nContent-Type: text/plain\r\n",1) rejects(lambda:read(multipart([http(),group([duplicate_type,http(204,"2")])],"b_fixture")),"contradictory duplicate MIME content type header") invalid = http().replace(b"HTTP/1.1 200 Fixture",b"NOT-HTTP") rejects(lambda:read(multipart([invalid],"b_fixture")),"invalid embedded status line") rejects(lambda:read(multipart([http(204)],"b_fixture")),"unexpected read status") rejects(lambda:read(multipart([http(),error,following],"b_fixture")),"later response unsupported after error in stop-on-error fixture policy") for code in [None,False,""]: rejects(lambda value=code:read(multipart([http(),http(400,body={"error":{"code":value}})],"b_fixture")),"invalid native-shaped error code") check(payload.startswith(b"--b_fixture\r\n") and payload.endswith(b"--b_fixture--\r\n"),"fixture has complete MIME framing") print(f"{count} local checks passed. Native SAP transactions: 0; HTTP calls: 0; ERP writes: 0.") if __name__ == "__main__": run()