test(ariadne): cover OpenSearch prune edges

tests/test_opensearch_prune.py

@@ -5,10 +5,23 @@ 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:
@@ -58,3 +71,118 @@ def test_prune_indices_deletes(monkeypatch) -> None:
 
     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"]