ariadne/tests/test_opensearch_prune.py

from __future__ import annotations

import types

import ariadne.services.opensearch_prune as prune_module


def _settings(**overrides):
    values = {
        "opensearch_url": "http://opensearch",
        "opensearch_limit_bytes": 5,
        "opensearch_index_patterns": "kube-*",
        "opensearch_timeout_sec": 5.0,
    }
    values.update(overrides)
    return types.SimpleNamespace(**values)


def test_parse_size() -> None:
    assert prune_module.parse_size("") == 0
    assert prune_module.parse_size("1gb") == 1024**3
    assert prune_module.parse_size("0") == 0
    assert prune_module.parse_size("bad") == 0
    assert prune_module.parse_size("1zb") == 0


def test_prune_indices_deletes(monkeypatch) -> None:
    dummy_settings = types.SimpleNamespace(
        opensearch_url="http://opensearch",
        opensearch_limit_bytes=5,
        opensearch_index_patterns="kube-*",
        opensearch_timeout_sec=5.0,
    )
    monkeypatch.setattr(prune_module, "settings", dummy_settings)

    class DummyResponse:
        def __init__(self, payload, status_code=200):
            self._payload = payload
            self.status_code = status_code

        def raise_for_status(self):
            return None

        def json(self):
            return self._payload

    class DummyClient:
        def __init__(self):
            self.deleted = []

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def get(self, url, params=None):
            return DummyResponse(
                [
                    {"index": "kube-1", "store.size": "10b", "creation.date": "1"},
                    {"index": "kube-2", "store.size": "1b", "creation.date": "2"},
                ]
            )

        def delete(self, url):
            self.deleted.append(url)
            return DummyResponse({}, 200)

    dummy = DummyClient()
    monkeypatch.setattr(prune_module.httpx, "Client", lambda *args, **kwargs: dummy)

    summary = prune_module.prune_indices()
    assert summary.deleted == 1


def test_fetch_indices_ignores_missing_pattern(monkeypatch) -> None:
    monkeypatch.setattr(prune_module, "settings", _settings())

    class DummyResponse:
        status_code = prune_module.HTTP_NOT_FOUND

        def raise_for_status(self):
            raise AssertionError("404 should be handled before raise_for_status")

    client = types.SimpleNamespace(get=lambda *_args, **_kwargs: DummyResponse())

    assert prune_module._fetch_indices(client, "missing-*") == []


def test_prune_indices_returns_when_no_patterns(monkeypatch) -> None:
    monkeypatch.setattr(prune_module, "settings", _settings(opensearch_index_patterns=" , "))

    summary = prune_module.prune_indices()

    assert summary.detail == "no patterns configured"
    assert summary.deleted == 0


def test_prune_indices_continues_after_fetch_failure(monkeypatch) -> None:
    monkeypatch.setattr(
        prune_module,
        "settings",
        _settings(opensearch_index_patterns="bad-*,kube-*", opensearch_limit_bytes=100),
    )

    class DummyResponse:
        status_code = 200

        def raise_for_status(self):
            return None

        def json(self):
            return [
                {"index": ".system", "store.size": "100b", "creation.date": "1"},
                {"index": "kube-1", "store.size": "1b", "creation.date": "2"},
            ]

    class DummyClient:
        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def get(self, url, params=None):
            if "bad-*" in url:
                raise RuntimeError("fetch failed")
            return DummyResponse()

        def delete(self, _url):
            raise AssertionError("within-limit result should not delete indices")

    monkeypatch.setattr(prune_module.httpx, "Client", lambda *args, **kwargs: DummyClient())

    summary = prune_module.prune_indices()

    assert summary.detail == "within limit"
    assert summary.total_before == 1


def test_prune_indices_logs_delete_failures_and_keeps_pruning(monkeypatch) -> None:
    monkeypatch.setattr(prune_module, "settings", _settings(opensearch_limit_bytes=5))

    class DummyResponse:
        status_code = 200

        def __init__(self, payload):
            self._payload = payload

        def raise_for_status(self):
            return None

        def json(self):
            return self._payload

    class DummyClient:
        def __init__(self):
            self.deleted: list[str] = []

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, tb):
            return False

        def get(self, _url, params=None):
            return DummyResponse(
                [
                    {"index": "kube-old", "store.size": "10b", "creation.date": "1"},
                    {"index": "kube-new", "store.size": "10b", "creation.date": "2"},
                ]
            )

        def delete(self, url):
            self.deleted.append(url)
            if url.endswith("/kube-old"):
                raise RuntimeError("delete failed")
            return DummyResponse({})

    dummy = DummyClient()
    monkeypatch.setattr(prune_module.httpx, "Client", lambda *args, **kwargs: dummy)

    summary = prune_module.prune_indices()

    assert summary.deleted == 1
    assert summary.total_before == 20
    assert summary.total_after == 10
    assert dummy.deleted == ["http://opensearch/kube-old", "http://opensearch/kube-new"]