ipng/kumacli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reformat with black

Browse Source
This commit is contained in:
Pim van Pelt
2025-08-03 11:03:38 +02:00
parent 81c7f9c9bd
commit c1d70cd9b6
8 changed files with 239 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/__init__.py
View File

@@ -1 +1 @@
# Test package for kumacli
# Test package for kumacli

21
tests/conftest.py
View File

@@ -2,8 +2,9 @@
import sys
import os
# Add the src directory to Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
import pytest
from unittest.mock import Mock, MagicMock
@@ -28,7 +29,7 @@ def mock_monitors():
"type": "http",
"url": "https://example.com",
"active": True,
"parent": None
"parent": None,
},
{
"id": 2,
@@ -36,14 +37,14 @@ def mock_monitors():
"type": "http",
"url": "https://test.com",
"active": False,
"parent": None
"parent": None,
},
{
"id": 3,
"name": "Group Monitor",
"type": "group",
"active": True,
"parent": None
"parent": None,
},
{
"id": 4,
@@ -51,8 +52,8 @@ def mock_monitors():
"type": "http",
"url": "https://child.com",
"active": False,
"parent": 3
}
"parent": 3,
},
]
@@ -65,13 +66,13 @@ def mock_maintenances():
"title": "Test Maintenance",
"description": "Test maintenance description",
"strategy": "single",
"active": True
"active": True,
},
{
"id": 2,
"title": "Inactive Maintenance",
"description": "Inactive maintenance description",
"strategy": "single",
"active": False
}
]
"active": False,
},
]

96
tests/test_cli_integration.py
View File

@@ -16,31 +16,31 @@ class TestCLIIntegration:
"""Test monitor parser setup"""
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
monitor_parser = setup_monitor_parser(subparsers)
# Verify parser is created
assert monitor_parser is not None
assert hasattr(setup_monitor_parser, '_parser')
assert hasattr(setup_monitor_parser, "_parser")
def test_maintenance_parser_setup(self):
"""Test maintenance parser setup"""
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
maintenance_parser = setup_maintenance_parser(subparsers)
# Verify parser is created
assert maintenance_parser is not None
assert hasattr(setup_maintenance_parser, '_parser')
assert hasattr(setup_maintenance_parser, "_parser")
def test_info_parser_setup(self):
"""Test info parser setup"""
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
info_parser = setup_info_parser(subparsers)
# Verify parser is created
assert info_parser is not None
@@ -49,14 +49,14 @@ class TestCLIIntegration:
# Setup
mock_args = Mock()
mock_args.monitor_action = None
# Setup parser reference
mock_parser = Mock()
setup_monitor_parser._parser = mock_parser
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is False
mock_parser.print_help.assert_called_once()
@@ -66,14 +66,14 @@ class TestCLIIntegration:
# Setup
mock_args = Mock()
mock_args.maintenance_action = None
# Setup parser reference
mock_parser = Mock()
setup_maintenance_parser._parser = mock_parser
# Execute
result = handle_maintenance_command(mock_args, mock_client)
# Verify
assert result is False
mock_parser.print_help.assert_called_once()
@@ -85,7 +85,7 @@ class TestCLIIntegration:
mock_args.monitor_action = "pause"
mock_args.monitor = ["test*"]
mock_args.group = ["web-services"]
# Mock client methods
mock_client.find_monitors_by_pattern.return_value = [
{"id": 1, "name": "test-monitor"}
@@ -94,10 +94,10 @@ class TestCLIIntegration:
{"id": 2, "name": "web-service-monitor"}
]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.find_monitors_by_pattern.assert_called_once_with(["test*"])
@@ -113,13 +113,13 @@ class TestCLIIntegration:
mock_args.monitor = None
mock_args.group = None
mock_args.all = True
mock_client.api.get_monitors.return_value = mock_monitors
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.api.get_monitors.assert_called_once()
@@ -135,9 +135,11 @@ class TestArgumentParsing:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
setup_monitor_parser(subparsers)
# Test with monitor patterns
args = parser.parse_args(["monitor", "list", "--monitor", "web*", "--monitor", "api*"])
args = parser.parse_args(
["monitor", "list", "--monitor", "web*", "--monitor", "api*"]
)
assert args.resource == "monitor"
assert args.monitor_action == "list"
assert args.monitor == ["web*", "api*"]
@@ -147,7 +149,7 @@ class TestArgumentParsing:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
setup_monitor_parser(subparsers)
# Test with group patterns
args = parser.parse_args(["monitor", "pause", "--group", "production"])
assert args.resource == "monitor"
@@ -159,7 +161,7 @@ class TestArgumentParsing:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
setup_monitor_parser(subparsers)
# Test with --all flag
args = parser.parse_args(["monitor", "resume", "--all"])
assert args.resource == "monitor"
@@ -171,15 +173,22 @@ class TestArgumentParsing:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
setup_maintenance_parser(subparsers)
# Test maintenance add
args = parser.parse_args([
"maintenance", "add",
"--title", "Server Update",
"--description", "Updating server software",
"--duration", "2h",
"--monitor", "server*"
])
args = parser.parse_args(
[
"maintenance",
"add",
"--title",
"Server Update",
"--description",
"Updating server software",
"--duration",
"2h",
"--monitor",
"server*",
]
)
assert args.resource == "maintenance"
assert args.maintenance_action == "add"
assert args.title == "Server Update"
@@ -192,7 +201,7 @@ class TestArgumentParsing:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
setup_maintenance_parser(subparsers)
# Test delete by ID
args = parser.parse_args(["maintenance", "delete", "--id", "123"])
assert args.resource == "maintenance"
@@ -210,7 +219,7 @@ class TestArgumentParsing:
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="resource")
setup_info_parser(subparsers)
# Test info command
args = parser.parse_args(["info"])
assert args.resource == "info"
@@ -224,31 +233,34 @@ class TestErrorHandling:
mock_args.monitor_action = "pause"
mock_args.monitor = ["nonexistent*"]
mock_args.group = None
# Mock no matches found
mock_client.find_monitors_by_pattern.return_value = []
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify error handling
assert result is True # Command completes even with no matches
captured = capsys.readouterr()
assert "Error: No monitors found matching the specified patterns or groups" in captured.out
assert (
"Error: No monitors found matching the specified patterns or groups"
in captured.out
)
def test_maintenance_command_resilience(self, mock_client, capsys):
"""Test maintenance command handles API errors"""
# Setup
mock_args = Mock()
mock_args.maintenance_action = "list"
# Mock API error
mock_client.api.get_maintenances.side_effect = Exception("Connection timeout")
# Execute
result = handle_maintenance_command(mock_args, mock_client)
# Verify error handling
assert result is True # Command completes even with error
captured = capsys.readouterr()
assert "Error listing maintenances: Connection timeout" in captured.out
assert "Error listing maintenances: Connection timeout" in captured.out

88
tests/test_client.py
View File

@@ -11,7 +11,7 @@ class TestKumaClient:
def test_parse_duration_minutes(self):
"""Test parsing duration in minutes"""
client = KumaClient("http://test.com")
assert client.parse_duration("90m") == 5400 # 90 * 60
assert client.parse_duration("1m") == 60
assert client.parse_duration("120m") == 7200
@@ -19,7 +19,7 @@ class TestKumaClient:
def test_parse_duration_hours(self):
"""Test parsing duration in hours"""
client = KumaClient("http://test.com")
assert client.parse_duration("1h") == 3600 # 1 * 3600
assert client.parse_duration("2h") == 7200
assert client.parse_duration("24h") == 86400
@@ -27,7 +27,7 @@ class TestKumaClient:
def test_parse_duration_seconds(self):
"""Test parsing duration in seconds"""
client = KumaClient("http://test.com")
assert client.parse_duration("3600s") == 3600
assert client.parse_duration("60s") == 60
assert client.parse_duration("1s") == 1
@@ -35,42 +35,42 @@ class TestKumaClient:
def test_parse_duration_default(self):
"""Test parsing duration with default value"""
client = KumaClient("http://test.com")
assert client.parse_duration(None) == 5400 # Default 90 minutes
assert client.parse_duration("") == 5400
def test_parse_duration_invalid(self):
"""Test parsing invalid duration format"""
client = KumaClient("http://test.com")
with pytest.raises(ValueError, match="Invalid duration format"):
client.parse_duration("invalid")
with pytest.raises(ValueError, match="Invalid duration format"):
client.parse_duration("90x")
with pytest.raises(ValueError, match="Invalid duration format"):
client.parse_duration("90")
def test_parse_start_time_none(self):
"""Test parsing start time with None (current time)"""
client = KumaClient("http://test.com")
before = datetime.utcnow()
result = client.parse_start_time(None)
after = datetime.utcnow()
assert before <= result <= after
def test_parse_start_time_iso_format(self):
"""Test parsing ISO format start time"""
client = KumaClient("http://test.com")
# Test ISO format with Z
result = client.parse_start_time("2023-12-25T10:30:00Z")
expected = datetime(2023, 12, 25, 10, 30, 0)
assert result == expected
# Test ISO format with timezone
result = client.parse_start_time("2023-12-25T10:30:00+00:00")
assert result == expected
@@ -78,17 +78,17 @@ class TestKumaClient:
def test_parse_start_time_common_formats(self):
"""Test parsing common date/time formats"""
client = KumaClient("http://test.com")
# Full datetime
result = client.parse_start_time("2023-12-25 10:30:00")
expected = datetime(2023, 12, 25, 10, 30, 0)
assert result == expected
# Date and hour:minute
result = client.parse_start_time("2023-12-25 10:30")
expected = datetime(2023, 12, 25, 10, 30, 0)
assert result == expected
# Date only
result = client.parse_start_time("2023-12-25")
expected = datetime(2023, 12, 25, 0, 0, 0)
@@ -97,10 +97,10 @@ class TestKumaClient:
def test_parse_start_time_invalid(self):
"""Test parsing invalid start time format"""
client = KumaClient("http://test.com")
with pytest.raises(ValueError, match="Invalid start time format"):
client.parse_start_time("invalid-date")
with pytest.raises(ValueError, match="Invalid start time format"):
client.parse_start_time("2023-13-45")
@@ -108,20 +108,20 @@ class TestKumaClient:
"""Test finding monitors by pattern successfully"""
client = KumaClient("http://test.com")
client.api = Mock()
mock_monitors = [
{"id": 1, "name": "Web Server"},
{"id": 2, "name": "API Server"},
{"id": 3, "name": "Database"},
{"id": 4, "name": "Web Frontend"}
{"id": 4, "name": "Web Frontend"},
]
client.api.get_monitors.return_value = mock_monitors
# Test exact match
result = client.find_monitors_by_pattern(["Web Server"])
assert len(result) == 1
assert result[0]["name"] == "Web Server"
# Test wildcard pattern
result = client.find_monitors_by_pattern(["Web*"])
assert len(result) == 2
@@ -133,13 +133,13 @@ class TestKumaClient:
"""Test finding monitors by pattern is case insensitive"""
client = KumaClient("http://test.com")
client.api = Mock()
mock_monitors = [
{"id": 1, "name": "Web Server"},
{"id": 2, "name": "API Server"}
{"id": 2, "name": "API Server"},
]
client.api.get_monitors.return_value = mock_monitors
# Test case insensitive matching
result = client.find_monitors_by_pattern(["web*"])
assert len(result) == 1
@@ -149,12 +149,10 @@ class TestKumaClient:
"""Test finding monitors with no matches"""
client = KumaClient("http://test.com")
client.api = Mock()
mock_monitors = [
{"id": 1, "name": "Web Server"}
]
mock_monitors = [{"id": 1, "name": "Web Server"}]
client.api.get_monitors.return_value = mock_monitors
result = client.find_monitors_by_pattern(["Database*"])
assert len(result) == 0
@@ -162,12 +160,10 @@ class TestKumaClient:
"""Test finding monitors removes duplicates"""
client = KumaClient("http://test.com")
client.api = Mock()
mock_monitors = [
{"id": 1, "name": "Web Server"}
]
mock_monitors = [{"id": 1, "name": "Web Server"}]
client.api.get_monitors.return_value = mock_monitors
# Same monitor should match both patterns
result = client.find_monitors_by_pattern(["Web*", "*Server"])
assert len(result) == 1
@@ -177,41 +173,41 @@ class TestKumaClient:
"""Test finding monitors handles API errors"""
client = KumaClient("http://test.com")
client.api = Mock()
client.api.get_monitors.side_effect = Exception("API Error")
result = client.find_monitors_by_pattern(["Web*"])
assert len(result) == 0
captured = capsys.readouterr()
assert "Error finding monitors: API Error" in captured.out
@patch('kumacli.client.UptimeKumaApi')
@patch("kumacli.client.UptimeKumaApi")
def test_connect_success(self, mock_api_class, capsys):
"""Test successful connection"""
mock_api = Mock()
mock_api_class.return_value = mock_api
mock_api.login.return_value = True
client = KumaClient("http://test.com", "user", "pass")
result = client.connect()
assert result is True
assert client.api is mock_api
mock_api_class.assert_called_once_with("http://test.com")
mock_api.login.assert_called_once_with("user", "pass")
captured = capsys.readouterr()
assert "Connected to http://test.com" in captured.out
@patch('kumacli.client.UptimeKumaApi')
@patch("kumacli.client.UptimeKumaApi")
def test_connect_failure(self, mock_api_class, capsys):
"""Test connection failure"""
mock_api_class.side_effect = Exception("Connection failed")
client = KumaClient("http://test.com", "user", "pass")
result = client.connect()
assert result is False
captured = capsys.readouterr()
assert "Failed to connect: Connection failed" in captured.out
@@ -220,7 +216,7 @@ class TestKumaClient:
"""Test disconnection"""
client = KumaClient("http://test.com")
client.api = Mock()
client.disconnect()
client.api.disconnect.assert_called_once()
client.api.disconnect.assert_called_once()

30
tests/test_info.py
View File

@@ -15,15 +15,15 @@ class TestInfoCommands:
mock_info_data = {
"version": "1.23.0",
"hostname": "kuma-server",
"primaryBaseURL": "https://status.example.com"
"primaryBaseURL": "https://status.example.com",
}
mock_client.api.info.return_value = mock_info_data
info_commands = InfoCommands(mock_client)
# Execute
info_commands.get_info()
# Verify
mock_client.api.info.assert_called_once()
captured = capsys.readouterr()
@@ -36,12 +36,12 @@ class TestInfoCommands:
"""Test info command with empty response"""
# Setup
mock_client.api.info.return_value = None
info_commands = InfoCommands(mock_client)
# Execute
info_commands.get_info()
# Verify
mock_client.api.info.assert_called_once()
captured = capsys.readouterr()
@@ -51,12 +51,12 @@ class TestInfoCommands:
"""Test info command with API error"""
# Setup
mock_client.api.info.side_effect = Exception("Connection failed")
info_commands = InfoCommands(mock_client)
# Execute
info_commands.get_info()
# Verify
mock_client.api.info.assert_called_once()
captured = capsys.readouterr()
@@ -70,10 +70,10 @@ class TestInfoCommandHandler:
mock_args = Mock()
mock_info_data = {"version": "1.23.0"}
mock_client.api.info.return_value = mock_info_data
# Execute
result = handle_info_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.api.info.assert_called_once()
@@ -83,10 +83,10 @@ class TestInfoCommandHandler:
# Setup
mock_args = Mock()
mock_client.api.info.side_effect = Exception("API Error")
# Execute
result = handle_info_command(mock_args, mock_client)
# Verify
assert result is True # Handler always returns True
mock_client.api.info.assert_called_once()
mock_client.api.info.assert_called_once()

83
tests/test_maintenance.py
View File

@@ -13,12 +13,12 @@ class TestMaintenanceCommands:
"""Test successful maintenance listing"""
# Setup
mock_client.api.get_maintenances.return_value = mock_maintenances
maintenance_commands = MaintenanceCommands(mock_client)
# Execute
maintenance_commands.list_maintenances()
# Verify
mock_client.api.get_maintenances.assert_called_once()
captured = capsys.readouterr()
@@ -31,12 +31,12 @@ class TestMaintenanceCommands:
"""Test maintenance listing with no maintenances"""
# Setup
mock_client.api.get_maintenances.return_value = []
maintenance_commands = MaintenanceCommands(mock_client)
# Execute
maintenance_commands.list_maintenances()
# Verify
captured = capsys.readouterr()
assert "No maintenances found" in captured.out
@@ -45,12 +45,12 @@ class TestMaintenanceCommands:
"""Test maintenance listing with API error"""
# Setup
mock_client.api.get_maintenances.side_effect = Exception("API Error")
maintenance_commands = MaintenanceCommands(mock_client)
# Execute
maintenance_commands.list_maintenances()
# Verify
captured = capsys.readouterr()
assert "Error listing maintenances: API Error" in captured.out
@@ -60,30 +60,37 @@ class TestMaintenanceCommands:
# Setup
mock_maintenance = {"id": 1, "title": "Test Maintenance"}
mock_client.api.get_maintenance.return_value = mock_maintenance
mock_client.api.delete_maintenance.return_value = {"msg": "Deleted Successfully"}
mock_client.api.delete_maintenance.return_value = {
"msg": "Deleted Successfully"
}
maintenance_commands = MaintenanceCommands(mock_client)
# Execute
maintenance_commands.delete_maintenance(maintenance_id=1)
# Verify
mock_client.api.get_maintenance.assert_called_once_with(1)
mock_client.api.delete_maintenance.assert_called_once_with(1)
captured = capsys.readouterr()
assert "Successfully deleted maintenance 'Test Maintenance' (ID: 1)" in captured.out
assert (
"Successfully deleted maintenance 'Test Maintenance' (ID: 1)"
in captured.out
)
def test_delete_all_maintenances(self, mock_client, mock_maintenances, capsys):
"""Test deleting all maintenances"""
# Setup
mock_client.api.get_maintenances.return_value = mock_maintenances
mock_client.api.delete_maintenance.return_value = {"msg": "Deleted Successfully"}
mock_client.api.delete_maintenance.return_value = {
"msg": "Deleted Successfully"
}
maintenance_commands = MaintenanceCommands(mock_client)
# Execute
maintenance_commands.delete_maintenance(delete_all=True)
# Verify
assert mock_client.api.delete_maintenance.call_count == 2
captured = capsys.readouterr()
@@ -93,13 +100,16 @@ class TestMaintenanceCommands:
def test_delete_maintenance_no_params(self, mock_client, capsys):
"""Test deleting maintenance without parameters"""
maintenance_commands = MaintenanceCommands(mock_client)
# Execute
maintenance_commands.delete_maintenance()
# Verify
captured = capsys.readouterr()
assert "Error: Either --id or --all flag is required for delete operation" in captured.out
assert (
"Error: Either --id or --all flag is required for delete operation"
in captured.out
)
class TestMaintenanceCommandHandler:
@@ -108,15 +118,15 @@ class TestMaintenanceCommandHandler:
# Setup
mock_args = Mock()
mock_args.maintenance_action = None
# Mock the parser setup
with patch('kumacli.cmd.maintenance.setup_maintenance_parser') as mock_setup:
with patch("kumacli.cmd.maintenance.setup_maintenance_parser") as mock_setup:
mock_parser = Mock()
mock_setup._parser = mock_parser
# Execute
result = handle_maintenance_command(mock_args, mock_client)
# Verify
assert result is False
@@ -126,10 +136,10 @@ class TestMaintenanceCommandHandler:
mock_args = Mock()
mock_args.maintenance_action = "list"
mock_client.api.get_maintenances.return_value = mock_maintenances
# Execute
result = handle_maintenance_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.api.get_maintenances.assert_called_once()
@@ -141,14 +151,16 @@ class TestMaintenanceCommandHandler:
mock_args.maintenance_action = "delete"
mock_args.id = 1
mock_args.all = False
mock_maintenance = {"id": 1, "title": "Test Maintenance"}
mock_client.api.get_maintenance.return_value = mock_maintenance
mock_client.api.delete_maintenance.return_value = {"msg": "Deleted Successfully"}
mock_client.api.delete_maintenance.return_value = {
"msg": "Deleted Successfully"
}
# Execute
result = handle_maintenance_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.api.delete_maintenance.assert_called_once_with(1)
@@ -158,11 +170,14 @@ class TestMaintenanceCommandHandler:
# Setup
mock_args = Mock()
mock_args.maintenance_action = "unknown"
# Execute
result = handle_maintenance_command(mock_args, mock_client)
# Verify
assert result is False
captured = capsys.readouterr()
assert "Unknown maintenance action. Use --help for usage information." in captured.out
assert (
"Unknown maintenance action. Use --help for usage information."
in captured.out
)

114
tests/test_monitor.py
View File

@@ -5,7 +5,11 @@ from unittest.mock import Mock, patch
from io import StringIO
import sys
from kumacli.cmd.monitor import MonitorCommands, handle_monitor_command, setup_monitor_parser
from kumacli.cmd.monitor import (
MonitorCommands,
handle_monitor_command,
setup_monitor_parser,
)
class TestMonitorCommands:
@@ -15,21 +19,21 @@ class TestMonitorCommands:
mock_client.api.get_monitors.return_value = mock_monitors
mock_client.find_monitors_by_pattern.return_value = [
{"id": 1, "name": "Test Monitor 1"},
{"id": 2, "name": "Test Monitor 2"}
{"id": 2, "name": "Test Monitor 2"},
]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.pause_monitors(monitor_patterns=["Test*"])
# Verify
mock_client.find_monitors_by_pattern.assert_called_once_with(["Test*"])
assert mock_client.api.pause_monitor.call_count == 2
mock_client.api.pause_monitor.assert_any_call(1)
mock_client.api.pause_monitor.assert_any_call(2)
captured = capsys.readouterr()
assert "Found 2 matching monitors to pause:" in captured.out
assert "Paused monitor 'Test Monitor 1' (ID: 1)" in captured.out
@@ -43,16 +47,16 @@ class TestMonitorCommands:
{"id": 4, "name": "Child Monitor"}
]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.pause_monitors(group_patterns=["Group*"])
# Verify
mock_client.get_monitors_in_groups.assert_called_once_with(["Group*"])
mock_client.api.pause_monitor.assert_called_once_with(4)
captured = capsys.readouterr()
assert "Found 1 matching monitors to pause:" in captured.out
assert "Paused monitor 'Child Monitor' (ID: 4)" in captured.out
@@ -60,10 +64,10 @@ class TestMonitorCommands:
def test_pause_monitors_no_patterns(self, mock_client, capsys):
"""Test pausing monitors without patterns"""
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.pause_monitors()
# Verify
captured = capsys.readouterr()
assert "Error: Either --monitor or --group flag is required." in captured.out
@@ -72,15 +76,18 @@ class TestMonitorCommands:
"""Test pausing monitors with no matches"""
# Setup
mock_client.find_monitors_by_pattern.return_value = []
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.pause_monitors(monitor_patterns=["NonExistent*"])
# Verify
captured = capsys.readouterr()
assert "Error: No monitors found matching the specified patterns or groups" in captured.out
assert (
"Error: No monitors found matching the specified patterns or groups"
in captured.out
)
def test_pause_monitors_api_error(self, mock_client, capsys):
"""Test pausing monitors with API error"""
@@ -89,12 +96,12 @@ class TestMonitorCommands:
{"id": 1, "name": "Test Monitor 1"}
]
mock_client.api.pause_monitor.side_effect = Exception("API Error")
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.pause_monitors(monitor_patterns=["Test*"])
# Verify
captured = capsys.readouterr()
assert "Failed to pause monitor 'Test Monitor 1': API Error" in captured.out
@@ -107,15 +114,15 @@ class TestMonitorCommands:
{"id": 2, "name": "Test Monitor 2"}
]
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.resume_monitors(monitor_patterns=["Test*"])
# Verify
mock_client.api.resume_monitor.assert_called_once_with(2)
captured = capsys.readouterr()
assert "Found 1 matching monitors to resume:" in captured.out
assert "Resumed monitor 'Test Monitor 2' (ID: 2)" in captured.out
@@ -125,18 +132,18 @@ class TestMonitorCommands:
# Setup
mock_client.api.get_monitors.return_value = mock_monitors
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.resume_monitors(resume_all=True)
# Verify
# Should resume monitors with active=False (monitors 2 and 4)
assert mock_client.api.resume_monitor.call_count == 2
mock_client.api.resume_monitor.assert_any_call(2)
mock_client.api.resume_monitor.assert_any_call(4)
captured = capsys.readouterr()
assert "Found 2 paused monitors to resume:" in captured.out
assert "Successfully resumed 2 out of 2 monitors" in captured.out
@@ -144,16 +151,14 @@ class TestMonitorCommands:
def test_resume_monitors_all_no_paused(self, mock_client, capsys):
"""Test resuming all paused monitors when none are paused"""
# Setup
active_monitors = [
{"id": 1, "name": "Active Monitor", "active": True}
]
active_monitors = [{"id": 1, "name": "Active Monitor", "active": True}]
mock_client.api.get_monitors.return_value = active_monitors
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.resume_monitors(resume_all=True)
# Verify
captured = capsys.readouterr()
assert "No paused monitors found to resume" in captured.out
@@ -161,13 +166,16 @@ class TestMonitorCommands:
def test_resume_monitors_no_args(self, mock_client, capsys):
"""Test resuming monitors without any arguments"""
monitor_commands = MonitorCommands(mock_client)
# Execute
monitor_commands.resume_monitors()
# Verify
captured = capsys.readouterr()
assert "Error: Either --monitor, --group, or --all flag is required." in captured.out
assert (
"Error: Either --monitor, --group, or --all flag is required."
in captured.out
)
class TestMonitorCommandHandler:
@@ -176,15 +184,15 @@ class TestMonitorCommandHandler:
# Setup
mock_args = Mock()
mock_args.monitor_action = None
# Mock the parser setup to avoid importing issues
with patch('kumacli.cmd.monitor.setup_monitor_parser') as mock_setup:
with patch("kumacli.cmd.monitor.setup_monitor_parser") as mock_setup:
mock_parser = Mock()
mock_setup._parser = mock_parser
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is False
@@ -195,15 +203,15 @@ class TestMonitorCommandHandler:
mock_args.monitor_action = "pause"
mock_args.monitor = ["test*"]
mock_args.group = None
mock_client.find_monitors_by_pattern.return_value = [
{"id": 1, "name": "test monitor"}
]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.api.pause_monitor.assert_called_once_with(1)
@@ -216,15 +224,15 @@ class TestMonitorCommandHandler:
mock_args.monitor = ["test*"]
mock_args.group = None
mock_args.all = False
mock_client.find_monitors_by_pattern.return_value = [
{"id": 1, "name": "test monitor"}
]
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is True
mock_client.api.resume_monitor.assert_called_once_with(1)
@@ -237,13 +245,13 @@ class TestMonitorCommandHandler:
mock_args.monitor = None
mock_args.group = None
mock_args.all = True
mock_client.api.get_monitors.return_value = mock_monitors
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is True
# Should resume paused monitors (monitors 2 and 4)
@@ -254,11 +262,13 @@ class TestMonitorCommandHandler:
# Setup
mock_args = Mock()
mock_args.monitor_action = "unknown"
# Execute
result = handle_monitor_command(mock_args, mock_client)
# Verify
assert result is False
captured = capsys.readouterr()
assert "Unknown monitor action. Use --help for usage information." in captured.out
assert (
"Unknown monitor action. Use --help for usage information." in captured.out
)