86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
|
"""Tests for the textfile-collector renderer."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from mail_imap_sync.prom import (
|
||
|
|
MetricFamily,
|
||
|
|
Sample,
|
||
|
|
read_prev_value,
|
||
|
|
render_prom,
|
||
|
|
write_prom_atomic,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _family(samples, name="kb_mail_sync_test", mtype="gauge"):
|
||
|
|
return MetricFamily(name, mtype, "a help string", samples)
|
||
|
|
|
||
|
|
|
||
|
|
class TestRender:
|
||
|
|
def test_emits_help_and_type_before_samples(self):
|
||
|
|
text = render_prom([_family([Sample(3)])])
|
||
|
|
assert text.splitlines()[:3] == [
|
||
|
|
"# HELP kb_mail_sync_test a help string",
|
||
|
|
"# TYPE kb_mail_sync_test gauge",
|
||
|
|
"kb_mail_sync_test 3",
|
||
|
|
]
|
||
|
|
|
||
|
|
def test_labels_are_sorted_and_quoted(self):
|
||
|
|
text = render_prom([_family([Sample(1, {"folder": "INBOX", "account": "gmail"})])])
|
||
|
|
assert 'kb_mail_sync_test{account="gmail",folder="INBOX"} 1' in text
|
||
|
|
|
||
|
|
def test_label_values_are_escaped(self):
|
||
|
|
text = render_prom([_family([Sample(1, {"folder": '[Gmail]/"All"\\Mail'})])])
|
||
|
|
assert 'folder="[Gmail]/\\"All\\"\\\\Mail"' in text
|
||
|
|
|
||
|
|
def test_integers_render_without_a_trailing_zero(self):
|
||
|
|
assert "kb_mail_sync_test 3\n" in render_prom([_family([Sample(3.0)])])
|
||
|
|
|
||
|
|
def test_floats_keep_their_precision(self):
|
||
|
|
assert "kb_mail_sync_test 1.5" in render_prom([_family([Sample(1.5)])])
|
||
|
|
|
||
|
|
def test_family_with_no_samples_is_omitted_entirely(self):
|
||
|
|
# "no accounts reported" and "accounts reported zero" are different facts.
|
||
|
|
assert render_prom([_family([])]) == "\n"
|
||
|
|
|
||
|
|
def test_output_ends_with_a_newline(self):
|
||
|
|
assert render_prom([_family([Sample(1)])]).endswith("\n")
|
||
|
|
|
||
|
|
|
||
|
|
class TestReadPrevValue:
|
||
|
|
def test_absent_file_returns_none(self, tmp_path: Path):
|
||
|
|
assert read_prev_value(tmp_path / "nope.prom", "kb_mail_sync_x") is None
|
||
|
|
|
||
|
|
def test_absent_metric_returns_none(self, tmp_path: Path):
|
||
|
|
path = tmp_path / "x.prom"
|
||
|
|
path.write_text("kb_mail_sync_other 5\n")
|
||
|
|
assert read_prev_value(path, "kb_mail_sync_x") is None
|
||
|
|
|
||
|
|
def test_reads_a_scientific_notation_value(self, tmp_path: Path):
|
||
|
|
path = tmp_path / "x.prom"
|
||
|
|
path.write_text("kb_mail_sync_x 1.786e9\n")
|
||
|
|
assert read_prev_value(path, "kb_mail_sync_x") == 1.786e9
|
||
|
|
|
||
|
|
def test_help_lines_are_not_mistaken_for_samples(self, tmp_path: Path):
|
||
|
|
path = tmp_path / "x.prom"
|
||
|
|
path.write_text("# HELP kb_mail_sync_x 42 things\nkb_mail_sync_x 7\n")
|
||
|
|
assert read_prev_value(path, "kb_mail_sync_x") == 7.0
|
||
|
|
|
||
|
|
|
||
|
|
class TestWriteAtomic:
|
||
|
|
def test_creates_the_directory_and_the_file(self, tmp_path: Path):
|
||
|
|
path = tmp_path / "node-exporter" / "kb-mail-sync.prom"
|
||
|
|
write_prom_atomic(path, "kb_mail_sync_x 1\n")
|
||
|
|
assert path.read_text() == "kb_mail_sync_x 1\n"
|
||
|
|
|
||
|
|
def test_leaves_no_temp_file_behind(self, tmp_path: Path):
|
||
|
|
path = tmp_path / "kb-mail-sync.prom"
|
||
|
|
write_prom_atomic(path, "kb_mail_sync_x 1\n")
|
||
|
|
assert [p.name for p in tmp_path.iterdir()] == ["kb-mail-sync.prom"]
|
||
|
|
|
||
|
|
def test_overwrites_in_place(self, tmp_path: Path):
|
||
|
|
path = tmp_path / "kb-mail-sync.prom"
|
||
|
|
write_prom_atomic(path, "kb_mail_sync_x 1\n")
|
||
|
|
write_prom_atomic(path, "kb_mail_sync_x 2\n")
|
||
|
|
assert path.read_text() == "kb_mail_sync_x 2\n"
|