atlas-iac/scripts/tests/test_availability_rollup.py

"""Unit tests for the Atlas availability publisher."""

import importlib.util
import json
from pathlib import Path

import pytest


def load_module():
    """Load the service-owned rollup module without packaging it."""
    path = (
        Path(__file__).resolve().parents[2]
        / "services/monitoring/scripts/availability_rollup.py"
    )
    spec = importlib.util.spec_from_file_location("availability_rollup", path)
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(module)
    return module


def test_parse_export_deduplicates_replay_boundaries() -> None:
    """Keep only the final value when replay chunks share a timestamp."""
    mod = load_module()
    lines = [
        (json.dumps({"timestamps": [1000, 2000], "values": [2, 3]}) + "\n").encode(),
        (json.dumps({"timestamps": [2000, 3000], "values": [3, 5]}) + "\n").encode(),
    ]

    assert mod.parse_export(lines) == {1000: 2.0, 2000: 3.0, 3000: 5.0}


def test_calculate_availability_uses_all_server_failures() -> None:
    """Calculate one bounded successful-request ratio."""
    mod = load_module()

    assert mod.calculate_availability(1000, 2) == pytest.approx(0.998)
    with pytest.raises(ValueError):
        mod.calculate_availability(0, 0)
    with pytest.raises(ValueError):
        mod.calculate_availability(100, -1)


def test_render_metric_publishes_only_the_request_v4_series() -> None:
    """Render the two series selected by the Grafana overview panels."""
    mod = load_module()

    assert mod.render_metric(mod.OUTPUT_METRIC, 0.9995, 1234) == (
        'atlas:availability:ratio_365d{definition="request-v4",scope="atlas",'
        'rollup="yearly"} 0.999500000000 1234\n'
    )
    assert mod.render_metric(mod.COVERAGE_METRIC, 109, 1234) == (
        'atlas:availability:coverage_days_365d{definition="request-v4",scope="atlas",'
        'rollup="yearly"} 109.000000000000 1234\n'
    )