synta 0.1.12

#!/usr/bin/env python3
"""Benchmark synta Python bindings vs cryptography.x509 — test_x509.py port.

This file is a direct port of the pytest-benchmark suite at
``cryptography/tests/bench/test_x509.py``, adapted to use the same
``criterion_compat`` measurement infrastructure as ``bench_certificate.py``.
Results are written as Criterion-compatible JSON so they appear alongside the
Rust benchmark numbers in criterion-compare reports.

Cryptography tests ported
--------------------------
test_object_identifier_constructor
    Construct an ObjectIdentifier from a dotted OID string.
    synta: ``synta.ObjectIdentifier("1.3.6.1.4.1.11129.2.4.5")``
    cryptography: ``x509.ObjectIdentifier("1.3.6.1.4.1.11129.2.4.5")``

test_load_der_certificate
    Parse GoodCACert.crt from the PKITS test suite.
    synta (lazy):  ``Certificate.from_der(cert_bytes)`` — shallow 4-op scan.
    synta (eager): ``Certificate.full_from_der(cert_bytes)`` — full RFC 5280 decode.
    cryptography:  ``x509.load_der_x509_certificate(cert_bytes)``

test_load_pem_certificate
    Parse cryptography.io.pem.  PEM-header stripping is included in the
    measurement for both libraries so that the same total work is covered.
    synta: ``Certificate.from_pem(pem_bytes)`` — Rust PEM decoder, no Python base64.
    cryptography: ``x509.load_pem_x509_certificate(pem_bytes)``

Added (no cryptography.x509 equivalent in test_x509.py)
---------------------------------------------------------
test_encode_pem_certificate
    Serialize a pre-parsed certificate back to PEM.
    synta: ``Certificate.to_pem(cert)``
    cryptography: ``cert.public_bytes(Encoding.PEM)``
    Both libraries receive a pre-parsed object; only the encoding step is timed.

Not ported
----------
test_aki_public_bytes
    Benchmarks encoding an AuthorityKeyIdentifier extension object.
    synta has no extension-builder API; encoding is out of scope.

test_verify_docs_python_org
    Full certificate-path verification.
    synta does not implement path verification.

Output (mirrors Criterion benchmark IDs)
-----------------------------------------
    object_identifier_constructor/synta/1.3.6.1.4.1.11129.2.4.5   avg: N µs
    object_identifier_constructor/cryptography_x509/...             avg: N µs
    load_der_certificate/synta/GoodCACert                           avg: N µs
    load_der_certificate/synta_full/GoodCACert                      avg: N µs
    load_der_certificate/cryptography_x509/GoodCACert               avg: N µs
    load_pem_certificate/synta/cryptography.io                      avg: N µs
    load_pem_certificate/cryptography_x509/cryptography.io          avg: N µs
    encode_pem_certificate/synta/GoodCACert                         avg: N µs
    encode_pem_certificate/cryptography_x509/GoodCACert             avg: N µs

Usage
-----
    # Build the Python extension (release build required):
    cd synta-python && maturin develop --release && cd ..

    # Run from the repository root:
    python python/bench_x509.py

    # Save Criterion-compatible JSON (default dir: target/criterion):
    python python/bench_x509.py --save-criterion
    python python/bench_x509.py --save-criterion path/to/criterion
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Callable

from criterion_compat import measure as _criterion_measure
from criterion_compat import save_criterion_files as _criterion_save

# ── synta import ──────────────────────────────────────────────────────────────

try:
    import synta
    import synta.oids as _synta_oids
except ImportError:
    print(
        "ERROR: 'synta' Python module not found.\n"
        "Build and install it with:\n"
        "    cd synta-python && maturin develop --release && cd ..",
        file=sys.stderr,
    )
    sys.exit(1)

# ── cryptography.x509 import (optional) ───────────────────────────────────────

try:
    import cryptography.x509 as _cx509
    _HAS_CRYPTOGRAPHY = True
except ImportError:
    _HAS_CRYPTOGRAPHY = False
    print(
        "INFO: 'cryptography' package not installed — skipping cryptography_x509 comparison.\n"
        "Install with: pip install cryptography",
        file=sys.stderr,
    )

# ── Paths ─────────────────────────────────────────────────────────────────────

_SCRIPT_DIR = Path(__file__).resolve().parent
_REPO_ROOT = _SCRIPT_DIR.parent
_TEST_VECTORS_DIR = _REPO_ROOT / "tests" / "vectors"

_X509_DIR = (
    _TEST_VECTORS_DIR
    / "cryptography"
    / "vectors"
    / "cryptography_vectors"
    / "x509"
)
_PKITS_DIR = _X509_DIR / "PKITS_data" / "certs"

# ── CLI flags ─────────────────────────────────────────────────────────────────

_CRITERION_DIR: Path | None = None
if "--save-criterion" in sys.argv:
    _idx = sys.argv.index("--save-criterion")
    _next = sys.argv[_idx + 1] if _idx + 1 < len(sys.argv) else ""
    if _next and not _next.startswith("--"):
        _CRITERION_DIR = Path(_next)
    else:
        _CRITERION_DIR = _REPO_ROOT / "target" / "criterion"

# ── Timing configuration (matches Criterion defaults) ─────────────────────────

WARMUP_TIME_S = 3.0
MEASUREMENT_TIME_S = 5.0


def _measure(fn: Callable[[], object]) -> tuple[float, int, list[float], list[float]]:
    """Run one Criterion Linear-mode measurement, return (avg_us, total_iters, iters, times)."""
    avg_us, iters, times_ns = _criterion_measure(fn, WARMUP_TIME_S, MEASUREMENT_TIME_S)
    return avg_us, sum(int(x) for x in iters), iters, times_ns


def _run(
    group: str,
    function: str,
    value: str,
    fn: Callable[[], object],
) -> None:
    """Time *fn*, print the result, and optionally save Criterion JSON."""
    avg_us, total_iters, iters, times_ns = _measure(fn)
    print(f"{group}/{function}/{value}  avg: {avg_us:.3f} µs  ({total_iters} iterations)")
    if _CRITERION_DIR is not None:
        _criterion_save(_CRITERION_DIR, group, function, value, iters, times_ns)


# ── Benchmark functions ───────────────────────────────────────────────────────

_OID_STR = str(_synta_oids.CT_PRECERT_SCTS)  # 1.3.6.1.4.1.11129.2.4.5


def bench_object_identifier_constructor() -> None:
    """Port of test_object_identifier_constructor."""
    print()
    group = "object_identifier_constructor"
    value = _OID_STR

    _run(group, "synta", value, lambda: synta.ObjectIdentifier(_OID_STR))
    if _HAS_CRYPTOGRAPHY:
        _run(group, "cryptography_x509", value,
             lambda: _cx509.ObjectIdentifier(_OID_STR))


def bench_load_der_certificate(cert_bytes: bytes, value: str) -> None:
    """Port of test_load_der_certificate (plus synta_full variant)."""
    print()
    group = "load_der_certificate"

    _run(group, "synta", value,
         lambda: synta.Certificate.from_der(cert_bytes))
    _run(group, "synta_full", value,
         lambda: synta.Certificate.full_from_der(cert_bytes))
    if _HAS_CRYPTOGRAPHY:
        _run(group, "cryptography_x509", value,
             lambda: _cx509.load_der_x509_certificate(cert_bytes))


def bench_load_pem_certificate(pem_bytes: bytes, value: str) -> None:
    """Port of test_load_pem_certificate."""
    print()
    group = "load_pem_certificate"

    _run(group, "synta", value,
         lambda: synta.Certificate.from_pem(pem_bytes))
    if _HAS_CRYPTOGRAPHY:
        _run(group, "cryptography_x509", value,
             lambda: _cx509.load_pem_x509_certificate(pem_bytes))


def bench_encode_pem_certificate(cert_bytes: bytes, value: str) -> None:
    """Benchmark PEM serialization: Certificate.to_pem vs cert.public_bytes(PEM)."""
    print()
    group = "encode_pem_certificate"

    synta_cert = synta.Certificate.from_der(cert_bytes)
    _run(group, "synta", value,
         lambda: synta.Certificate.to_pem(synta_cert))

    if _HAS_CRYPTOGRAPHY:
        import cryptography.hazmat.primitives.serialization as _ser
        cx509_cert = _cx509.load_der_x509_certificate(cert_bytes)
        _run(group, "cryptography_x509", value,
             lambda: cx509_cert.public_bytes(_ser.Encoding.PEM))


# ── Main ─────────────────────────────────────────────────────────────────────

def main() -> None:
    if not _TEST_VECTORS_DIR.exists():
        print(
            f"ERROR: test vectors directory not found:\n  {_TEST_VECTORS_DIR}\n\n"
            "Run the Criterion benchmarks first to clone certificate repositories:\n"
            "    cargo bench -p synta-bench --bench bindings --no-run",
            file=sys.stderr,
        )
        sys.exit(1)

    if _CRITERION_DIR is not None:
        print(f"Criterion JSON output: {_CRITERION_DIR}", file=sys.stderr)

    # test_object_identifier_constructor
    bench_object_identifier_constructor()

    # test_load_der_certificate + encode_pem_certificate  (PKITS GoodCACert.crt)
    good_ca = _PKITS_DIR / "GoodCACert.crt"
    if good_ca.exists():
        good_ca_bytes = good_ca.read_bytes()
        bench_load_der_certificate(good_ca_bytes, "GoodCACert")
        bench_encode_pem_certificate(good_ca_bytes, "GoodCACert")
    else:
        print(
            f"\nWARNING: {good_ca} not found — skipping load_der_certificate.\n"
            "Run: cargo bench -p synta-bench --bench bindings --no-run",
            file=sys.stderr,
        )

    # test_load_pem_certificate  (cryptography.io.pem)
    cryptography_io = _X509_DIR / "cryptography.io.pem"
    if cryptography_io.exists():
        bench_load_pem_certificate(cryptography_io.read_bytes(), "cryptography.io")
    else:
        print(
            f"\nWARNING: {cryptography_io} not found — skipping load_pem_certificate.",
            file=sys.stderr,
        )

    print()


if __name__ == "__main__":
    main()