84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
import litefs_client
|
|
|
|
@patch("requests.get")
|
|
def test_get_node_status_primary(mock_get):
|
|
"""Test fetching LiteFS status for a primary node via /status."""
|
|
mock_status = {
|
|
"clusterID": "cid1",
|
|
"primary": True,
|
|
"uptime": 3600,
|
|
"advertiseURL": "http://192.168.1.101:20202"
|
|
}
|
|
|
|
m = MagicMock()
|
|
m.status_code = 200
|
|
m.json.return_value = mock_status
|
|
mock_get.return_value = m
|
|
|
|
status = litefs_client.get_node_status("192.168.1.101")
|
|
|
|
assert status["is_primary"] is True
|
|
assert status["uptime"] == 3600
|
|
assert status["advertise_url"] == "http://192.168.1.101:20202"
|
|
|
|
@patch("requests.get")
|
|
def test_get_node_status_fallback(mock_get):
|
|
"""Test fetching LiteFS status via /debug/vars fallback."""
|
|
def side_effect(url, **kwargs):
|
|
m = MagicMock()
|
|
if "/status" in url:
|
|
m.status_code = 404
|
|
return m
|
|
elif "/debug/vars" in url:
|
|
m.status_code = 200
|
|
m.json.return_value = {
|
|
"store": {"isPrimary": True}
|
|
}
|
|
return m
|
|
return m
|
|
|
|
mock_get.side_effect = side_effect
|
|
|
|
status = litefs_client.get_node_status("192.168.1.101")
|
|
|
|
assert status["is_primary"] is True
|
|
assert status["uptime"] == "N/A"
|
|
assert status["advertise_url"] == "http://192.168.1.101:20202"
|
|
|
|
@patch("requests.get")
|
|
def test_get_node_status_error(mock_get):
|
|
"""Test fetching LiteFS status with a connection error."""
|
|
mock_get.side_effect = Exception("Connection failed")
|
|
|
|
status = litefs_client.get_node_status("192.168.1.101")
|
|
|
|
assert "error" in status
|
|
assert status["is_primary"] is False
|
|
|
|
@patch("nomad_client.exec_command")
|
|
def test_get_node_status_nomad_exec(mock_exec):
|
|
"""Test fetching LiteFS status via nomad alloc exec."""
|
|
# Mock LiteFS status output (text format)
|
|
mock_status_output = """
|
|
Config:
|
|
Path: /etc/litefs.yml
|
|
...
|
|
Status:
|
|
Primary: true
|
|
Uptime: 1h5m10s
|
|
Replication Lag: 0s
|
|
"""
|
|
mock_exec.return_value = mock_status_output
|
|
|
|
# We need to mock requests.get to fail first
|
|
with patch("requests.get") as mock_get:
|
|
mock_get.side_effect = Exception("HTTP failed")
|
|
|
|
status = litefs_client.get_node_status("1.1.1.1", alloc_id="abc12345")
|
|
|
|
assert status["is_primary"] is True
|
|
assert status["uptime"] == "1h5m10s"
|
|
# Since it's primary, lag might not be shown or be 0
|
|
assert status["replication_lag"] == "0s" |