"""Original offline S3 witness. No account, HTTP connection or ERP write.

Unix setup: python3 -m venv .venv-s3
.venv-s3/bin/python -m pip install botocore==1.43.114
.venv-s3/bin/python -B multipart_completion_receipts.py

Installation uses the package registry; the witness itself closes the network.
Prepared before-send events exercise the real serializer, parser and S3 error
handlers. They do not establish AWS authorization, persistence or atomicity.
"""

import hashlib
import json
import platform
import socket
from collections import deque
from dataclasses import dataclass
from importlib.metadata import version
from urllib.parse import parse_qs, urlsplit
from xml.etree import ElementTree as ET


CHECKS = []
CONNECTIONS = []
REAL_HTTP = []


def closed_connection(*args, **kwargs):
    CONNECTIONS.append("blocked socket connection")
    raise AssertionError("The offline witness must not open a connection")


socket.create_connection = closed_connection
socket.socket.connect = closed_connection
socket.socket.connect_ex = closed_connection

import botocore.session
from botocore import UNSIGNED
from botocore.awsrequest import AWSResponse
from botocore.config import Config
from botocore.exceptions import ClientError
from botocore.httpsession import URLLib3Session


def closed_http(*args, **kwargs):
    REAL_HTTP.append("blocked HTTP session")
    raise AssertionError("A prepared response was not supplied")


URLLib3Session.send = closed_http


def check(label, condition):
    if not condition:
        raise AssertionError(label)
    CHECKS.append(label)


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


BUCKET = "erp-fixture"
KEY = "exports/EXP-309.csv"
UPLOAD = "upload-fixture-309"
REVISION = "extract-fixture-4"
SOURCE = b"reference;quantite\n" + b"SKU-41;2\n" * 600000
PARTS = [SOURCE[:5 * 1024 * 1024], SOURCE[5 * 1024 * 1024:]]
ETAGS = ['"opaque-part-one"', '"opaque-part-two"']
SCOPE = (BUCKET, KEY, UPLOAD)
ERP = {"state": "awaiting_document", "document": None}


@dataclass(frozen=True)
class Receipt:
    number: int
    size: int
    etag: str
    slice_sha256: str


class RawResponse:
    def __init__(self, body):
        self.body = body

    def stream(self, amt=None, decode_content=False):
        yield self.body


class PreparedEvents:
    def __init__(self):
        self.responses = deque()
        self.calls = []

    def add(self, operation, status=200, body=b"", headers=None):
        self.responses.append((operation, status, body, headers or {}))

    def reply(self, request, operation, **kwargs):
        if not self.responses:
            raise AssertionError("Unexpected additional SDK request")
        expected, status, body, headers = self.responses.popleft()
        check("prepared operation " + str(len(self.calls) + 1), expected == operation)
        request_body = request.body
        if hasattr(request_body, "read"):
            position = request_body.tell()
            request_body = request_body.read()
            request.body.seek(position)
        request_body = request_body or b""
        if isinstance(request_body, str):
            request_body = request_body.encode()
        self.calls.append({
            "operation": operation,
            "method": request.method,
            "url": request.url,
            "headers": {k.lower(): v.decode() if isinstance(v, bytes) else v
                        for k, v in request.headers.items()},
            "body_sha256": sha(request_body),
            "body_size": len(request_body),
            "body_xml": request_body.decode() if operation == "CompleteMultipartUpload" else None,
            "prepared_wire_status": status,
        })
        return AWSResponse(request.url, status, headers, RawResponse(body))


prepared = PreparedEvents()
client = botocore.session.Session().create_client(
    "s3", region_name="eu-west-1", endpoint_url="https://s3.invalid",
    aws_access_key_id="fixture-not-a-key", aws_secret_access_key="fixture-not-a-secret",
    config=Config(signature_version=UNSIGNED, retries={"total_max_attempts": 1, "mode": "standard"},
                  request_checksum_calculation="when_required",
                  response_checksum_validation="when_required",
                  s3={"addressing_style": "path"}),
)
for operation in ["CreateMultipartUpload", "UploadPart", "ListParts", "CompleteMultipartUpload", "HeadObject"]:
    def event(request, _operation=operation, **kwargs):
        return prepared.reply(request, _operation, **kwargs)
    client.meta.events.register("before-send.s3." + operation, event)


def list_xml(items, truncated=False, next_marker=None):
    root = ET.Element("ListPartsResult", xmlns="http://s3.amazonaws.com/doc/2006-03-01/")
    for tag, value in [("Bucket", BUCKET), ("Key", KEY), ("UploadId", UPLOAD),
                       ("IsTruncated", str(truncated).lower())]:
        ET.SubElement(root, tag).text = value
    if next_marker is not None:
        ET.SubElement(root, "NextPartNumberMarker").text = str(next_marker)
    for number, size, etag in items:
        part = ET.SubElement(root, "Part")
        for tag, value in [("PartNumber", number), ("Size", size), ("ETag", etag)]:
            ET.SubElement(part, tag).text = str(value)
    return ET.tostring(root)


def collect_parts():
    found, seen, marker = [], set(), None
    while True:
        args = {"Bucket": BUCKET, "Key": KEY, "UploadId": UPLOAD, "MaxParts": 1000}
        if marker is not None:
            args["PartNumberMarker"] = marker
        result = client.list_parts(**args)
        found.extend(result.get("Parts", []))
        if not result["IsTruncated"]:
            return found
        next_marker = result.get("NextPartNumberMarker")
        if not isinstance(next_marker, int) or next_marker in seen or (marker is not None and next_marker <= marker):
            raise ValueError("No advancing pagination marker")
        seen.add(next_marker)
        marker = next_marker


def completion_plan(scope, revision, source, receipts, inventory):
    if scope != SCOPE or revision != REVISION or sha(source) != sha(SOURCE):
        raise ValueError("Source or upload identity changed")
    ordered = sorted(receipts, key=lambda r: r.number)
    if [r.number for r in ordered] != [1, 2]:
        raise ValueError("Producer receipt set is incomplete or duplicated")
    expected = {r.number: (r.size, r.etag) for r in ordered}
    remote = {p["PartNumber"]: (p["Size"], p["ETag"]) for p in inventory}
    if len(remote) != len(inventory) or remote != expected:
        raise ValueError("Inventory does not match producer receipts")
    offset = 0
    for receipt in ordered:
        if sha(source[offset:offset + receipt.size]) != receipt.slice_sha256:
            raise ValueError("Source slice changed")
        offset += receipt.size
    if offset != len(source):
        raise ValueError("Receipts do not cover the source")
    return [{"PartNumber": r.number, "ETag": r.etag} for r in ordered]


def expect_local_rejection(label, action):
    before = len(prepared.calls)
    try:
        action()
    except ValueError:
        check(label, True)
    else:
        raise AssertionError(label)
    check(label + " sends no completion", len(prepared.calls) == before)


def complete(plan):
    return client.complete_multipart_upload(Bucket=BUCKET, Key=KEY, UploadId=UPLOAD,
                                          MultipartUpload={"Parts": plan}, IfNoneMatch="*")


prepared.add("CreateMultipartUpload", body=(
    '<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
    f'<Bucket>{BUCKET}</Bucket><Key>{KEY}</Key><UploadId>{UPLOAD}</UploadId>'
    '</InitiateMultipartUploadResult>').encode())
created = client.create_multipart_upload(Bucket=BUCKET, Key=KEY, ContentType="text/csv")
check("upload identity decoded", created["UploadId"] == UPLOAD)
check("first part satisfies general-purpose minimum fixture size", len(PARTS[0]) == 5 * 1024 * 1024)
check("last part is nonempty", 0 < len(PARTS[1]) < len(PARTS[0]))
receipts = []
for number, raw in enumerate(PARTS, 1):
    prepared.add("UploadPart", headers={"ETag": ETAGS[number - 1]})
    result = client.upload_part(Bucket=BUCKET, Key=KEY, UploadId=UPLOAD,
                                PartNumber=number, Body=raw, ContentLength=len(raw))
    receipts.append(Receipt(number, len(raw), result["ETag"], sha(raw)))
    call = prepared.calls[-1]
    check("source bytes on part " + str(number), call["body_sha256"] == sha(raw))
    check("part number on wire " + str(number), parse_qs(urlsplit(call["url"]).query)["partNumber"] == [str(number)])
    check("upload identity on wire " + str(number), parse_qs(urlsplit(call["url"]).query)["uploadId"] == [UPLOAD])
    check("opaque ETag retained " + str(number), receipts[-1].etag == ETAGS[number - 1])

prepared.add("ListParts", body=list_xml([(1, len(PARTS[0]), ETAGS[0])], True, 1))
prepared.add("ListParts", body=list_xml([(2, len(PARTS[1]), ETAGS[1])]))
inventory = collect_parts()
check("both pages decoded", len(inventory) == 2 and isinstance(inventory[0]["Size"], int))
check("second list request carries native marker", parse_qs(urlsplit(prepared.calls[-1]["url"]).query)["part-number-marker"] == ["1"])
plan = completion_plan(SCOPE, REVISION, SOURCE, list(reversed(receipts)), inventory)
check("producer receipts sorted by part number", [p["PartNumber"] for p in plan] == [1, 2])
for label, scope, revision, source in [
    ("different key", (BUCKET, "exports/other.csv", UPLOAD), REVISION, SOURCE),
    ("different upload", (BUCKET, KEY, "another-upload"), REVISION, SOURCE),
    ("different revision", SCOPE, "extract-fixture-5", SOURCE),
    ("same-length changed source", SCOPE, REVISION, SOURCE[:-1] + b"X"),
]:
    expect_local_rejection(label, lambda: completion_plan(scope, revision, source, receipts, inventory))
expect_local_rejection("missing producer receipt", lambda: completion_plan(SCOPE, REVISION, SOURCE, receipts[:1], inventory))

prepared.add("CompleteMultipartUpload", body=b"<Error><Code>InternalError</Code><Message>Prepared completion failure</Message><RequestId>fixture-309</RequestId></Error>")
try:
    complete(plan)
except ClientError as error:
    check("embedded error retains native code", error.response["Error"]["Code"] == "InternalError")
    check("pinned S3 handler reclassifies error status", error.response["ResponseMetadata"]["HTTPStatusCode"] == 500)
    check("no implicit retry", error.response["ResponseMetadata"]["RetryAttempts"] == 0)
else:
    raise AssertionError("HTTP 200 Error must not become a completion success")
check("prepared wire response was HTTP 200", prepared.calls[-1]["prepared_wire_status"] == 200)
check("ERP remains unresolved after completion error", ERP == {"state": "awaiting_document", "document": None})
completion_call = prepared.calls[-1]
root = ET.fromstring(completion_call["body_xml"])
submitted = [{child.tag.rsplit("}", 1)[-1]: child.text for child in part} for part in root]
check("native XML retains producer part order", [p["PartNumber"] for p in submitted] == ["1", "2"])
check("native XML retains exact opaque ETags", [p["ETag"] for p in submitted] == ETAGS)
check("conditional creation header emitted", completion_call["headers"]["if-none-match"] == "*")
check("completion targets the original upload", parse_qs(urlsplit(completion_call["url"]).query)["uploadId"] == [UPLOAD])

# A separately supplied success response is not evidence that AWS accepted a
# retry after the error above. The witness does not operate a storage server.
prepared.add("CompleteMultipartUpload", body=(
    '<CompleteMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
    f'<Location>https://s3.invalid/{BUCKET}/{KEY}</Location><Bucket>{BUCKET}</Bucket>'
    f'<Key>{KEY}</Key><ETag>"opaque-complete-two-parts"</ETag>'
    '</CompleteMultipartUploadResult>').encode())
accepted = complete(plan)
check("prepared success decoded", accepted["ETag"] == '"opaque-complete-two-parts"')
prepared.add("HeadObject", headers={"Content-Length": str(len(SOURCE)), "ETag": accepted["ETag"]})
head = client.head_object(Bucket=BUCKET, Key=KEY)
check("prepared object length decoded", head["ContentLength"] == len(SOURCE))
check("prepared object ETag decoded", head["ETag"] == accepted["ETag"])
check("opaque object ETag is not a claimed source digest", head["ETag"].strip('"') != hashlib.md5(SOURCE).hexdigest())
check("technical success does not write ERP", ERP["document"] is None)

for status, code in [(400, "InvalidPart"), (409, "ConditionalRequestConflict"), (412, "PreconditionFailed")]:
    prepared.add("CompleteMultipartUpload", status, f"<Error><Code>{code}</Code><Message>Prepared refusal</Message></Error>".encode())
    try:
        complete(plan)
    except ClientError as error:
        check("native refusal " + code, error.response["Error"]["Code"] == code)
        check("native refusal status " + code, error.response["ResponseMetadata"]["HTTPStatusCode"] == status)
    else:
        raise AssertionError(code)

bad_inventories = [
    ("missing remote part", [(1, len(PARTS[0]), ETAGS[0])]),
    ("foreign remote part", [(1, len(PARTS[0]), ETAGS[0]), (2, len(PARTS[1]), ETAGS[1]), (3, 9, '"foreign"')]),
    ("changed remote ETag", [(1, len(PARTS[0]), '"replacement"'), (2, len(PARTS[1]), ETAGS[1])]),
]
for label, items in bad_inventories:
    prepared.add("ListParts", body=list_xml(items))
    observed = collect_parts()
    expect_local_rejection(label, lambda: completion_plan(SCOPE, REVISION, SOURCE, receipts, observed))
prepared.add("ListParts", body=list_xml([], True))
try:
    collect_parts()
except ValueError:
    check("truncated page without marker rejected", True)
else:
    raise AssertionError("Missing pagination marker")

# Metadata checks do not observe stored bytes. This equal-length substitution
# leaves the prepared inventory untouched and is deliberately not certified.
changed_remote_bytes = bytearray(PARTS[0])
changed_remote_bytes[-1] ^= 1
check("same-size content counterexample", len(changed_remote_bytes) == receipts[0].size and sha(changed_remote_bytes) != receipts[0].slice_sha256)
check("inventory-only admission cannot prove remote bytes", completion_plan(SCOPE, REVISION, SOURCE, receipts, inventory) == plan)
check("all prepared replies consumed", not prepared.responses)
check("no real HTTP session invoked", not REAL_HTTP)
check("no socket connection attempted", not CONNECTIONS)
check("ERP document stays unpublished in all cases", ERP == {"state": "awaiting_document", "document": None})

print(json.dumps({
    "python": platform.python_version(), "botocore": version("botocore"),
    "checks_passed": len(CHECKS), "checks": CHECKS,
    "prepared_before_send_invocations": len(prepared.calls),
    "real_http_requests": len(REAL_HTTP), "connection_attempts": len(CONNECTIONS),
    "source_bytes": len(SOURCE), "source_sha256": sha(SOURCE),
    "part_sizes": [len(p) for p in PARTS], "producer_part_order": plan,
    "requests": prepared.calls, "erp_state": ERP,
    "limits": ["Prepared replies do not certify AWS authorization, storage, checksums or concurrency.",
               "Source and receipt state are in memory; no durable crash-recovery or ERP acceptance test.",
               "General-purpose bucket fixture; no directory-bucket claim.",
               "Automatic optional checksum calculation is disabled; source SHA-256 is a local binding only."]
}, ensure_ascii=False, indent=2))