requests 2.31.0

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify pkg:pypi/requests@2.31.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/requests@2.31.0

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:7884d68f3f8e8ac75b12ea1e79ed7307e72f4bb64a4c918a196dcf33781ce2c4","contract":["requests.get sends HTTP GET request with query params, custom headers and decodes JSON response","requests.post sends HTTP POST request with JSON payload and decodes response body","requests.Response status_code and raise_for_status validate successful responses and raise HTTPError on errors","requests.Session persists headers and cookies across multiple HTTP requests"],"goal":"verify pkg:pypi/requests@2.31.0","kind":"HOW","packages":["pkg:pypi/requests@2.31.0"],"schemaVersion":1},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/requests@2.31.0"],"schemaVersion":1,"subject":"pkg:pypi/requests@2.31.0","verifierAdapter":"python@1"}

requests==2.31.0

"""Basic HTTP operations using requests."""

from typing import Any, Dict, Optional
import requests


def get_json(
    url: str,
    params: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 10.0,
) -> requests.Response:
    """Send an HTTP GET request and return the Response object."""
    return requests.get(url, params=params, headers=headers, timeout=timeout)


def post_json(
    url: str,
    payload: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 10.0,
) -> requests.Response:
    """Send an HTTP POST request with a JSON payload and return the Response object."""
    return requests.post(url, json=payload, headers=headers, timeout=timeout)


def create_session() -> requests.Session:
    """Create and return a new requests Session instance."""
    return requests.Session()

{
  "schemaVersion": 1,
  "goal": "verify pkg:pypi/requests@2.31.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/requests@2.31.0"
  ]
}

"""Contract verification test for requests 2.31.0."""

import http.server
import json
import os
import sys
import threading

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import requests
from sample import get_json, post_json, create_session


class MockHTTPHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.startswith("/api/data"):
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Echo-Header", self.headers.get("X-Custom-Test", ""))
            cookie_hdr = self.headers.get("Cookie", "")
            if cookie_hdr:
                self.send_header("X-Echo-Cookie", cookie_hdr)
            self.end_headers()
            self.wfile.write(b'{"status": "ok", "action": "read"}')
        elif self.path == "/error/404":
            self.send_response(404)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            self.wfile.write(b"Not Found")
        else:
            self.send_response(400)
            self.end_headers()

    def do_POST(self):
        if self.path == "/api/submit":
            content_length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_length)
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            parsed_body = json.loads(body.decode("utf-8")) if body else {}
            response_payload = {
                "status": "created",
                "received": parsed_body,
            }
            self.wfile.write(json.dumps(response_payload).encode("utf-8"))
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        pass


def test_get_request(base_url: str) -> None:
    """Assertion 1: requests.get sends HTTP GET request with query params, custom headers and decodes JSON response"""
    url = f"{base_url}/api/data"
    params = {"query": "sample", "limit": 5}
    headers = {"X-Custom-Test": "custom-value"}

    resp = get_json(url, params=params, headers=headers, timeout=5.0)
    assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
    assert resp.headers.get("X-Echo-Header") == "custom-value"
    data = resp.json()
    assert data == {"status": "ok", "action": "read"}


def test_post_request(base_url: str) -> None:
    """Assertion 2: requests.post sends HTTP POST request with JSON payload and decodes response body"""
    url = f"{base_url}/api/submit"
    payload = {"key": "value", "count": 42}

    resp = post_json(url, payload=payload, timeout=5.0)
    assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
    data = resp.json()
    assert data["status"] == "created"
    assert data["received"] == {"key": "value", "count": 42}


def test_response_status_and_errors(base_url: str) -> None:
    """Assertion 3: requests.Response status_code and raise_for_status validate successful responses and raise HTTPError on errors"""
    ok_resp = requests.get(f"{base_url}/api/data", timeout=5.0)
    assert ok_resp.status_code == 200
    ok_resp.raise_for_status()

    err_resp = requests.get(f"{base_url}/error/404", timeout=5.0)
    assert err_resp.status_code == 404
    raised = False
    try:
        err_resp.raise_for_status()
    except requests.exceptions.HTTPError as exc:
        raised = True
        assert exc.response.status_code == 404

    assert raised, "Expected raise_for_status to raise HTTPError for 404 response"


def test_session_persistence(base_url: str) -> None:
    """Assertion 4: requests.Session persists headers and cookies across multiple HTTP requests"""
    session = create_session()
    try:
        session.headers["X-Custom-Test"] = "session-header-123"
        session.cookies.set("session_token", "token-xyz")

        resp1 = session.get(f"{base_url}/api/data", timeout=5.0)
        assert resp1.status_code == 200
        assert resp1.headers.get("X-Echo-Header") == "session-header-123"
        assert "session_token=token-xyz" in resp1.headers.get("X-Echo-Cookie", "")

        resp2 = session.get(f"{base_url}/api/data", timeout=5.0)
        assert resp2.status_code == 200
        assert resp2.headers.get("X-Echo-Header") == "session-header-123"
        assert "session_token=token-xyz" in resp2.headers.get("X-Echo-Cookie", "")
    finally:
        session.close()


def main() -> None:
    server = http.server.HTTPServer(("127.0.0.1", 0), MockHTTPHandler)
    port = server.server_address[1]
    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()

    base_url = f"http://127.0.0.1:{port}"
    try:
        test_get_request(base_url)
        test_post_request(base_url)
        test_response_status_and_errors(base_url)
        test_session_persistence(base_url)
        print("All contract assertions passed.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()