#!/usr/bin/env python3
"""
Local end-to-end test of the Amazon integration, driving the real service.

Boots the actual FastAPI app (fresh process, scratch MySQL database,
EGRESS_MODE=api / INGRESS_MODE=api so no SFTP or PGP is touched) against
three local stubs:

  - 1Rx API        (POST /orders -> 202, GET .../download-jpeg-prescription)
  - Amazon Cognito (POST /oauth2/token -> access token)
  - Amazon Shuttle (POST /v1/consultation/response -> validates payload,
                    downloads the pre-signed prescription ZIP from REAL S3,
                    answers ACKNOWLEDGED or a scripted ERRORED)

Scenarios:
  1. Creation ingress -> 1Rx -> APPROVED webhook -> Shuttle ACKNOWLEDGED,
     prescription ZIP staged in S3 and downloadable by "Amazon"
  2. FAILED consultation (no prescription, failureReason mapped)
  3. Shuttle answers ERRORED -> recorded, then recovered via the
     management resend endpoint with a new trackingId
  4. Duplicate creation request -> acknowledged without a second 1Rx call
  5. Real order (isTestData=false) with EGRESS_API_TEST_ORDERS_ONLY=true
     -> never reaches Shuttle; the PGP-encrypted response ZIP is uploaded
     to an in-process SFTP server, decrypted and inspected

Usage:  python scripts/e2e_consume_test.py
Requires: local MySQL (root/root), AWS credentials for onerx-test-bucket,
          a gpg binary (scenario 5 generates a throwaway keypair).
"""

import base64
import hashlib
import json
import os
import signal
import socket
import subprocess
import sys
import threading
import time
import urllib.request
import tempfile
import zipfile
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from io import BytesIO
from pathlib import Path
from urllib.parse import quote

import gnupg
import mysql.connector
import paramiko
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

REPO = Path(__file__).resolve().parents[1]
APP_PORT = 8112
RX_PORT = 9100
COGNITO_PORT = 9101
SHUTTLE_PORT = 9102
SFTP_PORT = 9103
SFTP_OUT_USER = "e2e-outgoing-user"

DB = dict(host="localhost", port=3306, user="root", password="root", database="rx_onerx_e2e")
S3_BUCKET = "onerx-test-bucket"
S3_PREFIX = "e2e-consume-test"
RX_API_KEY = "amazon-6697f65c-8a8f-482e-bd27-7f4e075609d7"

# Must match token_auth.py defaults (the token is validated by decrypting
# vendorUid, not by signature).
TOKEN_SECRET_KEY = "aw345-der856-45ty675eu-&8%$#-87967-#@!^7!"
TOKEN_SALT = "kte45%623@e#{}&&#$%67"

PRIMARY = "404-7777777#e2e42"
JPEG_BYTES = b"\xff\xd8\xff\xe0" + b"fake-jpeg-content-for-e2e" * 10

PASS, FAIL = [], []


def check(name, condition, detail=""):
    (PASS if condition else FAIL).append(name)
    print(f"  {'PASS' if condition else 'FAIL'}  {name}" + (f"  [{detail}]" if detail and not condition else ""))


# --------------------------------------------------------------------------
# Partner JWT forging (mirrors onerx-service token issuance)
# --------------------------------------------------------------------------

def encrypt_vendor_uid(api_key: str) -> str:
    key = hashlib.pbkdf2_hmac("sha256", TOKEN_SECRET_KEY.encode(), TOKEN_SALT.encode(), 65536, dklen=32)
    padder = padding.PKCS7(128).padder()
    data = padder.update(api_key.encode()) + padder.finalize()
    enc = Cipher(algorithms.AES(key), modes.CBC(bytes(16))).encryptor()
    return base64.b64encode(enc.update(data) + enc.finalize()).decode()


def forge_amazon_token() -> str:
    def b64url(obj):
        return base64.urlsafe_b64encode(json.dumps(obj).encode()).rstrip(b"=").decode()
    header = b64url({"alg": "HS256", "typ": "JWT"})
    payload = b64url({
        "exp": int(time.time()) + 3600,
        "vendorId": "amazon-e2e",
        "vendorUid": encrypt_vendor_uid(RX_API_KEY),
    })
    return f"{header}.{payload}.e2e-unverified-signature"


# --------------------------------------------------------------------------
# Stub servers
# --------------------------------------------------------------------------

STATE = {
    "rx_orders": [],           # payloads 1Rx received
    "shuttle_requests": [],    # payloads Shuttle received
    "shuttle_zip_checks": [],  # (downloaded_ok, zip_names)
    "shuttle_errored_next": 0, # answer ERRORED for the next N calls
    "cognito_tokens": 0,
    "sftp_logins": [],         # usernames that authenticated to the SFTP stub
}


class Stub1Rx(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        STATE["rx_orders"].append(body)
        out = json.dumps({"data": {"orderId": f"order-{len(STATE['rx_orders'])}"}}).encode()
        self.send_response(202)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(out)

    def do_GET(self):
        if "download-jpeg-prescription" in self.path:
            self.send_response(200)
            self.send_header("Content-Type", "image/jpeg")
            self.end_headers()
            self.wfile.write(JPEG_BYTES)
        else:
            self.send_response(404)
            self.end_headers()


class StubCognito(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        self.rfile.read(int(self.headers.get("Content-Length", 0)))
        STATE["cognito_tokens"] += 1
        out = json.dumps({
            "access_token": f"e2e-shuttle-token-{STATE['cognito_tokens']}",
            "token_type": "Bearer",
            "expires_in": 3600,
        }).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(out)


class StubShuttle(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        payload["_authorization"] = self.headers.get("Authorization")
        STATE["shuttle_requests"].append(payload)

        ids = payload.get("messageIdentifiers", {})
        # "Amazon" downloads every fileUrl it is given, like the real thing.
        for rx in (payload.get("data", {}).get("rxDetailFromConsultation") or []):
            for image in rx.get("rxImageDetails", []):
                try:
                    with urllib.request.urlopen(image["fileUrl"], timeout=20) as resp:
                        blob = resp.read()
                    names = zipfile.ZipFile(BytesIO(blob)).namelist()
                    STATE["shuttle_zip_checks"].append((True, names))
                except Exception as e:
                    STATE["shuttle_zip_checks"].append((False, str(e)))

        if STATE["shuttle_errored_next"] > 0:
            STATE["shuttle_errored_next"] -= 1
            body = {
                "messageIdentifiers": ids,
                "status": "ERRORED",
                "acknowledgementTimeStamp": int(time.time() * 1000),
                "errorMessage": "e2e scripted rejection",
            }
        else:
            body = {
                "messageIdentifiers": ids,
                "status": "ACKNOWLEDGED",
                "acknowledgementTimeStamp": int(time.time() * 1000),
            }
        out = json.dumps(body).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(out)


def start_stub(handler, port):
    server = ThreadingHTTPServer(("127.0.0.1", port), handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


class StubSSHServer(paramiko.ServerInterface):
    """Accepts any public key; records who logged in."""

    def check_auth_publickey(self, username, key):
        STATE["sftp_logins"].append(username)
        return paramiko.AUTH_SUCCESSFUL

    def get_allowed_auths(self, username):
        return "publickey"

    def check_channel_request(self, kind, chanid):
        return paramiko.OPEN_SUCCEEDED


class StubSFTPServer(paramiko.SFTPServerInterface):
    """Just enough SFTP for ``sftp.put``: open-for-write + stat, rooted at ROOT."""

    ROOT = None

    def _real(self, path):
        return os.path.join(self.ROOT, path.lstrip("/"))

    def open(self, path, flags, attr):
        real = self._real(path)
        os.makedirs(os.path.dirname(real), exist_ok=True)
        fd = os.open(real, flags | os.O_CREAT, 0o644)
        mode = "wb" if flags & (os.O_WRONLY | os.O_RDWR) else "rb"
        f = os.fdopen(fd, mode)
        handle = paramiko.SFTPHandle(flags)
        handle.filename = real
        handle.readfile = f
        handle.writefile = f
        return handle

    def stat(self, path):
        try:
            return paramiko.SFTPAttributes.from_stat(os.stat(self._real(path)))
        except OSError as e:
            return paramiko.SFTPServer.convert_errno(e.errno)

    lstat = stat


def start_sftp_stub(port, root):
    StubSFTPServer.ROOT = root
    host_key = paramiko.RSAKey.generate(2048)
    sock = socket.socket()
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("127.0.0.1", port))
    sock.listen(5)

    def accept_loop():
        while True:
            conn, _ = sock.accept()
            transport = paramiko.Transport(conn)
            transport.add_server_key(host_key)
            transport.set_subsystem_handler("sftp", paramiko.SFTPServer, StubSFTPServer)
            transport.start_server(server=StubSSHServer())

    threading.Thread(target=accept_loop, daemon=True).start()


def make_local_keys(workdir):
    """Throwaway SSH client key + PGP keypair for the SFTP/PGP egress path.
    Returns (ssh_key_path, pgp_public_path, pgp_private_path, gnupg_home)."""
    ssh_key = os.path.join(workdir, "id_rsa_e2e")
    paramiko.RSAKey.generate(2048).write_private_key_file(ssh_key)

    gnupg_home = os.path.join(workdir, "gnupg")
    os.makedirs(gnupg_home, mode=0o700)
    gpg = gnupg.GPG(gnupghome=gnupg_home)
    key = gpg.gen_key(gpg.gen_key_input(
        key_type="RSA", key_length=2048, name_email="e2e@example.invalid", no_protection=True,
    ))
    if not key.fingerprint:
        raise RuntimeError(f"gpg key generation failed: {key.stderr}")
    public_path = os.path.join(workdir, "PGPPublicKey.asc")
    private_path = os.path.join(workdir, "pgp_private.asc")
    pathlib_write(public_path, gpg.export_keys(key.fingerprint))
    pathlib_write(private_path, gpg.export_keys(key.fingerprint, secret=True, expect_passphrase=False))
    return ssh_key, public_path, private_path, gnupg_home


def pathlib_write(path, text):
    with open(path, "w") as f:
        f.write(text)


def decrypt_sftp_zip(encrypted_path, gnupg_home):
    gpg = gnupg.GPG(gnupghome=gnupg_home)
    with open(encrypted_path, "rb") as f:
        result = gpg.decrypt_file(f)
    if not result.ok:
        raise RuntimeError(f"decrypt failed: {result.status} {result.stderr[-300:]}")
    return zipfile.ZipFile(BytesIO(result.data))


# --------------------------------------------------------------------------
# Driving the real service
# --------------------------------------------------------------------------

def http(method, url, body=None, headers=None, timeout=15):
    req = urllib.request.Request(url, method=method, headers=headers or {})
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, data=data, timeout=timeout) as resp:
            return resp.status, json.loads(resp.read() or b"{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or b"{}")
    except (urllib.error.URLError, OSError):
        return 0, {}


def creation_request(consultation_id, tracking_id, is_test_data=True):
    return {
        "messageIdentifiers": {
            "primaryReferenceId": PRIMARY,
            "secondaryReferenceId": consultation_id,
            "trackingId": tracking_id,
            "isTestData": is_test_data,
        },
        "data": {
            "requestMetadata": {
                "messageId": f"msg-{tracking_id}",
                "messageSentTime": int(time.time() * 1000),
                "messageType": "REVALIDATION_REQUEST",
            },
            "consultationDetails": {"sla": int(time.time() * 1000) + 3600_000, "priority": "P0"},
            "rxDetailForConsultation": [{
                "rxHeaderDetailsForConsultation": {
                    "rxId": "rx-e2e-1",
                    "patientContactDetails": {"name": "E2E Patient", "contactNumber": "+919876543210"},
                },
                "rxItemDetailForConsultation": [{
                    "rxItemId": "DOLO_PARACETAMOL#650.0#MILLIGRAMS###1_TABLETS",
                    "medicineName": "Dolo 650",
                    "orderedMedicine": "https://www.amazon.in/dp/B07X1ZM63M/",
                    "numberOfUnitsOrdered": 10,
                    "dosage": {"dose": 1, "form": "TABLETS", "frequency": "2",
                               "frequencyUnit": "DAILY", "duration": "5", "durationUnit": "DAYS"},
                }],
            }],
        },
    }


def rx_webhook(consultation_id, status="APPROVED"):
    if status != "APPROVED":
        return {"status": "CANCELLED", "pharmacyOrderId": PRIMARY, "shipmentCode": consultation_id,
                "orderCancelReason": "Call not answered"}
    med_uuid = None
    for order in STATE["rx_orders"]:
        if order.get("shipmentCode") == consultation_id:
            med_uuid = order["medicines"][0]["uuid"]
    return {
        "status": "APPROVED",
        "pharmacyOrderId": PRIMARY,
        "shipmentCode": consultation_id,
        "approveTime": int(time.time() * 1000),
        "orderedMedicines": [{"uuid": med_uuid, "url": "https://www.amazon.in/dp/B07X1ZM63M/"}],
        "approvedMedicines": [{
            "id": "DOLO_PARACETAMOL#650.0#MILLIGRAMS###1_TABLETS", "uuid": med_uuid,
            "name": "Dolo 650", "quantity": 10, "dosage": "1-0-1", "type": "TABLETS",
            "days": 5, "comments": "after food",
        }],
        "doctor": {"name": "Dr. E2E Sharma", "phone": "+91-9876500000",
                   "registrationNumber": "MCI-E2E-1", "address": "Bengaluru, KA"},
        "patient": {"name": "E2E Patient", "gender": "M", "age": 33},
    }


def db_row(consultation_id):
    conn = mysql.connector.connect(**DB)
    cur = conn.cursor(dictionary=True)
    cur.execute("SELECT * FROM connectivity_requests WHERE consultation_id=%s", (consultation_id,))
    row = cur.fetchone()
    conn.close()
    return row


def wait_for(predicate, timeout=20, interval=0.3):
    deadline = time.time() + timeout
    while time.time() < deadline:
        value = predicate()
        if value:
            return value
        time.sleep(interval)
    return None


def serve_forever():
    """Sandbox mode for manual (Postman/curl) testing: keep the service and
    stubs running until Ctrl+C. Same wiring as the automated scenarios."""
    print(f"""
Sandbox up — point Postman at it:
  base_url            http://127.0.0.1:{APP_PORT}
  partner token       python scripts/generate_partner_token.py
  1Rx stub            http://127.0.0.1:{RX_PORT}  (accepts any order, serves a fake JPEG)
  Cognito/Shuttle stub http://127.0.0.1:{COGNITO_PORT} / :{SHUTTLE_PORT}  (always ACKNOWLEDGED)
  Database            mysql {DB['database']}.connectivity_requests (root/root)
  Logs                tail -f logs-e2e/service.log

Ctrl+C to stop.""")
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        return 0


def main():
    serve = "--serve" in sys.argv
    for port in (APP_PORT, RX_PORT, COGNITO_PORT, SHUTTLE_PORT):
        with socket.socket() as s:
            if s.connect_ex(("127.0.0.1", port)) == 0:
                print(f"Port {port} already in use; aborting")
                return 1

    for port in (SFTP_PORT,):
        with socket.socket() as s:
            if s.connect_ex(("127.0.0.1", port)) == 0:
                print(f"Port {port} already in use; aborting")
                return 1

    start_stub(Stub1Rx, RX_PORT)
    start_stub(StubCognito, COGNITO_PORT)
    start_stub(StubShuttle, SHUTTLE_PORT)
    workdir = tempfile.mkdtemp(prefix="e2e-consume-")
    sftp_root = os.path.join(workdir, "sftp")
    start_sftp_stub(SFTP_PORT, sftp_root)
    ssh_key, pgp_public, pgp_private, gnupg_home = make_local_keys(workdir)

    env = {
        **os.environ,
        "APP_ENV": "dev",
        "INGRESS_MODE": "api",
        "EGRESS_MODE": "api",
        "MYSQL_HOST": "localhost", "MYSQL_PORT": "3306", "MYSQL_DATABASE": DB["database"],
        "MYSQL_USERNAME": DB["user"], "MYSQL_PASSWORD": DB["password"],
        "RX_API_BASE_URL": f"http://127.0.0.1:{RX_PORT}/api/1rx/v1",
        "RX_API_KEY": RX_API_KEY,
        "AMAZON_COGNITO_TOKEN_URL": f"http://127.0.0.1:{COGNITO_PORT}/oauth2/token",
        "AMAZON_CLIENT_ID": "e2e-client", "AMAZON_CLIENT_SECRET": "e2e-secret",
        "AMAZON_OAUTH_SCOPE": "shuttle-service-e2e-api/consultation",
        "AMAZON_CONSULTATION_RESPONSE_URL": f"http://127.0.0.1:{SHUTTLE_PORT}/v1/consultation/response",
        "AMAZON_PROVIDER_NAME": "myrx",
        "S3_BUCKET": S3_BUCKET, "S3_REGION": "ap-south-1", "S3_PREFIX": S3_PREFIX,
        # Real orders fall back to SFTP (EGRESS_API_TEST_ORDERS_ONLY defaults
        # to true): point that path at the in-process SFTP stub + local PGP keys.
        "SFTP_HOST": "127.0.0.1", "SFTP_PORT": str(SFTP_PORT),
        "SFTP_USERNAME": "e2e-incoming-user", "OUTGOING_SFTP_USERNAME": SFTP_OUT_USER,
        "SFTP_PRIVATE_KEY_PATH": ssh_key, "OUTGOING_SFTP_PATH": "/consultationresponses",
        "PGP_PUBLIC_KEY_PATH": pgp_public, "PGP_PRIVATE_KEY_PATH": pgp_private,
        "GNUPGHOME": gnupg_home,
    }
    app = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app", "--port", str(APP_PORT), "--log-level", "warning"],
        cwd=REPO, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
    )
    try:
        base = f"http://127.0.0.1:{APP_PORT}"
        up = wait_for(lambda: http("GET", f"{base}/openapi.json")[0] == 200 if app.poll() is None else None, timeout=30)
        if not up:
            print(app.stdout.read().decode()[-4000:])
            print("Service failed to start")
            return 1
        print(f"Service up on :{APP_PORT} (scratch DB {DB['database']}, egress=api, test orders only)\n")
        if serve:
            return serve_forever()
        token = forge_amazon_token()
        auth = {"Authorization": f"Bearer {token}"}

        # ---- Scenario 1: happy path ------------------------------------
        print("Scenario 1: creation -> 1Rx -> APPROVED webhook -> Shuttle ACK")
        cid = f"e2e-cons-{int(time.time())}"
        status, ack = http("POST", f"{base}/connector/amazon/orders", creation_request(cid, "trk-e2e-1"), auth)
        check("ingress acknowledged CREATED", status == 200 and ack.get("consultationCreationStatus") == "CREATED", str(ack))
        check("ingress echoed identifiers", ack.get("messageIdentifiers", {}).get("trackingId") == "trk-e2e-1")
        check("1Rx received the order", wait_for(lambda: any(o.get("shipmentCode") == cid for o in STATE["rx_orders"])) is not None)
        row = db_row(cid)
        check("row stored with trackingId/isTestData", row and row["tracking_id"] == "trk-e2e-1" and row["is_test_data"] == 1, str(row and (row["tracking_id"], row["is_test_data"])))

        status, body = http("POST", f"{base}/webhook/1rx", rx_webhook(cid), timeout=30)
        check("webhook accepted", status == 200 and body.get("status") == "accepted", str(body))
        row = wait_for(lambda: (r := db_row(cid)) and r.get("egress_api_status") and r)
        check("egress recorded acknowledged", row and row["egress_api_status"] == "acknowledged", str(row and row["egress_api_status"]))
        check("processing_complete", row and row["status"] == "processing_complete", str(row and row["status"]))

        sent = STATE["shuttle_requests"][-1] if STATE["shuttle_requests"] else {}
        ids = sent.get("messageIdentifiers", {})
        consultation = sent.get("data", {}).get("consultationResponse", {})
        detail = (sent.get("data", {}).get("rxDetailFromConsultation") or [{}])[0]
        check("Shuttle got Bearer token from Cognito stub", str(sent.get("_authorization", "")).startswith("Bearer e2e-shuttle-token"))
        check("payload echoes primary/secondary/isTestData",
              ids.get("primaryReferenceId") == PRIMARY and ids.get("secondaryReferenceId") == cid and ids.get("isTestData") is True, str(ids))
        check("providerName + fresh trackingId", ids.get("providerName") == "myrx" and ids.get("trackingId") not in (None, "trk-e2e-1"))
        check("SUCCESS with prescriber details", consultation.get("status") == "SUCCESS"
              and detail.get("rxHeaderDetailsFromConsultation", {}).get("prescriberDetails", {}).get("registrationNumber") == "MCI-E2E-1")
        zip_ok = STATE["shuttle_zip_checks"] and STATE["shuttle_zip_checks"][-1][0]
        check("Amazon downloaded prescription ZIP from S3", bool(zip_ok), str(STATE["shuttle_zip_checks"][-1:]))
        if zip_ok:
            check("ZIP contains the prescription", STATE["shuttle_zip_checks"][-1][1] == [f"{cid}.jpg"], str(STATE["shuttle_zip_checks"][-1][1]))

        # ---- Scenario 2: FAILED consultation ---------------------------
        print("Scenario 2: FAILED consultation (no prescription)")
        cid2 = f"{cid}-failed"
        http("POST", f"{base}/connector/amazon/orders", creation_request(cid2, "trk-e2e-2"), auth)
        wait_for(lambda: any(o.get("shipmentCode") == cid2 for o in STATE["rx_orders"]))
        before = len(STATE["shuttle_zip_checks"])
        http("POST", f"{base}/webhook/1rx", rx_webhook(cid2, status="CANCELLED"), timeout=30)
        row2 = wait_for(lambda: (r := db_row(cid2)) and r.get("egress_api_status") and r)
        sent2 = STATE["shuttle_requests"][-1]
        check("FAILED delivered + acknowledged", row2 and row2["egress_api_status"] == "acknowledged")
        check("failureReason mapped", sent2["data"]["consultationResponse"].get("failureReason") == "USER_NOT_ANSWERING_CALL", str(sent2["data"]["consultationResponse"]))
        check("no rxDetail / no ZIP for FAILED", "rxDetailFromConsultation" not in sent2["data"] and len(STATE["shuttle_zip_checks"]) == before)

        # ---- Scenario 3: ERRORED then manual resend --------------------
        print("Scenario 3: Shuttle ERRORED -> management resend recovers")
        cid3 = f"{cid}-err"
        http("POST", f"{base}/connector/amazon/orders", creation_request(cid3, "trk-e2e-3"), auth)
        wait_for(lambda: any(o.get("shipmentCode") == cid3 for o in STATE["rx_orders"]))
        STATE["shuttle_errored_next"] = 1
        http("POST", f"{base}/webhook/1rx", rx_webhook(cid3), timeout=30)
        row3 = wait_for(lambda: (r := db_row(cid3)) and r.get("egress_api_status") and r)
        check("ERRORED recorded with message", row3 and row3["egress_api_status"] == "errored" and "scripted rejection" in (row3["egress_api_error"] or ""), str(row3 and (row3["egress_api_status"], row3["egress_api_error"])))
        errored_tracking = row3 and row3["egress_tracking_id"]

        status, resend = http("POST", f"{base}/management/consultations/{quote(cid3, safe='')}/resend", {})
        row3 = db_row(cid3)
        check("resend acknowledged", status == 200 and resend.get("acknowledged") is True, str(resend))
        check("resend used a NEW trackingId", row3["egress_tracking_id"] and row3["egress_tracking_id"] != errored_tracking)
        check("row now acknowledged", row3["egress_api_status"] == "acknowledged")

        # ---- Scenario 4: duplicate creation request --------------------
        print("Scenario 4: duplicate creation request")
        rx_before = len(STATE["rx_orders"])
        status, dup = http("POST", f"{base}/connector/amazon/orders", creation_request(cid, "trk-e2e-1-retry"), auth)
        check("duplicate acknowledged CREATED", status == 200 and dup.get("consultationCreationStatus") == "CREATED", str(dup))
        time.sleep(1)
        check("no second 1Rx order for duplicate", len(STATE["rx_orders"]) == rx_before)

        # ---- Scenario 5: real order -> SFTP, never the API --------------
        print("Scenario 5: real order (isTestData=false) -> SFTP only, Shuttle untouched")
        cid5 = f"{cid}-real"
        status, ack5 = http("POST", f"{base}/connector/amazon/orders", creation_request(cid5, "trk-e2e-5", is_test_data=False), auth)
        check("real order ingress acknowledged CREATED", status == 200 and ack5.get("consultationCreationStatus") == "CREATED", str(ack5))
        wait_for(lambda: any(o.get("shipmentCode") == cid5 for o in STATE["rx_orders"]))
        row5 = db_row(cid5)
        check("row stored with isTestData=0", row5 and row5["is_test_data"] == 0, str(row5 and row5["is_test_data"]))
        shuttle_before, cognito_before = len(STATE["shuttle_requests"]), STATE["cognito_tokens"]
        status, body5 = http("POST", f"{base}/webhook/1rx", rx_webhook(cid5), timeout=60)
        check("webhook accepted", status == 200 and body5.get("status") == "accepted", str(body5))
        row5 = wait_for(lambda: (r := db_row(cid5)) and r["status"] in ("processing_complete", "failed_processing") and r, timeout=60)
        check("processing_complete via SFTP", row5 and row5["status"] == "processing_complete", str(row5 and (row5["status"], row5.get("error_message"))))
        check("Shuttle never called for the real order", len(STATE["shuttle_requests"]) == shuttle_before and STATE["cognito_tokens"] == cognito_before)
        check("no egress_api status recorded", row5 and row5.get("egress_api_status") is None and row5.get("egress_tracking_id") is None, str(row5 and (row5.get("egress_api_status"), row5.get("egress_tracking_id"))))
        out_dir = os.path.join(sftp_root, "consultationresponses")
        uploaded = sorted(os.listdir(out_dir)) if os.path.isdir(out_dir) else []
        check("encrypted ZIP uploaded to /consultationresponses", len(uploaded) == 1 and uploaded[0].startswith(f"{PRIMARY}_") and uploaded[0].endswith(".zip"), str(uploaded))
        check("uploaded as the OUTGOING sftp user", STATE["sftp_logins"][-1:] == [SFTP_OUT_USER], str(STATE["sftp_logins"]))
        if len(uploaded) == 1:
            try:
                zf = decrypt_sftp_zip(os.path.join(out_dir, uploaded[0]), gnupg_home)
                names = sorted(zf.namelist())
                check("ZIP decrypts with the partner key and holds response.json + JPEG", names == sorted(["response.json", f"{cid5}.jpg"]), str(names))
                response = json.loads(zf.read("response.json"))
                check("response.json is the legacy SFTP schema for this order", cid5 in json.dumps(response) and PRIMARY in json.dumps(response) and "messageIdentifiers" not in response, str(list(response.keys())))
                check("JPEG in ZIP is the downloaded prescription", zf.read(f"{cid5}.jpg") == JPEG_BYTES)
            except Exception as e:
                check("ZIP decrypts with the partner key", False, str(e))

        # ---- Auth negative check ---------------------------------------
        status, _ = http("POST", f"{base}/connector/amazon/orders", creation_request("x", "t"), {"Authorization": "Bearer not-a-token"})
        check("bad token rejected 401", status == 401)

        print(f"\n{'=' * 60}\nRESULT: {len(PASS)} passed, {len(FAIL)} failed")
        if FAIL:
            print("Failed checks:", *[f"  - {name}" for name in FAIL], sep="\n")
        return 1 if FAIL else 0
    finally:
        app.send_signal(signal.SIGINT)
        try:
            app.wait(timeout=10)
        except subprocess.TimeoutExpired:
            app.kill()


if __name__ == "__main__":
    sys.exit(main())