pytest Plugin¶

Install¶

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

Fixtures¶

fixture rampa.pytest_plugin.rampa_result → RunResult

¶

fixture

fixture

fixture rampa.pytest_plugin.rampa_result → RunResult

¶

Run a rampa scenario defined by @pytest.mark.rampa_scenario.

Depends on:

request

The marker accepts the same kwargs as ScenarioConfig plus an optional thresholds dict. The fixture runs the scenario, waits for completion, and returns the RunResult.

If thresholds are configured and any fail, the test fails with a descriptive message.

Param request:

Pytest request object.

Type request:

pytest.FixtureRequest

Returns:

• RunResult – The completed test run result.

• >>> import rampa.pytest_plugin

Quick start¶

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 ScenarioConfig plus:

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

Complete example¶

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