69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
|
"""Tests for materialize_from_api's actions mirror.
|
||
|
|
|
||
|
|
Covers the fix where actions/ (pending/approved/...) were never mirrored from
|
||
|
|
the VPS control-plane API, leaving a PIHA-side webui panel reading its own
|
||
|
|
always-empty local ACTIONS_DIR.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
|
import materializer as materializer_module
|
||
|
|
|
||
|
|
|
||
|
|
FAKE_ACTIONS = {
|
||
|
|
"pending": [{"id": "a1", "status": "pending", "type": "redeploy"}],
|
||
|
|
"approved": [], "running": [], "completed": [], "failed": [], "rejected": [],
|
||
|
|
}
|
||
|
|
|
||
|
|
ALL_ENDPOINTS_DATA = {
|
||
|
|
"/nodes": [], "/services": [], "/incidents": [], "/deployments": [],
|
||
|
|
"/recommendations": [], "/summary": {}, "/events": [],
|
||
|
|
"/actions": FAKE_ACTIONS,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def test_materialize_from_api_fetches_and_writes_actions_json(tmp_path, monkeypatch):
|
||
|
|
monkeypatch.setattr(materializer_module, "CONTROL_PLANE_URL", "http://cp.example")
|
||
|
|
monkeypatch.setattr(materializer_module, "WORLD_DIR", str(tmp_path))
|
||
|
|
|
||
|
|
requested = []
|
||
|
|
|
||
|
|
def fake_fetch(url):
|
||
|
|
requested.append(url)
|
||
|
|
path = url[len("http://cp.example"):]
|
||
|
|
return ALL_ENDPOINTS_DATA[path]
|
||
|
|
|
||
|
|
monkeypatch.setattr(materializer_module, "_fetch_json", fake_fetch)
|
||
|
|
|
||
|
|
ok = materializer_module.materialize_from_api()
|
||
|
|
|
||
|
|
assert ok is True
|
||
|
|
assert "http://cp.example/actions" in requested
|
||
|
|
|
||
|
|
written = json.loads((tmp_path / "actions.json").read_text())
|
||
|
|
assert written == FAKE_ACTIONS
|
||
|
|
|
||
|
|
|
||
|
|
def test_materialize_from_api_aborts_if_actions_endpoint_fails(tmp_path, monkeypatch):
|
||
|
|
monkeypatch.setattr(materializer_module, "CONTROL_PLANE_URL", "http://cp.example")
|
||
|
|
monkeypatch.setattr(materializer_module, "WORLD_DIR", str(tmp_path))
|
||
|
|
|
||
|
|
def fake_fetch(url):
|
||
|
|
if url.endswith("/actions"):
|
||
|
|
return None
|
||
|
|
path = url[len("http://cp.example"):]
|
||
|
|
return ALL_ENDPOINTS_DATA[path]
|
||
|
|
|
||
|
|
monkeypatch.setattr(materializer_module, "_fetch_json", fake_fetch)
|
||
|
|
|
||
|
|
ok = materializer_module.materialize_from_api()
|
||
|
|
|
||
|
|
assert ok is False
|
||
|
|
assert not (tmp_path / "actions.json").exists()
|