httptools 0.8.0: parse HTTP requests, responses, and URLs using HttpRequestParser, HttpResponseParser, and parse_url

# httptools Request/Response Streaming and URL Parsing Contracts

## Prior Solution Found
`search_known_solution` returned `NO_SAFE_MATCH` for `pkg:pypi/httptools@0.8.0`.

This sample establishes the baseline contract for parsing HTTP requests, responses, chunked transfer encoding, WebSocket upgrades, URL segmentation, and associated parser exception types in `httptools`.

## What a Model Would Have Written
A model implementing protocol callbacks for `httptools.HttpRequestParser` or `httptools.HttpResponseParser` typically assumes:
1. `on_url` and `on_status` are invoked once with the full payload, storing `self.url = url` directly instead of accumulating streaming byte slices.
2. `feed_data()` returns normally on WebSocket upgrade headers, expecting upgrade state to be checked solely via `parser.should_upgrade()` without catching the `httptools.HttpParserUpgrade` exception.
3. `httptools.parse_url` accepts string inputs or returns decoded string attributes instead of raw `bytes`.

## How the Wrong Version Fails
1. **Truncated Request Targets / Statuses**: When network TCP packets fragment HTTP request lines or status lines, `on_url` and `on_status` receive partial byte slices. Re-assigning rather than accumulating results in corrupted/truncated URL paths (e.g. `b''` or the last character) and broken routing.
2. **Unhandled Upgrade Exception**: Feeding an `Upgrade: websocket` HTTP header triggers `httptools.HttpParserUpgrade` immediately inside `feed_data()`. Unhandled exceptions abort connection handshakes rather than allowing the server to transition the socket to raw WebSocket frames.
3. **Type and Validation Failures**: Passing non-byte URLs or malformed IPv6 host strings raises `httptools.HttpParserInvalidURLError`, and invalid HTTP method tokens raise `httptools.HttpParserInvalidMethodError`.

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: parse HTTP requests, responses, and URLs using HttpRequestParser, HttpResponseParser, and parse_url
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/httptools@0.8.0
Demonstrate these symbols/APIs:
  - httptools.HttpParserInvalidMethodError
  - httptools.HttpParserInvalidURLError
  - httptools.HttpRequestParser
  - httptools.HttpResponseParser
  - httptools.parse_url

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:97f1d70539705ea54c1af0b122ae82d28d86c202c914ed5b24a93e026e3a9901","contract":["assert parse_url parses standard URLs into schema, host, port, path, query, and fragment bytes/int attributes","assert parse_url returns None for omitted components on relative URLs","assert parse_url raises HttpParserInvalidURLError on malformed URLs","assert HttpRequestParser feeds incremental byte streams, triggers protocol callbacks, and extracts method, http version, and keep-alive status","assert HttpRequestParser raises HttpParserInvalidMethodError when fed an invalid HTTP method verb","assert HttpResponseParser handles chunked transfer encodings and emits chunk header, body chunks, and status callbacks","assert HttpRequestParser raises HttpParserUpgrade on WebSocket upgrade request and reports should_upgrade as True"],"goal":"parse HTTP requests, responses, and URLs using HttpRequestParser, HttpResponseParser, and parse_url","kind":"HOW","packages":["pkg:pypi/httptools@0.8.0"],"schemaVersion":1,"symbols":["httptools.HttpParserInvalidMethodError","httptools.HttpParserInvalidURLError","httptools.HttpRequestParser","httptools.HttpResponseParser","httptools.parse_url"]},"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/httptools@0.8.0"],"schemaVersion":1,"subject":"pkg:pypi/httptools@0.8.0","symbols":["httptools.HttpParserInvalidMethodError","httptools.HttpParserInvalidURLError","httptools.HttpRequestParser","httptools.HttpResponseParser","httptools.parse_url"],"verifierAdapter":"python@1"}

# Full dependency closure for sandbox environments installing with --no-deps
httptools==0.8.0

{
  "schemaVersion": 1,
  "goal": "parse HTTP requests, responses, and URLs using HttpRequestParser, HttpResponseParser, and parse_url",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/httptools@0.8.0"
  ],
  "symbols": [
    "httptools.HttpParserInvalidMethodError",
    "httptools.HttpParserInvalidURLError",
    "httptools.HttpRequestParser",
    "httptools.HttpResponseParser",
    "httptools.parse_url"
  ]
}

# src package

"""Robust HTTP request and response parser wrapper using httptools.

Demonstrates correct handling of streaming byte chunks in protocol callbacks,
URL component parsing, chunked response reconstitution, and WebSocket upgrade events.
"""

from typing import Dict, List, Optional, Tuple
import httptools


class StreamingRequestProtocol:
    """Protocol implementation accumulating streaming chunks from HttpRequestParser."""

    def __init__(self) -> None:
        self.message_started: bool = False
        self.url_chunks: List[bytes] = []
        self.headers: List[Tuple[bytes, bytes]] = []
        self.headers_completed: bool = False
        self.body_chunks: List[bytes] = []
        self.message_completed: bool = False

    def on_message_begin(self) -> None:
        self.message_started = True

    def on_url(self, url: bytes) -> None:
        self.url_chunks.append(url)

    def on_header(self, name: bytes, value: bytes) -> None:
        self.headers.append((name, value))

    def on_headers_complete(self) -> None:
        self.headers_completed = True

    def on_body(self, body: bytes) -> None:
        self.body_chunks.append(body)

    def on_message_complete(self) -> None:
        self.message_completed = True

    @property
    def url(self) -> bytes:
        return b"".join(self.url_chunks)

    @property
    def body(self) -> bytes:
        return b"".join(self.body_chunks)


class StreamingResponseProtocol:
    """Protocol implementation accumulating streaming chunks from HttpResponseParser."""

    def __init__(self) -> None:
        self.message_started: bool = False
        self.status_chunks: List[bytes] = []
        self.headers: List[Tuple[bytes, bytes]] = []
        self.headers_completed: bool = False
        self.chunk_headers_count: int = 0
        self.body_chunks: List[bytes] = []
        self.message_completed: bool = False

    def on_message_begin(self) -> None:
        self.message_started = True

    def on_status(self, status: bytes) -> None:
        self.status_chunks.append(status)

    def on_header(self, name: bytes, value: bytes) -> None:
        self.headers.append((name, value))

    def on_headers_complete(self) -> None:
        self.headers_completed = True

    def on_chunk_header(self) -> None:
        self.chunk_headers_count += 1

    def on_body(self, body: bytes) -> None:
        self.body_chunks.append(body)

    def on_message_complete(self) -> None:
        self.message_completed = True

    @property
    def status(self) -> bytes:
        return b"".join(self.status_chunks)

    @property
    def body(self) -> bytes:
        return b"".join(self.body_chunks)


def parse_http_url(raw_url: bytes):
    """Parses a raw URL byte string using httptools.parse_url."""
    return httptools.parse_url(raw_url)


def parse_request_stream(data_chunks: List[bytes]) -> Tuple[StreamingRequestProtocol, httptools.HttpRequestParser, bool]:
    """Feeds streaming chunks to HttpRequestParser, capturing WebSocket upgrades if present."""
    protocol = StreamingRequestProtocol()
    parser = httptools.HttpRequestParser(protocol)
    upgraded = False

    for chunk in data_chunks:
        try:
            parser.feed_data(chunk)
        except httptools.HttpParserUpgrade:
            upgraded = True
            break

    return protocol, parser, upgraded


def parse_response_stream(data_chunks: List[bytes]) -> Tuple[StreamingResponseProtocol, httptools.HttpResponseParser]:
    """Feeds streaming chunks to HttpResponseParser."""
    protocol = StreamingResponseProtocol()
    parser = httptools.HttpResponseParser(protocol)

    for chunk in data_chunks:
        parser.feed_data(chunk)

    return protocol, parser

from pathlib import Path
import sys

# Ensure project root is on sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import httptools
from src.parser_service import (
    StreamingRequestProtocol,
    StreamingResponseProtocol,
    parse_http_url,
    parse_request_stream,
    parse_response_stream,
)

# ============================================================================
# 1. URL parsing with parse_url and error validation
# ============================================================================
# Full URL with schema, host, port, path, query, and fragment
full_url = b"http://example.com:8080/api/v1/resource?filter=active&page=1#details"
parsed = parse_http_url(full_url)
assert parsed.schema == b"http"
assert parsed.host == b"example.com"
assert parsed.port == 8080
assert parsed.path == b"/api/v1/resource"
assert parsed.query == b"filter=active&page=1"
assert parsed.fragment == b"details"
assert parsed.userinfo is None

# Partial URL omitting schema, host, port, fragment
partial = parse_http_url(b"/relative/endpoint?key=value")
assert partial.schema is None
assert partial.host is None
assert partial.port is None
assert partial.path == b"/relative/endpoint"
assert partial.query == b"key=value"
assert partial.fragment is None
assert partial.userinfo is None

# Malformed URL triggers HttpParserInvalidURLError
try:
    httptools.parse_url(b"http://[invalid-ipv6-bracket")
    raise AssertionError("httptools.parse_url should have raised HttpParserInvalidURLError")
except httptools.HttpParserInvalidURLError:
    pass

# ============================================================================
# 2. HttpRequestParser streaming chunk feed, callbacks, and method parsing
# ============================================================================
body_data = b'{"action":"deploy","target":"slot2"}'
header_block = (
    "POST /api/v2/action?confirm=true HTTP/1.1\r\n"
    "Host: example.com\r\n"
    "Content-Type: application/json\r\n"
    f"Content-Length: {len(body_data)}\r\n"
    "Connection: keep-alive\r\n\r\n"
).encode("latin-1")
raw_request = header_block + body_data

# Stream byte-by-byte into request parser
byte_chunks = [raw_request[i:i+1] for i in range(len(raw_request))]
req_proto, req_parser, req_upgraded = parse_request_stream(byte_chunks)

assert req_proto.message_started is True
assert req_proto.url == b"/api/v2/action?confirm=true"
assert req_proto.headers_completed is True
assert (b"Host", b"example.com") in req_proto.headers
assert (b"Content-Type", b"application/json") in req_proto.headers
assert (b"Connection", b"keep-alive") in req_proto.headers
assert req_proto.body == body_data
assert req_proto.message_completed is True
assert req_parser.get_method() == b"POST"
assert req_parser.get_http_version() == "1.1"
assert req_parser.should_keep_alive() is True
assert req_upgraded is False

# Invalid HTTP method verb raises HttpParserInvalidMethodError
invalid_proto = StreamingRequestProtocol()
invalid_parser = httptools.HttpRequestParser(invalid_proto)
try:
    invalid_parser.feed_data(b"UNKNOWNVERB / HTTP/1.1\r\n\r\n")
    raise AssertionError("HttpRequestParser should have raised HttpParserInvalidMethodError")
except httptools.HttpParserInvalidMethodError:
    pass

# ============================================================================
# 3. HttpResponseParser with chunked transfer encoding
# ============================================================================
raw_response = (
    b"HTTP/1.1 200 OK\r\n"
    b"Transfer-Encoding: chunked\r\n"
    b"Content-Type: text/plain\r\n\r\n"
    b"4\r\nWiki\r\n"
    b"5\r\npedia\r\n"
    b"0\r\n\r\n"
)

# Stream response in arbitrary small slice chunks
resp_chunks = [raw_response[i:i+3] for i in range(0, len(raw_response), 3)]
resp_proto, resp_parser = parse_response_stream(resp_chunks)

assert resp_proto.message_started is True
assert resp_proto.status == b"OK"
assert resp_parser.get_status_code() == 200
assert resp_parser.get_http_version() == "1.1"
assert (b"Transfer-Encoding", b"chunked") in resp_proto.headers
assert resp_proto.chunk_headers_count == 3  # Chunk 1 (4), Chunk 2 (5), Chunk 3 (0 terminator)
assert resp_proto.body == b"Wikipedia"
assert resp_proto.message_completed is True

# ============================================================================
# 4. WebSocket Upgrade handshake handling and HttpParserUpgrade exception
# ============================================================================
raw_upgrade = (
    b"GET /chat/v1 HTTP/1.1\r\n"
    b"Host: example.com\r\n"
    b"Upgrade: websocket\r\n"
    b"Connection: Upgrade\r\n\r\n"
)

upgrade_proto, upgrade_parser, is_upgraded = parse_request_stream([raw_upgrade])

assert is_upgraded is True
assert upgrade_parser.should_upgrade() is True
assert upgrade_proto.url == b"/chat/v1"
assert (b"Upgrade", b"websocket") in upgrade_proto.headers

print(
    "CONTRACT PASS: httptools",
    getattr(httptools, "__version__", "0.8.0"),
    "request streaming, response chunking, URL parsing, and upgrade handling verified on python",
    ".".join(str(part) for part in sys.version_info[:2]),
)