"""Offline Sendcloud V2 cancellation-response experiment; no API credentials. Native Requests prepares requests and processes synthetic HTTP responses from a local adapter. The adapter never opens a socket. It is not a Sendcloud SDK or a Sendcloud sandbox. All parcel identities, bodies and warehouse facts are fixtures. """ from __future__ import annotations from dataclasses import dataclass import json import sys from urllib.parse import urlsplit import requests from requests.adapters import BaseAdapter EXPECTED_REQUESTS = "2.32.5" BASE = "https://panel.sendcloud.sc/api/v2" PARCEL = "990000107" OTHER_PARCEL = "990000108" AUDIT = {"socket_connection_attempts": 0} CHECKS: list[str] = [] def deny_connections(event: str, args: tuple) -> None: if event == "socket.connect": AUDIT["socket_connection_attempts"] += 1 raise RuntimeError("This experiment forbids network connections") sys.addaudithook(deny_connections) def check(name: str, condition: bool) -> None: if not condition: raise AssertionError(name) CHECKS.append(name) @dataclass(frozen=True) class Fixture: name: str http_status: int body: str expected_native_ok: bool expected_native_http_error: bool expected_decision: str expected_parcel: str = PARCEL FIXTURES = [ Fixture("cancelled_200", 200, '{"status":"cancelled","message":"Parcel has been cancelled"}', True, False, "cancellation_reported"), Fixture("queued_202_empty_body", 202, "", True, False, "pending"), Fixture("deleted_410", 410, '{"status":"deleted","message":"Parcel has been deleted"}', False, True, "deletion_reported"), Fixture("rejected_400", 400, '{"error":"fictional cancellation rejection"}', False, True, "rejected"), Fixture("missing_404", 404, "", False, True, "unresolved"), Fixture("remote_error_500", 500, "", False, True, "unresolved"), Fixture("unrecognised_200", 200, '{"status":"ready","message":"Fictional unsupported outcome"}', True, False, "unresolved"), Fixture("v3_envelope_at_v2_boundary", 200, '{"data":{"status":"cancelled","message":"Shipment cancelled"}}', True, False, "unresolved"), Fixture("invalid_json_200", 200, "not-json", True, False, "unresolved"), Fixture("queued_with_conflicting_body", 202, '{"status":"cancelled","message":"Synthetic inconsistent fixture"}', True, False, "pending"), Fixture("missing_message_200", 200, '{"status":"cancelled"}', True, False, "unresolved"), Fixture("array_200", 200, '[]', True, False, "unresolved"), Fixture("wrong_target_200", 200, '{"status":"cancelled","message":"Parcel has been cancelled"}', True, False, "wrong_target", OTHER_PARCEL), ] class FixtureAdapter(BaseAdapter): """Return one synthetic Response through the native Session.send path.""" def __init__(self, fixture: Fixture): super().__init__() self.fixture = fixture self.prepared: list[requests.PreparedRequest] = [] def send(self, request, **kwargs): if request.method != "POST": raise AssertionError("Only the fixture POST is supported") if request.url != f"{BASE}/parcels/{PARCEL}/cancel": raise AssertionError("Unexpected URL; no network fallback is available") if request.body != b"{}": raise AssertionError("Unexpected prepared JSON body") self.prepared.append(request) response = requests.Response() response.status_code = self.fixture.http_status response.reason = "Synthetic fixture" response.url = request.url response.request = request response.encoding = "utf-8" response.headers["Content-Type"] = "application/json" response._content = self.fixture.body.encode("utf-8") response._content_consumed = True return response def close(self): return None def local_decision(response: requests.Response, expected_parcel: str) -> str: """Our local interpretation, not a native SDK state machine.""" expected_path = f"/api/v2/parcels/{expected_parcel}/cancel" if urlsplit(response.request.url).path != expected_path: return "wrong_target" if response.status_code == 202: return "pending" if response.status_code == 400: return "rejected" if response.status_code not in (200, 410): return "unresolved" try: body = response.json() except requests.exceptions.JSONDecodeError: return "unresolved" if not isinstance(body, dict) or not isinstance(body.get("message"), str): return "unresolved" if not body["message"]: return "unresolved" if response.status_code == 200 and body.get("status") == "cancelled": return "cancellation_reported" if response.status_code == 410 and body.get("status") == "deleted": return "deletion_reported" return "unresolved" def handoff(decision: str, warehouse_stop_confirmed: bool) -> str: """An explicit warehouse acknowledgement is a separate fictional input.""" if decision == "cancellation_reported" and warehouse_stop_confirmed: return "ready_for_business_review" return "keep_stock_release_blocked" def run() -> dict: check("native_requests_version_pinned", requests.__version__ == EXPECTED_REQUESTS) observations = [] for fixture in FIXTURES: adapter = FixtureAdapter(fixture) with requests.Session() as session: session.trust_env = False session.mount("https://", adapter) session.mount("http://", adapter) response = session.post( f"{BASE}/parcels/{PARCEL}/cancel", json={}, timeout=5, allow_redirects=False, ) check(fixture.name + ":one_prepared_request", len(adapter.prepared) == 1) check(fixture.name + ":native_response_type", type(response) is requests.Response) native_ok = response.ok naive_status = "cancelled" if native_ok else "failed" check(fixture.name + ":native_ok", native_ok == fixture.expected_native_ok) native_error = None try: response.raise_for_status() except requests.exceptions.HTTPError as exc: native_error = type(exc).__name__ check(fixture.name + ":native_error_keeps_response", exc.response is response) check(fixture.name + ":native_http_error", (native_error is not None) == fixture.expected_native_http_error) decision = local_decision(response, fixture.expected_parcel) check(fixture.name + ":local_interpretation", decision == fixture.expected_decision) waiting = handoff(decision, warehouse_stop_confirmed=False) check(fixture.name + ":no_invented_warehouse_stop", waiting == "keep_stock_release_blocked") observations.append({ "fixture": fixture.name, "http_status": response.status_code, "native_ok": native_ok, "naive_status": naive_status, "native_raise_for_status_error": native_error, "local_decision": decision, "without_warehouse_ack": waiting, "method": adapter.prepared[0].method, "prepared_path": urlsplit(adapter.prepared[0].url).path, "prepared_body": adapter.prepared[0].body.decode("utf-8"), }) queued = next(x for x in observations if x["fixture"] == "queued_202_empty_body") check("naive_202_false_closure_reproduced", queued["naive_status"] == "cancelled" and queued["local_decision"] == "pending") check("cancelled_and_warehouse_ack_are_separate_inputs", handoff("cancellation_reported", True) == "ready_for_business_review") check("queued_even_with_warehouse_ack_is_not_cancelled", handoff("pending", True) == "keep_stock_release_blocked") check("deleted_is_not_a_cancelled_label", handoff("deletion_reported", True) == "keep_stock_release_blocked") check("missing_is_not_a_cancelled_label", handoff("unresolved", True) == "keep_stock_release_blocked") check("zero_socket_connection_attempts", AUDIT["socket_connection_attempts"] == 0) return { "python": sys.version.split()[0], "requests": requests.__version__, "checks_passed": len(CHECKS), "checks": CHECKS, "native_prepared_requests": len(observations), "synthetic_http_responses": len(observations), "real_connection_attempts": AUDIT["socket_connection_attempts"], "real_sendcloud_api_calls": 0, "real_erp_wms_calls": 0, "real_label_or_stock_writes": 0, "parcel_identities_are_fictional": True, "observations": observations, } if __name__ == "__main__": print(json.dumps(run(), ensure_ascii=False, indent=2))