requests 2.34.2: requests

# requests API

This sample pins `requests` 2.34.2 and verifies top-level `requests` APIs:
- `requests.post` sending JSON payloads and form data
- `requests.put`, `requests.patch`, and `requests.delete` HTTP method dispatching
- `requests.head` header retrieval without response body
- `requests.codes` status code constants
- `requests.exceptions.RequestException` base exception hierarchy

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 requests in pkg:pypi/requests@2.34.2
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/requests@2.34.2
Demonstrate these symbols/APIs:
  - requests

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:c3c18ee8b0510693a3d093f05a55cc77859da448b5d534c557c7e76d6813a9e2","contract":["requests.post sends JSON payload and retrieves parsed JSON response with status code","requests.post sends form-encoded data with custom headers and retrieves response","requests.put, requests.patch, and requests.delete dispatch corresponding HTTP methods and receive responses","requests.head retrieves headers without response body and requests.codes provides HTTP status constants","requests.exceptions exposes RequestException base class caught during connection failure"],"goal":"verify requests in pkg:pypi/requests@2.34.2","kind":"HOW","packages":["pkg:pypi/requests@2.34.2"],"schemaVersion":1,"symbols":["requests"]},"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.34.2"],"schemaVersion":1,"subject":"pkg:pypi/requests@2.34.2","symbols":["requests"],"verifierAdapter":"python@1"}

requests==2.34.2

"""Demonstration of core requests functions."""

from typing import Any, Dict, Optional
import requests


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


def send_form_post(
    url: str,
    data: Dict[str, Any],
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 10.0,
) -> requests.Response:
    """Send an HTTP POST request with form-encoded data."""
    return requests.post(url, data=data, headers=headers, timeout=timeout)


def send_put(
    url: str,
    data: Optional[bytes] = None,
    json_data: Optional[Dict[str, Any]] = None,
    timeout: float = 10.0,
) -> requests.Response:
    """Send an HTTP PUT request."""
    return requests.put(url, data=data, json=json_data, timeout=timeout)


def send_patch(
    url: str,
    json_data: Optional[Dict[str, Any]] = None,
    timeout: float = 10.0,
) -> requests.Response:
    """Send an HTTP PATCH request."""
    return requests.patch(url, json=json_data, timeout=timeout)


def send_delete(
    url: str,
    timeout: float = 10.0,
) -> requests.Response:
    """Send an HTTP DELETE request."""
    return requests.delete(url, timeout=timeout)


def fetch_headers_only(
    url: str,
    timeout: float = 10.0,
) -> requests.structures.CaseInsensitiveDict:
    """Send an HTTP HEAD request and return response headers."""
    response = requests.head(url, timeout=timeout)
    return response.headers

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

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

import http.server
import json
import os
import socket
import sys
import threading
import urllib.parse

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import requests
from sample import (
    send_json_post,
    send_form_post,
    send_put,
    send_patch,
    send_delete,
    fetch_headers_only,
)


class MockHTTPHandler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length)

        if self.path == "/api/items":
            assert "application/json" in self.headers.get("Content-Type", "")
            data = json.loads(body.decode("utf-8"))
            assert data.get("title") == "test-item"
            self.send_response(201)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"id": "item-123", "status": "created"}')
        elif self.path == "/api/login":
            assert "application/x-www-form-urlencoded" in self.headers.get("Content-Type", "")
            assert self.headers.get("X-Client-Id") == "client-99"
            parsed = urllib.parse.parse_qs(body.decode("utf-8"))
            assert parsed.get("username") == ["user1"]
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(b"login successful")
        else:
            self.send_response(404)
            self.end_headers()

    def do_PUT(self):
        if self.path == "/api/resource":
            content_length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_length)
            assert body == b"new binary data"
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"updated": true}')
        else:
            self.send_response(404)
            self.end_headers()

    def do_PATCH(self):
        if self.path == "/api/resource":
            content_length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_length)
            data = json.loads(body.decode("utf-8"))
            assert data.get("field") == "patched_value"
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"patched": true}')
        else:
            self.send_response(404)
            self.end_headers()

    def do_DELETE(self):
        if self.path == "/api/resource":
            self.send_response(204)
            self.end_headers()
        else:
            self.send_response(404)
            self.end_headers()

    def do_HEAD(self):
        if self.path == "/api/headers-only":
            self.send_response(200)
            self.send_header("X-Sample-Header", "verified-token")
            self.send_header("Content-Type", "text/plain")
            self.send_header("Content-Length", "100")
            self.end_headers()
        else:
            self.send_response(404)
            self.end_headers()

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


def test_requests_post_json(base_url: str) -> None:
    """Assertion 1: requests.post sends JSON payload and retrieves parsed JSON response with status code."""
    url = f"{base_url}/api/items"
    payload = {"title": "test-item"}

    resp = requests.post(url, json=payload, timeout=5.0)
    assert resp.status_code == requests.codes.created, f"Expected 201, got {resp.status_code}"
    assert resp.json() == {"id": "item-123", "status": "created"}

    # Verify helper function
    sample_resp = send_json_post(url, payload, timeout=5.0)
    assert sample_resp.status_code == 201
    assert sample_resp.json()["id"] == "item-123"


def test_requests_post_form(base_url: str) -> None:
    """Assertion 2: requests.post sends form-encoded data with custom headers and retrieves response."""
    url = f"{base_url}/api/login"
    data = {"username": "user1"}
    headers = {"X-Client-Id": "client-99"}

    resp = requests.post(url, data=data, headers=headers, timeout=5.0)
    assert resp.status_code == requests.codes.ok, f"Expected 200, got {resp.status_code}"
    assert resp.text == "login successful"

    # Verify helper function
    sample_resp = send_form_post(url, data, headers=headers, timeout=5.0)
    assert sample_resp.status_code == 200
    assert sample_resp.text == "login successful"


def test_requests_put_patch_delete(base_url: str) -> None:
    """Assertion 3: requests.put, requests.patch, and requests.delete dispatch corresponding HTTP methods and receive responses."""
    url = f"{base_url}/api/resource"

    # PUT
    put_resp = send_put(url, data=b"new binary data", timeout=5.0)
    assert put_resp.status_code == requests.codes.ok
    assert put_resp.json() == {"updated": True}

    # PATCH
    patch_resp = send_patch(url, json_data={"field": "patched_value"}, timeout=5.0)
    assert patch_resp.status_code == requests.codes.ok
    assert patch_resp.json() == {"patched": True}

    # DELETE
    del_resp = send_delete(url, timeout=5.0)
    assert del_resp.status_code == requests.codes.no_content


def test_requests_head_and_codes(base_url: str) -> None:
    """Assertion 4: requests.head retrieves headers without response body and requests.codes provides HTTP status constants."""
    url = f"{base_url}/api/headers-only"

    # HEAD
    head_resp = requests.head(url, timeout=5.0)
    assert head_resp.status_code == requests.codes.ok
    assert head_resp.headers.get("X-Sample-Header") == "verified-token"
    assert head_resp.text == ""

    # Helper function check
    headers = fetch_headers_only(url, timeout=5.0)
    assert headers["x-sample-header"] == "verified-token"
    assert headers["X-Sample-Header"] == "verified-token"

    # requests.codes assertions
    assert requests.codes.ok == 200
    assert requests.codes.created == 201
    assert requests.codes.no_content == 204
    assert requests.codes.not_found == 404
    assert requests.codes.internal_server_error == 500


def test_requests_exceptions_hierarchy() -> None:
    """Assertion 5: requests.exceptions exposes RequestException base class caught during connection failure."""
    # Exception inheritance checks
    assert issubclass(requests.exceptions.ConnectionError, requests.exceptions.RequestException)
    assert issubclass(requests.exceptions.HTTPError, requests.exceptions.RequestException)
    assert issubclass(requests.exceptions.Timeout, requests.exceptions.RequestException)

    # Runtime connection failure catch check
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    closed_port = sock.getsockname()[1]
    sock.close()

    caught = False
    try:
        requests.get(f"http://127.0.0.1:{closed_port}", timeout=1.0)
    except requests.exceptions.RequestException as exc:
        caught = True
        assert isinstance(exc, requests.exceptions.ConnectionError)

    assert caught, "Expected RequestException when connecting to an unreachable port"


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_requests_post_json(base_url)
        test_requests_post_form(base_url)
        test_requests_put_patch_delete(base_url)
        test_requests_head_and_codes(base_url)
        test_requests_exceptions_hierarchy()
        print("All contract assertions passed.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()