API Reference

HavocMonkey

The main class. All attacks and the campaign runner live on this object. Instantiate once per test run and reuse it: the RNG state carries through, so the same seed always produces the same sequence of attacks.

from havoc_monkey import HavocMonkey

monkey = HavocMonkey(seed=42)

After a campaign, the last report is available on monkey.last_report, useful for inspecting results outside the return value.

Context manager

HavocMonkey supports the context manager protocol. The instance is returned as-is on enter; exit is a no-op. This is useful when you want to be explicit about the scope of a monkey's lifetime.

with HavocMonkey(seed=42) as monkey:
    attacked = monkey.null_flood(df, cols=['amount'])
    report = monkey.campaign(df, attacks=['null_flood', 'schema_drift'], health_check=check)

print(report)  # still accessible outside the block

Attacks

Every attack takes a DataFrame, applies the failure to a copy, and returns that copy. The original DataFrame is never mutated. Each attack is available as a method on the HavocMonkey instance.

In a campaign, each attack runs against the original DataFrame, not the output of the previous attack. This keeps each result clean: one failure mode, measured in isolation.

null_flood

Injects nulls into the specified columns. Rows are chosen at random according to the seed.

attacked = monkey.null_flood(df, cols=['amount', 'score'], pct=0.15)

volume_shock

Alters the number of rows. Tests whether your pipeline handles empty batches, unexpected overflow, or truncated partials.

monkey.volume_shock(df, attack='empty')
monkey.volume_shock(df, attack='overflow', factor=5.0)
monkey.volume_shock(df, attack='truncate', ratio=0.1)

type_coerce

Silently changes the dtype of a single column without renaming it. Mimics what happens when an upstream team loosens their schema or changes a serialisation format.

monkey.type_coerce(df, col='amount', target='str')
monkey.type_coerce(df, col='count', target='float')

schema_drift

Modifies the column structure. The most common production failure in multi-team pipelines: someone renames or drops a column without telling you.

monkey.schema_drift(df, attack='drop', col='category')
monkey.schema_drift(df, attack='rename', col='amount', new_name='value')
monkey.schema_drift(df, attack='add')
monkey.schema_drift(df, attack='reorder')

outlier_inject

Pushes a proportion of values far outside the column's normal distribution. Integer columns are promoted to float automatically, since outlier values by definition can't be stored as integers.

monkey.outlier_inject(df, cols=['amount', 'count'], sigma=8.0, pct=0.02)

temporal

Corrupts timestamp data. Requires a datetime column; if the column isn't datetime already, havoc-monkey will warn and convert it automatically using pd.to_datetime.

monkey.temporal(df, col='ts', attack='out_of_order', pct=0.2)
monkey.temporal(df, col='ts', attack='late', delta='6h', pct=0.3)
monkey.temporal(df, col='ts', attack='future', delta='30d', pct=0.1)
monkey.temporal(df, col='ts', attack='missing_window', window=('2024-03-01', '2024-03-07'))
monkey.temporal(df, col='ts', attack='duplicate_ts', pct=0.1)

campaign()

Runs a sequence of attacks against a single DataFrame and returns a Report. Each attack gets a fresh copy of the original DataFrame, so results are independent of each other.

report = monkey.campaign(
    df,
    attacks=[
        'null_flood',   # short form, uses default params
        {'name': 'schema_drift', 'params': {'attack': 'drop', 'col': 'category'}},
    ],
    health_check=my_check,
)

health_check

A callable you write that defines what "working" means for your pipeline. It receives the attacked DataFrame and is expected to run your pipeline on it and assert any invariants you care about. Return any truthy value on success; raise on failure.

def health_check(df: pd.DataFrame) -> bool:
    result = my_pipeline(df)
    assert result['amount'].notnull().all(), "nulls leaked"
    assert len(result) > 0, "empty output"
    return True

Each outcome is classified based on what the health check does:

• PASSED: the health check returned without error. Your pipeline handled the attack.
• FAILED: the health check raised AssertionError. Your pipeline explicitly detected the bad data, but it still broke.
• ERROR: the health check raised any other exception. Your pipeline crashed entirely, worse than FAILED.
• UNKNOWN: no health check was provided. The report documents what changed, but can't classify impact without one.

Severity

Severity is derived automatically from the attack type and outcome:

Report

Returned by campaign(). Print it directly for a colour-coded terminal summary, or use the export methods for CI integration or sharing.

Each AttackResult carries:

Export

Three formats, each suited to a different use:

report.to_dict()      # serialisable dict, use with json.dumps or any downstream tool
report.to_markdown()  # GitHub-flavoured Markdown table, paste into PRs or CI artifacts
report.to_html()      # self-contained branded HTML file, open in a browser or email as-is

Pytest Plugin

Registered automatically via the pytest11 entry point when you install havoc-monkey. No imports needed. The fixture and marker are available in every test file the moment you run pytest.

Fixture: havoc_monkey_instance

A HavocMonkey instance seeded by --havoc-seed (default 42), scoped per test. Use it to run attacks directly inside a test.

def test_handles_nulls(havoc_monkey_instance, my_df):
    attacked = havoc_monkey_instance.null_flood(my_df, cols=['amount'])
    result = my_pipeline(attacked)
    assert result['total'].notnull().all()

Marker: @pytest.mark.havoc_monkey

Run a full campaign from inside a test. If the test fails, the campaign report is automatically attached to the test output, so you see exactly which attack caused the failure and why.

import pytest

@pytest.mark.havoc_monkey(attacks=['null_flood', 'schema_drift'])
def test_pipeline_resilience(havoc_monkey_instance, my_df):
    def hc(df):
        result = my_pipeline(df)
        assert len(result) > 0
        return True

    report = havoc_monkey_instance.campaign(
        my_df,
        attacks=['null_flood', 'schema_drift'],
        health_check=hc,
    )
    assert report.failed == 0

CLI option

Override the seed for the entire test run:

pytest --havoc-seed 99

CLI

havoc-monkey run

Loads a file, runs the specified attacks without a health check, and prints the report. All outcomes are UNKNOWN: the CLI is for quick exploration, not measurement. Use the Python API with a health_check for classification.

havoc-monkey run \
  --file data.csv \
  --attacks null_flood \
  --attacks schema_drift \
  --seed 42

# Save as a Markdown artifact for CI
havoc-monkey run --file data.parquet --attacks null_flood --output report.md

# Save as a shareable HTML report
havoc-monkey run --file data.csv --attacks null_flood --attacks schema_drift --output report.html

havoc-monkey list-attacks

Prints a quick-reference table of every attack, its subtypes, and its parameters.

havoc-monkey list-attacks