"""Offline Colissimo delegated-account request preparation on fictional parcels. Requests 2.32.5 builds real PreparedRequest objects; nothing is sent. ACCOUNT_NUMBER selects the fictional target; a shared fictional apiKey is not a parcel owner. No provider rights, parcel validity, bordereau, PDF or physical deposit is tested. """ from collections import Counter from copy import deepcopy from importlib.metadata import version import json import sys import requests ENDPOINT = "https://ws.colissimo.fr/sls-ws/SlsServiceWSRest/3.1/generateBordereauByParcelsNumbers" attempts = 0 prepared_count = 0 header_failures = 0 connections = 0 checks = [] def offline(event, _args): global connections if event == "socket.connect": connections += 1 raise AssertionError("No connections are permitted by this exercise") sys.addaudithook(offline) def check(name, condition): assert name not in checks, name assert condition, name checks.append(name) def prepare(account, parcels, api_key, account_fields=None): global attempts, prepared_count, header_failures attempts += 1 fields = ([{"key": "ACCOUNT_NUMBER", "value": account}] if account_fields is None else account_fields) body = {"generateBordereauParcelNumberList": {"parcelsNumbers": parcels}, "fields": {"field": fields}} try: request = requests.Request("POST", ENDPOINT, headers={"apiKey": api_key}, json=body).prepare() except requests.exceptions.InvalidHeader: header_failures += 1 raise prepared_count += 1 return request def inspect_request(request, inventory): if request.method != "POST" or request.url != ENDPOINT: return {"decision": "other_operation_review"} body = json.loads(request.body) fields = body.get("fields", {}).get("field", []) targets = [f.get("value") for f in fields if isinstance(f, dict) and f.get("key") == "ACCOUNT_NUMBER"] if len(targets) != 1 or not isinstance(targets[0], str) or not targets[0]: return {"decision": "target_account_field_review"} parcels = body.get("generateBordereauParcelNumberList", {}).get("parcelsNumbers") if not isinstance(parcels, list) or not parcels: return {"decision": "parcel_collection_review"} if any(not isinstance(p, str) for p in parcels): return {"decision": "parcel_collection_review"} if len(set(parcels)) != len(parcels): return {"decision": "duplicate_parcel_review"} if any(p not in inventory for p in parcels): return {"decision": "unknown_parcel_review"} target = targets[0] wrong = [p for p in parcels if inventory[p] != target] if wrong: return {"decision": "mixed_account_request_review", "targetAccount": target, "wrongAccountParcels": wrong} return {"decision": "candidate_for_bordereau_review", "targetAccount": target, "parcels": parcels, "parcelCount": len(parcels)} def inspect_wave(prepared, inventory): observations = [inspect_request(r, inventory) for r in prepared] if any(x["decision"] != "candidate_for_bordereau_review" for x in observations): return {"decision": "request_content_review", "requests": observations} seen = Counter(p for x in observations for p in x["parcels"]) if seen != Counter({p: 1 for p in inventory}): return {"decision": "prepared_wave_coverage_review"} return {"decision": "candidate_wave_for_review", "requests": observations} def plan(rows, profile, inventory): if not isinstance(rows, list) or not rows: return "empty_plan_review", [] if not isinstance(profile, dict) or not isinstance(profile.get("apiKey"), str) or not profile["apiKey"]: return "credential_profile_review", [] targets = profile.get("clientTargets") if not isinstance(targets, dict) or not isinstance(inventory, dict): return "client_profile_review", [] numbers = [] for row in rows: if not isinstance(row, dict) or not isinstance(row.get("parcel"), str): return "row_shape_review", [] numbers.append(row["parcel"]) if len(set(numbers)) != len(numbers): return "duplicate_parcel_review", [] if set(numbers) != set(inventory): return "prepared_inventory_set_review", [] if any(not isinstance(owner, str) or not owner for owner in inventory.values()): return "inventory_owner_review", [] groups = {} for row in rows: client = row.get("client") if not isinstance(client, str) or client not in targets: return "client_profile_review", [] target = targets[client] if not isinstance(target, str) or not target: return "target_profile_review", [] if inventory[row["parcel"]] != target: return "parcel_account_review", [] groups.setdefault(target, []).append(row["parcel"]) prepared = [] for target, parcels in sorted(groups.items()): try: prepared.append(prepare(target, parcels, profile["apiKey"])) except requests.exceptions.InvalidHeader: return "native_header_review", [] return "candidate_wave_for_review", prepared def main(): check("native_requests_pin", version("requests") == "2.32.5") # Independent fictional preparation inventory, not inferred from the apiKey. inventory = {"DEMO-A-1": "000001", "DEMO-A-2": "000001", "DEMO-A-3": "000001", "DEMO-B-1": "000002", "DEMO-B-2": "000002", "DEMO-B-3": "000002"} rows = [{"parcel": "DEMO-A-1", "client": "client-a"}, {"parcel": "DEMO-B-1", "client": "client-b"}, {"parcel": "DEMO-A-2", "client": "client-a"}, {"parcel": "DEMO-B-2", "client": "client-b"}, {"parcel": "DEMO-A-3", "client": "client-a"}, {"parcel": "DEMO-B-3", "client": "client-b"}] profile = {"apiKey": "FICTIONAL_SHARED_KEY_NOT_VALID", "clientTargets": {"client-a": "000001", "client-b": "000002"}} naive_request = prepare("000001", [r["parcel"] for r in rows], profile["apiKey"]) naive = inspect_request(naive_request, inventory) check("naive_native_prepared_mixed_accounts", naive["decision"] == "mixed_account_request_review") check("naive_wrong_client_parcels_exact", naive["wrongAccountParcels"] == ["DEMO-B-1", "DEMO-B-2", "DEMO-B-3"]) cases = [] def add(name, expected, change_rows=None, change_profile=None, change_inventory=None): rs, cfg, inv = deepcopy(rows), deepcopy(profile), deepcopy(inventory) if change_rows: change_rows(rs) if change_profile: change_profile(cfg) if change_inventory: change_inventory(inv) cases.append((name, rs, cfg, inv, expected)) add("two_delegated_clients", "candidate_wave_for_review") add("row_order_reversed", "candidate_wave_for_review", lambda r: r.reverse()) def keep_client_a(rs): rs[:] = [row for row in rs if row["client"] == "client-a"] def keep_inventory_a(inv): for parcel in tuple(inv): if inv[parcel] != "000001": del inv[parcel] def use_second_alias(rs): keep_client_a(rs) rs[0]["client"] = "client-a-alias" add("one_target_wave", "candidate_wave_for_review", keep_client_a, change_inventory=keep_inventory_a) add("two_aliases_one_carrier_account", "candidate_wave_for_review", use_second_alias, lambda p: p["clientTargets"].update({"client-a-alias": "000001"}), keep_inventory_a) add("missing_client", "client_profile_review", lambda r: r[0].pop("client")) add("unknown_client", "client_profile_review", lambda r: r[0].update(client="unknown")) add("missing_target_mapping", "client_profile_review", change_profile=lambda p: p["clientTargets"].pop("client-b")) add("numeric_target", "target_profile_review", change_profile=lambda p: p["clientTargets"].update({"client-a": 1})) add("wrong_target_mapping", "parcel_account_review", change_profile=lambda p: p["clientTargets"].update({"client-a": "000002"})) add("duplicate_input", "duplicate_parcel_review", lambda r: r.append(deepcopy(r[0]))) add("missing_prepared_parcel", "prepared_inventory_set_review", lambda r: r.pop()) add("unknown_parcel", "prepared_inventory_set_review", lambda r: r[0].update(parcel="DEMO-UNKNOWN")) add("empty_wave", "empty_plan_review", lambda r: r.clear()) add("unknown_owner", "inventory_owner_review", change_inventory=lambda i: i.update({"DEMO-A-1": None})) add("missing_key", "credential_profile_review", change_profile=lambda p: p.pop("apiKey")) add("invalid_header", "native_header_review", change_profile=lambda p: p.update(apiKey="FICTIONAL\nINVALID")) observations = [] nominal_prepared = None for name, rs, cfg, inv, expected in cases: decision, prepared = plan(rs, cfg, inv) check(name, decision == expected) item = {"case": name, "decision": decision, "preparedRequests": len(prepared)} if prepared: audited = inspect_wave(prepared, inv) check(name + "_serialized_coverage_and_ownership", audited["decision"] == "candidate_wave_for_review") item["requests"] = audited["requests"] check(name + "_one_shared_fictional_key", all(r.headers["apiKey"] == profile["apiKey"] for r in prepared)) if name == "two_delegated_clients": nominal_prepared = prepared observations.append(item) by_name = {x["case"]: x for x in observations} check("two_clients_two_requests_same_key", by_name["two_delegated_clients"]["preparedRequests"] == 2) check("same_carrier_target_one_request", by_name["two_aliases_one_carrier_account"]["preparedRequests"] == 1) check("leading_zero_account_preserved", by_name["two_delegated_clients"]["requests"][0]["targetAccount"] == "000001") lost = inspect_wave(nominal_prepared[:1], inventory) check("lost_account_request_detected", lost["decision"] == "prepared_wave_coverage_review") wrong_case = prepare("000001", ["DEMO-A-1"], profile["apiKey"], [{"key": "account_number", "value": "000001"}]) check("exact_target_marker_required_locally", inspect_request(wrong_case, inventory)["decision"] == "target_account_field_review") duplicate_field = prepare("000001", ["DEMO-A-1"], profile["apiKey"], [{"key": "ACCOUNT_NUMBER", "value": "000001"}, {"key": "ACCOUNT_NUMBER", "value": "000002"}]) check("ambiguous_target_markers_review", inspect_request(duplicate_field, inventory)["decision"] == "target_account_field_review") check("native_header_failure_once", header_failures == 1) check("offline", connections == 0) print(json.dumps({"checksPassed": len(checks), "checks": checks, "planCases": len(cases), "nativePreparationAttempts": attempts, "nativePreparedRequests": prepared_count, "expectedInvalidHeaderFailures": header_failures, "requests": version("requests"), "naiveObservation": naive, "observations": observations, "omittedAccountRequestDecision": lost["decision"], "networkConnectionAttempts": connections, "colissimoApiCalls": 0, "pdfFilesCreated": 0, "erpWrites": 0, "apiKeysPrinted": False}, indent=2)) if __name__ == "__main__": main()