From a08d3b46139e7e4ac49f8cdbaec681410edacf62 Mon Sep 17 00:00:00 2001 From: lucasheld Date: Sun, 10 Jul 2022 18:07:11 +0200 Subject: [PATCH] check argument conditions --- datacollection/build_notifications.py | 69 ++++++++++++---- uptime_kuma_api/__version__.py | 2 +- uptime_kuma_api/api.py | 114 +++++++++++++++++++------- uptime_kuma_api/converter.py | 3 +- 4 files changed, 140 insertions(+), 48 deletions(-) diff --git a/datacollection/build_notifications.py b/datacollection/build_notifications.py index a68ea76..c45edbf 100644 --- a/datacollection/build_notifications.py +++ b/datacollection/build_notifications.py @@ -1,9 +1,11 @@ import glob import re +from pprint import pprint import jinja2 +from bs4 import BeautifulSoup -notification_providers = [] +from uptime_kuma_api import convert_from_socket, params_map_notification_provider def deduplicate_list(l): @@ -14,24 +16,53 @@ def deduplicate_list(l): return out -for path in glob.glob('uptime-kuma/server/notification-providers/*'): - with open(path) as f: - content = f.read() - match = re.search(r'class [^ ]+ extends NotificationProvider {', content) - if match: - match = re.search(r'name = "([^"]+)";', content) - name = match.group(1) +def build_notification_providers(): + providers = [] + for path in glob.glob('uptime-kuma/server/notification-providers/*'): + with open(path) as f: + content = f.read() + match = re.search(r'class [^ ]+ extends NotificationProvider {', content) + if match: + match = re.search(r'name = "([^"]+)";', content) + name = match.group(1) - inputs = re.findall(r'notification\.([^ ,.;})\]]+)', content) - inputs = deduplicate_list(inputs) - inputs = [i.strip() for i in inputs] + inputs = re.findall(r'notification\.([^ ,.;})\]]+)', content) + inputs = deduplicate_list(inputs) + inputs = [i.strip() for i in inputs] - notification_providers.append({ - "name": name, - "inputs": inputs, - }) + providers.append({ + "name": name, + "inputs": inputs, + }) + return providers -print(notification_providers) + +def build_notification_provider_conditions(): + conditions = {} + for path in glob.glob('uptime-kuma/src/components/notifications/*'): + if path.endswith("index.js"): + continue + with open(path) as f: + content = f.read() + match = re.search(r'', content, re.MULTILINE) + html = match.group(0) + soup = BeautifulSoup(html, "html.parser") + inputs = soup.find_all("input") + for input in inputs: + condition = {} + attrs = input.attrs + v_model = attrs.get("v-model") + min_ = attrs.get("min") + max_ = attrs.get("max") + if min_: + condition["min"] = int(min_) + if max_: + condition["max"] = int(max_) + param_name = re.match(r'\$parent.notification.(.*)$', v_model).group(1) + if condition: + conditions[param_name] = condition + conditions = convert_from_socket(params_map_notification_provider, conditions) + return conditions def write_to_file(template, destination, **kwargs): @@ -41,5 +72,7 @@ def write_to_file(template, destination, **kwargs): with open(destination, "w") as f: f.write(rendered) - -write_to_file("notification_providers.py.j2", "./../uptimekumaapi/notification_providers.py", notification_providers=notification_providers) +conditions = build_notification_provider_conditions() +pprint(conditions) +# notification_providers = build_notification_providers() +# write_to_file("notification_providers.py.j2", "./../uptimekumaapi/notification_providers.py", notification_providers=notification_providers) diff --git a/uptime_kuma_api/__version__.py b/uptime_kuma_api/__version__.py index ff1bed1..47f8a24 100644 --- a/uptime_kuma_api/__version__.py +++ b/uptime_kuma_api/__version__.py @@ -1,5 +1,5 @@ __title__ = "uptime_kuma_api" -__version__ = "0.0.5" +__version__ = "0.0.6" __author__ = "Lucas Held" # __license__ = "" __copyright__ = "Copyright 2022 Lucas Held" diff --git a/uptime_kuma_api/api.py b/uptime_kuma_api/api.py index 120f45f..ff69fd8 100644 --- a/uptime_kuma_api/api.py +++ b/uptime_kuma_api/api.py @@ -13,7 +13,6 @@ from . import convert_from_socket, convert_to_socket, params_map_monitor, params params_map_settings from . import UptimeKumaException - params_map_notification_and_provider = {**params_map_notification, **params_map_notification_provider} @@ -168,7 +167,7 @@ def _build_monitor_data( return data -def _build_notification_data(name: str, type_: NotificationType, default: bool, **kwargs): +def _build_notification_data(name: str, type_: NotificationType, default: bool = False, apply_existing: bool = False, **kwargs): data = { "name": name, "type_": type_, @@ -251,7 +250,7 @@ def _build_status_page_data( return slug, config, img_data_url, public_group_list -def _raise_missing_arguments(required_params, kwargs, params_map): +def _check_missing_arguments(required_params, kwargs, params_map): missing_arguments = [] for required_param in required_params: required_param_sock = convert_to_socket(params_map, required_param) @@ -262,17 +261,31 @@ def _raise_missing_arguments(required_params, kwargs, params_map): raise TypeError(f"missing {len(missing_arguments)} required argument: {missing_arguments_str}") -def _check_required_arguments_monitor(kwargs): - required_params = [ +def _check_argument_conditions(valid_params, kwargs, params_map): + for valid_param in valid_params: + valid_param_sock = convert_to_socket(params_map, valid_param) + if valid_param_sock in kwargs: + value = kwargs[valid_param_sock] + conditions = valid_params[valid_param] + min_ = conditions.get("min") + max_ = conditions.get("max") + if min_ is not None and value < min_: + raise ValueError(f"the value of {valid_param} must not be less than {min_}") + if max_ is not None and value > max_: + raise ValueError(f"the value of {valid_param} must not be larger than {max_}") + + +def _check_arguments_monitor(kwargs): + required_args = [ "type_", "name", "heartbeat_interval", "retries", "heartbeat_retry_interval" ] - _raise_missing_arguments(required_params, kwargs, params_map_monitor) + _check_missing_arguments(required_args, kwargs, params_map_monitor) - required_params_type_specific = { + required_args_by_type = { MonitorType.HTTP: ["url", "max_redirects"], MonitorType.PORT: ["hostname", "port"], MonitorType.PING: ["hostname"], @@ -283,25 +296,70 @@ def _check_required_arguments_monitor(kwargs): MonitorType.MQTT: ["hostname", "port", "mqtt_topic"], MonitorType.SQLSERVER: [], } - type_key = convert_to_socket(params_map_monitor, "type") - required_params = required_params_type_specific[kwargs[type_key]] - _raise_missing_arguments(required_params, kwargs, params_map_monitor) + type_ = kwargs[convert_to_socket(params_map_monitor, "type")] + required_args = required_args_by_type[type_] + _check_missing_arguments(required_args, kwargs, params_map_monitor) + + conditions = { + "heartbeat_interval": { + "min": 20 + }, + "retries": { + "min": 0 + }, + "heartbeat_retry_interval": { + "min": 20 + }, + "max_redirects": { + "min": 0 + }, + "port": { + "min": 0, + "max": 65535 + } + } + _check_argument_conditions(conditions, kwargs, params_map_monitor) -def _check_required_arguments_notification(kwargs): - required_params = ["type_", "name"] - type_key = convert_to_socket(params_map_notification, "type") - additional_params_sock = notification_provider_options[kwargs[type_key]] - additional_params = convert_from_socket(params_map_notification_provider, additional_params_sock) - required_params.extend(additional_params) - _raise_missing_arguments(required_params, kwargs, params_map_notification_and_provider) +def _check_arguments_notification(kwargs): + required_args = ["type_", "name"] + _check_missing_arguments(required_args, kwargs, params_map_notification_and_provider) + + type_ = kwargs[convert_to_socket(params_map_notification, "type")] + required_args_sock = notification_provider_options[type_] + required_args = convert_from_socket(params_map_notification_provider, required_args_sock) + _check_missing_arguments(required_args, kwargs, params_map_notification_and_provider) + + provider_conditions = { + 'gotify_priority': { + 'max': 10, + 'min': 0 + }, + 'ntfy_priority': { + 'max': 5, + 'min': 1 + }, + 'smtp_smtp_port': { + 'max': 65535, + 'min': 0 + } + } + _check_argument_conditions(provider_conditions, kwargs, params_map_notification_and_provider) -def _check_required_arguments_proxy(kwargs): - required_params = ["protocol", "host", "port"] +def _check_arguments_proxy(kwargs): + required_args = ["protocol", "host", "port"] if "auth" in kwargs: - required_params.extend(["username", "password"]) - _raise_missing_arguments(required_params, kwargs, params_map_proxy) + required_args.extend(["username", "password"]) + _check_missing_arguments(required_args, kwargs, params_map_proxy) + + conditions = { + "port": { + "min": 0, + "max": 65535 + } + } + _check_argument_conditions(conditions, kwargs, params_map_proxy) class UptimeKumaApi(object): @@ -453,7 +511,7 @@ class UptimeKumaApi(object): def add_monitor(self, **kwargs): data = _build_monitor_data(**kwargs) - _check_required_arguments_monitor(data) + _check_arguments_monitor(data) r = self._call('add', data) r = convert_from_socket(params_map_monitor, r) @@ -464,7 +522,7 @@ class UptimeKumaApi(object): data.update(kwargs) data = convert_to_socket(params_map_monitor, data) - _check_required_arguments_monitor(data) + _check_arguments_monitor(data) r = self._call('editMonitor', data) r = convert_from_socket(params_map_monitor, r) @@ -506,13 +564,13 @@ class UptimeKumaApi(object): def test_notification(self, **kwargs): data = _build_notification_data(**kwargs) - _check_required_arguments_notification(data) + _check_arguments_notification(data) return self._call('testNotification', data) def add_notification(self, **kwargs): data = _build_notification_data(**kwargs) - _check_required_arguments_notification(data) + _check_arguments_notification(data) return self._call('addNotification', (data, None)) def edit_notification(self, id_: int, **kwargs): @@ -530,7 +588,7 @@ class UptimeKumaApi(object): notification.update(kwargs) notification = convert_to_socket(params_map_notification_and_provider, notification) - _check_required_arguments_notification(notification) + _check_arguments_notification(notification) return self._call('addNotification', (notification, id_)) def delete_notification(self, id_: int): @@ -557,7 +615,7 @@ class UptimeKumaApi(object): def add_proxy(self, **kwargs): data = _build_proxy_data(**kwargs) - _check_required_arguments_proxy(data) + _check_arguments_proxy(data) return self._call('addProxy', (data, None)) def edit_proxy(self, id_: int, **kwargs): @@ -565,7 +623,7 @@ class UptimeKumaApi(object): proxy.update(kwargs) proxy = convert_to_socket(params_map_proxy, proxy) - _check_required_arguments_proxy(proxy) + _check_arguments_proxy(proxy) return self._call('addProxy', (proxy, id_)) def delete_proxy(self, id_: int): diff --git a/uptime_kuma_api/converter.py b/uptime_kuma_api/converter.py index a666c38..0e6456e 100644 --- a/uptime_kuma_api/converter.py +++ b/uptime_kuma_api/converter.py @@ -34,7 +34,8 @@ params_map_monitor = { params_map_notification = { "type": "type_", "isDefault": "default", - "userId": "user_id" + "userId": "user_id", + "applyExisting": "apply_existing", } params_map_notification_provider = {