Testing with Markus

Markus comes with a markus.testing.MetricsMock that makes it easier to write tests and assert things about generated metrics.

When activating the markus.testing.MetricsMock context, it becomes a backend for all emitted metrics. It’ll contains a list of metrics emitted. You can assert things about the metrics collected.

There are a set of assert_* helper methods for simplifying that code, but you can also use markus.testing.MetricsMock.filter_records() directly.

class markus.testing.MetricsMock

Mock for recording metrics events and testing them.

Mimics a metrics backend as a context manager. Keeps records of what got metricfied so that you can print them out, filter them, assert various things about them, etc.

To use:

from markus.testing import MetricsMock

def test_something():
    with MetricsMock() as mm:
        # Do things that might record metrics here

        # Assert something about the metrics recorded
        mm.assert_incr_once(stat="some.random.key", value=1)

When using the assert_* helper methods, if the assertion fails, it’ll print the MetricsRecords that were emitted to stdout.

assert_gauge(stat, value=None, tags=None)

Asserts a gauge was emitted at least once.

assert_gauge_once(stat, value=None, tags=None)

Asserts a gauge was emitted exactly once.

assert_histogram(stat, value=None, tags=None)

Asserts a histogram was emitted at least once.

assert_histogram_once(stat, value=None, tags=None)

Asserts a histogram was emitted exactly once.

assert_incr(stat, value=1, tags=None)

Asserts an incr was emitted at least once.

assert_incr_once(stat, value=1, tags=None)

Asserts an incr was emitted exactly once.

assert_not_gauge(stat, value=None, tags=None)

Asserts a gauge was not emitted.

assert_not_histogram(stat, value=None, tags=None)

Asserts a histogram was not emitted.

assert_not_incr(stat, value=1, tags=None)

Asserts an incr was not emitted.

assert_not_timing(stat, value=None, tags=None)

Asserts a timing was not emitted.

assert_timing(stat, value=None, tags=None)

Asserts a timing was emitted at least once.

assert_timing_once(stat, value=None, tags=None)

Asserts a timing was emitted exactly once.

clear_records()

Clear the records list.

filter_records(fun_name=None, stat=None, value=None, tags=None)

Filter collected metrics records for ones that match specified criteria.

Filtering is done by ANDing the requirements together. For example:

with MetricsMock() as mm:
    # Do something that emits metrics

    assert mm.filter_records("incr", stat="some.key", tags=["color:blue"])

markus.testing.MetricsMock.filter_records() will return markus.main.MetricsRecord instances that are "incr" AND the stat is "some.key" AND the tags list is ["color:blue"].

Parameters:

• fun_name (str) – “incr”, “gauge”, “timing”, “histogram”, or None

• stat (str) – the stat emitted

• value (int/float) – the value

• tags (list) – the list of tag strings or [] or None

Returns:

list of markus.main.MetricsRecord instances

get_records()

Return list of MetricsRecord instances.

This is the list of markus.main.MetricsRecord instances that were emitted while this markus.testing.MetricsMock was active.

has_record(fun_name=None, stat=None, value=None, tags=None)

Return True/False regarding whether collected metrics match criteria.

Parameters:

• fun_name (str) – “incr”, “gauge”, “timing”, “histogram”, or None

• stat (str) – the stat emitted

• value (int/float) – the value

• tags (list) – the list of tag strings or [] or None

Returns:

bool

print_records()

Print all the collected metrics.