58 lines
1.7 KiB
Python
58 lines
1.7 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 |