API Overview¶

def resolve(self, host: str, port: int, timeout: float) -> tuple[str, str, float]:
    """Resolve host and return IP, family label, and elapsed milliseconds.

    Args:
        host: Hostname to resolve (e.g., "example.com").
        port: Port number used to hint the address lookup.
        timeout: Maximum time to wait for resolution, in seconds.

    Returns:
        Tuple of ``(ip_address, ip_family, elapsed_ms)`` where ``ip_family``
        is one of ``"IPv4"``, ``"IPv6"``, or ``"AF_<num>"`` for other
        address families, and ``elapsed_ms`` is the resolution time in
        milliseconds.

    Raises:
        DNSResolutionError: If resolution times out, the lookup fails, or no
            usable address record is returned.
    """
    start_time = time.perf_counter()

    addr_info: list[AddrInfo] | None = None
    worker_error: Exception | None = None

    def resolver_task() -> None:
        nonlocal addr_info, worker_error
        try:
            addr_info = cast(
                "list[AddrInfo]",
                socket.getaddrinfo(
                    host,
                    port,
                    family=socket.AF_UNSPEC,
                    type=socket.SOCK_STREAM,
                ),
            )
        except Exception as exc:  # pragma: no cover - handled below  # noqa: BLE001
            worker_error = exc

    thread = threading.Thread(target=resolver_task, daemon=True)
    thread.start()
    thread.join(timeout)

    if thread.is_alive():
        message = f"DNS resolution timed out for {host} after {timeout:.2f}s"
        raise DNSResolutionError(message)

    if worker_error:
        if isinstance(worker_error, socket.gaierror):
            message = f"DNS resolution failed for {host}: {worker_error}"
            raise DNSResolutionError(message) from worker_error
        details = f"Unexpected error during DNS resolution for {host}: {worker_error}"
        raise DNSResolutionError(details) from worker_error

    if addr_info is None:
        message = f"No address records for {host}"
        raise DNSResolutionError(message)

    if not addr_info:
        message = f"No address records for {host}"
        raise DNSResolutionError(message)

    records = _normalize_addrinfo(addr_info)
    if not records:
        message = f"No address records for {host}"
        raise DNSResolutionError(message)

    record = records[0]
    ip = str(record.sockaddr[0]) if record.sockaddr else None
    if not ip:
        message = f"Failed to extract IP address for {host}"
        raise DNSResolutionError(message)

    elapsed_ms = (time.perf_counter() - start_time) * MS_IN_SECOND
    ip_family = self._family_to_label(record.family)
    return ip, ip_family, elapsed_ms

def inspect(self, host: str, port: int, timeout: float) -> NetworkInfo:
    """Inspect TLS connection and extract metadata.

    Args:
        host: Hostname to connect to, also used for SNI.
        port: Port number (typically 443 for HTTPS).
        timeout: Connection timeout in seconds. The probe is additionally
            capped by ``TLS_PROBE_MAX_TIMEOUT_SECONDS``.

    Returns:
        A NetworkInfo populated with the resolved IP, negotiated TLS
        version and cipher, and leaf-certificate details when available.

    Raises:
        TLSInspectionError: If the connection or TLS handshake fails.
    """
    network_info = NetworkInfo()
    network_info.tls_verified = self._verify
    probe_timeout = min(timeout, TLS_PROBE_MAX_TIMEOUT_SECONDS)

    try:
        connection = socket.create_connection((host, port), timeout=probe_timeout)
        with closing(connection) as raw_sock:
            self._populate_network_info(raw_sock, network_info)

            # Diagnostic tool: intentionally allows TLSv1.0+ to inspect legacy servers.
            # This is NOT a security issue because httptap is used for troubleshooting,
            # not for transmitting sensitive data in production.
            context = create_ssl_context(verify_ssl=self._verify, ca_bundle_path=self._ca_bundle_path)
            with context.wrap_socket(raw_sock, server_hostname=host) as tls_sock:
                tls_version, cipher_suite, cert_info = extract_tls_info(tls_sock)
                network_info.tls_version = tls_version
                network_info.tls_cipher = cipher_suite

                if cert_info:
                    apply_certificate_info(network_info, cert_info)

    except Exception as exc:
        msg = f"TLS inspection failed for {host}:{port}: {exc}"
        raise TLSInspectionError(
            msg,
        ) from exc

    return network_info

def mark_dns_start(self) -> None:
    """Record the beginning of DNS resolution."""
    self._dns_start = time.perf_counter()

def mark_dns_end(self) -> None:
    """Record the completion of DNS resolution."""
    self._dns_end = time.perf_counter()

def mark_request_start(self) -> None:
    """Record the moment the HTTP request starts."""
    self._request_start = time.perf_counter()

def mark_ttfb(self) -> None:
    """Record when the first response byte is received."""
    self._ttfb_time = time.perf_counter()

def mark_request_end(self) -> None:
    """Record when the response body transfer completes."""
    self._end_time = time.perf_counter()

def get_metrics(self) -> TimingMetrics:
    """Build a TimingMetrics instance populated with collected timings.

    Returns:
        A TimingMetrics with ``dns_ms``, ``ttfb_ms``, and ``total_ms``
        computed from the recorded marks. Call ``calculate_derived()`` on
        the result to populate ``wait_ms`` and ``xfer_ms``.
    """
    timing = TimingMetrics()
    timing.dns_ms = (self._dns_end - self._dns_start) * MS_IN_SECOND
    timing.total_ms = (self._end_time - self._start_time) * MS_IN_SECOND
    timing.ttfb_ms = (self._ttfb_time - self._start_time) * MS_IN_SECOND
    return timing

def render(self, step: StepMetrics) -> None:
    """Render a waterfall timeline for the provided step if data is valid.

    Steps that errored or have a non-positive total time are skipped and
    produce no output.

    Args:
        step: Step metrics containing timing, network, and response data.
    """
    if step.has_error or step.timing.total_ms <= 0:
        return

    phases = self._get_phases(step)
    durations = [duration for _, duration, _ in phases]
    bar_widths = self._compute_phase_widths(durations)
    used_width = sum(bar_widths) or 1
    scale = step.timing.total_ms / used_width

    self.console.print("\n  [bold]Request Timeline:[/bold]")

    current_position_chars = 0
    for (label, duration, color), bar_width in zip(phases, bar_widths, strict=True):
        current_position_chars = self._render_phase(
            label,
            duration,
            color,
            current_position_chars,
            bar_width,
        )

    self._render_total(step.timing.total_ms, scale)

def export(
    self,
    steps: Sequence[StepMetrics],
    initial_url: str,
    output_path: str,
    *,
    slo_result: SLOResult | None = None,
) -> None:
    """Export analysis data to JSON file.

    Creates a structured JSON file with all step metrics,
    timing information, and summary data.

    Args:
        steps: Sequence of step metrics to export.
        initial_url: Initial URL that was analyzed.
        output_path: Path to output JSON file.
        slo_result: Optional SLO evaluation result to embed under
            ``summary.slo``.

    Raises:
        IOError: If file cannot be written.

    """
    data = self._build_export_data(steps, initial_url, slo_result=slo_result)
    self._write_json_file(data, output_path)
    self._print_success(output_path)

def execute(self, options: RequestOptions) -> RequestOutcome:
    """Perform an HTTP request using the default client.

    Args:
        options: Fully populated request parameters.

    Returns:
        A RequestOutcome bundling the timing, network, and response data.

    Raises:
        httptap.http_client.HTTPClientError: If the underlying HTTP request
            fails.
    """
    timing, network, response = make_request(
        options.url,
        options.timeout,
        method=options.method,
        content=options.content,
        http2=options.http2,
        verify_ssl=options.verify_ssl,
        ca_bundle_path=options.ca_bundle_path,
        proxy=options.proxy,
        noproxy=options.noproxy,
        dns_resolver=options.dns_resolver,
        tls_inspector=options.tls_inspector,
        timing_collector=options.timing_collector,
        force_new_connection=options.force_new_connection,
        headers=options.headers,
    )
    return RequestOutcome(timing=timing, network=network, response=response)

def to_dict(self) -> dict[str, Any]:
    """Serialize to a plain ``dict`` for JSON export.

    Threshold keys are emitted in alphabetical order so that two
    evaluations with the same user input always produce
    byte-identical JSON (useful for diffing and cache keys).

    Returns:
        Mapping with keys ``pass``, ``thresholds_ms``, ``violations``.

    """
    return {
        "pass": self.passed,
        "thresholds_ms": {key: self.thresholds_ms[key] for key in sorted(self.thresholds_ms)},
        "violations": [v.to_dict() for v in self.violations],
    }

def to_dict(self) -> dict[str, float | str]:
    """Serialize to a plain ``dict`` for JSON export.

    Returns:
        Mapping with keys ``key``, ``threshold_ms``, ``actual_ms``,
        ``delta_ms``.

    """
    return {
        "key": self.key,
        "threshold_ms": self.threshold_ms,
        "actual_ms": self.actual_ms,
        "delta_ms": self.delta_ms,
    }