"""The shared format filters and status_class (#5093): one way to write a size, a rate, a span and a moment, and one grading against Settings → Thresholds.""" from datetime import datetime, timedelta, timezone import jinja2 import pytest from steward.core import formatting as f from steward.core.settings import threshold_style_for, to_thresholds_cfg from tests._templates import REPO as _REPO, template_files as _templates TH = to_thresholds_cfg({}) NOW = datetime(2026, 10, 6, 18, 0, 0, tzinfo=timezone.utc) @pytest.mark.parametrize("n, text", [ (None, "—"), (0, "0 B"), (512, "512 B"), (1023.9, "1023 B"), (1024, "1.0 KiB"), (1536, "1.5 KiB"), (1024 ** 2, "1.0 MiB"), (int(1.5 * 1024 ** 3), "1.5 GiB"), (1024 ** 4, "1.0 TiB"), (3 * 1024 ** 5, "3.0 PiB"), (1024 ** 6, "1024.0 PiB"), ]) def test_bytes(n, text): assert f.human_bytes(n) == text @pytest.mark.parametrize("n, text", [(None, "—"), (0, "0 B/s"), (2.5 * 1024 ** 2, "2.5 MiB/s")]) def test_rate(n, text): assert f.human_rate(n) == text @pytest.mark.parametrize("secs, text", [ (None, "—"), (-5, "—"), (0, "0s"), (0.9, "0s"), (45, "45s"), (60, "1m"), (90, "1m 30s"), (18 * 60, "18m"), (3600, "1h"), (2 * 3600 + 5 * 60 + 59, "2h 5m"), (86400, "1d"), (3 * 86400 + 4 * 3600 + 30, "3d 4h"), (400 * 86400, "400d"), ]) def test_duration(secs, text): assert f.human_duration(secs) == text @pytest.mark.parametrize("when, text", [ (NOW - timedelta(minutes=3), "3m ago"), (NOW - timedelta(days=2, hours=1), "2d 1h ago"), (NOW.replace(tzinfo=None) - timedelta(seconds=45), "45s ago"), # naive is UTC (NOW + timedelta(minutes=5), "in 5m"), (NOW, "just now"), ]) def test_ago_reads_relative_and_titles_the_absolute_utc_time(when, text): out = str(f.ago(when, now=NOW)) stamp = when.replace(tzinfo=timezone.utc) if when.tzinfo is None else when assert out == (f'') def test_ago_of_nothing_and_of_another_zone(): assert str(f.ago(None)) == "—" est = timezone(timedelta(hours=-4)) assert 'title="2026-10-06 18:00:00 UTC"' in str(f.ago(NOW.astimezone(est), now=NOW)) @pytest.mark.parametrize("value, kind, cls", [ (None, "cpu", "unknown"), (10, "cpu", "ok"), (80, "cpu", "warn"), (89.9, "cpu", "warn"), (90, "cpu", "crit"), (49, "latency", "ok"), (50, "latency", "warn"), (250, "latency", "crit"), # Lower is worse: uptime % and days of TLS left. (99.9, "uptime", "ok"), (98, "uptime", "warn"), (90, "uptime", "crit"), (30, "tls_days", "ok"), (10, "tls_days", "warn"), (1, "tls_days", "crit"), ]) def test_status_class_against_the_default_thresholds(value, kind, cls): assert f.status_class(value, kind, TH) == cls def test_status_class_follows_settings_and_refuses_an_unknown_kind(): th = to_thresholds_cfg({"thresholds.cpu_warn": 50, "thresholds.cpu_crit": 60}) assert [f.status_class(v, "cpu", th) for v in (49, 50, 60)] == ["ok", "warn", "crit"] with pytest.raises(KeyError): f.status_class(10, "cpuu", TH) def test_threshold_style_is_graded_by_status_class(): assert threshold_style_for(10, "cpu", TH) == "" assert threshold_style_for(None, "cpu", TH) == "" assert "var(--yellow)" in threshold_style_for(85, "cpu", TH) assert "var(--red)" in threshold_style_for(95, "cpu", TH) assert "700" not in threshold_style_for(95, "cpu", TH) # 500 is the heaviest weight # ── every template compiles against the app's filters ──────────────────────── async def test_compiling_catches_an_unknown_filter(app): with pytest.raises(jinja2.TemplateAssertionError, match="no_such"): app.jinja_env.compile("{{ 1|no_such }}") app.jinja_env.compile("{{ 1|bytes }} {{ 1|rate }} {{ 1|duration }} {{ none|ago }}") async def test_every_template_compiles_with_the_app_filters(app): files = _templates() assert len(files) > 100 failed = {} for path in files: try: app.jinja_env.compile(path.read_text(), name=str(path.relative_to(_REPO))) except jinja2.TemplateError as exc: failed[str(path.relative_to(_REPO))] = str(exc) assert not failed, failed @pytest.mark.parametrize("key, label", [ ("metrics_url", "Metrics URL"), ("scrape_interval_seconds", "Scrape interval seconds"), ("geoip_db", "GeoIP database"), ("db", "Database"), ("verify_ssl", "Verify SSL"), ("enabled", "Enabled"), ("", ""), ]) def test_a_settings_key_reads_as_a_label(key, label): """#5212: plugin settings are labelled in words, not snake_case.""" from steward.core.formatting import field_label assert field_label(key) == label def test_metric_names_and_units_read_in_words(): from steward.core.formatting import metric_label, metric_unit, source_label assert metric_label("disk_used_pct_worst") == "Fullest disk" and metric_unit("disk_used_pct_worst") == "%" assert metric_label("hrStorageUsed") == "hrStorageUsed" and metric_unit("hrStorageUsed") == "" assert metric_label("if_in_octets") == "if in octets" assert metric_unit("cert_expiry_days") == "days" and metric_unit("seconds_since_seen") == "seconds" assert metric_unit("is_up") == "" and metric_unit("request_rate") == "req/s" assert source_label("host_agent") == "Host agent" and source_label("my_plugin") == "my plugin"