Compare commits

...

6 Commits

Author SHA1 Message Date
Pim van Pelt
7eafb1e68e Release v1.3.0 2025-08-03 11:51:55 +02:00
Pim van Pelt
11aad79720 Restore behavior where 'kumacli monitor' shows the subcommands 2025-08-03 11:50:44 +02:00
Pim van Pelt
64e1ca124f Fix tests after refactor 2025-08-03 11:43:53 +02:00
Pim van Pelt
f1d10458c6 Use UptimeKumaException exceptions 2025-08-03 11:27:36 +02:00
Pim van Pelt
1e3999eee2 Refactor: add find_monitors_by_globs() and find_and_get_monitors() 2025-08-03 11:21:10 +02:00
Pim van Pelt
2ddcc00cda Fix some pylint issues 2025-08-03 11:10:24 +02:00
12 changed files with 271 additions and 221 deletions

View File

@@ -29,6 +29,9 @@ kumacli --url http://localhost:3001 --username admin --password password monitor
### Monitor Commands ### Monitor Commands
```bash ```bash
# Show available subcommands
kumacli monitor
# List all monitors # List all monitors
kumacli monitor list kumacli monitor list
@@ -40,11 +43,23 @@ kumacli monitor list --group "production*"
# Combine filters # Combine filters
kumacli monitor list --monitor "*api*" --group "web*" kumacli monitor list --monitor "*api*" --group "web*"
# Pause monitors
kumacli monitor pause --monitor "*api*"
kumacli monitor pause --group "production*"
# Resume monitors
kumacli monitor resume --monitor "*api*"
kumacli monitor resume --group "production*"
kumacli monitor resume --all
``` ```
### Maintenance Commands ### Maintenance Commands
```bash ```bash
# Show available subcommands
kumacli maintenance
# Create maintenance for specific monitors (90 minutes, starting now) # Create maintenance for specific monitors (90 minutes, starting now)
kumacli maintenance add --monitor "*nextcloud*" kumacli maintenance add --monitor "*nextcloud*"

View File

@@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as fh:
setup( setup(
name="kumacli", name="kumacli",
version="1.2.0", version="1.3.0",
author="Uptime Kuma CLI", author="Uptime Kuma CLI",
description="A command-line interface for Uptime Kuma", description="A command-line interface for Uptime Kuma",
long_description=long_description, long_description=long_description,

View File

@@ -1,12 +1,15 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Uptime Kuma client wrapper for API operations."""
import fnmatch import fnmatch
import re import re
from datetime import datetime, timedelta from datetime import datetime
from uptime_kuma_api import UptimeKumaApi from uptime_kuma_api import UptimeKumaApi, UptimeKumaException
class KumaClient: class KumaClient:
"""Client wrapper for Uptime Kuma API operations."""
def __init__(self, url, username=None, password=None): def __init__(self, url, username=None, password=None):
self.url = url self.url = url
self.username = username self.username = username
@@ -29,9 +32,9 @@ class KumaClient:
if unit == "s": if unit == "s":
return value return value
elif unit == "m": if unit == "m":
return value * 60 return value * 60
elif unit == "h": if unit == "h":
return value * 3600 return value * 3600
raise ValueError(f"Invalid duration unit: {unit}") raise ValueError(f"Invalid duration unit: {unit}")
@@ -71,10 +74,10 @@ class KumaClient:
try: try:
self.api = UptimeKumaApi(self.url) self.api = UptimeKumaApi(self.url)
if self.username and self.password: if self.username and self.password:
result = self.api.login(self.username, self.password) self.api.login(self.username, self.password)
print(f"Connected to {self.url}") print(f"Connected to {self.url}")
return True return True
except Exception as e: except UptimeKumaException as e:
print(f"Failed to connect: {e}") print(f"Failed to connect: {e}")
return False return False
@@ -108,7 +111,7 @@ class KumaClient:
return unique_monitors return unique_monitors
except Exception as e: except UptimeKumaException as e:
print(f"Error finding monitors: {e}") print(f"Error finding monitors: {e}")
return [] return []
@@ -138,7 +141,7 @@ class KumaClient:
return unique_groups return unique_groups
except Exception as e: except UptimeKumaException as e:
print(f"Error finding groups: {e}") print(f"Error finding groups: {e}")
return [] return []
@@ -165,6 +168,108 @@ class KumaClient:
return group_members return group_members
except Exception as e: except UptimeKumaException as e:
print(f"Error getting group members: {e}") print(f"Error getting group members: {e}")
return [] return []
def find_monitors_by_globs(self, monitor_patterns=None, group_patterns=None):
"""Find monitor IDs by name patterns and/or group patterns.
Args:
monitor_patterns: List of monitor name patterns (supports wildcards)
group_patterns: List of group name patterns (supports wildcards)
Returns:
List of monitor IDs (integers) that match the criteria
"""
try:
# Check if we have either monitor patterns or group patterns
if not monitor_patterns and not group_patterns:
print(
"Error: Either monitor or group patterns required. "
"Specify at least one pattern."
)
return []
matched_monitors = []
# Find monitors by patterns if specified
if monitor_patterns:
pattern_monitors = self.find_monitors_by_pattern(monitor_patterns)
matched_monitors.extend(pattern_monitors)
# Find monitors by groups if specified
if group_patterns:
group_monitors = self.get_monitors_in_groups(group_patterns)
# Convert to same format as find_monitors_by_pattern
group_monitor_objs = [
{"id": m.get("id"), "name": m.get("name")} for m in group_monitors
]
matched_monitors.extend(group_monitor_objs)
# Remove duplicates while preserving order
seen = set()
unique_monitors = []
for monitor in matched_monitors:
if monitor["id"] not in seen:
seen.add(monitor["id"])
unique_monitors.append(monitor)
matched_monitors = unique_monitors
if not matched_monitors:
print(
"Error: No monitors found matching the specified patterns or groups"
)
return []
# Return list of monitor IDs
return [monitor["id"] for monitor in matched_monitors]
except UptimeKumaException as e:
print(f"Error finding monitors by globs: {e}")
return []
def get_monitor_details(self, monitor_ids):
"""Get monitor details for display purposes.
Args:
monitor_ids: List of monitor IDs
Returns:
List of dicts with 'id' and 'name' keys
"""
try:
all_monitors = self.api.get_monitors()
return [
{
"id": mid,
"name": next(
(
m.get("name", f"Monitor {mid}")
for m in all_monitors
if m.get("id") == mid
),
f"Monitor {mid}",
),
}
for mid in monitor_ids
]
except UptimeKumaException as e:
print(f"Error getting monitor details: {e}")
return []
def find_and_get_monitors(self, monitor_patterns=None, group_patterns=None):
"""Find monitors by patterns/groups and return detailed info.
Args:
monitor_patterns: List of monitor name patterns (supports wildcards)
group_patterns: List of group name patterns (supports wildcards)
Returns:
List of dicts with 'id' and 'name' keys, or empty list if none found
"""
monitor_ids = self.find_monitors_by_globs(monitor_patterns, group_patterns)
if not monitor_ids:
return []
return self.get_monitor_details(monitor_ids)

View File

@@ -1,9 +1,13 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Info command implementations for Uptime Kuma CLI."""
from uptime_kuma_api import UptimeKumaException
from ..client import KumaClient from ..client import KumaClient
class InfoCommands: class InfoCommands:
"""Commands for retrieving server information."""
def __init__(self, client: KumaClient): def __init__(self, client: KumaClient):
self.client = client self.client = client
@@ -19,7 +23,7 @@ class InfoCommands:
for key, value in info.items(): for key, value in info.items():
print(f" {key}: {value}") print(f" {key}: {value}")
except Exception as e: except UptimeKumaException as e:
print(f"Error getting server info: {e}") print(f"Error getting server info: {e}")
@@ -29,7 +33,7 @@ def setup_info_parser(subparsers):
return info_parser return info_parser
def handle_info_command(args, client): def handle_info_command(args, client): # pylint: disable=unused-argument
"""Handle info command execution""" """Handle info command execution"""
info_commands = InfoCommands(client) info_commands = InfoCommands(client)
info_commands.get_info() info_commands.get_info()

View File

@@ -1,10 +1,14 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Maintenance command implementations for Uptime Kuma CLI."""
from datetime import datetime, timedelta from datetime import datetime, timedelta
from uptime_kuma_api import UptimeKumaException
from ..client import KumaClient from ..client import KumaClient
class MaintenanceCommands: class MaintenanceCommands:
"""Commands for managing maintenance windows."""
def __init__(self, client: KumaClient): def __init__(self, client: KumaClient):
self.client = client self.client = client
@@ -32,7 +36,7 @@ class MaintenanceCommands:
f"{maintenance_id:<5} {title:<30} {strategy:<15} {active:<10} {description:<50}" f"{maintenance_id:<5} {title:<30} {strategy:<15} {active:<10} {description:<50}"
) )
except Exception as e: except UptimeKumaException as e:
print(f"Error listing maintenances: {e}") print(f"Error listing maintenances: {e}")
def add_maintenance( def add_maintenance(
@@ -46,45 +50,10 @@ class MaintenanceCommands:
): ):
"""Add a new maintenance""" """Add a new maintenance"""
try: try:
# Check if we have either monitor patterns or group patterns matched_monitors = self.client.find_and_get_monitors(
if not monitor_patterns and not group_patterns: monitor_patterns, group_patterns
print( )
"Error: Either --monitor or --group flag is required. Specify at least one pattern."
)
return
matched_monitors = []
# Find monitors by patterns if specified
if monitor_patterns:
pattern_monitors = self.client.find_monitors_by_pattern(
monitor_patterns
)
matched_monitors.extend(pattern_monitors)
# Find monitors by groups if specified
if group_patterns:
group_monitors = self.client.get_monitors_in_groups(group_patterns)
# Convert to the same format as find_monitors_by_pattern
group_monitor_objs = [
{"id": m.get("id"), "name": m.get("name")} for m in group_monitors
]
matched_monitors.extend(group_monitor_objs)
# Remove duplicates while preserving order
seen = set()
unique_monitors = []
for monitor in matched_monitors:
if monitor["id"] not in seen:
seen.add(monitor["id"])
unique_monitors.append(monitor)
matched_monitors = unique_monitors
if not matched_monitors: if not matched_monitors:
print(
"Error: No monitors found matching the specified patterns or groups"
)
return return
print(f"Found {len(matched_monitors)} matching monitors:") print(f"Found {len(matched_monitors)} matching monitors:")
@@ -102,7 +71,8 @@ class MaintenanceCommands:
end_dt = start_dt + timedelta(seconds=duration_seconds) end_dt = start_dt + timedelta(seconds=duration_seconds)
print( print(
f"Maintenance window: {start_dt.strftime('%Y-%m-%d %H:%M:%S UTC')} - {end_dt.strftime('%Y-%m-%d %H:%M:%S UTC')} ({duration})" f"Maintenance window: {start_dt.strftime('%Y-%m-%d %H:%M:%S UTC')} - "
f"{end_dt.strftime('%Y-%m-%d %H:%M:%S UTC')} ({duration})"
) )
# Create the maintenance with single strategy and date range # Create the maintenance with single strategy and date range
@@ -135,13 +105,13 @@ class MaintenanceCommands:
f"Successfully added {len(matched_monitors)} monitors to maintenance" f"Successfully added {len(matched_monitors)} monitors to maintenance"
) )
print(f"API response: {result}") print(f"API response: {result}")
except Exception as e: except UptimeKumaException as e:
print(f"Error: Failed to add monitors to maintenance: {e}") print(f"Error: Failed to add monitors to maintenance: {e}")
print( print(
"This might be due to API compatibility issues or server configuration" "This might be due to API compatibility issues or server configuration"
) )
except Exception as e: except UptimeKumaException as e:
print(f"Error adding maintenance: {e}") print(f"Error adding maintenance: {e}")
def delete_maintenance(self, maintenance_id=None, delete_all=False): def delete_maintenance(self, maintenance_id=None, delete_all=False):
@@ -157,7 +127,8 @@ class MaintenanceCommands:
print(f"Found {len(maintenances)} maintenances to delete:") print(f"Found {len(maintenances)} maintenances to delete:")
for maintenance in maintenances: for maintenance in maintenances:
print( print(
f" - {maintenance.get('title', 'N/A')} (ID: {maintenance.get('id', 'N/A')})" f" - {maintenance.get('title', 'N/A')} "
f"(ID: {maintenance.get('id', 'N/A')})"
) )
# Delete all maintenances # Delete all maintenances
@@ -168,10 +139,11 @@ class MaintenanceCommands:
maintenance.get("id") maintenance.get("id")
) )
print( print(
f"Deleted maintenance '{maintenance.get('title', 'N/A')}' (ID: {maintenance.get('id')})" f"Deleted maintenance '{maintenance.get('title', 'N/A')}' "
f"(ID: {maintenance.get('id')})"
) )
deleted_count += 1 deleted_count += 1
except Exception as e: except UptimeKumaException as e:
print( print(
f"Failed to delete maintenance '{maintenance.get('title', 'N/A')}': {e}" f"Failed to delete maintenance '{maintenance.get('title', 'N/A')}': {e}"
) )
@@ -189,18 +161,19 @@ class MaintenanceCommands:
result = self.client.api.delete_maintenance(maintenance_id) result = self.client.api.delete_maintenance(maintenance_id)
print( print(
f"Successfully deleted maintenance '{maintenance_title}' (ID: {maintenance_id})" f"Successfully deleted maintenance '{maintenance_title}' "
f"(ID: {maintenance_id})"
) )
print(f"API response: {result}") print(f"API response: {result}")
except Exception as e: except UptimeKumaException as e:
print(f"Failed to delete maintenance ID {maintenance_id}: {e}") print(f"Failed to delete maintenance ID {maintenance_id}: {e}")
else: else:
print( print(
"Error: Either --id or --all flag is required for delete operation" "Error: Either --id or --all flag is required for delete operation"
) )
except Exception as e: except UptimeKumaException as e:
print(f"Error during maintenance deletion: {e}") print(f"Error during maintenance deletion: {e}")
@@ -209,8 +182,7 @@ def setup_maintenance_parser(subparsers):
maintenance_parser = subparsers.add_parser( maintenance_parser = subparsers.add_parser(
"maintenance", help="Maintenance operations" "maintenance", help="Maintenance operations"
) )
# Store reference to parser for help display setup_maintenance_parser.parser = maintenance_parser
setup_maintenance_parser._parser = maintenance_parser
maintenance_subparsers = maintenance_parser.add_subparsers( maintenance_subparsers = maintenance_parser.add_subparsers(
dest="maintenance_action", help="Maintenance actions" dest="maintenance_action", help="Maintenance actions"
) )
@@ -250,12 +222,14 @@ def setup_maintenance_parser(subparsers):
add_maintenance_parser.add_argument( add_maintenance_parser.add_argument(
"--monitor", "--monitor",
action="append", action="append",
help="Monitor name pattern to add to maintenance (supports wildcards like *NextCloud*, can be repeated)", help="Monitor name pattern to add to maintenance "
"(supports wildcards like *NextCloud*, can be repeated)",
) )
add_maintenance_parser.add_argument( add_maintenance_parser.add_argument(
"--group", "--group",
action="append", action="append",
help="Group name pattern to add all group members to maintenance (supports wildcards, can be repeated)", help="Group name pattern to add all group members to maintenance "
"(supports wildcards, can be repeated)",
) )
return maintenance_parser return maintenance_parser
@@ -266,14 +240,9 @@ def handle_maintenance_command(args, client):
maintenance_commands = MaintenanceCommands(client) maintenance_commands = MaintenanceCommands(client)
if not args.maintenance_action: if not args.maintenance_action:
if hasattr(setup_maintenance_parser, "_parser"): setup_maintenance_parser.parser.print_help()
setup_maintenance_parser._parser.print_help()
else:
print(
"Error: No maintenance action specified. Use --help for usage information."
)
return False return False
elif args.maintenance_action == "list": if args.maintenance_action == "list":
maintenance_commands.list_maintenances() maintenance_commands.list_maintenances()
elif args.maintenance_action == "add": elif args.maintenance_action == "add":
title = ( title = (

View File

@@ -1,9 +1,13 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Monitor command implementations for Uptime Kuma CLI."""
from uptime_kuma_api import UptimeKumaException
from ..client import KumaClient from ..client import KumaClient
class MonitorCommands: class MonitorCommands:
"""Commands for managing monitors."""
def __init__(self, client: KumaClient): def __init__(self, client: KumaClient):
self.client = client self.client = client
@@ -61,54 +65,20 @@ class MonitorCommands:
parent_name = parent_monitor.get("name", f"Group {parent_id}") parent_name = parent_monitor.get("name", f"Group {parent_id}")
print( print(
f"{monitor_id:<5} {name:<25} {monitor_type:<12} {parent_name:<20} {url:<35} {active:<10}" f"{monitor_id:<5} {name:<25} {monitor_type:<12} "
f"{parent_name:<20} {url:<35} {active:<10}"
) )
except Exception as e: except UptimeKumaException as e:
print(f"Error listing monitors: {e}") print(f"Error listing monitors: {e}")
def pause_monitors(self, monitor_patterns=None, group_patterns=None): def pause_monitors(self, monitor_patterns=None, group_patterns=None):
"""Pause monitors by patterns and/or groups""" """Pause monitors by patterns and/or groups"""
try: try:
# Check if we have either monitor patterns or group patterns matched_monitors = self.client.find_and_get_monitors(
if not monitor_patterns and not group_patterns: monitor_patterns, group_patterns
print( )
"Error: Either --monitor or --group flag is required. Specify at least one pattern."
)
return
matched_monitors = []
# Find monitors by patterns if specified
if monitor_patterns:
pattern_monitors = self.client.find_monitors_by_pattern(
monitor_patterns
)
matched_monitors.extend(pattern_monitors)
# Find monitors by groups if specified
if group_patterns:
group_monitors = self.client.get_monitors_in_groups(group_patterns)
# Convert to the same format as find_monitors_by_pattern
group_monitor_objs = [
{"id": m.get("id"), "name": m.get("name")} for m in group_monitors
]
matched_monitors.extend(group_monitor_objs)
# Remove duplicates while preserving order
seen = set()
unique_monitors = []
for monitor in matched_monitors:
if monitor["id"] not in seen:
seen.add(monitor["id"])
unique_monitors.append(monitor)
matched_monitors = unique_monitors
if not matched_monitors: if not matched_monitors:
print(
"Error: No monitors found matching the specified patterns or groups"
)
return return
print(f"Found {len(matched_monitors)} matching monitors to pause:") print(f"Found {len(matched_monitors)} matching monitors to pause:")
@@ -119,17 +89,17 @@ class MonitorCommands:
paused_count = 0 paused_count = 0
for monitor in matched_monitors: for monitor in matched_monitors:
try: try:
result = self.client.api.pause_monitor(monitor["id"]) self.client.api.pause_monitor(monitor["id"])
print(f"Paused monitor '{monitor['name']}' (ID: {monitor['id']})") print(f"Paused monitor '{monitor['name']}' (ID: {monitor['id']})")
paused_count += 1 paused_count += 1
except Exception as e: except UptimeKumaException as e:
print(f"Failed to pause monitor '{monitor['name']}': {e}") print(f"Failed to pause monitor '{monitor['name']}': {e}")
print( print(
f"Successfully paused {paused_count} out of {len(matched_monitors)} monitors" f"Successfully paused {paused_count} out of {len(matched_monitors)} monitors"
) )
except Exception as e: except UptimeKumaException as e:
print(f"Error pausing monitors: {e}") print(f"Error pausing monitors: {e}")
def resume_monitors( def resume_monitors(
@@ -137,62 +107,36 @@ class MonitorCommands:
): ):
"""Resume monitors by patterns and/or groups, or all paused monitors""" """Resume monitors by patterns and/or groups, or all paused monitors"""
try: try:
# Check if we have either monitor patterns, group patterns, or --all flag
if not monitor_patterns and not group_patterns and not resume_all:
print("Error: Either --monitor, --group, or --all flag is required.")
return
matched_monitors = []
if resume_all: if resume_all:
# Get all monitors and filter for inactive (paused) ones # Get all monitors and filter for inactive (paused) ones
all_monitors = self.client.api.get_monitors() all_monitors = self.client.api.get_monitors()
paused_monitors = [ monitor_ids = [
{"id": m.get("id"), "name": m.get("name")} m.get("id") for m in all_monitors if not m.get("active", True)
for m in all_monitors
if not m.get("active", True)
] ]
matched_monitors.extend(paused_monitors) if not monitor_ids:
else:
# Find monitors by patterns if specified
if monitor_patterns:
pattern_monitors = self.client.find_monitors_by_pattern(
monitor_patterns
)
matched_monitors.extend(pattern_monitors)
# Find monitors by groups if specified
if group_patterns:
group_monitors = self.client.get_monitors_in_groups(group_patterns)
# Convert to the same format as find_monitors_by_pattern
group_monitor_objs = [
{"id": m.get("id"), "name": m.get("name")}
for m in group_monitors
]
matched_monitors.extend(group_monitor_objs)
# Remove duplicates while preserving order
seen = set()
unique_monitors = []
for monitor in matched_monitors:
if monitor["id"] not in seen:
seen.add(monitor["id"])
unique_monitors.append(monitor)
matched_monitors = unique_monitors
if not matched_monitors:
if resume_all:
print("No paused monitors found to resume") print("No paused monitors found to resume")
else: return
print( matched_monitors = [
"Error: No monitors found matching the specified patterns or groups" {
) "id": mid,
return "name": next(
(
if resume_all: m.get("name", f"Monitor {mid}")
for m in all_monitors
if m.get("id") == mid
),
f"Monitor {mid}",
),
}
for mid in monitor_ids
]
print(f"Found {len(matched_monitors)} paused monitors to resume:") print(f"Found {len(matched_monitors)} paused monitors to resume:")
else: else:
matched_monitors = self.client.find_and_get_monitors(
monitor_patterns, group_patterns
)
if not matched_monitors:
return
print(f"Found {len(matched_monitors)} matching monitors to resume:") print(f"Found {len(matched_monitors)} matching monitors to resume:")
for monitor in matched_monitors: for monitor in matched_monitors:
print(f" - {monitor['name']} (ID: {monitor['id']})") print(f" - {monitor['name']} (ID: {monitor['id']})")
@@ -201,25 +145,24 @@ class MonitorCommands:
resumed_count = 0 resumed_count = 0
for monitor in matched_monitors: for monitor in matched_monitors:
try: try:
result = self.client.api.resume_monitor(monitor["id"]) self.client.api.resume_monitor(monitor["id"])
print(f"Resumed monitor '{monitor['name']}' (ID: {monitor['id']})") print(f"Resumed monitor '{monitor['name']}' (ID: {monitor['id']})")
resumed_count += 1 resumed_count += 1
except Exception as e: except UptimeKumaException as e:
print(f"Failed to resume monitor '{monitor['name']}': {e}") print(f"Failed to resume monitor '{monitor['name']}': {e}")
print( print(
f"Successfully resumed {resumed_count} out of {len(matched_monitors)} monitors" f"Successfully resumed {resumed_count} out of {len(matched_monitors)} monitors"
) )
except Exception as e: except UptimeKumaException as e:
print(f"Error resuming monitors: {e}") print(f"Error resuming monitors: {e}")
def setup_monitor_parser(subparsers): def setup_monitor_parser(subparsers):
"""Setup monitor command parser""" """Setup monitor command parser"""
monitor_parser = subparsers.add_parser("monitor", help="Monitor operations") monitor_parser = subparsers.add_parser("monitor", help="Monitor operations")
# Store reference to parser for help display setup_monitor_parser.parser = monitor_parser
setup_monitor_parser._parser = monitor_parser
monitor_subparsers = monitor_parser.add_subparsers( monitor_subparsers = monitor_parser.add_subparsers(
dest="monitor_action", help="Monitor actions" dest="monitor_action", help="Monitor actions"
) )
@@ -280,14 +223,9 @@ def handle_monitor_command(args, client):
monitor_commands = MonitorCommands(client) monitor_commands = MonitorCommands(client)
if not args.monitor_action: if not args.monitor_action:
if hasattr(setup_monitor_parser, "_parser"): setup_monitor_parser.parser.print_help()
setup_monitor_parser._parser.print_help()
else:
print(
"Error: No monitor action specified. Use --help for usage information."
)
return False return False
elif args.monitor_action == "list": if args.monitor_action == "list":
monitor_commands.list_monitors( monitor_commands.list_monitors(
monitor_patterns=args.monitor, group_patterns=args.group monitor_patterns=args.monitor, group_patterns=args.group
) )

View File

@@ -1,9 +1,9 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Main CLI module for Uptime Kuma."""
import argparse import argparse
import os import os
import sys import sys
from datetime import datetime
# Handle both direct execution and package import # Handle both direct execution and package import
try: try:
@@ -24,6 +24,7 @@ except ImportError:
def main(): def main():
"""Main entry point for the CLI application."""
parser = argparse.ArgumentParser(description="Uptime Kuma CLI Client") parser = argparse.ArgumentParser(description="Uptime Kuma CLI Client")
parser.add_argument( parser.add_argument(
"--url", help="Uptime Kuma server URL (can also use KUMA_URL env var)" "--url", help="Uptime Kuma server URL (can also use KUMA_URL env var)"
@@ -40,9 +41,9 @@ def main():
subparsers = parser.add_subparsers(dest="resource", help="Resource to operate on") subparsers = parser.add_subparsers(dest="resource", help="Resource to operate on")
# Setup command parsers # Setup command parsers
monitor_parser = setup_monitor_parser(subparsers) setup_monitor_parser(subparsers)
maintenance_parser = setup_maintenance_parser(subparsers) setup_maintenance_parser(subparsers)
info_parser = setup_info_parser(subparsers) setup_info_parser(subparsers)
args = parser.parse_args() args = parser.parse_args()

View File

@@ -21,7 +21,7 @@ class TestCLIIntegration:
# Verify parser is created # Verify parser is created
assert monitor_parser is not None assert monitor_parser is not None
assert hasattr(setup_monitor_parser, "_parser") assert monitor_parser.prog.endswith("monitor")
def test_maintenance_parser_setup(self): def test_maintenance_parser_setup(self):
"""Test maintenance parser setup""" """Test maintenance parser setup"""
@@ -32,7 +32,7 @@ class TestCLIIntegration:
# Verify parser is created # Verify parser is created
assert maintenance_parser is not None assert maintenance_parser is not None
assert hasattr(setup_maintenance_parser, "_parser") assert maintenance_parser.prog.endswith("maintenance")
def test_info_parser_setup(self): def test_info_parser_setup(self):
"""Test info parser setup""" """Test info parser setup"""
@@ -50,9 +50,9 @@ class TestCLIIntegration:
mock_args = Mock() mock_args = Mock()
mock_args.monitor_action = None mock_args.monitor_action = None
# Setup parser reference # Setup parser reference to simulate having called setup_monitor_parser
mock_parser = Mock() mock_parser = Mock()
setup_monitor_parser._parser = mock_parser setup_monitor_parser.parser = mock_parser
# Execute # Execute
result = handle_monitor_command(mock_args, mock_client) result = handle_monitor_command(mock_args, mock_client)
@@ -67,9 +67,9 @@ class TestCLIIntegration:
mock_args = Mock() mock_args = Mock()
mock_args.maintenance_action = None mock_args.maintenance_action = None
# Setup parser reference # Setup parser reference to simulate having called setup_maintenance_parser
mock_parser = Mock() mock_parser = Mock()
setup_maintenance_parser._parser = mock_parser setup_maintenance_parser.parser = mock_parser
# Execute # Execute
result = handle_maintenance_command(mock_args, mock_client) result = handle_maintenance_command(mock_args, mock_client)
@@ -87,11 +87,9 @@ class TestCLIIntegration:
mock_args.group = ["web-services"] mock_args.group = ["web-services"]
# Mock client methods # Mock client methods
mock_client.find_monitors_by_pattern.return_value = [ mock_client.find_and_get_monitors.return_value = [
{"id": 1, "name": "test-monitor"} {"id": 1, "name": "test-monitor"},
] {"id": 2, "name": "web-service-monitor"},
mock_client.get_monitors_in_groups.return_value = [
{"id": 2, "name": "web-service-monitor"}
] ]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."} mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
@@ -100,8 +98,9 @@ class TestCLIIntegration:
# Verify # Verify
assert result is True assert result is True
mock_client.find_monitors_by_pattern.assert_called_once_with(["test*"]) mock_client.find_and_get_monitors.assert_called_once_with(
mock_client.get_monitors_in_groups.assert_called_once_with(["web-services"]) ["test*"], ["web-services"]
)
# Should pause both monitors (deduplicated) # Should pause both monitors (deduplicated)
assert mock_client.api.pause_monitor.call_count == 2 assert mock_client.api.pause_monitor.call_count == 2
@@ -235,7 +234,11 @@ class TestErrorHandling:
mock_args.group = None mock_args.group = None
# Mock no matches found # Mock no matches found
mock_client.find_monitors_by_pattern.return_value = [] 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 # Execute
result = handle_monitor_command(mock_args, mock_client) result = handle_monitor_command(mock_args, mock_client)
@@ -255,7 +258,11 @@ class TestErrorHandling:
mock_args.maintenance_action = "list" mock_args.maintenance_action = "list"
# Mock API error # Mock API error
mock_client.api.get_maintenances.side_effect = Exception("Connection timeout") from uptime_kuma_api import UptimeKumaException
mock_client.api.get_maintenances.side_effect = UptimeKumaException(
"Connection timeout"
)
# Execute # Execute
result = handle_maintenance_command(mock_args, mock_client) result = handle_maintenance_command(mock_args, mock_client)

View File

@@ -171,10 +171,12 @@ class TestKumaClient:
def test_find_monitors_by_pattern_api_error(self, capsys): def test_find_monitors_by_pattern_api_error(self, capsys):
"""Test finding monitors handles API errors""" """Test finding monitors handles API errors"""
from uptime_kuma_api import UptimeKumaException
client = KumaClient("http://test.com") client = KumaClient("http://test.com")
client.api = Mock() client.api = Mock()
client.api.get_monitors.side_effect = Exception("API Error") client.api.get_monitors.side_effect = UptimeKumaException("API Error")
result = client.find_monitors_by_pattern(["Web*"]) result = client.find_monitors_by_pattern(["Web*"])
assert len(result) == 0 assert len(result) == 0
@@ -203,7 +205,9 @@ class TestKumaClient:
@patch("kumacli.client.UptimeKumaApi") @patch("kumacli.client.UptimeKumaApi")
def test_connect_failure(self, mock_api_class, capsys): def test_connect_failure(self, mock_api_class, capsys):
"""Test connection failure""" """Test connection failure"""
mock_api_class.side_effect = Exception("Connection failed") from uptime_kuma_api import UptimeKumaException
mock_api_class.side_effect = UptimeKumaException("Connection failed")
client = KumaClient("http://test.com", "user", "pass") client = KumaClient("http://test.com", "user", "pass")
result = client.connect() result = client.connect()

View File

@@ -49,8 +49,10 @@ class TestInfoCommands:
def test_get_info_api_error(self, mock_client, capsys): def test_get_info_api_error(self, mock_client, capsys):
"""Test info command with API error""" """Test info command with API error"""
from uptime_kuma_api import UptimeKumaException
# Setup # Setup
mock_client.api.info.side_effect = Exception("Connection failed") mock_client.api.info.side_effect = UptimeKumaException("Connection failed")
info_commands = InfoCommands(mock_client) info_commands = InfoCommands(mock_client)
@@ -80,9 +82,11 @@ class TestInfoCommandHandler:
def test_handle_info_command_with_error(self, mock_client): def test_handle_info_command_with_error(self, mock_client):
"""Test info command handler with error""" """Test info command handler with error"""
from uptime_kuma_api import UptimeKumaException
# Setup # Setup
mock_args = Mock() mock_args = Mock()
mock_client.api.info.side_effect = Exception("API Error") mock_client.api.info.side_effect = UptimeKumaException("API Error")
# Execute # Execute
result = handle_info_command(mock_args, mock_client) result = handle_info_command(mock_args, mock_client)

View File

@@ -43,8 +43,10 @@ class TestMaintenanceCommands:
def test_list_maintenances_api_error(self, mock_client, capsys): def test_list_maintenances_api_error(self, mock_client, capsys):
"""Test maintenance listing with API error""" """Test maintenance listing with API error"""
from uptime_kuma_api import UptimeKumaException
# Setup # Setup
mock_client.api.get_maintenances.side_effect = Exception("API Error") mock_client.api.get_maintenances.side_effect = UptimeKumaException("API Error")
maintenance_commands = MaintenanceCommands(mock_client) maintenance_commands = MaintenanceCommands(mock_client)

View File

@@ -16,8 +16,7 @@ class TestMonitorCommands:
def test_pause_monitors_by_pattern(self, mock_client, mock_monitors, capsys): def test_pause_monitors_by_pattern(self, mock_client, mock_monitors, capsys):
"""Test pausing monitors by pattern""" """Test pausing monitors by pattern"""
# Setup # Setup
mock_client.api.get_monitors.return_value = mock_monitors mock_client.find_and_get_monitors.return_value = [
mock_client.find_monitors_by_pattern.return_value = [
{"id": 1, "name": "Test Monitor 1"}, {"id": 1, "name": "Test Monitor 1"},
{"id": 2, "name": "Test Monitor 2"}, {"id": 2, "name": "Test Monitor 2"},
] ]
@@ -29,7 +28,7 @@ class TestMonitorCommands:
monitor_commands.pause_monitors(monitor_patterns=["Test*"]) monitor_commands.pause_monitors(monitor_patterns=["Test*"])
# Verify # Verify
mock_client.find_monitors_by_pattern.assert_called_once_with(["Test*"]) mock_client.find_and_get_monitors.assert_called_once_with(["Test*"], None)
assert mock_client.api.pause_monitor.call_count == 2 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(1)
mock_client.api.pause_monitor.assert_any_call(2) mock_client.api.pause_monitor.assert_any_call(2)
@@ -43,7 +42,7 @@ class TestMonitorCommands:
def test_pause_monitors_by_group(self, mock_client, mock_monitors, capsys): def test_pause_monitors_by_group(self, mock_client, mock_monitors, capsys):
"""Test pausing monitors by group""" """Test pausing monitors by group"""
# Setup # Setup
mock_client.get_monitors_in_groups.return_value = [ mock_client.find_and_get_monitors.return_value = [
{"id": 4, "name": "Child Monitor"} {"id": 4, "name": "Child Monitor"}
] ]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."} mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
@@ -54,7 +53,7 @@ class TestMonitorCommands:
monitor_commands.pause_monitors(group_patterns=["Group*"]) monitor_commands.pause_monitors(group_patterns=["Group*"])
# Verify # Verify
mock_client.get_monitors_in_groups.assert_called_once_with(["Group*"]) mock_client.find_and_get_monitors.assert_called_once_with(None, ["Group*"])
mock_client.api.pause_monitor.assert_called_once_with(4) mock_client.api.pause_monitor.assert_called_once_with(4)
captured = capsys.readouterr() captured = capsys.readouterr()
@@ -63,19 +62,21 @@ class TestMonitorCommands:
def test_pause_monitors_no_patterns(self, mock_client, capsys): def test_pause_monitors_no_patterns(self, mock_client, capsys):
"""Test pausing monitors without patterns""" """Test pausing monitors without patterns"""
# Setup
mock_client.find_and_get_monitors.return_value = []
monitor_commands = MonitorCommands(mock_client) monitor_commands = MonitorCommands(mock_client)
# Execute # Execute
monitor_commands.pause_monitors() monitor_commands.pause_monitors()
# Verify # Verify
captured = capsys.readouterr() mock_client.find_and_get_monitors.assert_called_once_with(None, None)
assert "Error: Either --monitor or --group flag is required." in captured.out
def test_pause_monitors_no_matches(self, mock_client, capsys): def test_pause_monitors_no_matches(self, mock_client, capsys):
"""Test pausing monitors with no matches""" """Test pausing monitors with no matches"""
# Setup # Setup
mock_client.find_monitors_by_pattern.return_value = [] mock_client.find_and_get_monitors.return_value = []
monitor_commands = MonitorCommands(mock_client) monitor_commands = MonitorCommands(mock_client)
@@ -83,19 +84,19 @@ class TestMonitorCommands:
monitor_commands.pause_monitors(monitor_patterns=["NonExistent*"]) monitor_commands.pause_monitors(monitor_patterns=["NonExistent*"])
# Verify # Verify
captured = capsys.readouterr() mock_client.find_and_get_monitors.assert_called_once_with(
assert ( ["NonExistent*"], None
"Error: No monitors found matching the specified patterns or groups"
in captured.out
) )
def test_pause_monitors_api_error(self, mock_client, capsys): def test_pause_monitors_api_error(self, mock_client, capsys):
"""Test pausing monitors with API error""" """Test pausing monitors with API error"""
# Setup # Setup
mock_client.find_monitors_by_pattern.return_value = [ mock_client.find_and_get_monitors.return_value = [
{"id": 1, "name": "Test Monitor 1"} {"id": 1, "name": "Test Monitor 1"}
] ]
mock_client.api.pause_monitor.side_effect = Exception("API Error") from uptime_kuma_api import UptimeKumaException
mock_client.api.pause_monitor.side_effect = UptimeKumaException("API Error")
monitor_commands = MonitorCommands(mock_client) monitor_commands = MonitorCommands(mock_client)
@@ -110,7 +111,7 @@ class TestMonitorCommands:
def test_resume_monitors_by_pattern(self, mock_client, mock_monitors, capsys): def test_resume_monitors_by_pattern(self, mock_client, mock_monitors, capsys):
"""Test resuming monitors by pattern""" """Test resuming monitors by pattern"""
# Setup # Setup
mock_client.find_monitors_by_pattern.return_value = [ mock_client.find_and_get_monitors.return_value = [
{"id": 2, "name": "Test Monitor 2"} {"id": 2, "name": "Test Monitor 2"}
] ]
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."} mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}
@@ -121,6 +122,7 @@ class TestMonitorCommands:
monitor_commands.resume_monitors(monitor_patterns=["Test*"]) monitor_commands.resume_monitors(monitor_patterns=["Test*"])
# Verify # Verify
mock_client.find_and_get_monitors.assert_called_once_with(["Test*"], None)
mock_client.api.resume_monitor.assert_called_once_with(2) mock_client.api.resume_monitor.assert_called_once_with(2)
captured = capsys.readouterr() captured = capsys.readouterr()
@@ -165,17 +167,16 @@ class TestMonitorCommands:
def test_resume_monitors_no_args(self, mock_client, capsys): def test_resume_monitors_no_args(self, mock_client, capsys):
"""Test resuming monitors without any arguments""" """Test resuming monitors without any arguments"""
# Setup
mock_client.find_and_get_monitors.return_value = []
monitor_commands = MonitorCommands(mock_client) monitor_commands = MonitorCommands(mock_client)
# Execute # Execute
monitor_commands.resume_monitors() monitor_commands.resume_monitors()
# Verify # Verify
captured = capsys.readouterr() mock_client.find_and_get_monitors.assert_called_once_with(None, None)
assert (
"Error: Either --monitor, --group, or --all flag is required."
in captured.out
)
class TestMonitorCommandHandler: class TestMonitorCommandHandler:
@@ -204,7 +205,7 @@ class TestMonitorCommandHandler:
mock_args.monitor = ["test*"] mock_args.monitor = ["test*"]
mock_args.group = None mock_args.group = None
mock_client.find_monitors_by_pattern.return_value = [ mock_client.find_and_get_monitors.return_value = [
{"id": 1, "name": "test monitor"} {"id": 1, "name": "test monitor"}
] ]
mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."} mock_client.api.pause_monitor.return_value = {"msg": "Paused Successfully."}
@@ -225,7 +226,7 @@ class TestMonitorCommandHandler:
mock_args.group = None mock_args.group = None
mock_args.all = False mock_args.all = False
mock_client.find_monitors_by_pattern.return_value = [ mock_client.find_and_get_monitors.return_value = [
{"id": 1, "name": "test monitor"} {"id": 1, "name": "test monitor"}
] ]
mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."} mock_client.api.resume_monitor.return_value = {"msg": "Resumed Successfully."}