Testing with Markus¶
Asserting things about emitted metrics¶
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 capture 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.
pytest plugin¶
Markus includes a pytest plugin with a metricsmock
pytest fixture. It’s
implemented like this:
from markus.testing import MetricsMock
import pytest
@pytest.fixture
def metricsmock() -> MetricsMock:
with MetricsMock() as mm:
yield mm
Testing against tag values¶
You can assert against tags ignoring the actual values using
markus.testing.AnyTagValue
.
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(stat="some.random.key", value=1, tags=[AnyTagValue("host")])
Test module API¶
- class markus.testing.AnyTagValue(key)¶
Matches a markus metrics tag with any value
Example:
import markus from markus.testing import AnyTagValue, MetricsMock with MetricsMock() as mm: metrics = get_metrics("test") metrics.incr("key1", value=1, tags=["host:12345"]) mm.assert_incr(stat="test.key1", tags=[AnyTagValue("host")])
- Parameters:
key (str)
- 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.
- Parameters:
stat (str | None)
value (int | None)
tags (List[str | AnyTagValue] | None)
- assert_gauge_once(stat, value=None, tags=None)¶
Asserts a gauge was emitted exactly once.
- Parameters:
stat (str | None)
value (int | None)
tags (List[str | AnyTagValue] | None)
- assert_histogram(stat, value=None, tags=None)¶
Asserts a histogram was emitted at least once.
- Parameters:
stat (str | None)
value (float | None)
tags (List[str | AnyTagValue] | None)
- assert_histogram_once(stat, value=None, tags=None)¶
Asserts a histogram was emitted exactly once.
- Parameters:
stat (str | None)
value (float | None)
tags (List[str | AnyTagValue] | None)
- assert_incr(stat, value=1, tags=None)¶
Asserts an incr was emitted at least once.
- Parameters:
stat (str | None)
value (int | None)
tags (List[str | AnyTagValue] | None)
- assert_incr_once(stat, value=1, tags=None)¶
Asserts an incr was emitted exactly once.
- Parameters:
stat (str | None)
value (int | None)
tags (List[str | AnyTagValue] | None)
- assert_not_gauge(stat, value=None, tags=None)¶
Asserts a gauge was not emitted.
- Parameters:
stat (str | None)
value (int | None)
tags (List[str | AnyTagValue] | None)
- assert_not_histogram(stat, value=None, tags=None)¶
Asserts a histogram was not emitted.
- Parameters:
stat (str | None)
value (float | None)
tags (List[str | AnyTagValue] | None)
- assert_not_incr(stat, value=1, tags=None)¶
Asserts an incr was not emitted.
- Parameters:
stat (str | None)
value (int | None)
tags (List[str | AnyTagValue] | None)
- assert_not_timing(stat, value=None, tags=None)¶
Asserts a timing was not emitted.
- Parameters:
stat (str | None)
value (float | None)
tags (List[str | AnyTagValue] | None)
- assert_timing(stat, value=None, tags=None)¶
Asserts a timing was emitted at least once.
- Parameters:
stat (str | None)
value (float | None)
tags (List[str | AnyTagValue] | None)
- assert_timing_once(stat, value=None, tags=None)¶
Asserts a timing was emitted exactly once.
- Parameters:
stat (str | None)
value (float | None)
tags (List[str | AnyTagValue] | None)
- 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 returnmarkus.main.MetricsRecord
instances that are"incr"
AND the stat is"some.key"
AND the tags list is["color:blue"]
.- Parameters:
fun_name (str | None) – “incr”, “gauge”, “timing”, “histogram”, or
None
stat (str | None) – the stat emitted
value (int | float | None) – the value
tags (List[str | AnyTagValue] | None) – the list of tag strings or
[]
orNone
fun_name
stat
value
tags
- Returns:
list of
markus.main.MetricsRecord
instances- Return type:
List[MetricsRecord]
- get_records()¶
Return list of MetricsRecord instances.
This is the list of
markus.main.MetricsRecord
instances that were emitted while thismarkus.testing.MetricsMock
was active.- Return type:
List[MetricsRecord]
- has_record(fun_name=None, stat=None, value=None, tags=None)¶
Return True/False regarding whether collected metrics match criteria.
- Parameters:
fun_name (str | None) – “incr”, “gauge”, “timing”, “histogram”, or
None
stat (str | None) – the stat emitted
value (int | float | None) – the value
tags – the list of tag strings or
[]
orNone
fun_name
stat
value
- Returns:
bool
- Return type:
bool
- print_records()¶
Print all the collected metrics.