#!/usr/bin/env python3
"""Inspect Google SDK requests with explicitly prepared replies, without a network.

The byte sequence is synthetic, not a validated PDF. File IDs, metadata and
session URLs are fictional. This exercises the installed client's transitions;
it neither implements a Drive server nor creates or attaches a remote file.
Install google-api-python-client==2.201.0 separately before running this file.
"""

from __future__ import annotations

import hashlib
import importlib.metadata
import io
import json
import socket
from urllib.parse import parse_qs, urlsplit


SDK_VERSION = "2.201.0"
DISCOVERY_SHA256 = "9a15f3b0e37c5f1a99ee517f5e2372c4e12476207e4959dc6c36d083227d46ef"
TOTAL = 786432
CHUNK = 524288
ACKNOWLEDGED = 262144
FILE_A = "identifiant-reserve-fictif-A"
FILE_B = "identifiant-reserve-fictif-B"
PARENT = "dossier-fictif-ADV"
NAME = "BL-4182-r1.pdf"
FIELDS = "id,name,mimeType,parents,size,md5Checksum"
SESSION_A = "https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable&upload_id=SESSION_FICTIVE_A"
seed = b"OCTETS SYNTHETIQUES BL-4182 REVISION R1 - PAS UN PDF VALIDE\n"
PAYLOAD = (seed * (TOTAL // len(seed) + 1))[:TOTAL]


def digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def metadata(file_id: str = FILE_A) -> dict:
    return {
        "id": file_id,
        "name": NAME,
        "mimeType": "application/pdf",
        "parents": [PARENT],
        "size": str(TOTAL),
        "md5Checksum": hashlib.md5(PAYLOAD, usedforsecurity=False).hexdigest(),
    }


class Checks:
    def __init__(self) -> None:
        self.labels: list[str] = []

    def check(self, label: str, condition: bool) -> None:
        if not condition:
            raise AssertionError(label)
        self.labels.append(label)


checks = Checks()
network = {"connection_attempts": 0, "real_http_requests": 0}


def forbidden_connection(*args, **kwargs):
    network["connection_attempts"] += 1
    raise RuntimeError("Network connection forbidden in this offline exercise")


def forbidden_http(*args, **kwargs):
    network["real_http_requests"] += 1
    raise RuntimeError("Real httplib2 transport forbidden in this exercise")


class PreparedTransport:
    """Consume a script of replies; never forward a request to an HTTP client."""

    def __init__(self, steps: list[dict]) -> None:
        self.steps = list(steps)
        self.requests: list[dict] = []

    def request(self, uri, method="GET", body=None, headers=None, **kwargs):
        if not self.steps:
            raise RuntimeError("Unexpected SDK transport invocation")
        step = self.steps.pop(0)
        if method != step["method"] or urlsplit(uri).path != step["path"]:
            raise RuntimeError("SDK request differs from the prepared scenario")
        normalized = {key.lower(): value for key, value in (headers or {}).items()}
        if hasattr(body, "read"):
            data = body.read()
        elif isinstance(body, str):
            data = body.encode("utf-8")
        else:
            data = body or b""
        parsed_body = None
        if normalized.get("content-type", "").startswith("application/json") and data:
            parsed_body = json.loads(data)
        self.requests.append({
            "method": method,
            "path": urlsplit(uri).path,
            "query": parse_qs(urlsplit(uri).query),
            "session": "A" if uri == SESSION_A else None,
            "headers": normalized,
            "body_length": len(data),
            "body_sha256": digest(data),
            "metadata_body": parsed_body,
        })
        if step.get("timeout_after_read"):
            # Bytes have been consumed by this local transport, not by Drive.
            raise TimeoutError("Prepared loss of a response, no network involved")
        response = httplib2.Response({
            "status": str(step["status"]),
            "content-type": "application/json",
            **step.get("headers", {}),
        })
        content = step.get("body")
        return response, b"" if content is None else json.dumps(content).encode("utf-8")

    def exhausted(self) -> bool:
        return not self.steps


UPLOAD_PATH = "/upload/drive/v3/files"


def initiation(session=SESSION_A) -> dict:
    return {"method": "POST", "path": UPLOAD_PATH, "status": 200,
            "headers": {"location": session}}


def data_reply(status: int, *, headers=None, body=None) -> dict:
    return {"method": "PUT", "path": UPLOAD_PATH, "status": status,
            "headers": headers or {}, "body": body}


def lost_reply() -> dict:
    return {"method": "PUT", "path": UPLOAD_PATH, "timeout_after_read": True}


def client(http):
    return build_from_document(discovery_document, http=http)


def upload(http, file_id=FILE_A):
    return client(http).files().create(
        body={"id": file_id, "name": NAME, "mimeType": "application/pdf", "parents": [PARENT]},
        media_body=MediaIoBaseUpload(io.BytesIO(PAYLOAD), mimetype="application/pdf",
                                     chunksize=CHUNK, resumable=True),
        fields=FIELDS,
    )


def expect_timeout(request, label: str) -> None:
    try:
        request.next_chunk(num_retries=0)
    except TimeoutError:
        checks.check(label, True)
    else:
        checks.check(label, False)


def is_status_probe(record: dict) -> bool:
    return (record["method"] == "PUT" and record["body_length"] == 0
            and record["headers"].get("content-range") == f"bytes */{TOTAL}"
            and record["headers"].get("content-length") == "0")


def metadata_differences(observed: dict) -> list[str]:
    """Compare five supplied fields; names, permissions and ERP acceptance differ."""
    expected = metadata()
    return [key for key in ("id", "mimeType", "parents", "size", "md5Checksum")
            if observed.get(key) != expected[key]]


def main() -> dict:
    checks.check("Pinned Google client version", importlib.metadata.version("google-api-python-client") == SDK_VERSION)
    checks.check("Pinned bundled Drive discovery", digest(discovery_document.encode()) == DISCOVERY_SHA256)
    checks.check("Pinned Drive discovery revision", json.loads(discovery_document)["revision"] == "20260916")
    checks.check("Synthetic source byte count", len(PAYLOAD) == TOTAL)

    reserve = PreparedTransport([{
        "method": "GET", "path": "/drive/v3/files/generateIds", "status": 200,
        "body": {"kind": "drive#generatedIds", "ids": [FILE_A], "space": "drive"},
    }])
    reserved = client(reserve).files().generateIds(count=1, space="drive", type="files").execute(num_retries=0)
    checks.check("SDK parses supplied fictional ID", reserved["ids"] == [FILE_A])
    checks.check("Reservation query", reserve.requests[0]["query"] == {"count": ["1"], "space": ["drive"], "type": ["files"], "alt": ["json"]})
    checks.check("Reservation has no request body", reserve.requests[0]["body_length"] == 0)
    checks.check("Reservation script exhausted", reserve.exhausted())

    # The first prepared acknowledgment confirms only half the attempted chunk.
    partial = PreparedTransport([
        initiation(),
        data_reply(308, headers={"range": f"bytes=0-{ACKNOWLEDGED - 1}"}),
        data_reply(200, body=metadata()),
    ])
    request = upload(partial, reserved["ids"][0])
    checks.check("Resumable creation is POST", request.method == "POST")
    checks.check("Resumable URI options", parse_qs(urlsplit(request.uri).query) == {"fields": [FIELDS], "alt": ["json"], "uploadType": ["resumable"]})
    checks.check("Reserved ID is in actual SDK metadata body", json.loads(request.body)["id"] == FILE_A)
    checks.check("Binary MIME, without Workspace conversion", json.loads(request.body)["mimeType"] == "application/pdf")
    status, result = request.next_chunk(num_retries=0)
    checks.check("Initial200 and first308 do not return a completed File", result is None)
    checks.check("Progress comes from acknowledged bytes", status.resumable_progress == ACKNOWLEDGED)
    checks.check("SDK attempted a larger chunk", partial.requests[1]["body_length"] == CHUNK)
    checks.check("First data range", partial.requests[1]["headers"]["content-range"] == f"bytes 0-{CHUNK - 1}/{TOTAL}")
    checks.check("Initial request declares full data length", partial.requests[0]["headers"]["x-upload-content-length"] == str(TOTAL))
    status, result = request.next_chunk(num_retries=0)
    checks.check("SDK starts next chunk at acknowledged offset", partial.requests[2]["headers"]["content-range"] == f"bytes {ACKNOWLEDGED}-{TOTAL - 1}/{TOTAL}")
    checks.check("Second chunk reads that exact source slice", partial.requests[2]["body_sha256"] == digest(PAYLOAD[ACKNOWLEDGED:]))
    checks.check("Both chunks use the same session", [x["session"] for x in partial.requests[1:]] == ["A", "A"])
    checks.check("Prepared final File is parsed", status is None and result == metadata())
    checks.check("Partial acknowledgment script exhausted", partial.exhausted())

    # A lost response does not tell this offline transport what a real server stored.
    completed = PreparedTransport([
        initiation(), data_reply(308, headers={"range": f"bytes=0-{CHUNK - 1}"}),
        lost_reply(), data_reply(200, body=metadata()),
    ])
    request = upload(completed)
    status, result = request.next_chunk(num_retries=0)
    checks.check("Completed-case first chunk is acknowledged", result is None and status.resumable_progress == CHUNK)
    expect_timeout(request, "Prepared final-response loss is observed")
    checks.check("Final source chunk was actually supplied before timeout", completed.requests[2]["headers"]["content-range"] == f"bytes {CHUNK}-{TOTAL - 1}/{TOTAL}" and completed.requests[2]["body_sha256"] == digest(PAYLOAD[CHUNK:]))
    status, result = request.next_chunk(num_retries=0)
    checks.check("Next call probes the uncertain session", is_status_probe(completed.requests[3]))
    checks.check("Prepared completed probe returns File", status is None and result == metadata())
    checks.check("No further data PUT after completed probe", len(completed.requests) == 4 and completed.exhausted())

    incomplete = PreparedTransport([
        initiation(), lost_reply(),
        data_reply(308, headers={"range": f"bytes=0-{ACKNOWLEDGED - 1}"}),
        data_reply(201, body=metadata()),
    ])
    request = upload(incomplete)
    expect_timeout(request, "Prepared incomplete-response loss is observed")
    status, result = request.next_chunk(num_retries=0)
    checks.check("Incomplete case probes before sending data", is_status_probe(incomplete.requests[2]))
    checks.check("One next_chunk can probe then send", len(incomplete.requests) == 4)
    checks.check("Probe acknowledgment controls subsequent range", incomplete.requests[3]["headers"]["content-range"] == f"bytes {ACKNOWLEDGED}-{TOTAL - 1}/{TOTAL}")
    checks.check("Resumed source slice is unchanged", incomplete.requests[3]["body_sha256"] == digest(PAYLOAD[ACKNOWLEDGED:]))
    checks.check("Prepared201 returns parsed metadata", status is None and result == metadata() and incomplete.exhausted())

    empty_ack = PreparedTransport([
        initiation(), lost_reply(), data_reply(308),
        data_reply(308, headers={"range": f"bytes=0-{CHUNK - 1}"}),
        data_reply(200, body=metadata()),
    ])
    request = upload(empty_ack)
    expect_timeout(request, "Prepared no-Range response loss is observed")
    status, result = request.next_chunk(num_retries=0)
    checks.check("No-Range case first probes", is_status_probe(empty_ack.requests[2]))
    checks.check("No-Range probe restarts from byte0", empty_ack.requests[3]["headers"]["content-range"] == f"bytes 0-{CHUNK - 1}/{TOTAL}")
    checks.check("No-Range retry resends the first source slice", empty_ack.requests[3]["body_sha256"] == digest(PAYLOAD[:CHUNK]))
    checks.check("Progress then follows the new supplied acknowledgment", status.resumable_progress == CHUNK and result is None)
    status, result = request.next_chunk(num_retries=0)
    checks.check("Remaining bytes after new acknowledgment", empty_ack.requests[4]["headers"]["content-range"] == f"bytes {CHUNK}-{TOTAL - 1}/{TOTAL}")
    checks.check("No-Range script reaches its prepared final result", result == metadata() and empty_ack.exhausted())

    expired = PreparedTransport([
        initiation(), lost_reply(),
        data_reply(404, body={"error": {"code": 404, "message": "Session fictive expirée"}}),
    ])
    request = upload(expired)
    expect_timeout(request, "Prepared expired-session response loss is observed")
    try:
        request.next_chunk(num_retries=0)
    except HttpError as error:
        checks.check("Expired-session status surfaces as HttpError404", error.resp.status == 404)
    else:
        checks.check("Expired-session status surfaces as HttpError404", False)
    checks.check("Expired session was probed, not silently replaced", is_status_probe(expired.requests[2]))
    checks.check("SDK did not generate a new file identity after404", len(expired.requests) == 3 and expired.exhausted())

    conflict = PreparedTransport([{
        "method": "POST", "path": UPLOAD_PATH, "status": 409,
        "body": {"error": {"code": 409, "message": "Conflit fictif de création"}},
    }])
    request = upload(conflict)
    try:
        request.next_chunk(num_retries=0)
    except ResumableUploadError as error:
        checks.check("Initial409 remains a resumable creation exception", error.resp.status == 409)
    else:
        checks.check("Initial409 remains a resumable creation exception", False)
    checks.check("New creation request preserves original reserved ID", conflict.requests[0]["metadata_body"]["id"] == FILE_A)
    checks.check("Conflict produced no session or data PUT", request.resumable_uri is None and len(conflict.requests) == 1 and conflict.exhausted())

    unused = PreparedTransport([])
    wrong_new_request = upload(unused, FILE_B)
    checks.check("Counterexample asks for a different file ID", json.loads(wrong_new_request.body)["id"] != FILE_A)
    checks.check("Counterexample keeps exactly the same document name", json.loads(wrong_new_request.body)["name"] == NAME)
    checks.check("Counterexample was constructed but never sent", unused.requests == [])

    replies = [metadata(), {**metadata(), "parents": ["autre-dossier-fictif"]},
               {key: value for key, value in metadata().items() if key != "md5Checksum"}]
    inspect_file = PreparedTransport([
        {"method": "GET", "path": "/drive/v3/files/" + FILE_A,
         "status": 200, "body": reply} for reply in replies
    ])
    service = client(inspect_file)
    received = service.files().get(fileId=FILE_A, fields=FIELDS).execute(num_retries=0)
    checks.check("Reconciliation requests the reserved ID", inspect_file.requests[0]["path"].endswith("/" + FILE_A))
    checks.check("Reconciliation requests explicit File fields", inspect_file.requests[0]["query"] == {"fields": [FIELDS], "alt": ["json"]})
    checks.check("SDK preserves File.size as a JSON string", received["size"] == str(TOTAL) and isinstance(received["size"], str))
    checks.check("Prepared binary checksum matches synthetic content", received["md5Checksum"] == hashlib.md5(PAYLOAD, usedforsecurity=False).hexdigest())
    checks.check("Prepared parent and MIME stay observable", received["parents"] == [PARENT] and received["mimeType"] == "application/pdf")
    checks.check("Five supplied fields match this local expectation", metadata_differences(received) == [])
    wrong_parent = service.files().get(fileId=FILE_A, fields=FIELDS).execute(num_retries=0)
    missing_checksum = service.files().get(fileId=FILE_A, fields=FIELDS).execute(num_retries=0)
    checks.check("Different supplied parent is reported after SDK parsing", metadata_differences(wrong_parent) == ["parents"])
    checks.check("Omitted checksum is reported after SDK parsing", metadata_differences(missing_checksum) == ["md5Checksum"])
    checks.check("Reconciliation reply script exhausted", inspect_file.exhausted())

    transports = {"reservation": reserve, "partial_ack": partial,
                  "lost_reply_completed": completed, "lost_reply_incomplete": incomplete,
                  "probe_without_range": empty_ack, "expired_session": expired,
                  "creation_conflict": conflict, "reconciliation": inspect_file}
    records = [record for transport in transports.values() for record in transport.requests]
    checks.check("All scripted requests omit credentials", all(not ({"authorization", "cookie", "proxy-authorization"} & set(r["headers"])) for r in records))
    checks.check("No real HTTP request observed", network["real_http_requests"] == 0)
    checks.check("No network connection attempt observed", network["connection_attempts"] == 0)
    return {
        "checks_passed": len(checks.labels),
        "check_labels": checks.labels,
        "sdk_version": SDK_VERSION,
        "drive_discovery_revision": "20260916",
        "drive_discovery_sha256": DISCOVERY_SHA256,
        "source_bytes": TOTAL,
        "source_sha256": digest(PAYLOAD),
        "first_attempted_chunk_bytes": CHUNK,
        "first_acknowledged_bytes": ACKNOWLEDGED,
        "next_zero_based_offset": ACKNOWLEDGED,
        "prepared_transport_invocations": len(records),
        "constructed_but_unsent_different_id_requests": 1,
        "supplied_metadata_differences": [metadata_differences(x) for x in [received, wrong_parent, missing_checksum]],
        "network_observations": network,
        "traces": {name: transport.requests for name, transport in transports.items()},
        "limits": [
            "Explicitly prepared replies, never captured Drive responses or a server emulator.",
            "Synthetic bytes declared application/pdf, without PDF validation.",
            "No OAuth account, Drive file, native permission, sharing or ERP attachment tested.",
            "Durable reservation transactions, concurrent workers and process restarts are not exercised.",
            "The documented native pre-generated-ID retry guarantee is not proven by these fake responses.",
            "MD5 comparison is not an authenticity, malware or business-acceptance test.",
        ],
    }


if __name__ == "__main__":
    originals = (socket.socket.connect, socket.socket.connect_ex, socket.create_connection)
    socket.socket.connect = forbidden_connection
    socket.socket.connect_ex = forbidden_connection
    socket.create_connection = forbidden_connection
    original_http = None
    try:
        import httplib2
        original_http = httplib2.Http.request
        httplib2.Http.request = forbidden_http
        from googleapiclient.discovery import build_from_document
        from googleapiclient.discovery_cache import get_static_doc
        from googleapiclient.errors import HttpError, ResumableUploadError
        from googleapiclient.http import MediaIoBaseUpload
        discovery_document = get_static_doc("drive", "v3")
        print(json.dumps(main(), ensure_ascii=False, indent=2))
    finally:
        socket.socket.connect, socket.socket.connect_ex, socket.create_connection = originals
        if original_http is not None:
            httplib2.Http.request = original_http