274 lines
9.5 KiB
Python
274 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
import argparse
|
|
from io import StringIO
|
|
import sys
|
|
|
|
from kumacli.cmd.monitor import setup_monitor_parser, handle_monitor_command
|
|
from kumacli.cmd.maintenance import setup_maintenance_parser, handle_maintenance_command
|
|
from kumacli.cmd.info import setup_info_parser, handle_info_command
|
|
|
|
|
|
class TestCLIIntegration:
|
|
def test_monitor_parser_setup(self):
|
|
"""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 monitor_parser.prog.endswith("monitor")
|
|
|
|
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 maintenance_parser.prog.endswith("maintenance")
|
|
|
|
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
|
|
|
|
def test_monitor_help_message(self, mock_client, capsys):
|
|
"""Test monitor command shows help when no action specified"""
|
|
# Setup
|
|
mock_args = Mock()
|
|
mock_args.monitor_action = None
|
|
|
|
# Setup parser reference to simulate having called setup_monitor_parser
|
|
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()
|
|
|
|
def test_maintenance_help_message(self, mock_client, capsys):
|
|
"""Test maintenance command shows help when no action specified"""
|
|
# Setup
|
|
mock_args = Mock()
|
|
mock_args.maintenance_action = None
|
|
|
|
# Setup parser reference to simulate having called setup_maintenance_parser
|
|
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()
|
|
|
|
def test_monitor_command_with_full_args(self, mock_client):
|
|
"""Test monitor command with complete argument structure"""
|
|
# Setup
|
|
mock_args = Mock()
|
|
mock_args.monitor_action = "pause"
|
|
mock_args.monitor = ["test*"]
|
|
mock_args.group = ["web-services"]
|
|
|
|
# Mock client methods
|
|
mock_client.find_and_get_monitors.return_value = [
|
|
{"id": 1, "name": "test-monitor"},
|
|
{"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_and_get_monitors.assert_called_once_with(
|
|
["test*"], ["web-services"]
|
|
)
|
|
# Should pause both monitors (deduplicated)
|
|
assert mock_client.api.pause_monitor.call_count == 2
|
|
|
|
def test_resume_all_monitors_integration(self, mock_client, mock_monitors):
|
|
"""Test resume all monitors integration"""
|
|
# Setup
|
|
mock_args = Mock()
|
|
mock_args.monitor_action = "resume"
|
|
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()
|
|
# Should resume only paused monitors (ID 2 and 4 from mock_monitors)
|
|
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)
|
|
|
|
|
|
class TestArgumentParsing:
|
|
def test_monitor_list_arguments(self):
|
|
"""Test monitor list command argument parsing"""
|
|
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*"]
|
|
)
|
|
assert args.resource == "monitor"
|
|
assert args.monitor_action == "list"
|
|
assert args.monitor == ["web*", "api*"]
|
|
|
|
def test_monitor_pause_arguments(self):
|
|
"""Test monitor pause command argument parsing"""
|
|
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"
|
|
assert args.monitor_action == "pause"
|
|
assert args.group == ["production"]
|
|
|
|
def test_monitor_resume_all_arguments(self):
|
|
"""Test monitor resume all command argument parsing"""
|
|
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"
|
|
assert args.monitor_action == "resume"
|
|
assert args.all is True
|
|
|
|
def test_maintenance_add_arguments(self):
|
|
"""Test maintenance add command argument parsing"""
|
|
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*",
|
|
]
|
|
)
|
|
assert args.resource == "maintenance"
|
|
assert args.maintenance_action == "add"
|
|
assert args.title == "Server Update"
|
|
assert args.description == "Updating server software"
|
|
assert args.duration == "2h"
|
|
assert args.monitor == ["server*"]
|
|
|
|
def test_maintenance_delete_arguments(self):
|
|
"""Test maintenance delete command argument parsing"""
|
|
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"
|
|
assert args.maintenance_action == "delete"
|
|
assert args.id == 123
|
|
|
|
# Test delete all
|
|
args = parser.parse_args(["maintenance", "delete", "--all"])
|
|
assert args.resource == "maintenance"
|
|
assert args.maintenance_action == "delete"
|
|
assert args.all is True
|
|
|
|
def test_info_arguments(self):
|
|
"""Test info command argument parsing"""
|
|
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"
|
|
|
|
|
|
class TestErrorHandling:
|
|
def test_monitor_command_resilience(self, mock_client, capsys):
|
|
"""Test monitor command handles various error conditions"""
|
|
# Setup
|
|
mock_args = Mock()
|
|
mock_args.monitor_action = "pause"
|
|
mock_args.monitor = ["nonexistent*"]
|
|
mock_args.group = None
|
|
|
|
# Mock no matches found
|
|
def mock_find_and_get_monitors(*args, **kwargs):
|
|
print("Error: No monitors found matching the specified patterns or groups")
|
|
return []
|
|
|
|
mock_client.find_and_get_monitors.side_effect = mock_find_and_get_monitors
|
|
|
|
# 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
|
|
)
|
|
|
|
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
|
|
from uptime_kuma_api import UptimeKumaException
|
|
|
|
mock_client.api.get_maintenances.side_effect = UptimeKumaException(
|
|
"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
|