批量接口返回 200，为什么还有一项没查清？

"""Real loopback HTTP, fictional read-only batch contract; no model or remote API."""
from copy import deepcopy
import csv
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path
from threading import Thread
from urllib.error import HTTPError
from urllib.request import Request, ProxyHandler, build_opener

REQUESTS = [{"id": "q1", "resource": "R1"}, {"id": "q2", "resource": "R2"}, {"id": "q3", "resource": "R3"}]
MODES = ("mixed", "all_found", "missing", "duplicate", "unknown", "outer_failure")


def fixture(mode):
    rows = [{"id": "q3", "status": 429, "body": {"reason": "busy"}},
            {"id": "q1", "status": 200, "body": {"resource": "R1", "name": "Tea"}},
            {"id": "q2", "status": 404, "body": {"reason": "not_found"}}]
    if mode == "all_found":
        rows = [{"id": q["id"], "status": 200, "body": {"resource": q["resource"], "name": "Fixture"}}
                for q in reversed(REQUESTS)]
    if mode == "missing":
        rows.pop()
    if mode == "duplicate":
        rows.append(deepcopy(rows[1]))
    if mode == "unknown":
        rows.append({"id": "qX", "status": 200, "body": {"resource": "RX", "name": "Other"}})
    return {"responses": rows}


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_POST(self):
        mode = self.path.removeprefix("/")
        request = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        assert request == {"requests": REQUESTS}
        status = 503 if mode == "outer_failure" else 200
        body = {"error": "unavailable"} if status == 503 else fixture(mode)
        encoded = json.dumps(body).encode("utf-8")
        self.server.seen.append({"mode": mode, "outer_status": status})
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)


def reconcile(requests, outer_status, payload):
    requested = {q["id"]: q for q in requests}
    if len(requested) != len(requests):
        raise ValueError("duplicate_request_id")
    problems, responses = [], {}
    if outer_status != 200:
        problems.append("outer_request_failed")
    elif not isinstance(payload, dict) or set(payload) != {"responses"} or not isinstance(payload["responses"], list):
        problems.append("invalid_envelope")
    else:
        for row in payload["responses"]:
            if (not isinstance(row, dict) or set(row) != {"id", "status", "body"}
                or not isinstance(row["id"], str) or type(row["status"]) is not int or not isinstance(row["body"], dict)):
                problems.append("invalid_response_shape")
                continue
            key = row["id"]
            if key not in requested:
                problems.append("unknown_response_id:" + key)
            elif key in responses:
                problems.append("duplicate_response_id:" + key)
            else:
                responses[key] = row
        problems.extend("missing_response_id:" + key for key in requested if key not in responses)
    ledger = []
    for request in requests:  # Present results in request order, not response order.
        row = responses.get(request["id"])
        if problems:
            verdict = "unverified"
        elif row["status"] == 200:
            verdict = "found" if row["body"].get("resource") == request["resource"] else "resource_mismatch"
        elif row["status"] == 404:
            verdict = "not_found"  # Defined negative answer in THIS fictional contract.
        elif row["status"] == 429:
            verdict = "temporarily_unavailable"
        else:
            verdict = "unsupported_status"
        ledger.append({**request, "status": row["status"] if row else None, "verdict": verdict})
    resolved = sum(row["verdict"] in {"found", "not_found"} for row in ledger)
    return {"status": "invalid_batch" if problems else ("resolved" if resolved == len(requests) else "incomplete"),
            "requested": len(requests), "resolved": resolved, "unresolved": len(requests) - resolved,
            "problems": problems, "ledger": ledger}


def main():
    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    server.seen = []
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    opener = build_opener(ProxyHandler({}))
    cases = []
    try:
        for mode in MODES:
            request = Request(f"http://127.0.0.1:{server.server_port}/{mode}",
                              data=json.dumps({"requests": REQUESTS}).encode("utf-8"),
                              headers={"Content-Type": "application/json"}, method="POST")
            try:
                response = opener.open(request, timeout=5)
            except HTTPError as error:
                response = error
            with response:
                status, payload = response.status, json.load(response)
            cases.append({"mode": mode, "outer_status": status, "raw": payload,
                          "report": reconcile(REQUESTS, status, payload)})
    finally:
        server.shutdown()
        thread.join(timeout=5)
        server.server_close()
    assert not thread.is_alive() and len(server.seen) == 6
    mixed = cases[0]["report"]
    assert mixed["status"] == "incomplete" and mixed["resolved"] == 2 and mixed["unresolved"] == 1
    assert [r["verdict"] for r in mixed["ledger"]] == ["found", "not_found", "temporarily_unavailable"]
    naive = [{"request_id": q["id"], "paired_response_id": row["id"], "status": row["status"]}
             for q, row in zip(REQUESTS, cases[0]["raw"]["responses"])]
    assert all(r["request_id"] != r["paired_response_id"] for r in naive)
    assert cases[1]["report"]["status"] == "resolved"
    assert all(c["report"]["status"] == "invalid_batch" and c["report"]["resolved"] == 0 for c in cases[2:])
    assert cases[-1]["outer_status"] == 503
    assert cases[2]["report"]["problems"] == ["missing_response_id:q2"]
    assert cases[3]["report"]["problems"] == ["duplicate_response_id:q1"]
    assert cases[4]["report"]["problems"] == ["unknown_response_id:qX"]
    wrong_resource = fixture("mixed")
    wrong_resource["responses"][1]["body"]["resource"] = "R9"
    assert reconcile(REQUESTS, 200, wrong_resource)["ledger"][0]["verdict"] == "resource_mismatch"
    bad_status = fixture("mixed")
    bad_status["responses"][0]["status"] = True
    assert "invalid_response_shape" in reconcile(REQUESTS, 200, bad_status)["problems"]
    try:
        reconcile(REQUESTS + [REQUESTS[0]], 200, fixture("mixed"))
    except ValueError:
        pass
    else:
        raise AssertionError("duplicate request accepted")
    output = {"scope": "six real loopback requests; fixed read-only fixtures; no retries",
              "naive_position_pairing": naive, "server_seen": server.seen, "cases": cases, "assertions": "passed"}
    Path("batch-lookup-report.json").write_text(json.dumps(output, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    with Path("batch-lookup-ledger.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=["mode", "id", "resource", "status", "verdict"])
        writer.writeheader()
        for case in cases:
            writer.writerows({"mode": case["mode"], **row} for row in case["report"]["ledger"])
    for case in cases:
        result = case["report"]
        print(f"{case['mode']}: HTTP {case['outer_status']}; {result['status']}; resolved={result['resolved']}/3")
    print("PASS: 6 loopback batches; item identity, status and completeness checked; no retries")


if __name__ == "__main__":
    main()