65 lines
2.3 KiB
Python
65 lines
2.3 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
import consul_client
|
|
|
|
@patch("requests.get")
|
|
def test_get_cluster_services(mock_get):
|
|
"""Test fetching healthy services from Consul."""
|
|
# Mock responses for navidrome and replica-navidrome
|
|
mock_navidrome = [
|
|
{
|
|
"Node": {"Node": "node1", "Address": "192.168.1.101"},
|
|
"Service": {"Service": "navidrome", "Port": 4533, "ID": "navidrome-1"},
|
|
"Checks": [{"Status": "passing"}]
|
|
}
|
|
]
|
|
mock_replicas = [
|
|
{
|
|
"Node": {"Node": "node2", "Address": "192.168.1.102"},
|
|
"Service": {"Service": "replica-navidrome", "Port": 4533, "ID": "replica-1"},
|
|
"Checks": [{"Status": "passing"}]
|
|
},
|
|
{
|
|
"Node": {"Node": "node3", "Address": "192.168.1.103"},
|
|
"Service": {"Service": "replica-navidrome", "Port": 4533, "ID": "replica-2"},
|
|
"Checks": [{"Status": "critical"}] # One failing check
|
|
}
|
|
]
|
|
|
|
def side_effect(url, params=None, timeout=None):
|
|
if "health/service/navidrome" in url:
|
|
m = MagicMock()
|
|
m.json.return_value = mock_navidrome
|
|
m.raise_for_status.return_value = None
|
|
return m
|
|
elif "health/service/replica-navidrome" in url:
|
|
m = MagicMock()
|
|
m.json.return_value = mock_replicas
|
|
m.raise_for_status.return_value = None
|
|
return m
|
|
return MagicMock()
|
|
|
|
mock_get.side_effect = side_effect
|
|
|
|
consul_url = "http://consul:8500"
|
|
services = consul_client.get_cluster_services(consul_url)
|
|
|
|
# Should find 3 nodes total (node1 primary, node2 healthy replica, node3 critical replica)
|
|
assert len(services) == 3
|
|
|
|
# Check node1 (primary)
|
|
node1 = next(s for s in services if s["node"] == "node1")
|
|
assert node1["role"] == "primary"
|
|
assert node1["status"] == "passing"
|
|
assert node1["address"] == "192.168.1.101"
|
|
|
|
# Check node2 (healthy replica)
|
|
node2 = next(s for s in services if s["node"] == "node2")
|
|
assert node2["role"] == "replica"
|
|
assert node2["status"] == "passing"
|
|
|
|
# Check node3 (critical replica)
|
|
node3 = next(s for s in services if s["node"] == "node3")
|
|
assert node3["role"] == "replica"
|
|
assert node3["status"] == "critical"
|