139 lines
5.3 KiB
Python
139 lines
5.3 KiB
Python
"""
|
|
Regression tests for daily-infra-report.py fixes (PR #64).
|
|
|
|
Tests:
|
|
(a) Asserts the nested zulip read feeds the agent-card fields
|
|
(b) Asserts an unreachable pve_get renders labelled-unreachable, not "0/0"
|
|
"""
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
# Add scripts to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
|
|
import importlib.util
|
|
|
|
def load_script():
|
|
"""Load the daily-infra-report script as a module."""
|
|
script_path = Path(__file__).parent.parent / "scripts" / "daily-infra-report.py"
|
|
spec = importlib.util.spec_from_file_location("daily_infra_report", script_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_nested_zulip_read_feeds_agent_card():
|
|
"""Test that Zulip state is read from the nested 'zulip' key and feeds agent-card fields."""
|
|
# Mock the http_get_body response with nested structure
|
|
mock_health_response = json.dumps({
|
|
"status": "ok",
|
|
"platform": "pi",
|
|
"agent": "abiba",
|
|
"zulip": {
|
|
"connected": True,
|
|
"queue_id": "test-queue-id",
|
|
"messages_processed": 42,
|
|
"skipped": 5,
|
|
"last_error": None
|
|
}
|
|
})
|
|
|
|
# Import and patch
|
|
report_mod = load_script()
|
|
|
|
with patch.object(report_mod, 'http_get_body', return_value=mock_health_response):
|
|
# Simulate the collect() function's Zulip section
|
|
zulip_health = json.loads(report_mod.http_get_body("http://localhost:9200/health"))
|
|
zulip_state = zulip_health.get("zulip", {})
|
|
|
|
# Assert the nested key is read correctly
|
|
assert zulip_state.get("connected") == True, "Zulip connected should be True from nested key"
|
|
assert zulip_state.get("messages_processed") == 42, "messages_processed should be 42 from nested key"
|
|
assert zulip_state.get("queue_id") == "test-queue-id", "queue_id should be read from nested key"
|
|
|
|
# Simulate the agent card field population
|
|
agent_card = {
|
|
"zulip_connected": zulip_state.get("connected", False),
|
|
"zulip_processed": zulip_state.get("messages_processed", 0),
|
|
}
|
|
|
|
assert agent_card["zulip_connected"] == True, "Agent card should show Zulip connected"
|
|
assert agent_card["zulip_processed"] == 42, "Agent card should show 42 processed messages"
|
|
|
|
|
|
def test_unreachable_pve_get_renders_labelled_unreachable():
|
|
"""Test that an unreachable PVE API renders 'unreachable' instead of '0/0'."""
|
|
# Import and patch
|
|
report_mod = load_script()
|
|
|
|
# Test pve_get returns None on error
|
|
with patch.object(report_mod.subprocess, 'run') as mock_run:
|
|
mock_run.return_value.returncode = 7 # Connection failure
|
|
result = report_mod.pve_get("/api2/json/nodes")
|
|
assert result is None, "pve_get should return None on connection failure"
|
|
|
|
# Test the render logic
|
|
report = {
|
|
"nodes": {},
|
|
"node_count": 0,
|
|
"nodes_online": 0,
|
|
"pve_probe_status": "unreachable",
|
|
"total_vms": 0,
|
|
"running_vms": 0,
|
|
}
|
|
|
|
# The render should show "unreachable" not "0/0"
|
|
pve_status_label = "unreachable" if report.get('pve_probe_status') == 'unreachable' else f"{report['nodes_online']}/{report['node_count']}"
|
|
|
|
assert pve_status_label == "unreachable", "PVE status should show 'unreachable' when probe fails, not '0/0'"
|
|
|
|
|
|
def test_unreachable_resources_renders_labelled_unreachable():
|
|
"""Test that unreachable resources probe renders 'unreachable' instead of '0/0'."""
|
|
report_mod = load_script()
|
|
|
|
# Test resources probe returns None
|
|
with patch.object(report_mod.subprocess, 'run') as mock_run:
|
|
mock_run.return_value.returncode = 7
|
|
result = report_mod.pve_get("/api2/json/cluster/resources")
|
|
assert result is None, "pve_get for resources should return None on connection failure"
|
|
|
|
# Test the render logic
|
|
report = {
|
|
"resources_probe_status": "unreachable",
|
|
"total_vms": 0,
|
|
"running_vms": 0,
|
|
}
|
|
|
|
resources_label = "unreachable" if report.get('resources_probe_status') == 'unreachable' else f"{report['running_vms']}/{report['total_vms']}"
|
|
|
|
assert resources_label == "unreachable", "Resources status should show 'unreachable' when probe fails, not '0/0'"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Running tests...")
|
|
try:
|
|
test_nested_zulip_read_feeds_agent_card()
|
|
print("✓ test_nested_zulip_read_feeds_agent_card passed")
|
|
except AssertionError as e:
|
|
print(f"✗ test_nested_zulip_read_feeds_agent_card failed: {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
test_unreachable_pve_get_renders_labelled_unreachable()
|
|
print("✓ test_unreachable_pve_get_renders_labelled_unreachable passed")
|
|
except AssertionError as e:
|
|
print(f"✗ test_unreachable_pve_get_renders_labelled_unreachable failed: {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
test_unreachable_resources_renders_labelled_unreachable()
|
|
print("✓ test_unreachable_resources_renders_labelled_unreachable passed")
|
|
except AssertionError as e:
|
|
print(f"✗ test_unreachable_resources_renders_labelled_unreachable failed: {e}")
|
|
sys.exit(1)
|
|
|
|
print("All tests passed!")
|