ariadne/tests/test_hermes_storage_signals.py

"""Tests for recognising Jenkins workspace storage exhaustion."""

from __future__ import annotations

from ariadne.services import hermes_storage_signals as module


def _bundle(*lines: str, tail: str = "") -> dict:
    return {"jenkins": {"console_failures": [{"text": "\n".join(lines)}], "console_tail": tail}}


def test_a_full_workspace_volume_is_recognised() -> None:
    matched, marker = module.has_storage_exhaustion_signature(
        _bundle("cp: error writing '/home/jenkins/agent/workspace/x': No space left on device")
    )
    assert matched is True
    assert marker == "no space left on device"


def test_the_tail_is_searched_too() -> None:
    matched, _ = module.has_storage_exhaustion_signature(
        _bundle("nothing here", tail="workspace volume full: disk quota exceeded")
    )
    assert matched is True


def test_a_full_disk_without_a_workspace_hint_is_not_this_failure() -> None:
    """Reclaiming Jenkins workspaces would not address it."""

    matched, marker = module.has_storage_exhaustion_signature(
        _bundle("etcd: No space left on device")
    )
    assert (matched, marker) == (False, None)


def test_a_workspace_mention_without_a_storage_marker_does_not_match() -> None:
    matched, _ = module.has_storage_exhaustion_signature(_bundle("workspace checkout failed"))
    assert matched is False


def test_never_raises_on_hostile_input() -> None:
    for bad in (None, 7, {}, {"jenkins": None}, {"jenkins": {"console_failures": "no"}}):
        assert module.has_storage_exhaustion_signature(bad) == (False, None)


def test_agent_provisioning_failures_are_recognised() -> None:
    """The build never got an agent, rather than failing once running."""

    from ariadne.services import hermes_infra_signals

    for text in (
        "All nodes of label 'lesavka_583-abc' are offline",
        "Error in provisioning; agent=KubernetesSlave[x]",
        "Still waiting to schedule task",
    ):
        matched, marker = hermes_infra_signals.has_agent_provisioning_signature(_bundle(text))
        assert matched is True, text
        assert marker


def test_an_ordinary_failure_is_not_an_agent_problem() -> None:
    from ariadne.services import hermes_infra_signals

    matched, _ = hermes_infra_signals.has_agent_provisioning_signature(
        _bundle("AssertionError: expected 1 got 2")
    )
    assert matched is False


def test_agent_signature_never_raises() -> None:
    from ariadne.services import hermes_infra_signals

    for bad in (None, 7, {}, {"jenkins": {"console_failures": "no"}}):
        assert hermes_infra_signals.has_agent_provisioning_signature(bad) == (False, None)