61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import types
|
||
|
|
|
||
|
|
import ariadne.services.opensearch_prune as prune_module
|
||
|
|
|
||
|
|
|
||
|
|
def test_parse_size() -> None:
|
||
|
|
assert prune_module.parse_size("1gb") == 1024**3
|
||
|
|
assert prune_module.parse_size("0") == 0
|
||
|
|
assert prune_module.parse_size("bad") == 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
|