# ADR 001: Pure Python/Rust Accelerator Module Compatibility Requirements
Source: https://rampa.git-pull.com/adrs/001-pure-python-rust-accelerator-module-compatibility-requirements/

# ADR 001: Pure Python/Rust Accelerator Module Compatibility Requirements

Status: Accepted
Date: 2026-05-28

## Context

This project is a Python project that may use Rust to accelerate selected modules, functions, or classes.

Native acceleration creates a compatibility risk: the Rust implementation can accidentally become the real implementation, while the Python implementation becomes incomplete, stale, or semantically different. That harms portability, testability, and user trust. It can also make the project harder to install in environments without a Rust toolchain or without compatible binary wheels.

Python's standard library has a similar policy in PEP 399 for pure Python modules with native accelerators. This ADR adapts that idea for this project: pure Python remains the reference implementation, and Rust exists as an optional drop-in accelerator.

## Decision

Every public API must have a pure Python implementation unless this project explicitly grants an exemption.

The pure Python implementation is the semantic source of truth. Rust acceleration may be added for performance, but it must behave as a drop-in replacement for the Python implementation as far as reasonably possible.

The Rust accelerator must pass the same behavioral tests as the pure Python implementation. Rust-specific tests may be added, but they do not replace shared compatibility tests.

The package must remain usable without the Rust extension.

## Scope

This ADR applies to:

- Public functions
- Public classes
- Public methods and attributes
- Public constants whose value or type is part of the API contract
- Public module behavior
- Serialization, equality, hashing, ordering, iteration, context-manager, and async behavior where relevant
- Error behavior that users can observe

This ADR also applies to private Rust code when that code affects public behavior.

## Requirements

### 1. Pure Python first

New public behavior must be implemented in Python before it is accelerated in Rust.

Rust must not be the only implementation of a public API unless an exemption is approved and documented.

Acceptable exemption cases are narrow. Examples include:

- APIs whose only purpose is to expose a Rust-only subsystem.
- Functionality that cannot reasonably be implemented in Python.
- Internal diagnostics, build hooks, or development-only helpers that are not public API.

Exemptions must be documented in the pull request and in the relevant module or package documentation.

### 2. Rust as companion accelerator

Rust acceleration is a companion implementation, not an independent API surface.

Rust may replace selected Python functions, classes, or internals only after the Python implementation has defined the expected public behavior.

Rust must not introduce:

- New public functions
- New public classes
- New public methods or attributes
- New accepted argument forms
- New return shapes
- Different validation rules
- Different mutation side effects
- Different ordering, equality, hashing, or serialization behavior
- Different exception classes for the same invalid input, unless approved and documented

### 3. Optional accelerator

The project must import and run without the Rust extension.

When the Rust extension is unavailable, the package should fall back to Python:

```python
from ._module_py import parse, normalize

_HAS_RUST_ACCELERATOR = False

try:
    from ._native import parse as parse
    from ._native import normalize as normalize
except ImportError:
    pass
else:
    _HAS_RUST_ACCELERATOR = True
```

Fallback code should catch `ImportError`, not broad `Exception`, unless there is a specific and documented reason. Tests should not hide unexpected Rust import failures.

### 4. Shared compatibility tests

Every accelerated API must be tested against both implementations.

The shared test suite must run against:

1. The pure Python implementation with Rust disabled or absent.
2. The Rust-accelerated implementation when Rust is available.

Recommended `pytest` structure:

```python
import pytest

from package_name import _module_py

try:
    from package_name import _native
except ImportError:
    _native = None


@pytest.fixture(params=[_module_py, _native], ids=["python", "rust"])
def impl(request):
    if request.param is None:
        pytest.skip("Rust accelerator is not available")
    return request.param


def test_empty_input(impl):
    assert impl.parse("") == []


def test_invalid_input(impl):
    with pytest.raises(ValueError):
        impl.parse("\x00")
```

Tests must cover the behavior users rely on, including:

- Normal inputs
- Empty inputs
- Boundary values
- Invalid inputs
- Subclasses and duck-typed inputs where relevant
- Mutation and aliasing behavior
- Repeated calls
- Large inputs
- Unicode or binary edge cases where relevant
- Error paths
- Resource cleanup paths
- Serialization, equality, hashing, ordering, iteration, context-manager, or async behavior where relevant

### 5. Duck typing preservation

Rust must preserve the input contract of the Python implementation.

If Python accepts any iterable, mapping, sequence, path-like object, buffer-like object, subclass, or file-like object, Rust must not narrow that behavior to a concrete type only.

Fast paths are allowed, but they must retain a correct generic path.

Acceptable:

```text
Rust uses a fast path for list[str], then falls back to generic iterable handling.
```

Unacceptable:

```text
Python accepts any iterable[str], but Rust accepts only list[str].
```

### 6. Error behavior

Rust must raise the same Python exception classes as the Python implementation wherever practical.

Rust panics must not cross the Python FFI boundary. Internal Rust errors must be converted into Python exceptions.

The compatibility tests must verify important error paths.

### 7. Documentation and type hints

Public documentation describes the public Python API, not the Rust implementation.

Type hints, overloads, and stubs must remain accurate for the public API regardless of whether Rust is installed.

Rust-only signatures must not leak into user-facing documentation or stubs.

### 8. Packaging

The package must remain usable in environments without a Rust compiler or compatible native wheel unless the project explicitly approves a Rust-required feature.

Packaging must support:

- Python-only operation
- Rust-accelerated operation when available
- Clear fallback behavior
- No import-time failure solely because Rust is unavailable

### 9. CI

CI must include both code paths.

Minimum required jobs:

```text
Python-only job:
  - install without Rust or force the Python fallback
  - run the full shared behavioral test suite

Rust-enabled job:
  - build/install the Rust extension
  - run the same shared behavioral test suite
  - run Rust-specific tests, if any
```

The Python-only job is mandatory. A passing Rust-enabled job does not compensate for a failing Python-only job.

### 10. Unsafe Rust

`unsafe` Rust is allowed only when necessary.

Every `unsafe` block must have a nearby `SAFETY:` comment explaining:

1. Why `unsafe` is needed.
2. What invariants make it sound.
3. How those invariants are enforced.
4. Which tests cover the relevant edge cases, when applicable.

Example:

```rust
// SAFETY:
// `idx` is checked against `items.len()` immediately above.
// `items` is not mutated between the bounds check and access.
unsafe {
    items.get_unchecked(idx)
}
```

## Consequences

### Positive consequences

- The project remains portable across environments where Rust is unavailable.
- Users receive the same behavior whether or not acceleration is installed.
- The Python implementation remains complete and useful for debugging, documentation, and alternative runtimes.
- Rust acceleration can be added without creating a second public API.
- CI detects semantic drift between Python and Rust implementations.

### Tradeoffs

- Contributors must maintain two implementations for accelerated behavior.
- Tests must be structured to exercise both paths.
- Some performance optimizations may be rejected if they narrow Python semantics.
- Build and packaging workflows must account for both Python-only and Rust-enabled modes.

### Risks

The main risk is semantic drift: Rust and Python implementations may diverge over time. The mitigation is mandatory shared compatibility testing and Python-first development.

Another risk is hidden fallback: broad exception handling can mask Rust defects. The mitigation is narrow import fallback in runtime code and stricter behavior in tests.

## Implementation guidance

Preferred module layout:

```text
src/
  package_name/
    __init__.py
    module.py          # public API and accelerator selection
    _module_py.py      # pure Python reference implementation
    _native.*          # compiled Rust extension artifact
rust/
  Cargo.toml
  src/
    lib.rs
tests/
  test_module.py
  test_module_compat.py
```

Preferred public-module pattern:

```python
from ._module_py import Token, parse, normalize

_HAS_RUST_ACCELERATOR = False

try:
    from ._native import parse as parse
    from ._native import normalize as normalize
except ImportError:
    pass
else:
    _HAS_RUST_ACCELERATOR = True
```

Public Rust-only names must not be re-exported from the public module.

## Pull request checklist

A pull request that adds or modifies Rust acceleration must confirm:

```text
[ ] Public behavior exists first in pure Python.
[ ] Shared tests cover the Python implementation.
[ ] The same shared tests pass with Rust enabled.
[ ] The package imports and runs without Rust.
[ ] Rust exposes no extra public API.
[ ] Rust preserves duck-typed inputs accepted by Python.
[ ] Rust error behavior matches Python error behavior.
[ ] Type hints and documentation remain accurate.
[ ] Packaging impact is described.
[ ] Benchmarks or a clear performance rationale justify the accelerator.
[ ] Unsafe Rust, if any, is documented with SAFETY comments.
```

## Final position

Rust may make this project faster. Rust must not make the project less Pythonic, less portable, less tested, less predictable, or less compatible.

The Python implementation defines the meaning of the public API. The Rust implementation may make that meaning faster.

---

# Architecture Decision Records
Source: https://rampa.git-pull.com/adrs/

(adrs)=

# Architecture Decision Records

Significant design decisions and their rationale.

```{toctree}
:hidden:

001-pure-python-rust-accelerator-module-compatibility-requirements
```

---

# rampa check
Source: https://rampa.git-pull.com/cli/check/

(cli-check)=

# rampa check

Validate a test script without running it. Discovers scenarios,
validates executor configurations, and reports a summary.

```console
$ rampa check load_test.py
```

## Command

```{eval-rst}
.. argparse::
   :module: rampa.cli
   :func: build_docs_parser
   :prog: rampa
   :path: check
   :nodescription:
```

## Example output

```text
scenarios: 2 found
  - smoke (constant-vus, 5 VUs, 10s)
  - load (ramping-vus)
thresholds: 2 configured
setup: yes
teardown: no
status: valid
```

## What it checks

- Script imports and loads without errors
- `@scenario` decorators are discovered
- Executor names are valid (with fuzzy suggestions for typos)
- Setup and teardown functions are detected
- Threshold expressions are listed

## Exit codes

| Code | Meaning |
|------|---------|
| 0 | Script is valid |
| 1 | Validation error (no scenarios, bad executor, import failure) |
| 2 | File not found |

---

# rampa doctor
Source: https://rampa.git-pull.com/cli/doctor/

(cli-doctor)=

# rampa doctor

Report the runtime environment — Python version, rampa version,
platform, and optional dependency availability.

```console
$ rampa doctor
```

## Command

```{eval-rst}
.. argparse::
   :module: rampa.cli
   :func: build_docs_parser
   :prog: rampa
   :path: doctor
   :nodescription:
```

## Example output

```text
python: 3.14.0
rampa: 0.0.1
platform: linux (x86_64)
aiohttp: 3.13.5
uvloop: not installed
textual: not installed
fastmcp: 3.3.1
```

## Checked dependencies

| Dependency | Extra | Purpose |
|-----------|-------|---------|
| aiohttp | core | HTTP client |
| uvloop | performance | Event loop acceleration |
| textual | tui | TUI dashboard (planned) |
| fastmcp | mcp | MCP server |

---

# CLI
Source: https://rampa.git-pull.com/cli/

(cli)=

# CLI

The `rampa` CLI runs load tests from the terminal and produces
structured output for scripts and CI.

```{cli-install}
```

::::{grid} 1 2 2 4
:gutter: 2 2 3 3

:::{grid-item-card} rampa run
:link: run
:link-type: doc
Execute a load test script with metric output and exit codes.
:::

:::{grid-item-card} rampa check
:link: check
:link-type: doc
Validate a script without running it.
:::

:::{grid-item-card} rampa doctor
:link: doctor
:link-type: doc
Report environment and dependency status.
:::

:::{grid-item-card} CLI Reference
:link: reference
:link-type: doc
Auto-generated reference for all commands and options.
:::

::::

```{toctree}
:hidden:

run
check
doctor
reference
```

---

# CLI Reference
Source: https://rampa.git-pull.com/cli/reference/

(cli-reference)=

# CLI Reference

Auto-generated reference for all `rampa` commands.

## Command: `rampa`

```{eval-rst}
.. argparse::
   :module: rampa.cli
   :func: build_docs_parser
   :prog: rampa
```

## Entry Points

```{eval-rst}
.. autofunction:: rampa.cli.create_parser
.. autofunction:: rampa.cli.build_docs_parser
.. autofunction:: rampa.cli.main
```

---

# rampa run
Source: https://rampa.git-pull.com/cli/run/

(cli-run)=

# rampa run

Execute a load test script.

```console
$ rampa run load_test.py
```

## Command

```{eval-rst}
.. argparse::
   :module: rampa.cli
   :func: build_docs_parser
   :prog: rampa
   :path: run
   :nodescription:
```

## Examples

Run with 20 VUs for 1 minute:

```console
$ rampa run load_test.py --vus 20 --duration 1m
```

Run a specific scenario:

```console
$ rampa run load_test.py --scenario smoke
```

Save results as JSON:

```console
$ rampa run load_test.py --out results.json --quiet
```

Capture the event stream for postmortem analysis:

```console
$ rampa run load_test.py --event-log events.jsonl
```

## Exit codes

| Code | Meaning |
|------|---------|
| 0 | All thresholds passed |
| 1 | Threshold breach |
| 2 | Iteration exception |
| 3 | Invalid configuration |
| 4 | Aborted (SIGINT) |
| 5 | Setup failure |
| 6 | Output failure |
| 7 | Teardown failure |

---

# Benchmarks
Source: https://rampa.git-pull.com/dev/benchmark/

(benchmarks)=

# Benchmarks

Four benchmark scripts measure different layers of the framework.
All produce JSON output for CI regression tracking.

## Throughput

Measures raw iterations-per-second with a no-op worker:

```console
$ uv run python scripts/bench_throughput.py
```

Environment variables: `BENCH_VUS` (default 50), `BENCH_DURATION`
(default 2).

## Scheduler precision

Measures arrival-rate scheduling accuracy — how closely actual
inter-arrival times match the target:

```console
$ uv run python scripts/bench_scheduler.py
```

## Metric engine

Measures sample ingestion rate and snapshot latency:

```console
$ uv run python scripts/bench_metrics.py
```

## HTTP overhead

Measures per-request overhead against a local loopback server:

```console
$ uv run python scripts/bench_http_local.py
```

## JSON output

All benchmarks accept `--json-output <path>` to write structured
results for automated comparison:

```console
$ uv run python scripts/bench_throughput.py --json-output bench.json
```

---

# Development
Source: https://rampa.git-pull.com/dev/

(development)=

# Development

Contributing to rampa and understanding its internals.

## Setup

```console
$ git clone https://github.com/tony/rampa.git
```

```console
$ cd rampa && uv sync --all-extras --all-groups
```

## Quality gates

Run before every commit:

```console
$ rm -rf docs/_build; uv run ruff check . --fix --show-fixes; \
  uv run ruff format .; uv run ty check; \
  uv run py.test --reruns 0 -vvv; just build-docs;
```

## Test suite

```console
$ uv run pytest
```

Continuous testing with pytest-watcher:

```console
$ uv run ptw .
```

## Architecture

```text
User script (@scenario)
        │
    Loader ──→ TestPlan
        │
    Engine ──→ RunController
      │  │         │
      │  │     EventBus ──→ CLI / MCP / pytest / JSONL
      │  │
      │  └──→ Executors (6 types)
      │            │
      │        Workers ──→ HttpClient ──→ target
      │            │
      └──→ MetricEngine (thread)
               │
           SinkProtocol ──→ Thresholds ──→ exit code
```

The engine is headless — it owns execution and cleanup. Frontends
own presentation, format, and exit behavior. The `EventBus`
broadcasts typed events to concurrent subscribers.

The metric engine runs in a dedicated `threading.Thread`, draining
samples from a `queue.SimpleQueue` on a 50ms timer. The
`SinkProtocol` is a structural protocol (not ABC) designed as the
future Rust/PyO3 seam.

::::{grid} 1 1 2 2
:gutter: 2

:::{grid-item-card} Benchmarks
:link: benchmark
:link-type: doc
Throughput, scheduling, metric engine, and HTTP benchmarks.
:::

:::{grid-item-card} ADRs
:link: ../adrs/index
:link-type: doc
Architecture decision records — significant design choices and rationale.
:::

::::

```{toctree}
:hidden:

benchmark
../adrs/index
```

---

# Getting Started
Source: https://rampa.git-pull.com/getting-started/

(getting-started)=

# Getting Started

Write and run your first load test in 60 seconds.

## Install

```{library-install}
```

Verify:

```console
$ rampa doctor
```

## Write a scenario

Create `load_test.py`:

```python
import asyncio
import rampa


@rampa.scenario(executor="constant-vus", vus=5, duration="10s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://httpbin.org/get")
    worker.check(resp, {
        "status is 200": lambda r: r.status == 200,
    })
```

## Run it

```console
$ rampa run load_test.py
```

The console summary shows iteration count, request timing percentiles
(p90, p95, p99), check pass/fail rates, and data transfer totals.

## Add thresholds

Thresholds enforce performance criteria. A breach produces exit code 1.

```python
config = rampa.Config(
    thresholds={
        "http_req_duration": ["p(95)<500"],
        "http_req_failed": ["rate<0.01"],
    },
)
```

## Save results

```console
$ rampa run load_test.py --out results.json
```

```console
$ rampa run load_test.py --event-log events.jsonl
```

## Next steps

- {doc}`../cli/index` — all CLI flags and commands
- {doc}`../library/executors` — choosing the right executor
- {doc}`../library/metrics` — built-in and custom metrics
- {doc}`../pytest/index` — load tests in your test suite

---

# Changelog
Source: https://rampa.git-pull.com/history/

(changes)=
(changelog)=
(history)=

```{include} ../CHANGES

```

---

# rampa
Source: https://rampa.git-pull.com/

(index)=

# rampa

Async Python load testing, inspired by k6.

Write an async scenario function, run it, get request metrics with
percentiles, checks, thresholds, and correct exit codes.

```{cli-install}
:variant: compact
```

```{mcp-install}
:variant: compact
```

## Your first load test

```python
import asyncio
import rampa

@rampa.scenario(executor="constant-vus", vus=10, duration="30s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://httpbin.org/get")
    worker.check(resp, {"status is 200": lambda r: r.status == 200})
```

```console
$ rampa run load_test.py
```

::::{grid} 1 1 2 3
:gutter: 2 2 3 3

:::{grid-item-card} Quickstart
:link: getting-started/index
:link-type: doc
Write and run your first scenario in 60 seconds.
:::

:::{grid-item-card} CLI
:link: cli/index
:link-type: doc
`rampa run`, `rampa check`, `rampa doctor` from the terminal.
:::

:::{grid-item-card} Library
:link: library/index
:link-type: doc
Executors, metrics, thresholds, and the Python API.
:::

:::{grid-item-card} pytest
:link: pytest/index
:link-type: doc
Run load tests inside your test suite with `@pytest.mark.rampa_scenario`.
:::

:::{grid-item-card} MCP
:link: mcp/index
:link-type: doc
Start, stop, and query load tests from AI agents.
:::

:::{grid-item-card} Development
:link: dev/index
:link-type: doc
Contributing, benchmarks, and architecture.
:::

::::

## What you get

### Six executor types

Closed-model (VU-based) and open-model (arrival-rate) scheduling,
matching k6's executor vocabulary.

| Executor | Model | Use when |
|----------|-------|----------|
| `constant-vus` | Closed | Fixed concurrency for a duration |
| `ramping-vus` | Closed | Ramp concurrency up/down through stages |
| `shared-iterations` | Closed | Run exactly N total iterations |
| `per-vu-iterations` | Closed | Each VU runs exactly N iterations |
| `constant-arrival-rate` | Open | Maintain a fixed request rate |
| `ramping-arrival-rate` | Open | Ramp request rate through stages |

### Automatic HTTP metrics

Every HTTP request auto-emits timing metrics with per-phase
decomposition (blocked, connecting, sending, waiting, receiving),
failure classification, and data transfer counters.

### Threshold expressions

```python
config = rampa.Config(
    thresholds={
        "http_req_duration": ["p(95)<500", "avg<200"],
        "http_req_failed": ["rate<0.01"],
    },
)
```

Threshold breaches produce exit code 1 for CI integration.

### Multiple frontends and outputs

- **CLI** — `rampa run` with `--progress` live status or `--tui` dashboard
- **TUI** — interactive Textual dashboard with live metrics and keyboard control
- **pytest plugin** — `@pytest.mark.rampa_scenario` for test suites
- **unittest mixin** — `RampaTestCase` for unittest integration
- **MCP server** — `rampa-mcp` for AI agent integration
- **Output backends** — CSV, InfluxDB, Prometheus, OTEL, webhooks via `--output`
- **CI comparison** — `python -m rampa.ci.compare` for benchmark diffs

```{toctree}
:hidden:

getting-started/index
cli/index
library/index
pytest/index
mcp/index
dev/index
history
GitHub <https://github.com/tony/rampa>
```

---

# CI integration
Source: https://rampa.git-pull.com/library/ci/

(ci)=

# CI integration

rampa integrates with CI pipelines for automated benchmarking and
performance regression detection.

## GitHub Action

Use the built-in composite action:

```yaml
- uses: tony/rampa/.github/actions/rampa-benchmark@main
  with:
    script: tests/load_test.py
    vus: 10
    duration: 30s
```

The action:
1. Installs rampa
2. Runs the load test
3. Uploads results as an artifact
4. Writes a markdown summary to `$GITHUB_STEP_SUMMARY`
5. Fails the step if thresholds breach (configurable)

## Result comparison

Compare two result files from the CLI:

```console
$ python -m rampa.ci.compare \
  --baseline baseline.json \
  --current current.json \
  --format markdown
```

Output formats: `text`, `markdown` (for PR comments), `json`.

Metrics with >5% degradation are flagged as regressions.

## GitHub Actions output backend

Add the `github` output backend to emit threshold annotations:

```console
$ rampa run load_test.py --output github
```

When running in GitHub Actions (`GITHUB_ACTIONS=true`), this emits:
- `::error::` annotations for failed thresholds
- A markdown summary to `$GITHUB_STEP_SUMMARY`

---

# Distributed execution
Source: https://rampa.git-pull.com/library/distributed/

(distributed)=

# Distributed execution

rampa supports splitting load tests across multiple machines for
higher throughput. The coordinator manages workers and aggregates
metrics centrally.

## Architecture

```
Coordinator (your machine)
  ├── MetricEngine (aggregated)
  ├── Threshold evaluation (centralized)
  └── WebSocket server
       ├── Worker 0 (SSH / Lambda / ECS)
       ├── Worker 1
       └── Worker N
```

Workers connect to the coordinator, receive work segments, run the
test locally, and stream samples back.

## Execution segments

Work is deterministically partitioned across workers without central
assignment:

```python
from rampa.distributed.segment import ExecutionSegment

seg = ExecutionSegment(index=0, total=3)
seg.vu_range(30)       # range(0, 10)
seg.scale_rate(1000.0)  # 333.3
```

Each worker independently computes its share from its index and the
total worker count.

## Test archives

Self-contained `.rampa` zip bundles contain everything a remote
worker needs:

```console
$ rampa archive create load_test.py -o test.rampa
```

Contents: script, data files, `requirements.txt`, `manifest.json`.

```python
from rampa.distributed.archive import create_archive, extract_archive

create_archive("load_test.py", "test.rampa", requirements=["aiohttp>=3.9"])
manifest = extract_archive("test.rampa", "/tmp/work")
```

## Wire protocol

Coordinator and workers communicate via WebSocket using MessagePack
(JSON fallback). Message types: `register`, `assign`, `samples`,
`stop`, `heartbeat_req/resp`, `threshold_breach`.

---

# Executors
Source: https://rampa.git-pull.com/library/executors/

(executors)=

# Executors

Executors control how iterations are scheduled — how many virtual
users run, for how long, and at what rate. rampa provides six executor
types matching k6's scheduling vocabulary.

## Which executor should I use?

| I want to... | Executor | Model |
|--------------|----------|-------|
| Run N users for a fixed duration | `constant-vus` | Closed |
| Ramp users up and down | `ramping-vus` | Closed |
| Run exactly N total iterations | `shared-iterations` | Closed |
| Run N iterations per user | `per-vu-iterations` | Closed |
| Maintain a fixed request rate | `constant-arrival-rate` | Open |
| Ramp request rate up and down | `ramping-arrival-rate` | Open |

**Closed model** — each VU waits for the previous iteration to finish
before starting the next. The system under test controls the effective
rate. Use this when you want to simulate a fixed number of concurrent
users.

**Open model** — iterations start at a fixed rate regardless of
response time. If the system slows down, VUs pile up. Use this when
you want to measure behavior under a specific request rate. If VU
capacity is exhausted, the iteration is counted as `dropped_iterations`.

## constant-vus

Run a fixed number of VUs for a duration.

```python
@rampa.scenario(executor="constant-vus", vus=10, duration="30s")
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

## ramping-vus

Linearly interpolate VU count between stages.

```python
@rampa.scenario(
    executor="ramping-vus",
    stages=[
        rampa.Stage(duration="30s", target=50),
        rampa.Stage(duration="1m", target=100),
        rampa.Stage(duration="30s", target=0),
    ],
)
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

## shared-iterations

N VUs share a pool of M total iterations.

```python
@rampa.scenario(
    executor="shared-iterations",
    vus=10,
    iterations=1000,
)
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

## per-vu-iterations

Each VU runs exactly N iterations independently.

```python
@rampa.scenario(
    executor="per-vu-iterations",
    vus=10,
    iterations=100,
)
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

## constant-arrival-rate

Maintain a fixed request rate. Iterations that cannot start (all VUs
busy) are counted as `dropped_iterations`.

```python
@rampa.scenario(
    executor="constant-arrival-rate",
    rate=100,
    duration="1m",
    pre_allocated_vus=10,
    max_vus=50,
)
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

## ramping-arrival-rate

Ramp the request rate linearly through stages.

```python
@rampa.scenario(
    executor="ramping-arrival-rate",
    stages=[
        rampa.Stage(duration="30s", target=50),
        rampa.Stage(duration="1m", target=200),
        rampa.Stage(duration="30s", target=0),
    ],
    pre_allocated_vus=10,
    max_vus=100,
)
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

---

# Library
Source: https://rampa.git-pull.com/library/

(library)=

# Library

Use rampa as a Python library to build, configure, and run load tests
programmatically.

```{library-install}
```

::::{grid} 1 1 2 2
:gutter: 2

:::{grid-item-card} Tutorial
:link: tutorial
:link-type: doc
Write your first scenario, add checks and thresholds.
:::

:::{grid-item-card} Executors
:link: executors
:link-type: doc
Choose the right scheduling model for your test.
:::

:::{grid-item-card} Metrics
:link: metrics
:link-type: doc
Built-in metrics, custom metrics, and the metric pipeline.
:::

:::{grid-item-card} Thresholds
:link: thresholds
:link-type: doc
Expression syntax and pass/fail evaluation.
:::

:::{grid-item-card} Outputs
:link: outputs
:link-type: doc
Send results to CSV, InfluxDB, Prometheus, OTEL, and more.
:::

:::{grid-item-card} TUI Dashboard
:link: tui
:link-type: doc
Live terminal dashboard with metrics, timing, and thresholds.
:::

:::{grid-item-card} Protocols
:link: protocols
:link-type: doc
HTTP, WebSocket, gRPC, and custom protocol clients.
:::

:::{grid-item-card} Distributed
:link: distributed
:link-type: doc
Split tests across machines with execution segments.
:::

:::{grid-item-card} CI Integration
:link: ci
:link-type: doc
GitHub Actions, result comparison, and benchmarking.
:::

:::{grid-item-card} API Reference
:link: reference
:link-type: doc
Public API: Engine, RunController, Worker, Config.
:::

::::

```{toctree}
:hidden:

tutorial
executors
metrics
thresholds
outputs
tui
protocols
distributed
ci
reference
```

---

# Metrics
Source: https://rampa.git-pull.com/library/metrics/

(metrics)=

# Metrics

rampa automatically collects metrics for every iteration and HTTP
request. You can also emit custom metrics from your scenario code.

## Built-in metrics

### Execution metrics

| Metric | Type | Description |
|--------|------|-------------|
| `iterations` | Counter | Completed iterations |
| `iteration_duration` | Trend | Time per iteration (ms) |
| `iteration_errors` | Counter | Failed iterations |
| `dropped_iterations` | Counter | Skipped iterations (arrival-rate) |
| `vus` | Gauge | Active virtual users |
| `vus_max` | Gauge | Peak virtual users |

### HTTP metrics

| Metric | Type | Description |
|--------|------|-------------|
| `http_reqs` | Counter | Completed HTTP requests |
| `http_req_duration` | Trend | Total request time (ms) |
| `http_req_failed` | Rate | Failure ratio |
| `data_sent` | Counter | Bytes sent |
| `data_received` | Counter | Bytes received |

### HTTP phase timing

| Metric | Type | Description |
|--------|------|-------------|
| `http_req_blocked` | Trend | DNS + connection queue (ms) |
| `http_req_connecting` | Trend | TCP connection (ms) |
| `http_req_sending` | Trend | Request send (ms) |
| `http_req_waiting` | Trend | Time to first byte (ms) |
| `http_req_receiving` | Trend | Response receive (ms) |

### Check metrics

| Metric | Type | Description |
|--------|------|-------------|
| `checks` | Rate | Check pass ratio |

## Metric types

| Type | Tracks | Aggregations |
|------|--------|-------------|
| **Counter** | Cumulative total | count, rate |
| **Gauge** | Current value | value, min, max |
| **Rate** | Success ratio | rate, passes, fails |
| **Trend** | Distribution | count, avg, min, max, med, p(90), p(95), p(99) |

## Custom metrics

Emit custom metrics from your scenario:

```python
@rampa.scenario(executor="constant-vus", vus=5, duration="30s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://api.example.com/items")
    items = resp.json()

    worker.counter("api_calls")
    worker.gauge("items_returned", float(len(items)))
    worker.trend("processing_time", 42.5)
```

Custom metrics appear in the console summary and JSON output alongside
built-in metrics.

---

# Output backends
Source: https://rampa.git-pull.com/library/outputs/

(outputs)=

# Output backends

rampa ships metric samples to output backends during and after a test run.
Use `--output` to send results to multiple destinations simultaneously.

## Built-in backends

| Backend | Destination | Dependencies |
|---------|------------|-------------|
| `console` | Terminal summary (default) | None |
| `json` | JSON file | None |
| `csv` | CSV file | None |
| `influxdb` | InfluxDB HTTP API | aiohttp (included) |
| `prometheus` | Prometheus remote write | aiohttp; optional `python-snappy` |
| `otel` | OpenTelemetry collector | aiohttp (included) |
| `webhook` | Any HTTP endpoint | aiohttp (included) |

## CLI usage

```console
$ rampa run load_test.py --output csv=results.csv
```

Multiple outputs in one run:

```console
$ rampa run load_test.py \
  --output csv=results.csv \
  --output influxdb=http://localhost:8086/api/v2/write?org=myorg&bucket=rampa
```

The `--out` flag is shorthand for `--output json=<path>`.

## CSV

One row per sample. Tag keys become columns.

```console
$ rampa run load_test.py --output csv=metrics.csv
```

Output:

```
timestamp,metric,value,method,scenario,status
1716691200,http_reqs,1.0,GET,smoke,200
1716691201,http_req_duration,45.2,GET,smoke,200
```

## InfluxDB

Pushes samples as [line protocol](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/) over HTTP.

```console
$ rampa run load_test.py \
  --output influxdb=http://localhost:8086/api/v2/write?org=myorg&bucket=rampa
```

Each sample becomes one line:

```
http_req_duration,method=GET,scenario=smoke value=45.2 1716691200000000000
```

Tags map to InfluxDB tags, the sample value maps to the `value` field,
and the monotonic nanosecond timestamp maps to the InfluxDB timestamp.

## Prometheus

Pushes metrics to Prometheus via the remote write API. Uses hand-crafted
protobuf v1 encoding with snappy compression (falls back to gzip if
`python-snappy` is not installed).

```console
$ rampa run load_test.py \
  --output prometheus=http://localhost:9090/api/v1/write
```

Each sample becomes a Prometheus TimeSeries with `__name__` set to the
metric name and sample tags as labels. Feeds Grafana dashboards directly.

Install `python-snappy` for optimal compression:

```console
$ uv add python-snappy
```

## OpenTelemetry

Exports metrics via OTLP/HTTP+JSON to any OpenTelemetry-compatible
collector (Grafana Alloy, OTEL Collector, Jaeger, etc.). Zero additional
dependencies.

```console
$ rampa run load_test.py --output otel=http://localhost:4318
```

The `/v1/metrics` path is appended automatically. Uses the JSON wire
format (proto3 standard JSON mapping) — no protobuf compiler needed.

## Webhook

POST sample batches as JSON to any HTTP endpoint.

```console
$ rampa run load_test.py --output webhook=https://example.com/hook
```

Payload shape:

```json
{
  "samples": [
    {"metric": "http_reqs", "value": 1.0, "timestamp": 1716691200, "tags": {"method": "GET"}}
  ]
}
```

## Programmatic usage

```python
from rampa.outputs import get_output

csv_out = get_output("csv", "results.csv")
influx_out = get_output("influxdb", "http://localhost:8086/api/v2/write")
```

All outputs implement the {class}`~rampa.output.Output` protocol:
`start()`, `add_samples(batch)`, `stop(error)`.

---

# Protocol clients
Source: https://rampa.git-pull.com/library/protocols/

(protocols)=

# Protocol clients

rampa provides protocol-specific clients that auto-emit metrics.
Each client is lazily initialized via a Worker property.

## HTTP (built-in)

```python
@rampa.scenario(vus=10, duration="30s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://example.com/api")
    worker.check(resp, {"status 200": lambda r: r.status == 200})
```

Metrics: `http_reqs`, `http_req_duration`, `http_req_failed`, phase timings.

## WebSocket (built-in)

```python
@rampa.scenario(vus=50, duration="1m")
async def websocket_load(worker: rampa.Worker) -> None:
    async with worker.ws.connect("wss://echo.example.com") as session:
        await session.send('{"type": "ping"}')
        response = await session.receive()
        worker.check(response, {"got pong": lambda r: "pong" in r})
```

Metrics: `ws_sessions`, `ws_connecting`, `ws_session_duration`,
`ws_messages_sent`, `ws_messages_received`, `ws_errors`.

## gRPC (optional)

Install: `pip install rampa[grpc]`

```python
@rampa.scenario(vus=20, duration="30s")
async def grpc_load(worker: rampa.Worker) -> None:
    resp = await worker.grpc.unary(
        "localhost:50051",
        "/myservice.MyService/GetUser",
        request=b"\x08\x01",  # serialized protobuf
    )
    worker.check(resp, {"ok": lambda r: r.ok})
```

Metrics: `grpc_reqs`, `grpc_req_duration`, `grpc_req_failed`,
`grpc_streams_opened`, `grpc_messages_received`.

Supports `unary()` and `server_stream()` call patterns.

## Custom protocols

For protocols without a built-in client, use raw async code with
custom metric emission:

```python
@rampa.scenario(vus=10, duration="30s")
async def tcp_load(worker: rampa.Worker) -> None:
    import asyncio, time

    start = time.monotonic()
    reader, writer = await asyncio.open_connection("localhost", 9999)
    worker.trend("tcp_connect", (time.monotonic() - start) * 1000)

    writer.write(b"PING\n")
    await writer.drain()
    data = await reader.readline()

    worker.trend("tcp_roundtrip", (time.monotonic() - start) * 1000)
    worker.counter("tcp_messages")
    worker.check(data, {"got pong": lambda d: d.strip() == b"PONG"})

    writer.close()
    await writer.wait_closed()
```

---

# API Reference
Source: https://rampa.git-pull.com/library/reference/

(api-reference)=

# API Reference

## Configuration

```{eval-rst}
.. automodule:: rampa.config
   :members: Config, ScenarioConfig, Stage, Options, parse_duration
```

## Engine

```{eval-rst}
.. automodule:: rampa.engine
   :members: Engine, EngineOptions, RunController
```

## Events

```{eval-rst}
.. automodule:: rampa.events
   :members: RunStatus, RunResult, EngineEvent, PhaseEvent, SnapshotEvent, ThresholdEvent
```

## Worker

```{eval-rst}
.. automodule:: rampa.worker
   :members: Worker, ExecutionInfo
```

## HTTP Client

```{eval-rst}
.. automodule:: rampa.http
   :members: HttpClient, Response
```

## Metrics

```{eval-rst}
.. automodule:: rampa.metrics
   :members: MetricSnapshot, MetricRegistry, MetricEngine, SinkProtocol
```

## Thresholds

```{eval-rst}
.. automodule:: rampa.thresholds
   :members: ThresholdResult, parse_threshold, evaluate_thresholds
```

## Loader

```{eval-rst}
.. automodule:: rampa.loader
   :members: TestPlan, scenario, load_test
```

---

# Thresholds
Source: https://rampa.git-pull.com/library/thresholds/

(thresholds)=

# Thresholds

Thresholds define pass/fail criteria for your load test. When any
threshold breaches, `rampa run` exits with code 1.

## Syntax

```
<aggregation>[(<parameter>)] <operator> <value>
```

## Aggregation functions

| Function | Metric types | Description |
|----------|-------------|-------------|
| `avg` | Trend | Mean value |
| `min` | Trend | Minimum value |
| `max` | Trend | Maximum value |
| `med` | Trend | Median (p50) |
| `p(N)` | Trend | Nth percentile (e.g. `p(95)`) |
| `count` | Counter | Total count |
| `rate` | Rate | Pass ratio (0.0–1.0) |
| `value` | Gauge | Current value |

## Operators

`<`, `<=`, `>`, `>=`, `==`, `!=`

## Examples

```python
config = rampa.Config(
    thresholds={
        "http_req_duration": [
            "p(95)<500",    # 95th percentile under 500ms
            "avg<200",      # Average under 200ms
            "max<2000",     # No request over 2 seconds
        ],
        "http_req_failed": [
            "rate<0.01",    # Less than 1% failure rate
        ],
        "checks": [
            "rate>0.99",    # At least 99% of checks pass
        ],
        "iterations": [
            "count>=100",   # At least 100 iterations completed
        ],
    },
)
```

## In a script

Define `config` at module level:

```python
import rampa

config = rampa.Config(
    thresholds={
        "http_req_duration": ["p(95)<500"],
    },
)


@rampa.scenario(executor="constant-vus", vus=10, duration="30s")
async def default(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/data")
```

---

# TUI dashboard
Source: https://rampa.git-pull.com/library/tui/

(tui)=

# TUI dashboard

rampa includes a live terminal dashboard for real-time monitoring
during load tests.

## Installation

```console
$ pip install rampa[tui]
```

## Usage

```console
$ rampa run load_test.py --tui
```

The dashboard shows:

- **Phase indicator** — setup, executing, paused, teardown, complete
- **Execution metrics** — VU count, iteration count/rate, error count, duration
- **HTTP timing** — request count/rate, p50/p90/p95/p99 latency
- **Threshold status** — pass/fail for each configured threshold

## Keyboard shortcuts

| Key | Action |
|-----|--------|
| `q` | Quit |
| `p` | Pause / Resume |
| `s` | Stop test |

## Progressive display options

rampa offers three levels of live output:

```console
$ rampa run load_test.py                # Console summary after completion
```

```console
$ rampa run load_test.py --progress     # Single-line live status on stderr
```

```console
$ rampa run load_test.py --tui          # Full interactive dashboard
```

The `--progress` flag requires no additional dependencies.
The `--tui` flag requires `textual>=3.0` (installed via `rampa[tui]`).

---

# Tutorial
Source: https://rampa.git-pull.com/library/tutorial/

(tutorial)=

# Tutorial

Build a load test step by step — from a minimal scenario to a
complete test with checks, thresholds, and structured output.

## Minimal scenario

A rampa scenario is an async function decorated with `@rampa.scenario`.
It receives a {class}`~rampa.Worker` and runs one iteration of your
workload.

```python
import asyncio
import rampa


@rampa.scenario(executor="constant-vus", vus=1, duration="5s")
async def default(worker: rampa.Worker) -> None:
    await asyncio.sleep(0.01)
```

```console
$ rampa run load_test.py
```

## HTTP requests

The worker provides an HTTP client that auto-emits timing metrics:

```python
@rampa.scenario(executor="constant-vus", vus=5, duration="30s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://httpbin.org/get")
```

Every request emits `http_reqs`, `http_req_duration`, `http_req_failed`,
data transfer counters, and per-phase timing metrics automatically.

## Checks

Checks validate response properties. Each condition emits a pass/fail
sample on the `checks` metric:

```python
@rampa.scenario(executor="constant-vus", vus=5, duration="30s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://httpbin.org/get")
    worker.check(resp, {
        "status is 200": lambda r: r.status == 200,
        "body is JSON": lambda r: r.json() is not None,
    })
```

## Thresholds

Thresholds define pass/fail criteria. A breach produces exit code 1:

```python
config = rampa.Config(
    thresholds={
        "http_req_duration": ["p(95)<500", "avg<200"],
        "http_req_failed": ["rate<0.01"],
        "checks": ["rate>0.99"],
    },
)
```

## Custom metrics

Emit your own counters, gauges, and trends:

```python
@rampa.scenario(executor="constant-vus", vus=5, duration="30s")
async def default(worker: rampa.Worker) -> None:
    resp = await worker.http.get("https://api.example.com/items")
    items = resp.json()
    worker.gauge("items_returned", float(len(items)))
    worker.counter("api_calls")
```

## Setup and teardown

Module-level `setup()` and `teardown()` functions run once:

```python
async def setup():
    return {"token": "abc123"}


async def teardown():
    pass


@rampa.scenario(executor="constant-vus", vus=5, duration="30s")
async def default(worker: rampa.Worker) -> None:
    token = worker.setup_data["token"]
    await worker.http.get(
        "https://api.example.com/data",
        headers={"Authorization": f"Bearer {token}"},
    )
```

## Multiple scenarios

A single script can define multiple scenarios:

```python
@rampa.scenario(
    name="smoke",
    executor="constant-vus",
    vus=1,
    duration="10s",
)
async def smoke(worker: rampa.Worker) -> None:
    await worker.http.get("https://api.example.com/health")


@rampa.scenario(
    name="load",
    executor="ramping-vus",
    stages=[
        rampa.Stage(duration="30s", target=50),
        rampa.Stage(duration="1m", target=100),
        rampa.Stage(duration="30s", target=0),
    ],
)
async def load(worker: rampa.Worker) -> None:
    await worker.http.post(
        "https://api.example.com/data",
        json={"key": "value"},
    )
```

Run a specific scenario:

```console
$ rampa run load_test.py --scenario smoke
```

---

# MCP Server
Source: https://rampa.git-pull.com/mcp/

(mcp)=

# MCP Server

The `rampa-mcp` server lets AI agents start, stop, and query load
tests via the Model Context Protocol.

```{mcp-install}
```

::::{grid} 1 1 3 3
:gutter: 2 2 3 3

:::{grid-item-card} Tools
:link: tools
:link-type: doc
Start, stop, and query load test runs.
:::

:::{grid-item-card} Resources
:link: resources
:link-type: doc
URI templates for runs, metrics, and thresholds.
:::

:::{grid-item-card} API Reference
:link: reference
:link-type: doc
Server factory, run registry, and models.
:::

::::

## What you can do

### Load Testing

Start and manage load test runs from AI agents.
{ref}`fastmcp-tool-start-run` · {ref}`fastmcp-tool-stop-run` · {ref}`fastmcp-tool-get-status` · {ref}`fastmcp-tool-list-runs`

### Observability

Query metrics and threshold results for completed runs.
{ref}`fastmcp-tool-get-metrics` · {ref}`fastmcp-tool-get-thresholds`

```{toctree}
:hidden:

tools
resources
reference
```

---

# API Reference
Source: https://rampa.git-pull.com/mcp/reference/

(mcp-reference)=

# API Reference

FastMCP server factory, run registry, event models, and
configuration types used by the MCP tools and resources.

## Server

```{eval-rst}
.. autofunction:: rampa.mcp.server.build_mcp_server
.. autofunction:: rampa.mcp.server.main
```

## Registry

```{eval-rst}
.. autoclass:: rampa.mcp.registry.RunRecord
   :members:

.. autoclass:: rampa.mcp.registry.RuntimeRun
   :members:

.. autoclass:: rampa.mcp.registry.RunRegistry
   :members:
```

## Events

```{eval-rst}
.. autoclass:: rampa.events.RunResult
   :members:
   :no-index:

.. autoclass:: rampa.events.RunStatus
   :members:
   :undoc-members:
   :no-index:

.. autoclass:: rampa.events.PhaseEvent
   :members:
   :no-index:

.. autoclass:: rampa.events.SnapshotEvent
   :members:
   :no-index:

.. autoclass:: rampa.events.ThresholdEvent
   :members:
   :no-index:
```

## Configuration

```{eval-rst}
.. autoclass:: rampa.config.Config
   :members:
   :no-index:

.. autoclass:: rampa.config.ScenarioConfig
   :members:
   :no-index:

.. autoclass:: rampa.config.Stage
   :members:
   :no-index:
```

## Metrics

```{eval-rst}
.. autoclass:: rampa.metrics.MetricSnapshot
   :members:
   :no-index:

.. autoclass:: rampa.thresholds.ThresholdResult
   :members:
   :no-index:

.. autoclass:: rampa.thresholds.ThresholdExpression
   :members:
   :no-index:
```

---

# Resources
Source: https://rampa.git-pull.com/mcp/resources/

(mcp-resources)=

# Resources

MCP resources expose passive read-only data at `rampa://` URIs.
Clients read them with `resources/read`.

::::{grid} 1 2 3 3
:gutter: 2 2 3 3

:::{grid-item-card} All Runs
:link: fastmcp-resource-all-runs
:link-type: ref
List every active and completed run.
:::

:::{grid-item-card} Run Details
:link: fastmcp-resource-template-run-details
:link-type: ref
Status, script path, event count for one run.
:::

:::{grid-item-card} Run Metrics
:link: fastmcp-resource-template-run-metrics
:link-type: ref
All metrics for a run.
:::

:::{grid-item-card} Specific Metric
:link: fastmcp-resource-template-run-metric
:link-type: ref
Single metric by name.
:::

:::{grid-item-card} Threshold Results
:link: fastmcp-resource-template-run-thresholds
:link-type: ref
Pass/fail results for each threshold expression.
:::

:::{grid-item-card} Event Log
:link: fastmcp-resource-template-run-events
:link-type: ref
Accumulated event history for a run.
:::

::::

## All runs

```{fastmcp-resource} all_runs
```

Read `rampa://runs` to list every active and completed run with
their `run_id`, `script_path`, and current status.

## Run details

```{fastmcp-resource-template} run_details
```

Read `rampa://runs/{run_id}` for status, script path, completion
state, and event count.

## Run metrics

```{fastmcp-resource-template} run_metrics
```

Read `rampa://runs/{run_id}/metrics` for all metric values including
timing percentiles, counters, and rates.

## Specific metric

```{fastmcp-resource-template} run_metric
```

Read `rampa://runs/{run_id}/metrics/{name}` to retrieve a single
metric by name (e.g. `http_req_duration`, `http_reqs`).

## Threshold results

```{fastmcp-resource-template} run_thresholds
```

Read `rampa://runs/{run_id}/thresholds` for pass/fail results with
`source`, `passed`, `lhs` (actual), and `rhs` (expected) for each
threshold expression.

## Event log

```{fastmcp-resource-template} run_events
```

Read `rampa://runs/{run_id}/events` for the accumulated event
history including `PhaseEvent`, `SnapshotEvent`, and
`ThresholdEvent` entries.

---

# Tools
Source: https://rampa.git-pull.com/mcp/tools/

(mcp-tools)=

# Tools

The rampa MCP server provides six tools for load test lifecycle
management, metric retrieval, and threshold evaluation.

::::{grid} 1 2 3 3
:gutter: 2 2 3 3

:::{grid-item-card} Start Run
:link: fastmcp-tool-start-run
:link-type: ref
Start a new load test from a script path.
:::

:::{grid-item-card} Stop Run
:link: fastmcp-tool-stop-run
:link-type: ref
Gracefully stop a running test.
:::

:::{grid-item-card} Get Status
:link: fastmcp-tool-get-status
:link-type: ref
Poll whether a run is active or completed.
:::

:::{grid-item-card} List Runs
:link: fastmcp-tool-list-runs
:link-type: ref
List all active and completed runs.
:::

:::{grid-item-card} Get Metrics
:link: fastmcp-tool-get-metrics
:link-type: ref
Retrieve metric snapshots with percentiles.
:::

:::{grid-item-card} Get Thresholds
:link: fastmcp-tool-get-thresholds
:link-type: ref
Evaluate threshold pass/fail results.
:::

::::

## Run Lifecycle

```{fastmcp-tool} start_run
```

Start a new load test from a script path. Returns the `run_id` and
initial status.

```{fastmcp-tool-input} start_run
```

```{fastmcp-tool} stop_run
```

Gracefully stop a running test. Idempotent — calling on a completed
run returns `already_completed`.

```{fastmcp-tool-input} stop_run
```

```{fastmcp-tool} get_status
```

Poll whether a run is still active or has completed.

```{fastmcp-tool-input} get_status
```

```{fastmcp-tool} list_runs
```

List all active and completed runs with their `run_id`, `status`,
and `script_path`.

## Metrics

```{fastmcp-tool} get_metrics
```

Retrieve the latest metric snapshot for a run. Without a
`metric_name` filter, returns all metrics including timing
percentiles, counters, and rates.

```{fastmcp-tool-input} get_metrics
```

## Thresholds

```{fastmcp-tool} get_thresholds
```

Evaluate threshold results for a completed run. Returns pass/fail
status with actual vs expected values for each threshold expression.

```{fastmcp-tool-input} get_thresholds
```

---

# pytest Plugin
Source: https://rampa.git-pull.com/pytest/

(pytest-plugin)=

# pytest Plugin

Run load tests inside your existing test suite.

::::{grid} 1 1 2 2
:gutter: 2

:::{grid-item-card} API Reference
:link: reference
:link-type: doc
Function and class reference.
:::

::::

## Install

rampa registers as a pytest plugin via the `pytest11` entry point.
No configuration needed — install rampa and the plugin activates.

## Fixtures

```{eval-rst}
.. autofixture:: rampa.pytest_plugin.rampa_result
```

## Quick start

```python
import pytest
from rampa.events import RunResult, RunStatus
from rampa.worker import Worker


async def my_worker(w: Worker) -> None:
    await w.http.get("https://httpbin.org/get")


@pytest.mark.rampa_scenario(
    executor="constant-vus",
    vus=2,
    duration="500ms",
    worker_fn=my_worker,
)
def test_api_performance(rampa_result: RunResult) -> None:
    assert rampa_result.status == RunStatus.PASSED
```

## Marker: `@pytest.mark.rampa_scenario`

The marker accepts the same keyword arguments as
{class}`~rampa.config.ScenarioConfig` plus:

| Kwarg | Type | Description |
|-------|------|-------------|
| `worker_fn` | async callable | The scenario function (required) |
| `thresholds` | dict | Threshold expressions per metric |

All `ScenarioConfig` fields work: `executor`, `vus`, `duration`,
`iterations`, `stages`, `rate`, `max_vus`.

## Complete example

```python
import asyncio
import pytest
from rampa.events import RunResult, RunStatus
from rampa.worker import Worker


async def api_worker(w: Worker) -> None:
    await asyncio.sleep(0.001)
    w.counter("requests")


@pytest.mark.rampa_scenario(
    executor="constant-vus",
    vus=2,
    duration="200ms",
    worker_fn=api_worker,
    thresholds={"iteration_duration": ["avg<100"]},
)
def test_api_under_load(rampa_result: RunResult) -> None:
    assert rampa_result.status == RunStatus.PASSED
    assert rampa_result.snapshot is not None
```

```{toctree}
:hidden:

reference
```

---

# API Reference
Source: https://rampa.git-pull.com/pytest/reference/

(pytest-reference)=

# API Reference

## Plugin hooks

```{eval-rst}
.. autofunction:: rampa.pytest_plugin.pytest_configure
```

## Fixture implementation

```{eval-rst}
.. autofunction:: rampa.pytest_plugin.rampa_result
   :no-index:
```

## Internal

```{eval-rst}
.. autofunction:: rampa.pytest_plugin._run_plan
```

---