Compare commits

..

No commits in common. "master" and "0.2.0" have entirely different histories.

56 changed files with 785 additions and 5964 deletions

1
.gitignore vendored
View file

@ -4,4 +4,3 @@ venv
__pycache__
*.egg-info
dist
docs/_build

View file

@ -1,6 +0,0 @@
version: 2
python:
install:
- method: pip
path: .

View file

@ -1,172 +1,12 @@
## Changelog
### Release 1.2.1
#### Bugfixes
- drop first info event without a version
### Release 1.2.0
#### Features
- add support for uptime kuma 1.23.0 and 1.23.1
#### Bugfixes
- remove `name` from maintenance monitors and status pages
- rstip url globally
- convert sendUrl from bool to int
- validate accepted status codes types
### Release 1.1.0
#### Features
- add support for uptime kuma 1.22.0 and 1.22.1
### Release 1.0.1
#### Bugfixes
- fix ValueError if monitor authMethod is None
### Release 1.0.0
#### Features
- add `ssl_verify` parameter
- add `wait_events` parameter
- implement context manager for UptimeKumaApi class
- drop Python 3.6 support
- implement `get_monitor_status` helper method
- implement timeouts for all methods (`timeout` parameter)
- add support for uptime kuma 1.21.3
- drop support for Uptime Kuma versions < 1.21.3
- check for required notification arguments
- raise exception when deleting an element that does not exist
- replace raw return values with enum values
#### Bugfixes
- adjust monitor `status` type to allow all used values
- fix memory leak
#### BREAKING CHANGES
- Python 3.7+ required
- maintenance parameter `timezone` renamed to `timezoneOption`
- Removed the `wait_timeout` parameter. Use the new `timeout` parameter instead. The `timeout` parameter specifies how many seconds the client should wait for the connection, an expected event or a server response.
- changed return values of methods `get_heartbeats`, `get_important_heartbeats`, `avg_ping`, `uptime`, `cert_info`
- Uptime Kuma versions < 1.21.3 are not supported in uptime-kuma-api 1.0.0+
- Removed the `get_heartbeat` method. This method was never intended to retrieve information. Use `get_heartbeats` or `get_important_heartbeats` instead.
- Types of return values changed to enum values:
- monitor: `type` (str -> MonitorType), `status` (bool -> MonitorStatus), `authMethod` (str -> AuthMethod)
- notification: `type` (str -> NotificationType)
- docker host: `dockerType` (str -> DockerType)
- status page: `style` (str -> IncidentStyle)
- maintenance: `strategy` (str -> MaintenanceStrategy)
- proxy: `protocol` (str -> ProxyProtocol)
### Release 0.13.0
#### Feature
- add support for uptime kuma 1.21.2
- implement custom socketio headers
#### Bugfix
- do not wait for events that have already arrived
### Release 0.12.0
#### Feature
- add support for uptime kuma 1.21.1
### Release 0.11.0
#### Feature
- add support for uptime kuma 1.21.0
### Release 0.10.0
#### Feature
- add support for uptime kuma 1.20.0
### Release 0.9.0
#### Feature
- add support for uptime kuma 1.19.5
### Release 0.8.0
#### Feature
- add support for uptime kuma 1.19.3
### Release 0.7.1
#### Bugfix
- remove unsupported type hints on old python versions
### Release 0.7.0
#### Feature
- add support for uptime kuma 1.19.2
#### Bugfix
- skip condition check for None values
### Release 0.6.0
#### Feature
- add parameter `wait_timeout` to adjust connection timeout
### Release 0.5.2
#### Bugfix
- add type to notification provider options
### Release 0.5.1
#### Bugfix
- remove required notification provider args check
### Release 0.5.0
#### Feature
- support for uptime kuma 1.18.3
### Release 0.4.0
#### Feature
- support for uptime kuma 1.18.1 / 1.18.2
#### Bugfix
- update event list data after changes
### Release 0.3.0
#### Feature
- support autoLogin for enabled disableAuth
#### Bugfix
- set_settings password is only required if disableAuth is enabled
- increase event wait time to receive the slow statusPageList event
### Release 0.2.2
#### Bugfix
- remove `tags` from monitor input
- convert monitor notificationIDList only once
### Release 0.2.1
#### Bugfix
- generate pushToken on push monitor save
- convert monitor notificationIDList return value
### Release 0.2.0
#### Feature
- support for uptime kuma 1.18.0
#### Bugfix
- convert values on monitor edit
### Release 0.1.1
#### Bugfix
- implement 2FA login
- allow to add monitors to status pages
- do not block certain methods

View file

@ -1,6 +1,6 @@
MIT License
Copyright (c) 2023 Lucas Held
Copyright (c) 2022 Lucas Held
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

View file

@ -6,14 +6,9 @@ uptime-kuma-api is a Python wrapper for the [Uptime Kuma](https://github.com/lou
This package was developed to configure Uptime Kuma with Ansible. The Ansible collection can be found at https://github.com/lucasheld/ansible-uptime-kuma.
Python version 3.7+ is required.
Python version 3.6+ is required.
Supported Uptime Kuma versions:
| Uptime Kuma | uptime-kuma-api |
|-----------------|-----------------|
| 1.21.3 - 1.23.2 | 1.0.0 - 1.2.1 |
| 1.17.0 - 1.21.2 | 0.1.0 - 0.13.0 |
Supported Uptime Kuma versions: 1.18, 1.17
Installation
---
@ -25,11 +20,7 @@ You can install it using pip:
pip install uptime-kuma-api
```
Documentation
---
The API Reference is available on [Read the Docs](https://uptime-kuma-api.readthedocs.io).
Example
Examples
---
Once you have installed the python package, you can use it to communicate with an Uptime Kuma instance.
@ -54,17 +45,3 @@ At the end, the connection to the API must be disconnected so that the program d
```python
>>> api.disconnect()
```
With a context manager, the disconnect method is called automatically:
```python
from uptime_kuma_api import UptimeKumaApi
with UptimeKumaApi('INSERT_URL') as api:
api.login('INSERT_USERNAME', 'INSERT_PASSWORD')
api.add_monitor(
type=MonitorType.HTTP,
name="Google",
url="https://google.com"
)
```

View file

@ -1,4 +0,0 @@
Sphinx==5.3.0
pyotp==2.8.0
Jinja2==3.1.2
BeautifulSoup4==4.12.2

View file

@ -1,20 +0,0 @@
# Minimal makefile for Sphinx documentation
#
# You can set these variables from the command line, and also
# from the environment for the first two.
SPHINXOPTS ?=
SPHINXBUILD ?= sphinx-build
SOURCEDIR = .
BUILDDIR = _build
# Put it first so that "make" without argument is like "make help".
help:
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
.PHONY: help Makefile
# Catch-all target: route all unknown targets to Sphinx using the new
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
%: Makefile
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)

View file

@ -1,46 +0,0 @@
.. _api:
Main Interface
--------------
.. module:: uptime_kuma_api
.. autoclass:: UptimeKumaApi
:inherited-members:
Enums
-----
.. autoclass:: AuthMethod
:members:
.. autoclass:: MonitorType
:members:
.. autoclass:: MonitorStatus
:members:
.. autoclass:: NotificationType
:members:
.. autoclass:: ProxyProtocol
:members:
.. autoclass:: IncidentStyle
:members:
.. autoclass:: DockerType
:members:
.. autoclass:: MaintenanceStrategy
:members:
Exceptions
----------
.. autoexception:: UptimeKumaException
.. autoexception:: Timeout

View file

@ -1,53 +0,0 @@
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
import os
import sys
sys.path.insert(0, os.path.abspath(".."))
import uptime_kuma_api
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
project = 'uptime-kuma-api'
copyright = '2023, Lucas Held'
author = 'Lucas Held'
version = uptime_kuma_api.__version__
release = uptime_kuma_api.__version__
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
extensions = [
"sphinx.ext.autodoc"
]
templates_path = ['_templates']
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']
toc_object_entries_show_parents = "hide"
autodoc_member_order = "bysource"
add_module_names = False
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
html_theme = 'alabaster'
html_static_path = ['_static']
html_theme_options = {
"show_powered_by": False,
"github_user": "lucasheld",
"github_repo": "uptime-kuma-api",
"github_banner": True,
"show_related": False,
"note_bg": "#FFF59C",
"github_button": True,
"github_type": "star"
}

View file

@ -1,32 +0,0 @@
.. uptime-kuma-api documentation master file, created by
sphinx-quickstart on Thu Dec 15 11:58:11 2022.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
uptime-kuma-api
===============
Release v\ |version|. (:ref:`Installation <install>`)
A Python wrapper for the Uptime Kuma Socket.IO API
-------------------
Indices and tables
------------------
* :ref:`genindex`
* :ref:`search`
API Reference
-------------
If you are looking for information on a specific function, class, or method,
this part of the documentation is for you.
.. toctree::
:maxdepth: 3
api

View file

@ -1,8 +0,0 @@
.. _install:
Installation
------------
To install uptime-kuma-api, run this command in your terminal::
$ pip install uptime-kuma-api

View file

@ -1,35 +0,0 @@
@ECHO OFF
pushd %~dp0
REM Command file for Sphinx documentation
if "%SPHINXBUILD%" == "" (
set SPHINXBUILD=sphinx-build
)
set SOURCEDIR=.
set BUILDDIR=_build
%SPHINXBUILD% >NUL 2>NUL
if errorlevel 9009 (
echo.
echo.The 'sphinx-build' command was not found. Make sure you have Sphinx
echo.installed, then set the SPHINXBUILD environment variable to point
echo.to the full path of the 'sphinx-build' executable. Alternatively you
echo.may add the Sphinx directory to PATH.
echo.
echo.If you don't have Sphinx installed, grab it from
echo.https://www.sphinx-doc.org/
exit /b 1
)
if "%1" == "" goto help
%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
goto end
:help
%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
:end
popd

View file

@ -1,29 +0,0 @@
#!/bin/sh
version="$1"
if [ $version ]
then
versions=("$version")
else
versions=(1.23.2 1.23.0 1.22.1 1.22.0 1.21.3)
fi
for version in ${versions[*]}
do
echo "Starting uptime kuma $version..."
docker run -d -it --rm -p 3001:3001 --name uptimekuma "louislam/uptime-kuma:$version" > /dev/null
while [[ "$(curl -s -L -o /dev/null -w ''%{http_code}'' 127.0.0.1:3001)" != "200" ]]
do
sleep 0.5
done
echo "Running tests..."
python -m unittest discover -s tests
echo "Stopping uptime kuma..."
docker stop uptimekuma > /dev/null
sleep 1
echo ''
done

1
scripts/.gitignore vendored
View file

@ -1,2 +1 @@
uptime-kuma
uptime-kuma-old

View file

@ -1,57 +1,34 @@
import glob
import re
from bs4 import BeautifulSoup
from utils import deduplicate_list, parse_vue_template, diff, type_html_to_py
input_ignores = {
"settings": [
"$root.styleElapsedTime"
]
}
from utils import deduplicate_list, parse_vue_template
def parse_model_keys(content, object_name):
soup = BeautifulSoup(content, "html.parser")
soup_inputs = soup.find_all(attrs={"v-model": True})
inputs = []
for soup_input in soup_inputs:
key = soup_input["v-model"]
if key in input_ignores.get(object_name, []):
continue
else:
key = re.sub(r'^' + object_name + r'\.', "", key)
type_ = soup_input.get("type", "text")
type_ = type_html_to_py(type_)
if type_ == "bool":
value = True if soup_input.get("checked") else False
else:
value = soup_input.get("value")
inputs.append({
"key": key,
"type": type_,
"default": value,
})
return inputs
match = re.findall(object_name + r"\.[0-9a-zA-Z_$]+", content)
keys = []
for m in match:
key = m.replace(object_name + ".", "")
keys.append(key)
keys = deduplicate_list(keys)
return keys
def parse_proxy_keys(root):
content = parse_vue_template(f"{root}/src/components/ProxyDialog.vue")
def parse_proxy_keys():
content = parse_vue_template("uptime-kuma/src/components/ProxyDialog.vue")
keys = parse_model_keys(content, "proxy")
return keys
def parse_notification_keys(root):
content = parse_vue_template(f"{root}/src/components/NotificationDialog.vue")
def parse_notification_keys():
content = parse_vue_template("uptime-kuma/src/components/NotificationDialog.vue")
keys = parse_model_keys(content, "notification")
return keys
def parse_settings_keys(root):
def parse_settings_keys():
all_keys = []
for path in glob.glob(f'{root}/src/components/settings/*'):
for path in glob.glob('uptime-kuma/src/components/settings/*'):
content = parse_vue_template(path)
keys = parse_model_keys(content, "settings")
all_keys.extend(keys)
@ -59,20 +36,20 @@ def parse_settings_keys(root):
return all_keys
def parse_monitor_keys(root):
content = parse_vue_template(f"{root}/src/pages/EditMonitor.vue")
def parse_monitor_keys():
content = parse_vue_template("uptime-kuma/src/pages/EditMonitor.vue")
keys = parse_model_keys(content, "monitor")
return keys
def parse_status_page_keys(root):
def parse_status_page_keys():
all_keys = ["id"]
content = parse_vue_template(f"{root}/src/pages/StatusPage.vue")
content = parse_vue_template("uptime-kuma/src/pages/StatusPage.vue")
keys = parse_model_keys(content, "config")
all_keys.extend(keys)
content = parse_vue_template(f"{root}/src/pages/ManageStatusPage.vue")
content = parse_vue_template("uptime-kuma/src/pages/ManageStatusPage.vue")
keys = parse_model_keys(content, "statusPage")
all_keys.extend(keys)
@ -80,28 +57,21 @@ def parse_status_page_keys(root):
return all_keys
def parse_maintenance_keys(root):
content = parse_vue_template(f"{root}/src/pages/EditMaintenance.vue")
keys = parse_model_keys(content, "maintenance")
return keys
def main():
root_old = "uptime-kuma-old"
root_new = "uptime-kuma"
proxy_keys = parse_proxy_keys()
print("proxy:", proxy_keys)
for name, func in [
["proxy", parse_proxy_keys],
["notification", parse_notification_keys],
["settings", parse_settings_keys],
["monitor", parse_monitor_keys],
["status_page", parse_status_page_keys],
["maintenance", parse_maintenance_keys],
]:
keys_old = func(root_old)
keys_new = func(root_new)
print(f"{name}:")
diff(keys_old, keys_new)
notification_keys = parse_notification_keys()
print("notification:", notification_keys)
settings_keys = parse_settings_keys()
print("settings:", settings_keys)
monitor_keys = parse_monitor_keys()
print("monitor:", monitor_keys)
status_page_keys = parse_status_page_keys()
print("status_page:", status_page_keys)
if __name__ == "__main__":

View file

@ -1,6 +1,7 @@
import re
from pprint import pprint
from utils import deduplicate_list, diff
from utils import deduplicate_list
def parse_json_keys(data):
@ -18,8 +19,17 @@ def parse_json_keys(data):
return keys
def parse_heartbeat(root):
with open(f'{root}/server/model/heartbeat.js') as f:
# def parse_object_keys(code, object_name):
# match = re.findall(object_name + r'\.[0-9a-zA-Z_$]+', code)
# keys = []
# for m in match:
# key = m.replace(object_name + ".", "")
# keys.append(key)
# return list(set(keys))
def parse_heartbeat():
with open('uptime-kuma/server/model/heartbeat.js') as f:
content = f.read()
all_keys = []
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
@ -34,8 +44,8 @@ def parse_heartbeat(root):
return all_keys
def parse_incident(root):
with open(f'{root}/server/model/incident.js') as f:
def parse_incident():
with open('uptime-kuma/server/model/incident.js') as f:
content = f.read()
match = re.search(r'toPublicJSON\(\) {\s+return.*{([^}]+)}', content)
data = match.group(1)
@ -43,9 +53,9 @@ def parse_incident(root):
return keys
def parse_monitor(root):
def parse_monitor():
# todo: toPublicJSON ???
with open(f'{root}/server/model/monitor.js') as f:
with open('uptime-kuma/server/model/monitor.js') as f:
content = f.read()
matches = re.findall(r'data = {([^}]+)}', content)
all_keys = []
@ -57,8 +67,8 @@ def parse_monitor(root):
return all_keys
def parse_proxy(root):
with open(f'{root}/server/model/proxy.js') as f:
def parse_proxy():
with open('uptime-kuma/server/model/proxy.js') as f:
content = f.read()
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
data = match.group(1)
@ -89,7 +99,7 @@ def parse_proxy(root):
# # input (add, edit proxy)
# def parse_proxy2():
# with open(f'{root}/server/proxy.js') as f:
# with open('uptime-kuma/server/proxy.js') as f:
# content = f.read()
#
# code = parse_function(r'async save\([^)]+\) ', content)
@ -97,8 +107,8 @@ def parse_proxy(root):
# return keys
def parse_status_page(root):
with open(f'{root}/server/model/status_page.js') as f:
def parse_status_page():
with open('uptime-kuma/server/model/status_page.js') as f:
content = f.read()
all_keys = []
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
@ -113,8 +123,8 @@ def parse_status_page(root):
return all_keys
def parse_tag(root):
with open(f'{root}/server/model/tag.js') as f:
def parse_tag():
with open('uptime-kuma/server/model/tag.js') as f:
content = f.read()
match = re.search(r'toJSON\(\) {\s+return.*{([^}]+)}', content)
data = match.group(1)
@ -122,22 +132,33 @@ def parse_tag(root):
return keys
if __name__ == "__main__":
root_old = "uptime-kuma-old"
root_new = "uptime-kuma"
print("heartbeat")
pprint(parse_heartbeat())
print("")
for name, func in [
["heartbeat", parse_heartbeat],
["incident", parse_incident],
["monitor", parse_monitor],
["proxy", parse_proxy],
["status page", parse_status_page],
["tag", parse_tag],
]:
keys_old = func(root_old)
keys_new = func(root_new)
print(f"{name}:")
diff(keys_old, keys_new)
print("incident")
pprint(parse_incident())
print("")
print("monitor")
pprint(parse_monitor())
print("")
print("proxy")
pprint(parse_proxy())
print("")
# print("prox2")
# pprint(parse_proxy2())
# print("")
print("status page")
pprint(parse_status_page())
print("")
print("tag")
pprint(parse_tag())
print("")
# TODO:

View file

@ -3,32 +3,6 @@ from bs4 import BeautifulSoup
from utils import parse_vue_template, write_to_file
titles = {
"http": "HTTP(s)",
"port": "TCP Port",
"ping": "Ping",
"keyword": "HTTP(s) - Keyword",
"grpc-keyword": "gRPC(s) - Keyword",
"dns": "DNS",
"docker": "Docker Container",
"push": "Push",
"steam": "Steam Game Server",
"gamedig": "GameDig",
"mqtt": "MQTT",
"sqlserver": "Microsoft SQL Server",
"postgres": "PostgreSQL",
"mysql": "MySQL/MariaDB",
"mongodb": "MongoDB",
"radius": "Radius",
"redis": "Redis",
"group": "Group",
"json-query": "HTTP(s) - Json Query",
"real-browser": "HTTP(s) - Browser Engine (Chrome/Chromium)",
"kafka-producer": "Kafka Producer",
"tailscale-ping": "Tailscale Ping"
}
def parse_monitor_types():
content = parse_vue_template("uptime-kuma/src/pages/EditMonitor.vue")
@ -36,13 +10,10 @@ def parse_monitor_types():
select = soup.find("select", id="type")
options = select.find_all("option")
types = {}
types = []
for o in options:
type_ = o.attrs["value"]
types[type_] = {
"value": type_,
"title": titles[type_]
}
types.append(type_)
return types

View file

@ -1,109 +0,0 @@
from uptime_kuma_api import notification_provider_options
data = {
"lunaseaTarget": """Allowed values: "device", "user".""",
"lunaseaUserID": """User ID.""",
"lunaseaDevice": """Device ID.""",
"pagertreeAutoResolve": """
Available values are:
- ``0``: Do Nothing
- ``resolve``: Auto Resolve""",
"pagertreeUrgency": """
Available values are:
- ``silent``: Silent
- ``low``: Low
- ``medium``: Medium
- ``high``: High
- ``critical``: Critical""",
"promosmsAllowLongSMS": "Allow long SMS.",
"promosmsPhoneNumber": "Phone number (for Polish recipient You can skip area codes).",
"promosmsSMSType": """
Available values are:
- ``0``: SMS FLASH - Message will automatically show on recipient device. Limited only to Polish recipients.
- ``1``: SMS ECO - cheap but slow and often overloaded. Limited only to Polish recipients.
- ``3``: SMS FULL - Premium tier of SMS, You can use your Sender Name (You need to register name first). Reliable for alerts.
- ``4``: SMS SPEED - Highest priority in system. Very quick and reliable but costly (about twice of SMS FULL price).""",
"smseagleEncoding": "True to send messages in unicode.",
"smseaglePriority": "Message priority (0-9, default = 0).",
"smseagleRecipientType": """Recipient type.
Available values are:
- ``smseagle-to``: Phone number(s)
- ``smseagle-group``: Phonebook group name(s)
- ``smseagle-contact``: Phonebook contact name(s)""",
"smseagleToken": "API Access token.",
"smseagleRecipient": "Recipient(s) (multiple must be separated with comma).",
"smseagleUrl": "Your SMSEagle device URL.",
"splunkAutoResolve": """Auto resolve or acknowledged.
Available values are:
- ``0``: do nothing
- ``ACKNOWLEDGEMENT``: auto acknowledged
- ``RECOVERY``: auto resolve""",
"splunkSeverity": """Severity.
Available values are:
- ``INFO``
- ``WARNING``
- ``CRITICAL``""",
"splunkRestURL": "Splunk Rest URL.",
"opsgeniePriority": "Priority. Available values are numbers between ``1`` and ``5``.",
"opsgenieRegion": """Region. Available values are:
- ``us``: US (Default)
- ``eu``: EU""",
"opsgenieApiKey": "API Key.",
"twilioAccountSID": "Account SID.",
"twilioAuthToken": "Auth Token.",
"twilioToNumber": "To Number.",
"twilioFromNumber": "From Number.",
"pushoverttl": "Message TTL (Seconds).",
"ntfyaccesstoken": "Access Token.",
"ntfyAuthenticationMethod": "Authentication Method.",
}
for provider in notification_provider_options:
provider_options = notification_provider_options[provider]
for option in provider_options:
type_ = provider_options[option]["type"]
required = provider_options[option]["required"]
text = data.get(option)
line = f":param {type_}{', optional' if required else ''} {option}: Notification option for ``type`` :attr:`~.NotificationType.{provider.name}`."
if text:
line += f" {text}"
print(line)

View file

@ -3,96 +3,12 @@ import re
from bs4 import BeautifulSoup
from utils import deduplicate_list, write_to_file, type_html_to_py
# deprecated or wrong inputs
ignored_inputs = {
"slack": [
"slackbutton"
],
"rocket.chat": [
"rocketbutton"
],
"octopush": [
"octopushDMLogin",
"octopushDMAPIKey",
"octopushDMPhoneNumber",
"octopushDMSenderName",
"octopushDMSMSType"
],
"Splunk": [
"pagerdutyIntegrationKey"
]
}
input_overwrites = {
"showAdditionalHeadersField": "webhookAdditionalHeaders"
}
titles = {
"alerta": "Alerta",
"AlertNow": "AlertNow",
"apprise": "Apprise (Support 50+ Notification services)",
"Bark": "Bark",
"clicksendsms": "ClickSend SMS",
"discord": "Discord",
"GoogleChat": "Google Chat (Google Workspace)",
"gorush": "Gorush",
"gotify": "Gotify",
"HomeAssistant": "Home Assistant",
"Kook": "Kook",
"line": "LINE Messenger",
"LineNotify": "LINE Notify",
"lunasea": "LunaSea",
"matrix": "Matrix",
"mattermost": "Mattermost",
"ntfy": "Ntfy",
"octopush": "Octopush",
"OneBot": "OneBot",
"Opsgenie": "Opsgenie",
"PagerDuty": "PagerDuty",
"PagerTree": "PagerTree",
"pushbullet": "Pushbullet",
"PushByTechulus": "Push by Techulus",
"pushover": "Pushover",
"pushy": "Pushy",
"rocket.chat": "Rocket.Chat",
"signal": "Signal",
"slack": "Slack",
"squadcast": "SquadCast",
"SMSEagle": "SMSEagle",
"smtp": "Email (SMTP)",
"stackfield": "Stackfield",
"teams": "Microsoft Teams",
"telegram": "Telegram",
"twilio": "Twilio",
"Splunk": "Splunk",
"webhook": "Webhook",
"GoAlert": "GoAlert",
"ZohoCliq": "ZohoCliq",
"AliyunSMS": "AliyunSMS",
"DingDing": "DingDing",
"Feishu": "Feishu",
"FreeMobile": "FreeMobile (mobile.free.fr)",
"PushDeer": "PushDeer",
"promosms": "PromoSMS",
"serwersms": "SerwerSMS.pl",
"SMSManager": "SmsManager (smsmanager.cz)",
"WeCom": "WeCom",
"ServerChan": "ServerChan",
"nostr": "Nostr",
"FlashDuty": "FlashDuty",
"smsc": "SMSC",
}
from utils import deduplicate_list, write_to_file
def build_notification_providers():
root = "uptime-kuma"
providers = {}
# get providers and input names
for path in sorted(glob.glob(f'{root}/server/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)
@ -100,20 +16,20 @@ def build_notification_providers():
match = re.search(r'name = "([^"]+)";', content)
name = match.group(1)
inputs = re.findall(r'notification\??\.([^ ,.;})\]]+)', content)
inputs = re.findall(r'notification\.([^ ,.;})\]]+)', content)
inputs = deduplicate_list(inputs)
inputs = [i.strip() for i in inputs]
providers[name] = {
"title": titles[name],
"inputs": {},
}
for input_ in inputs:
if input_ not in ignored_inputs.get(name, []):
providers[name]["inputs"][input_] = {}
providers.append({
"name": name,
"inputs": inputs,
})
return providers
# get inputs
for path in glob.glob(f'{root}/src/components/notifications/*'):
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:
@ -121,59 +37,25 @@ def build_notification_providers():
match = re.search(r'<template>[\s\S]+</template>', content, re.MULTILINE)
html = match.group(0)
soup = BeautifulSoup(html, "html.parser")
inputs = soup.find_all(attrs={"v-model": True})
for input_ in inputs:
conditions = {}
attrs = input_.attrs
inputs = soup.find_all("input")
for input in inputs:
condition = {}
attrs = input.attrs
v_model = attrs.get("v-model")
v_model_overwrite = input_overwrites.get(v_model)
if v_model_overwrite:
param_name = v_model_overwrite
else:
param_name = re.match(r'\$parent.notification.(.*)$', v_model).group(1)
type_ = attrs.get("type")
type_ = type_html_to_py(type_)
required_true_values = ['', 'true']
if attrs.get("required") in required_true_values or attrs.get(":required") in required_true_values:
required = True
else:
required = False
min_ = attrs.get("min")
if min_:
conditions["min"] = int(min_)
max_ = attrs.get("max")
if min_:
condition["min"] = int(min_)
if max_:
conditions["max"] = int(max_)
# find provider inputs dict
input_found = False
for name in list(providers.keys()):
inputs = providers[name]["inputs"]
for provider_input in inputs:
if provider_input == param_name:
input_found = True
providers[name]["inputs"][provider_input] = {
"conditions": conditions,
"type": type_,
"required": required
}
assert input_found
return providers
condition["max"] = int(max_)
param_name = re.match(r'\$parent.notification.(.*)$', v_model).group(1)
if condition:
conditions[param_name] = condition
return conditions
notification_providers = build_notification_providers()
notification_provider_conditions = {}
for notification_provider in notification_providers:
for notification_provider_input_name in notification_providers[notification_provider]["inputs"]:
notification_provider_input = notification_providers[notification_provider]["inputs"][notification_provider_input_name]
if notification_provider_input["conditions"]:
notification_provider_conditions[notification_provider_input_name] = notification_provider_input["conditions"]
notification_provider_conditions = build_notification_provider_conditions()
write_to_file(
"notification_providers.py.j2", "./../uptime_kuma_api/notification_providers.py",

View file

@ -0,0 +1,6 @@
notification = [
"type",
"isDefault",
"userId",
"applyExisting",
]

View file

@ -2,10 +2,7 @@ from enum import Enum
class MonitorType(str, Enum):
"""Enumerate monitor types."""
{{""}}
{%- for type_ in monitor_types.values() %}
{{ type_["value"].upper().replace("-", "_") }} = "{{ type_["value"] }}"
"""{{ type_["title"] }}"""
{% endfor -%}
{%- for name in monitor_types %}
{{ name.upper() }} = "{{ name }}"
{%- endfor %}

View file

@ -2,30 +2,30 @@ from enum import Enum
class NotificationType(str, Enum):
"""Enumerate notification types."""
{% for provider in notification_providers %}
{{ provider.upper().replace(".", "_") }} = "{{ provider }}"
"""{{ notification_providers[provider]["title"] }}"""
{% endfor %}
{%- for notification_provider in notification_providers %}
{%- set name = notification_provider["name"] %}
{{ name.upper().replace(".", "_") }} = "{{ name }}"
{%- endfor %}
notification_provider_options = {
{%- for provider in notification_providers %}
NotificationType.{{ provider.upper().replace(".", "_") }}: dict(
{%- for input_name in notification_providers[provider]["inputs"] %}
{%- set input = notification_providers[provider]["inputs"][input_name] %}
{{ input_name }}=dict(type="{{ input["type"] }}", required={{ input["required"] }}),
{%- for notification_provider in notification_providers %}
{%- set name = notification_provider["name"] %}
NotificationType.{{ name.upper().replace(".", "_") }}: [
{%- for input in notification_provider["inputs"] %}
"{{ input }}",
{%- endfor %}
),
],
{%- endfor %}
}
notification_provider_conditions = dict(
notification_provider_conditions = {
{%- for provider in notification_provider_conditions %}
{{ provider }}=dict(
"{{ provider }}": {
{%- for key, value in notification_provider_conditions[provider].items() %}
{{ key }}={{ value }},
"{{ key }}": {{ value }},
{%- endfor %}
),
},
{%- endfor %}
)
}

2
scripts/requirenents.txt Normal file
View file

@ -0,0 +1,2 @@
Jinja2==3.1.2
BeautifulSoup4==4.11.1

View file

@ -24,23 +24,3 @@ def write_to_file(template, destination, **kwargs):
rendered = template.render(**kwargs)
with open(destination, "w") as f:
f.write(rendered)
def diff(old, new):
for i in new:
if i not in old:
print("+", i)
for i in old:
if i not in new:
print("-", i)
print("")
def type_html_to_py(type_):
if type_ == "number":
type_ = "int"
elif type_ == "checkbox":
type_ = "bool"
else:
type_ = "str"
return type_

View file

@ -29,13 +29,13 @@ setup(
author_email="lucasheld@hotmail.de",
license=info["__license__"],
packages=["uptime_kuma_api"],
python_requires=">=3.7, <4",
python_requires=">=3.6, <4",
install_requires=[
"python-socketio[client]>=5.0.0",
"packaging"
],
classifiers=[
"Development Status :: 5 - Production/Stable",
"Development Status :: 1 - Planning",
"Environment :: Web Environment",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
@ -43,6 +43,7 @@ setup(
"Operating System :: OS Independent",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",

1
tests/requirements.txt Normal file
View file

@ -0,0 +1 @@
pyotp==2.6.0

View file

@ -1,59 +0,0 @@
import unittest
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase
class TestApiKey(UptimeKumaTestCase):
def test_api_key(self):
# get empty list to make sure that future accesses will also work
self.api.get_api_keys()
expected = {
"name": "name 1",
"expires": "2023-03-30 12:20:00",
"active": True
}
# add api key
r = self.api.add_api_key(**expected)
self.assertEqual(r["msg"], "Added Successfully.")
api_key_id = r["keyID"]
# get api key
api_key = self.api.get_api_key(api_key_id)
self.compare(api_key, expected)
# get api keys
api_keys = self.api.get_api_keys()
api_key = self.find_by_id(api_keys, api_key_id)
self.assertIsNotNone(api_key)
self.compare(api_key, expected)
# disable api key
r = self.api.disable_api_key(api_key_id)
self.assertEqual(r["msg"], "Disabled Successfully.")
api_key = self.api.get_api_key(api_key_id)
expected["active"] = False
self.compare(api_key, expected)
# enable api key
r = self.api.enable_api_key(api_key_id)
self.assertEqual(r["msg"], "Enabled Successfully")
api_key = self.api.get_api_key(api_key_id)
expected["active"] = True
self.compare(api_key, expected)
# delete api key
r = self.api.delete_api_key(api_key_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_api_key(api_key_id)
def test_delete_not_existing_api_key(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_api_key(42)
if __name__ == '__main__':
unittest.main()

View file

@ -6,10 +6,13 @@ from uptime_kuma_test_case import UptimeKumaTestCase
class TestDockerHost(UptimeKumaTestCase):
def test_docker_host(self):
# get empty list to make sure that future accesses will also work
self.api.get_docker_hosts()
def setUp(self):
super(TestDockerHost, self).setUp()
if parse_version(self.api.version) < parse_version("1.18"):
super(TestDockerHost, self).tearDown()
self.skipTest("Unsupported in this Uptime Kuma version")
def test_docker_host(self):
expected_docker_host = {
"name": "name 1",
"dockerType": DockerType.SOCKET,
@ -17,10 +20,8 @@ class TestDockerHost(UptimeKumaTestCase):
}
# test docker host
if parse_version(self.api.version) != parse_version("1.23.0"):
# test_docker_host does not work in 1.23.0 (https://github.com/louislam/uptime-kuma/issues/3605)
with self.assertRaisesRegex(UptimeKumaException, r'connect ENOENT /var/run/docker.sock'):
self.api.test_docker_host(**expected_docker_host)
with self.assertRaisesRegex(UptimeKumaException, r'connect ENOENT /var/run/docker.sock'):
self.api.test_docker_host(**expected_docker_host)
# add docker host
r = self.api.add_docker_host(**expected_docker_host)
@ -33,7 +34,6 @@ class TestDockerHost(UptimeKumaTestCase):
# get docker hosts
docker_hosts = self.api.get_docker_hosts()
self.assertTrue(type(docker_hosts[0]["dockerType"]) == DockerType)
docker_host = self.find_by_id(docker_hosts, docker_host_id)
self.assertIsNotNone(docker_host)
self.compare(docker_host, expected_docker_host)
@ -51,10 +51,6 @@ class TestDockerHost(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException):
self.api.get_docker_host(docker_host_id)
def test_delete_not_existing_docker_host(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_docker_host(42)
if __name__ == '__main__':
unittest.main()

View file

@ -1,13 +0,0 @@
import unittest
from uptime_kuma_test_case import UptimeKumaTestCase
class TestGameList(UptimeKumaTestCase):
def test_game_list(self):
game_list = self.api.get_game_list()
self.assertTrue("keys" in game_list[0])
if __name__ == '__main__':
unittest.main()

View file

@ -1,19 +1,14 @@
import unittest
from uptime_kuma_api import MonitorStatus
from uptime_kuma_test_case import UptimeKumaTestCase
class TestHeartbeat(UptimeKumaTestCase):
def test_get_heartbeats(self):
self.add_monitor()
r = self.api.get_heartbeats()
self.assertTrue(type(list(r.values())[0][0]["status"]) == MonitorStatus)
self.api.get_heartbeats()
def test_get_important_heartbeats(self):
self.add_monitor()
r = self.api.get_important_heartbeats()
self.assertTrue(type(list(r.values())[0][0]["status"]) == MonitorStatus)
self.api.get_important_heartbeats()
if __name__ == '__main__':

View file

@ -1,15 +0,0 @@
import unittest
from uptime_kuma_api import MonitorStatus
from uptime_kuma_test_case import UptimeKumaTestCase
class TestHelperMethods(UptimeKumaTestCase):
def test_monitor_status(self):
monitor_id = self.add_monitor()
status = self.api.get_monitor_status(monitor_id)
self.assertTrue(type(status) == MonitorStatus)
if __name__ == '__main__':
unittest.main()

View file

@ -1,6 +1,5 @@
import unittest
from uptime_kuma_api import UptimeKumaApi
from uptime_kuma_test_case import UptimeKumaTestCase
@ -10,16 +9,6 @@ class TestInfo(UptimeKumaTestCase):
self.assertIn("version", info)
self.assertIn("latestVersion", info)
def test_info_with_version(self):
# If wait_events is set to 0, the first info event is normally used.
# The info event handler needs to drop this first event without a version.
self.api.logout()
self.api.disconnect()
self.api = UptimeKumaApi(self.url, wait_events=0)
self.api.login(self.username, self.password)
info = self.api.info()
self.assertIn("version", info)
if __name__ == '__main__':
unittest.main()

View file

@ -1,31 +0,0 @@
import unittest
from uptime_kuma_api import UptimeKumaApi
from uptime_kuma_test_case import UptimeKumaTestCase
class TestLogin(UptimeKumaTestCase):
def test_auto_login(self):
# disable auth
r = self.api.set_settings(self.password, disableAuth=True)
self.assertEqual(r["msg"], "Saved")
# login again without username and password
self.api.logout()
self.api.disconnect()
self.api = UptimeKumaApi(self.url)
self.api.login()
r = self.api.get_settings()
self.assertTrue(r["disableAuth"])
# enable auth again
r = self.api.set_settings(disableAuth=False)
self.assertEqual(r["msg"], "Saved")
r = self.api.get_settings()
self.assertFalse(r["disableAuth"])
if __name__ == '__main__':
unittest.main()

View file

@ -1,269 +0,0 @@
import unittest
from uptime_kuma_api import UptimeKumaException, MaintenanceStrategy
from uptime_kuma_test_case import UptimeKumaTestCase
class TestMaintenance(UptimeKumaTestCase):
def test_maintenance(self):
expected_maintenance = {
"title": "maintenance 1",
"description": "test",
"strategy": MaintenanceStrategy.SINGLE,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:36:00",
"2022-12-29 22:36:00"
],
"weekdays": [],
"daysOfMonth": [],
"timezoneOption": "Europe/Berlin"
}
# add maintenance
r = self.api.add_maintenance(**expected_maintenance)
self.assertEqual(r["msg"], "Added Successfully.")
maintenance_id = r["maintenanceID"]
# get maintenance
maintenance = self.api.get_maintenance(maintenance_id)
self.assertTrue(type(maintenance["strategy"]) == MaintenanceStrategy)
self.compare(maintenance, expected_maintenance)
# get maintenances
maintenances = self.api.get_maintenances()
self.assertTrue(type(maintenances[0]["strategy"]) == MaintenanceStrategy)
maintenance = self.find_by_id(maintenances, maintenance_id)
self.assertIsNotNone(maintenance)
self.compare(maintenance, expected_maintenance)
# edit maintenance
expected_maintenance["strategy"] = MaintenanceStrategy.RECURRING_INTERVAL
expected_maintenance["title"] = "maintenance 1 new"
r = self.api.edit_maintenance(maintenance_id, **expected_maintenance)
self.assertEqual(r["msg"], "Saved.")
maintenance = self.api.get_maintenance(maintenance_id)
self.compare(maintenance, expected_maintenance)
# pause maintenance
r = self.api.pause_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Paused Successfully.")
# resume maintenance
r = self.api.resume_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Resume Successfully")
# add monitor maintenance
monitor_name = "monitor 1"
monitor_id = self.add_monitor(monitor_name)
monitors = [
{
"id": monitor_id
},
]
r = self.api.add_monitor_maintenance(maintenance_id, monitors)
self.assertEqual(r["msg"], "Added Successfully.")
# get monitor maintenance
monitors = self.api.get_monitor_maintenance(maintenance_id)
monitor = self.find_by_id(monitors, monitor_id)
self.assertIsNotNone(monitor)
# add status page maintenance
status_page_title = "status page 1"
status_page_id = self.add_status_page(status_page_title)
status_pages = [
{
"id": status_page_id
}
]
r = self.api.add_status_page_maintenance(maintenance_id, status_pages)
self.assertEqual(r["msg"], "Added Successfully.")
# get status page maintenance
status_pages = self.api.get_status_page_maintenance(maintenance_id)
status_page = self.find_by_id(status_pages, status_page_id)
self.assertIsNotNone(status_page)
# delete maintenance
r = self.api.delete_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_maintenance(maintenance_id)
def test_maintenance_strategy_manual(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.MANUAL,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 00:00:00"
],
"weekdays": [],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_single(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.SINGLE,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:36:00",
"2022-12-29 22:36:00"
],
"weekdays": [],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_recurring_interval(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.RECURRING_INTERVAL,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:37:00",
"2022-12-31 22:37:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_recurring_weekday(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.RECURRING_WEEKDAY,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:38:00",
"2022-12-31 22:38:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [
1,
3,
5,
0
],
"daysOfMonth": []
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_recurring_day_of_month(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.RECURRING_DAY_OF_MONTH,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:39:00",
"2022-12-31 22:39:00"
],
"timeRange": [
{
"hours": 2,
"minutes": 0
},
{
"hours": 3,
"minutes": 0
}
],
"weekdays": [],
"daysOfMonth": [
1,
10,
20,
30,
"lastDay1"
]
}
self.do_test_maintenance_strategy(expected_maintenance)
def test_maintenance_strategy_cron(self):
expected_maintenance = {
"title": "test",
"description": "test",
"strategy": MaintenanceStrategy.CRON,
"active": True,
"intervalDay": 1,
"dateRange": [
"2022-12-27 22:37:00",
"2022-12-31 22:37:00"
],
"weekdays": [],
"daysOfMonth": [],
"cron": "50 5 * * *",
"durationMinutes": 120,
"timezoneOption": "Europe/Berlin"
}
self.do_test_maintenance_strategy(expected_maintenance)
def do_test_maintenance_strategy(self, expected_maintenance):
# add maintenance
r = self.api.add_maintenance(**expected_maintenance)
self.assertEqual(r["msg"], "Added Successfully.")
maintenance_id = r["maintenanceID"]
# get maintenance
maintenance = self.api.get_maintenance(maintenance_id)
self.compare(maintenance, expected_maintenance)
# get maintenances
maintenances = self.api.get_maintenances()
maintenance = self.find_by_id(maintenances, maintenance_id)
self.assertIsNotNone(maintenance)
self.compare(maintenance, expected_maintenance)
# edit maintenance
r = self.api.edit_maintenance(maintenance_id, title="name 2")
self.assertEqual(r["msg"], "Saved.")
maintenance = self.api.get_maintenance(maintenance_id)
expected_maintenance["title"] = "name 2"
self.compare(maintenance, expected_maintenance)
# delete maintenance
r = self.api.delete_maintenance(maintenance_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_maintenance(maintenance_id)
def test_delete_not_existing_maintenance(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_maintenance(42)
if __name__ == '__main__':
unittest.main()

View file

@ -1,28 +1,16 @@
import unittest
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod, MonitorStatus
from uptime_kuma_api import UptimeKumaException, MonitorType
from uptime_kuma_test_case import UptimeKumaTestCase
class TestMonitor(UptimeKumaTestCase):
def test_monitor(self):
# get empty list to make sure that future accesses will also work
self.api.get_monitors()
notification_id_1 = self.add_notification()
notification_id_2 = self.add_notification()
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"interval": 60,
"retryInterval": 60,
"maxretries": 0,
"notificationIDList": [notification_id_1, notification_id_2],
"upsideDown": False,
"url": "http://127.0.0.1",
"resendInterval": 0
"url": "http://127.0.0.1"
}
# add monitor
@ -36,16 +24,12 @@ class TestMonitor(UptimeKumaTestCase):
# get monitors
monitors = self.api.get_monitors()
self.assertTrue(type(monitors[0]["type"]) == MonitorType)
self.assertTrue(type(monitors[0]["authMethod"]) == AuthMethod)
monitor = self.find_by_id(monitors, monitor_id)
self.assertTrue(type(monitor["type"]) == MonitorType)
self.assertTrue(type(monitor["authMethod"]) == AuthMethod)
self.assertIsNotNone(monitor)
self.compare(monitor, expected_monitor)
# edit monitor
expected_monitor["type"] = MonitorType.PING
expected_monitor["type"] = "ping"
expected_monitor["name"] = "monitor 1 new"
expected_monitor["hostname"] = "127.0.0.10"
del expected_monitor["url"]
@ -63,8 +47,7 @@ class TestMonitor(UptimeKumaTestCase):
self.assertEqual(r["msg"], "Resumed Successfully.")
# get monitor beats
r = self.api.get_monitor_beats(monitor_id, 6)
self.assertTrue(type(r[0]["status"]) == MonitorStatus)
self.api.get_monitor_beats(monitor_id, 6)
# delete monitor
r = self.api.delete_monitor(monitor_id)
@ -80,68 +63,14 @@ class TestMonitor(UptimeKumaTestCase):
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
expected_monitor.update({
"name": "monitor 2"
})
r = self.api.edit_monitor(monitor_id, **expected_monitor)
self.assertEqual(r["msg"], "Saved.")
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
return monitor
def test_monitor_type_http(self):
proxy_id = self.add_proxy()
json_data = '{"key": "value"}'
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"expiryNotification": False,
"ignoreTls": False,
"maxredirects": 10,
"accepted_statuscodes": ["200-299"],
"proxyId": proxy_id,
"method": "GET",
"body": json_data,
"headers": json_data,
"authMethod": AuthMethod.NONE
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_auth_method(self):
for auth_method in [AuthMethod.HTTP_BASIC, AuthMethod.NTLM]:
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"authMethod": auth_method,
"basic_auth_user": "auth user",
"basic_auth_pass": "auth pass",
}
self.do_test_monitor_type(expected_monitor)
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"authMethod": AuthMethod.NTLM,
"authDomain": "auth domain",
"authWorkstation": "auth workstation",
}
self.do_test_monitor_type(expected_monitor)
expected_monitor = {
"type": MonitorType.HTTP,
"name": "monitor 1",
"url": "http://127.0.0.1",
"authMethod": AuthMethod.MTLS,
"tlsCert": "cert",
"tlsKey": "key",
"tlsCa": "ca",
"headers": json_data
}
self.do_test_monitor_type(expected_monitor)
@ -158,8 +87,7 @@ class TestMonitor(UptimeKumaTestCase):
expected_monitor = {
"type": MonitorType.PING,
"name": "monitor 1",
"hostname": "127.0.0.1",
"packetSize": 56
"hostname": "127.0.0.1"
}
self.do_test_monitor_type(expected_monitor)
@ -172,35 +100,13 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_grpc_keyword(self):
expected_monitor = {
"type": MonitorType.GRPC_KEYWORD,
"name": "monitor 1",
"grpcUrl": "127.0.0.1",
"keyword": "healthy",
"grpcServiceName": "health",
"grpcMethod": "check",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_dns(self):
expected_monitor = {
"type": MonitorType.DNS,
"name": "monitor 1",
"hostname": "127.0.0.1",
"port": 8888,
"dns_resolve_server": "1.1.1.1",
"dns_resolve_type": "A"
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_docker(self):
docker_host_id = self.add_docker_host()
expected_monitor = {
"type": MonitorType.DOCKER,
"name": "monitor 1",
"docker_container": "test",
"docker_host": docker_host_id
"dns_resolve_server": "1.1.1.1"
}
self.do_test_monitor_type(expected_monitor)
@ -209,10 +115,7 @@ class TestMonitor(UptimeKumaTestCase):
"type": MonitorType.PUSH,
"name": "monitor 1"
}
monitor = self.do_test_monitor_type(expected_monitor)
# https://github.com/lucasheld/ansible-uptime-kuma/issues/5
self.assertIsNotNone(monitor["pushToken"])
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_steam(self):
expected_monitor = {
@ -223,28 +126,13 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_gamedig(self):
game_list = self.api.get_game_list()
game = game_list[0]["keys"][0]
expected_monitor = {
"type": MonitorType.GAMEDIG,
"name": "monitor 1",
"hostname": "127.0.0.1",
"port": 8888,
"game": game
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_mqtt(self):
expected_monitor = {
"type": MonitorType.MQTT,
"name": "monitor 1",
"hostname": "127.0.0.1",
"port": 8888,
"mqttUsername": "mqtt username",
"mqttPassword": "mqtt password",
"mqttTopic": "mqtt topic",
"mqttSuccessMessage": "mqtt success message"
"mqttTopic": "test"
}
self.do_test_monitor_type(expected_monitor)
@ -259,6 +147,9 @@ class TestMonitor(UptimeKumaTestCase):
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_postgres(self):
if parse_version(self.api.version) < parse_version("1.18"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.POSTGRES,
"name": "monitor 1",
@ -267,24 +158,23 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_mysql(self):
expected_monitor = {
"type": MonitorType.MYSQL,
"name": "monitor 1",
"databaseConnectionString": "mysql://username:password@host:port/database",
"databaseQuery": "select getdate()"
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_docker(self):
if parse_version(self.api.version) < parse_version("1.18"):
self.skipTest("Unsupported in this Uptime Kuma version")
def test_monitor_type_mongodb(self):
docker_host_id = self.add_docker_host()
expected_monitor = {
"type": MonitorType.MONGODB,
"type": MonitorType.DOCKER,
"name": "monitor 1",
"databaseConnectionString": "mongodb://username:password@host:port/database"
"docker_container": "test",
"docker_host": docker_host_id
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_radius(self):
if parse_version(self.api.version) < parse_version("1.18"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.RADIUS,
"name": "monitor 1",
@ -296,84 +186,20 @@ class TestMonitor(UptimeKumaTestCase):
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_redis(self):
expected_monitor = {
"type": MonitorType.REDIS,
"name": "monitor 1",
"databaseConnectionString": "redis://user:password@host:port"
}
self.do_test_monitor_type(expected_monitor)
def test_edit_notification_id_list(self):
# https://github.com/lucasheld/uptime-kuma-api/issues/3
def test_monitor_type_group(self):
if parse_version(self.api.version) < parse_version("1.22"):
self.skipTest("Unsupported in this Uptime Kuma version")
monitor_id = self.add_monitor()
notification_id = self.add_notification()
# create monitor group
expected_monitor = {
"type": MonitorType.GROUP,
"name": "monitor 1"
}
group_monitor = self.do_test_monitor_type(expected_monitor)
group_monitor_id = group_monitor["id"]
expected_monitor = self.api.get_monitor(monitor_id)
expected_monitor["notificationIDList"] = {str(notification_id): True}
# use monitor group as parent for another monitor
expected_monitor = {
"type": MonitorType.PUSH,
"name": "monitor 1",
"parent": group_monitor_id
}
self.do_test_monitor_type(expected_monitor)
r = self.api.edit_monitor(id_=monitor_id, notificationIDList=[notification_id])
self.assertEqual(r["msg"], "Saved.")
def test_monitor_type_json_query(self):
if parse_version(self.api.version) < parse_version("1.23"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.JSON_QUERY,
"name": "monitor 1",
"url": "http://127.0.0.1",
"jsonPath": "address.country",
"expectedValue": "germany",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_real_browser(self):
if parse_version(self.api.version) < parse_version("1.23"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.REAL_BROWSER,
"name": "monitor 1",
"url": "http://127.0.0.1",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_kafka_producer(self):
if parse_version(self.api.version) < parse_version("1.23"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.KAFKA_PRODUCER,
"name": "monitor 1",
"kafkaProducerTopic": "topic",
"kafkaProducerMessage": "message",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_tailscale_ping(self):
if parse_version(self.api.version) < parse_version("1.23"):
self.skipTest("Unsupported in this Uptime Kuma version")
expected_monitor = {
"type": MonitorType.TAILSCALE_PING,
"name": "monitor 1",
"hostname": "127.0.0.1"
}
self.do_test_monitor_type(expected_monitor)
def test_delete_not_existing_monitor(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_monitor(42)
monitor = self.api.get_monitor(monitor_id)
self.compare(monitor, expected_monitor)
if __name__ == '__main__':

View file

@ -1,6 +1,5 @@
import unittest
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase
@ -19,24 +18,10 @@ class TestMonitorTag(UptimeKumaTestCase):
r = self.api.add_monitor_tag(**expected_monitor_tag)
self.assertEqual(r["msg"], "Added Successfully.")
# check if tag is listed in monitor tags
monitors = self.api.get_monitors()
monitor = self.find_by_id(monitors, monitor_id)
self.assertEqual(monitor["tags"][0]["tag_id"], tag_id)
# delete monitor tag
r = self.api.delete_monitor_tag(**expected_monitor_tag)
self.assertEqual(r["msg"], "Deleted Successfully.")
# check if tag is not listed in monitor tags
monitors = self.api.get_monitors()
monitor = self.find_by_id(monitors, monitor_id)
self.assertEqual(monitor["tags"], [])
def test_delete_not_existing_monitor_tag(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_monitor_tag(42, 42, 42)
if __name__ == '__main__':
unittest.main()

View file

@ -1,25 +1,21 @@
import unittest
from uptime_kuma_api import UptimeKumaException, NotificationType
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase
class TestNotification(UptimeKumaTestCase):
def test_notification(self):
# get empty list to make sure that future accesses will also work
self.api.get_notifications()
expected_notification = {
"name": "notification 1",
"isDefault": True,
"default": True,
"applyExisting": True,
"type": NotificationType.TELEGRAM,
"telegramChatID": "123456789",
"telegramBotToken": "987654321"
"type": "PushByTechulus",
"pushAPIKey": "123456789"
}
# test notification
with self.assertRaisesRegex(UptimeKumaException, r'Not Found'):
with self.assertRaisesRegex(UptimeKumaException, r'Invalid API key'):
self.api.test_notification(**expected_notification)
# add notification
@ -33,9 +29,7 @@ class TestNotification(UptimeKumaTestCase):
# get notifications
notifications = self.api.get_notifications()
self.assertTrue(type(notifications[0]["type"]) == NotificationType)
notification = self.find_by_id(notifications, notification_id)
self.assertTrue(type(notification["type"]) == NotificationType)
self.assertIsNotNone(notification)
self.compare(notification, expected_notification)
@ -43,10 +37,9 @@ class TestNotification(UptimeKumaTestCase):
expected_notification["name"] = "notification 1 new"
expected_notification["default"] = False
expected_notification["applyExisting"] = False
expected_notification["type"] = NotificationType.PUSHDEER
expected_notification["type"] = "PushDeer"
expected_notification["pushdeerKey"] = "987654321"
del expected_notification["telegramChatID"]
del expected_notification["telegramBotToken"]
del expected_notification["pushAPIKey"]
r = self.api.edit_notification(notification_id, **expected_notification)
self.assertEqual(r["msg"], "Saved")
notification = self.api.get_notification(notification_id)
@ -59,10 +52,6 @@ class TestNotification(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException):
self.api.delete_notification(notification_id)
def test_delete_not_existing_notification(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_notification(42)
if __name__ == '__main__':
unittest.main()

View file

@ -1,27 +1,20 @@
import unittest
from uptime_kuma_api import UptimeKumaException, ProxyProtocol
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase
class TestProxy(UptimeKumaTestCase):
def test_proxy(self):
# get empty list to make sure that future accesses will also work
self.api.get_proxies()
expected_proxy = {
"protocol": ProxyProtocol.HTTP,
"protocol": "http",
"host": "127.0.0.1",
"port": 8080,
"auth": True,
"username": "username",
"password": "password",
"active": True,
"default": False
"active": True
}
# add proxy
r = self.api.add_proxy(applyExisting=False, **expected_proxy)
r = self.api.add_proxy(**expected_proxy)
self.assertEqual(r["msg"], "Saved")
proxy_id = r["id"]
@ -31,13 +24,12 @@ class TestProxy(UptimeKumaTestCase):
# get proxies
proxies = self.api.get_proxies()
self.assertTrue(type(proxies[0]["protocol"]) == ProxyProtocol)
proxy = self.find_by_id(proxies, proxy_id)
self.assertIsNotNone(proxy)
self.compare(proxy, expected_proxy)
# edit proxy
expected_proxy["protocol"] = ProxyProtocol.HTTPS
expected_proxy["protocol"] = "https"
expected_proxy["host"] = "127.0.0.2"
expected_proxy["port"] = 8888
r = self.api.edit_proxy(proxy_id, **expected_proxy)
@ -51,10 +43,6 @@ class TestProxy(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException):
self.api.get_proxy(proxy_id)
def test_delete_not_existing_proxy(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_proxy(42)
if __name__ == '__main__':
unittest.main()

View file

@ -1,37 +1,18 @@
import json
import unittest
import json
from uptime_kuma_test_case import UptimeKumaTestCase
class TestSettings(UptimeKumaTestCase):
def test_settings(self):
expected_settings = {
"checkUpdate": False,
"checkBeta": False,
"keepDataPeriodDays": 180,
"entryPage": "dashboard",
"searchEngineIndex": False,
"primaryBaseURL": "",
"steamAPIKey": "",
"tlsExpiryNotifyDays": [7, 14, 21],
"disableAuth": False,
"trustProxy": False,
"serverTimezone": "Europe/Berlin",
"dnsCache": True
}
settings_before = self.api.get_settings()
# set settings
r = self.api.set_settings(self.password, **expected_settings)
self.assertEqual(r["msg"], "Saved")
expected_check_update = not settings_before.get("checkUpdate", True)
self.api.set_settings(self.password, checkUpdate=expected_check_update)
# set settings without password
r = self.api.set_settings(**expected_settings)
self.assertEqual(r["msg"], "Saved")
# get settings
settings = self.api.get_settings()
self.compare(settings, expected_settings)
self.assertEqual(settings["checkUpdate"], expected_check_update)
def test_change_password(self):
new_password = "321terces"

View file

@ -6,9 +6,6 @@ from uptime_kuma_test_case import UptimeKumaTestCase
class TestStatusPage(UptimeKumaTestCase):
def test_status_page(self):
# get empty list to make sure that future accesses will also work
self.api.get_status_pages()
monitor_id = self.add_monitor()
slug = "slug1"
@ -16,14 +13,7 @@ class TestStatusPage(UptimeKumaTestCase):
"slug": slug,
"title": "status page 1",
"description": "description 1",
"theme": "light",
"published": True,
"showTags": False,
"domainNameList": [],
"customCSS": "",
"footerText": None,
"showPoweredBy": False,
"icon": "/icon.svg",
"publicGroupList": [
{
'name': 'Services',
@ -34,10 +24,15 @@ class TestStatusPage(UptimeKumaTestCase):
}
]
}
],
"googleAnalyticsId": ""
]
}
# slug must be unique
try:
self.api.delete_status_page(slug)
except UptimeKumaException:
pass
# add status page
r = self.api.add_status_page(slug, expected_status_page["title"])
self.assertEqual(r["msg"], "OK!")
@ -71,11 +66,9 @@ class TestStatusPage(UptimeKumaTestCase):
"style": IncidentStyle.DANGER
}
incident = self.api.post_incident(slug, **incident_expected)
self.assertTrue(type(incident["style"]) == IncidentStyle)
self.compare(incident, incident_expected)
status_page = self.api.get_status_page(slug)
self.compare(status_page["incident"], incident)
self.assertTrue(type(status_page["incident"]["style"]) == IncidentStyle)
# unpin incident
self.api.unpin_incident(slug)
@ -87,14 +80,6 @@ class TestStatusPage(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException):
self.api.get_status_page(slug)
status_pages = self.api.get_status_pages()
status_page = self.find_by_id(status_pages, slug, "slug")
self.assertIsNone(status_page)
def test_delete_not_existing_status_page(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_status_page("slug42")
if __name__ == '__main__':
unittest.main()

View file

@ -26,24 +26,12 @@ class TestTag(UptimeKumaTestCase):
self.assertIsNotNone(tag)
self.compare(tag, expected_tag)
# edit tag
expected_tag["name"] = "tag 1 new"
expected_tag["color"] = "#000000"
r = self.api.edit_tag(tag_id, **expected_tag)
self.assertEqual(r["msg"], "Saved")
tag = self.api.get_tag(tag_id)
self.compare(tag, expected_tag)
# delete tag
r = self.api.delete_tag(tag_id)
self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException):
self.api.get_tag(tag_id)
def test_delete_not_existing_tag(self):
with self.assertRaises(UptimeKumaException):
self.api.delete_tag(42)
if __name__ == '__main__':
unittest.main()

View file

@ -1,7 +1,7 @@
import json
import unittest
import warnings
from uptime_kuma_api import UptimeKumaApi, MonitorType, DockerType
from uptime_kuma_api import UptimeKumaApi, Event, MonitorType, DockerType
token = None
@ -35,9 +35,7 @@ class UptimeKumaTestCase(unittest.TestCase):
password = "secret123"
def setUp(self):
warnings.simplefilter("ignore", ResourceWarning)
self.api = UptimeKumaApi(self.url, timeout=1, wait_events=0.01)
self.api = UptimeKumaApi(self.url)
global token
if not token:
@ -48,50 +46,17 @@ class UptimeKumaTestCase(unittest.TestCase):
self.api.login_by_token(token)
# delete monitors
monitors = self.api.get_monitors()
for monitor in monitors:
self.api.delete_monitor(monitor["id"])
data = {
"version": "1.17.1",
"notificationList": [],
"monitorList": [],
"proxyList": []
}
data_str = json.dumps(data)
r = self.api.upload_backup(data_str, "overwrite")
self.assertEqual(r["msg"], "Backup successfully restored.")
# delete notifications
notifications = self.api.get_notifications()
for notification in notifications:
self.api.delete_notification(notification["id"])
# delete proxies
proxies = self.api.get_proxies()
for proxy in proxies:
self.api.delete_proxy(proxy["id"])
# delete tags
tags = self.api.get_tags()
for tag in tags:
self.api.delete_tag(tag["id"])
# delete status pages
status_pages = self.api.get_status_pages()
for status_page in status_pages:
self.api.delete_status_page(status_page["slug"])
# delete docker hosts
docker_hosts = self.api.get_docker_hosts()
for docker_host in docker_hosts:
self.api.delete_docker_host(docker_host["id"])
# delete maintenances
maintenances = self.api.get_maintenances()
for maintenance in maintenances:
self.api.delete_maintenance(maintenance["id"])
# delete api keys
api_keys = self.api.get_api_keys()
for api_key in api_keys:
self.api.delete_api_key(api_key["id"])
# login again to receive initial messages
self.api.disconnect()
self.api = UptimeKumaApi(self.url)
self.api.login_by_token(token)
self.api._event_data[Event.MONITOR_LIST] = {}
def tearDown(self):
self.api.disconnect()
@ -104,52 +69,26 @@ class UptimeKumaTestCase(unittest.TestCase):
if obj[key] == value:
return obj
def add_monitor(self, name="monitor 1"):
r = self.api.add_monitor(
type=MonitorType.HTTP,
name=name,
url="http://127.0.0.1"
)
def add_monitor(self):
r = self.api.add_monitor(type=MonitorType.HTTP, name="monitor 1", url="http://127.0.0.1")
monitor_id = r["monitorID"]
return monitor_id
def add_tag(self):
r = self.api.add_tag(
name="tag 1",
color="#ffffff"
)
r = self.api.add_tag(name="tag 1", color="#ffffff")
tag_id = r["id"]
return tag_id
def add_notification(self):
r = self.api.add_notification(
name="notification 1",
type="PushByTechulus",
pushAPIKey="123456789"
)
r = self.api.add_notification(name="notification 1", type="PushByTechulus", pushAPIKey="123456789")
notification_id = r["id"]
return notification_id
def add_proxy(self):
r = self.api.add_proxy(
protocol="http",
host="127.0.0.1",
port=8080,
active=True
)
proxy_id = r["id"]
return proxy_id
def add_docker_host(self):
r = self.api.add_docker_host(
name="docker host 1",
dockerType=DockerType.SOCKET
)
expected_docker_host = {
"name": "docker host 1",
"dockerType": DockerType.SOCKET
}
r = self.api.add_docker_host(**expected_docker_host)
docker_host_id = r["id"]
return docker_host_id
def add_status_page(self, title="status page 1"):
slug = "statuspage1"
self.api.add_status_page(slug, title)
r = self.api.get_status_page(slug)
return r["id"]

View file

@ -1,12 +1,10 @@
from .__version__ import __title__, __version__, __author__, __copyright__
from .auth_method import AuthMethod
from .monitor_status import MonitorStatus
from .monitor_type import MonitorType
from .notification_providers import NotificationType, notification_provider_options, notification_provider_conditions
from .proxy_protocol import ProxyProtocol
from .incident_style import IncidentStyle
from .docker_type import DockerType
from .maintenance_strategy import MaintenanceStrategy
from .exceptions import UptimeKumaException, Timeout
from .exceptions import UptimeKumaException
from .event import Event
from .api import UptimeKumaApi

View file

@ -1,5 +1,5 @@
__title__ = "uptime_kuma_api"
__version__ = "1.2.1"
__version__ = "0.2.0"
__author__ = "Lucas Held"
__license__ = "MIT"
__copyright__ = "Copyright 2023 Lucas Held"
__copyright__ = "Copyright 2022 Lucas Held"

File diff suppressed because one or more lines are too long

View file

@ -2,19 +2,6 @@ from enum import Enum
class AuthMethod(str, Enum):
"""Enumerate authentication methods for monitors."""
NONE = ""
"""Authentication is disabled."""
HTTP_BASIC = "basic"
"""HTTP Basic Authentication."""
NTLM = "ntlm"
"""NTLM Authentication."""
MTLS = "mtls"
"""mTLS Authentication."""
OAUTH2_CC = "oauth2-cc"
"""OAuth2: Client Credentials"""

View file

@ -2,10 +2,5 @@ from enum import Enum
class DockerType(str, Enum):
"""Enumerate docker connection types."""
SOCKET = "socket"
"""Socket"""
TCP = "tcp"
"""TCP"""

View file

@ -1,387 +0,0 @@
def append_docstring(value):
def _doc(func):
# inserts the value into the existing docstring before the :return: line
split_value = ":return:"
splitted = func.__doc__.split(split_value)
part1 = splitted[0]
line = [i for i in part1.split("\n") if i][0]
indent = len(line) - len(line.lstrip())
line_start = " " * indent
part2 = split_value + line_start.join(splitted[1:])
func.__doc__ = part1 + "\n" + line_start + value + "\n" + line_start + part2
return func
return _doc
def monitor_docstring(mode) -> str:
return f"""
:param MonitorType{", optional" if mode == "edit" else ""} type: Monitor Type
:param str{", optional" if mode == "edit" else ""} name: Friendly Name
:param str, optional parent: Id of the monitor group, defaults to None
:param str, optional description: Description, defaults to None
:param int, optional interval: Heartbeat Interval, defaults to 60
:param int, optional retryInterval: Retry every X seconds, defaults to 60
:param int, optional resendInterval: Resend every X times, defaults to 0
:param int, optional maxretries: Retries. Maximum retries before the service is marked as down and a notification is sent., defaults to 0
:param bool, optional upsideDown: Upside Down Mode. Flip the status upside down. If the service is reachable, it is DOWN., defaults to False
:param list, optional notificationIDList: Notifications, defaults to None
:param str, optional url: URL, defaults to None
:param bool, optional expiryNotification: Certificate Expiry Notification, defaults to False
:param bool, optional ignoreTls: Ignore TLS/SSL error for HTTPS websites, defaults to False
:param int, optional maxredirects: Max. Redirects. Maximum number of redirects to follow. Set to 0 to disable redirects., defaults to 10
:param list, optional accepted_statuscodes: Accepted Status Codes. Select status codes which are considered as a successful response., defaults to None
:param int, optional proxyId: Proxy, defaults to None
:param str, optional method: Method, defaults to "GET"
:param str, optional httpBodyEncoding: Body Encoding, defaults to "json". Allowed values: "json", "xml".
:param str, optional body: Body, defaults to None
:param str, optional headers: Headers, defaults to None
:param AuthMethod, optional authMethod: Method, defaults to :attr:`~.AuthMethod.NONE`
:param str, optional tlsCert: Cert for ``authMethod`` :attr:`~.AuthMethod.MTLS`, defaults to None.
:param str, optional tlsKey: Key for ``authMethod`` :attr:`~.AuthMethod.MTLS`, defaults to None.
:param str, optional tlsCa: Ca for ``authMethod`` :attr:`~.AuthMethod.MTLS`, defaults to None.
:param str, optional basic_auth_user: Username for ``authMethod`` :attr:`~.AuthMethod.HTTP_BASIC` and :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional basic_auth_pass: Password for ``authMethod`` :attr:`~.AuthMethod.HTTP_BASIC` and :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional authDomain: Domain for ``authMethod`` :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional authWorkstation: Workstation for ``authMethod`` :attr:`~.AuthMethod.NTLM`, defaults to None
:param str, optional oauth_auth_method: Authentication Method, defaults to None
:param str, optional oauth_token_url: OAuth Token URL, defaults to None
:param str, optional oauth_client_id: Client ID, defaults to None
:param str, optional oauth_client_secret: Client Secret, defaults to None
:param str, optional oauth_scopes: OAuth Scope, defaults to None
:param int, optional timeout: Request Timeout, defaults to None
:param str, optional keyword: Keyword. Search keyword in plain HTML or JSON response. The search is case-sensitive., defaults to None
:param bool, optional invertKeyword: Invert Keyword. Look for the keyword to be absent rather than present., defaults to False
:param str, optional hostname: Hostname, defaults to None
:param int, optional packetSize: Packet Size, defaults to None
:param int, optional port: Port, ``type`` :attr:`~.MonitorType.DNS` defaults to ``53`` and ``type`` :attr:`~.MonitorType.RADIUS` defaults to ``1812``
:param str, optional dns_resolve_server: Resolver Server, defaults to "1.1.1.1"
:param str, optional dns_resolve_type: Resource Record Type, defaults to "A". Available values are:
- "A"
- "AAAA"
- "CAA"
- "CNAME"
- "MX"
- "NS"
- "PTR"
- "SOA"
- "SRV"
- "TXT"
:param str, optional mqttUsername: MQTT Username, defaults to None
:param str, optional mqttPassword: MQTT Password, defaults to None
:param str, optional mqttTopic: MQTT Topic, defaults to None
:param str, optional mqttSuccessMessage: MQTT Success Message, defaults to None
:param str, optional databaseConnectionString: Connection String, defaults to None
:param str, optional databaseQuery: Query, defaults to None
:param str, optional docker_container: Container Name / ID, defaults to ""
:param int, optional docker_host: Docker Host, defaults to None
:param str, optional radiusUsername: Radius Username, defaults to None
:param str, optional radiusPassword: Radius Password, defaults to None
:param str, optional radiusSecret: Radius Secret. Shared Secret between client and server., defaults to None
:param str, optional radiusCalledStationId: Called Station Id. Identifier of the called device., defaults to None
:param str, optional radiusCallingStationId: Calling Station Id. Identifier of the calling device., defaults to None
:param str, optional game: Game, defaults to None
:param bool, optional gamedigGivenPortOnly: Gamedig: Guess Port. The port used by Valve Server Query Protocol may be different from the client port. Try this if the monitor cannot connect to your server., defaults to False
:param str, optional jsonPath: Json Query, defaults to None
:param str, optional expectedValue: Expected Value, defaults to None
:param str, optional kafkaProducerBrokers: Kafka Broker list, defaults to None
:param str, optional kafkaProducerTopic: Kafka Topic Name, defaults to None
:param str, optional kafkaProducerMessage: Kafka Producer Message, defaults to None
:param bool, optional kafkaProducerSsl: Enable Kafka SSL, defaults to False
:param bool, optional kafkaProducerAllowAutoTopicCreation: Enable Kafka Producer Auto Topic Creation, defaults to False
:param dict, optional kafkaProducerSaslOptions: Kafka SASL Options
- **mechanism** (*str*, *optional*): Mechanism, defaults to "None". Available values are:
- "None"
- "plain"
- "scram-sha-256"
- "scram-sha-512"
- "aws"
- **username** (*str*, *optional*): Username, defaults to None
- **password** (*str*, *optional*): Password, defaults to None
- **authorizationIdentity** (*str*, *optional*): Authorization Identity, defaults to None
- **accessKeyId** (*str*, *optional*): AccessKey Id, defaults to None
- **secretAccessKey** (*str*, *optional*): Secret AccessKey, defaults to None
- **sessionToken** (*str*, *optional*): Session Token, defaults to None
"""
def notification_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} name: Friendly Name
:param NotificationType{", optional" if mode == "edit" else ""} type: Notification Type
:param bool, optional isDefault: Default enabled. This notification will be enabled by default for new monitors. You can still disable the notification separately for each monitor., defaults to False
:param bool, optional applyExisting: Apply on all existing monitors, defaults to False
:param str, optional alertaApiEndpoint: Notification option for ``type`` :attr:`~.NotificationType.ALERTA`.
:param str, optional alertaApiKey: Notification option for ``type`` :attr:`~.NotificationType.ALERTA`.
:param str, optional alertaEnvironment: Notification option for ``type`` :attr:`~.NotificationType.ALERTA`.
:param str, optional alertaAlertState: Notification option for ``type`` :attr:`~.NotificationType.ALERTA`.
:param str, optional alertaRecoverState: Notification option for ``type`` :attr:`~.NotificationType.ALERTA`.
:param str, optional alertNowWebhookURL: Notification option for ``type`` :attr:`~.NotificationType.ALERTNOW`.
:param str, optional phonenumber: Notification option for ``type`` :attr:`~.NotificationType.ALIYUNSMS`.
:param str, optional templateCode: Notification option for ``type`` :attr:`~.NotificationType.ALIYUNSMS`.
:param str, optional signName: Notification option for ``type`` :attr:`~.NotificationType.ALIYUNSMS`.
:param str, optional accessKeyId: Notification option for ``type`` :attr:`~.NotificationType.ALIYUNSMS`.
:param str, optional secretAccessKey: Notification option for ``type`` :attr:`~.NotificationType.ALIYUNSMS`.
:param str, optional appriseURL: Notification option for ``type`` :attr:`~.NotificationType.APPRISE`.
:param str title: Notification option for ``type`` :attr:`~.NotificationType.APPRISE`.
:param str, optional barkEndpoint: Notification option for ``type`` :attr:`~.NotificationType.BARK`.
:param str, optional barkGroup: Notification option for ``type`` :attr:`~.NotificationType.BARK`.
:param str, optional barkSound: Notification option for ``type`` :attr:`~.NotificationType.BARK`.
:param str, optional clicksendsmsLogin: Notification option for ``type`` :attr:`~.NotificationType.CLICKSENDSMS`.
:param str, optional clicksendsmsPassword: Notification option for ``type`` :attr:`~.NotificationType.CLICKSENDSMS`.
:param str, optional clicksendsmsToNumber: Notification option for ``type`` :attr:`~.NotificationType.CLICKSENDSMS`.
:param str clicksendsmsSenderName: Notification option for ``type`` :attr:`~.NotificationType.CLICKSENDSMS`.
:param str, optional webHookUrl: Notification option for ``type`` :attr:`~.NotificationType.DINGDING`.
:param str, optional secretKey: Notification option for ``type`` :attr:`~.NotificationType.DINGDING`.
:param str discordUsername: Notification option for ``type`` :attr:`~.NotificationType.DISCORD`.
:param str, optional discordWebhookUrl: Notification option for ``type`` :attr:`~.NotificationType.DISCORD`.
:param str discordPrefixMessage: Notification option for ``type`` :attr:`~.NotificationType.DISCORD`.
:param str, optional feishuWebHookUrl: Notification option for ``type`` :attr:`~.NotificationType.FEISHU`.
:param str, optional flashdutySeverity: Notification option for ``type`` :attr:`~.NotificationType.FLASHDUTY`.
:param str flashdutyIntegrationKey: Notification option for ``type`` :attr:`~.NotificationType.FLASHDUTY`.
:param str, optional freemobileUser: Notification option for ``type`` :attr:`~.NotificationType.FREEMOBILE`.
:param str, optional freemobilePass: Notification option for ``type`` :attr:`~.NotificationType.FREEMOBILE`.
:param str, optional goAlertBaseURL: Notification option for ``type`` :attr:`~.NotificationType.GOALERT`.
:param str, optional goAlertToken: Notification option for ``type`` :attr:`~.NotificationType.GOALERT`.
:param str, optional googleChatWebhookURL: Notification option for ``type`` :attr:`~.NotificationType.GOOGLECHAT`.
:param str, optional gorushDeviceToken: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param str gorushPlatform: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param str gorushTitle: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param str gorushPriority: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param int gorushRetry: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param str gorushTopic: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param str, optional gorushServerURL: Notification option for ``type`` :attr:`~.NotificationType.GORUSH`.
:param str, optional gotifyserverurl: Notification option for ``type`` :attr:`~.NotificationType.GOTIFY`.
:param str, optional gotifyapplicationToken: Notification option for ``type`` :attr:`~.NotificationType.GOTIFY`.
:param int, optional gotifyPriority: Notification option for ``type`` :attr:`~.NotificationType.GOTIFY`.
:param str notificationService: Notification option for ``type`` :attr:`~.NotificationType.HOMEASSISTANT`.
:param str, optional homeAssistantUrl: Notification option for ``type`` :attr:`~.NotificationType.HOMEASSISTANT`.
:param str, optional longLivedAccessToken: Notification option for ``type`` :attr:`~.NotificationType.HOMEASSISTANT`.
:param str, optional kookGuildID: Notification option for ``type`` :attr:`~.NotificationType.KOOK`.
:param str, optional kookBotToken: Notification option for ``type`` :attr:`~.NotificationType.KOOK`.
:param str, optional lineChannelAccessToken: Notification option for ``type`` :attr:`~.NotificationType.LINE`.
:param str, optional lineUserID: Notification option for ``type`` :attr:`~.NotificationType.LINE`.
:param str, optional lineNotifyAccessToken: Notification option for ``type`` :attr:`~.NotificationType.LINENOTIFY`.
:param str, optional lunaseaTarget: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. Allowed values: "device", "user".
:param str lunaseaUserID: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. User ID.
:param str lunaseaDevice: Notification option for ``type`` :attr:`~.NotificationType.LUNASEA`. Device ID.
:param str, optional internalRoomId: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`.
:param str, optional accessToken: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`.
:param str, optional homeserverUrl: Notification option for ``type`` :attr:`~.NotificationType.MATRIX`.
:param str mattermostusername: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str, optional mattermostWebhookUrl: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str mattermostchannel: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str mattermosticonemo: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str mattermosticonurl: Notification option for ``type`` :attr:`~.NotificationType.MATTERMOST`.
:param str, optional sender: Notification option for ``type`` :attr:`~.NotificationType.NOSTR`.
:param str, optional recipients: Notification option for ``type`` :attr:`~.NotificationType.NOSTR`.
:param str, optional relays: Notification option for ``type`` :attr:`~.NotificationType.NOSTR`.
:param str ntfyAuthenticationMethod: Notification option for ``type`` :attr:`~.NotificationType.NTFY`. Authentication Method.
:param str ntfyusername: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfypassword: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfyaccesstoken: Notification option for ``type`` :attr:`~.NotificationType.NTFY`. Access Token.
:param str, optional ntfytopic: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param int, optional ntfyPriority: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str, optional ntfyserverurl: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str ntfyIcon: Notification option for ``type`` :attr:`~.NotificationType.NTFY`.
:param str octopushVersion: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str, optional octopushAPIKey: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str, optional octopushLogin: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str, optional octopushPhoneNumber: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str octopushSMSType: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str octopushSenderName: Notification option for ``type`` :attr:`~.NotificationType.OCTOPUSH`.
:param str, optional httpAddr: Notification option for ``type`` :attr:`~.NotificationType.ONEBOT`.
:param str, optional accessToken: Notification option for ``type`` :attr:`~.NotificationType.ONEBOT`.
:param str msgType: Notification option for ``type`` :attr:`~.NotificationType.ONEBOT`.
:param str, optional recieverId: Notification option for ``type`` :attr:`~.NotificationType.ONEBOT`.
:param int opsgeniePriority: Notification option for ``type`` :attr:`~.NotificationType.OPSGENIE`. Priority. Available values are numbers between ``1`` and ``5``.
:param str, optional opsgenieRegion: Notification option for ``type`` :attr:`~.NotificationType.OPSGENIE`. Region. Available values are:
- ``us``: US (Default)
- ``eu``: EU
:param str, optional opsgenieApiKey: Notification option for ``type`` :attr:`~.NotificationType.OPSGENIE`. API Key.
:param str pagerdutyAutoResolve: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`.
:param str pagerdutyIntegrationUrl: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`.
:param str pagerdutyPriority: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`.
:param str, optional pagerdutyIntegrationKey: Notification option for ``type`` :attr:`~.NotificationType.PAGERDUTY`.
:param str pagertreeAutoResolve: Notification option for ``type`` :attr:`~.NotificationType.PAGERTREE`.
Available values are:
- ``0``: Do Nothing
- ``resolve``: Auto Resolve
:param str pagertreeIntegrationUrl: Notification option for ``type`` :attr:`~.NotificationType.PAGERTREE`.
:param str pagertreeUrgency: Notification option for ``type`` :attr:`~.NotificationType.PAGERTREE`.
Available values are:
- ``silent``: Silent
- ``low``: Low
- ``medium``: Medium
- ``high``: High
- ``critical``: Critical
:param bool promosmsAllowLongSMS: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`. Allow long SMS.
:param str, optional promosmsLogin: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`.
:param str, optional promosmsPassword: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`.
:param str, optional promosmsPhoneNumber: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`. Phone number (for Polish recipient You can skip area codes).
:param str promosmsSMSType: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`.
Available values are:
- ``0``: SMS FLASH - Message will automatically show on recipient device. Limited only to Polish recipients.
- ``1``: SMS ECO - cheap but slow and often overloaded. Limited only to Polish recipients.
- ``3``: SMS FULL - Premium tier of SMS, You can use your Sender Name (You need to register name first). Reliable for alerts.
- ``4``: SMS SPEED - Highest priority in system. Very quick and reliable but costly (about twice of SMS FULL price).
:param str promosmsSenderName: Notification option for ``type`` :attr:`~.NotificationType.PROMOSMS`.
:param str, optional pushbulletAccessToken: Notification option for ``type`` :attr:`~.NotificationType.PUSHBULLET`.
:param str pushdeerServer: Notification option for ``type`` :attr:`~.NotificationType.PUSHDEER`.
:param str, optional pushdeerKey: Notification option for ``type`` :attr:`~.NotificationType.PUSHDEER`.
:param str, optional pushoveruserkey: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str, optional pushoverapptoken: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str pushoversounds: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str pushoverpriority: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str pushovertitle: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param str pushoverdevice: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`.
:param int pushoverttl: Notification option for ``type`` :attr:`~.NotificationType.PUSHOVER`. Message TTL (Seconds).
:param str, optional pushyAPIKey: Notification option for ``type`` :attr:`~.NotificationType.PUSHY`.
:param str, optional pushyToken: Notification option for ``type`` :attr:`~.NotificationType.PUSHY`.
:param str rocketchannel: Notification option for ``type`` :attr:`~.NotificationType.ROCKET_CHAT`.
:param str rocketusername: Notification option for ``type`` :attr:`~.NotificationType.ROCKET_CHAT`.
:param str rocketiconemo: Notification option for ``type`` :attr:`~.NotificationType.ROCKET_CHAT`.
:param str, optional rocketwebhookURL: Notification option for ``type`` :attr:`~.NotificationType.ROCKET_CHAT`.
:param str, optional serverChanSendKey: Notification option for ``type`` :attr:`~.NotificationType.SERVERCHAN`.
:param str, optional serwersmsUsername: Notification option for ``type`` :attr:`~.NotificationType.SERWERSMS`.
:param str, optional serwersmsPassword: Notification option for ``type`` :attr:`~.NotificationType.SERWERSMS`.
:param str, optional serwersmsPhoneNumber: Notification option for ``type`` :attr:`~.NotificationType.SERWERSMS`.
:param str serwersmsSenderName: Notification option for ``type`` :attr:`~.NotificationType.SERWERSMS`.
:param str, optional signalNumber: Notification option for ``type`` :attr:`~.NotificationType.SIGNAL`.
:param str, optional signalRecipients: Notification option for ``type`` :attr:`~.NotificationType.SIGNAL`.
:param str, optional signalURL: Notification option for ``type`` :attr:`~.NotificationType.SIGNAL`.
:param bool slackchannelnotify: Notification option for ``type`` :attr:`~.NotificationType.SLACK`.
:param str slackchannel: Notification option for ``type`` :attr:`~.NotificationType.SLACK`.
:param str slackusername: Notification option for ``type`` :attr:`~.NotificationType.SLACK`.
:param str slackiconemo: Notification option for ``type`` :attr:`~.NotificationType.SLACK`.
:param str, optional slackwebhookURL: Notification option for ``type`` :attr:`~.NotificationType.SLACK`.
:param str smscTranslit: Notification option for ``type`` :attr:`~.NotificationType.SMSC`.
:param str, optional smscLogin: Notification option for ``type`` :attr:`~.NotificationType.SMSC`.
:param str, optional smscPassword: Notification option for ``type`` :attr:`~.NotificationType.SMSC`.
:param str, optional smscToNumber: Notification option for ``type`` :attr:`~.NotificationType.SMSC`.
:param str smscSenderName: Notification option for ``type`` :attr:`~.NotificationType.SMSC`.
:param bool smseagleEncoding: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. True to send messages in unicode.
:param int smseaglePriority: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Message priority (0-9, default = 0).
:param str smseagleRecipientType: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Recipient type.
Available values are:
- ``smseagle-to``: Phone number(s)
- ``smseagle-group``: Phonebook group name(s)
- ``smseagle-contact``: Phonebook contact name(s)
:param str, optional smseagleToken: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. API Access token.
:param str, optional smseagleRecipient: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Recipient(s) (multiple must be separated with comma).
:param str, optional smseagleUrl: Notification option for ``type`` :attr:`~.NotificationType.SMSEAGLE`. Your SMSEagle device URL.
:param str smsmanagerApiKey: Notification option for ``type`` :attr:`~.NotificationType.SMSMANAGER`.
:param str numbers: Notification option for ``type`` :attr:`~.NotificationType.SMSMANAGER`.
:param str messageType: Notification option for ``type`` :attr:`~.NotificationType.SMSMANAGER`.
:param str, optional smtpHost: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param int, optional smtpPort: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpSecure: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param bool smtpIgnoreTLSError: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpDkimDomain: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpDkimKeySelector: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpDkimPrivateKey: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpDkimHashAlgo: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpDkimheaderFieldNames: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpDkimskipFields: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpUsername: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpPassword: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str customSubject: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str, optional smtpFrom: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpCC: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpBCC: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str smtpTo: Notification option for ``type`` :attr:`~.NotificationType.SMTP`.
:param str splunkAutoResolve: Notification option for ``type`` :attr:`~.NotificationType.SPLUNK`. Auto resolve or acknowledged.
Available values are:
- ``0``: do nothing
- ``ACKNOWLEDGEMENT``: auto acknowledged
- ``RECOVERY``: auto resolve
:param str splunkSeverity: Notification option for ``type`` :attr:`~.NotificationType.SPLUNK`. Severity.
Available values are:
- ``INFO``
- ``WARNING``
- ``CRITICAL``
:param str, optional splunkRestURL: Notification option for ``type`` :attr:`~.NotificationType.SPLUNK`. Splunk Rest URL.
:param str, optional squadcastWebhookURL: Notification option for ``type`` :attr:`~.NotificationType.SQUADCAST`.
:param str, optional stackfieldwebhookURL: Notification option for ``type`` :attr:`~.NotificationType.STACKFIELD`.
:param str, optional webhookUrl: Notification option for ``type`` :attr:`~.NotificationType.TEAMS`.
:param str, optional pushAPIKey: Notification option for ``type`` :attr:`~.NotificationType.PUSHBYTECHULUS`.
:param str, optional telegramChatID: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`.
:param bool telegramSendSilently: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`.
:param bool telegramProtectContent: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`.
:param str telegramMessageThreadID: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`.
:param str, optional telegramBotToken: Notification option for ``type`` :attr:`~.NotificationType.TELEGRAM`.
:param str, optional twilioAccountSID: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. Account SID.
:param str twilioApiKey: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`.
:param str, optional twilioAuthToken: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. Auth Token.
:param str, optional twilioToNumber: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. To Number.
:param str, optional twilioFromNumber: Notification option for ``type`` :attr:`~.NotificationType.TWILIO`. From Number.
:param str, optional webhookContentType: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`.
:param str webhookCustomBody: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`.
:param str webhookAdditionalHeaders: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`.
:param str, optional webhookURL: Notification option for ``type`` :attr:`~.NotificationType.WEBHOOK`.
:param str, optional weComBotKey: Notification option for ``type`` :attr:`~.NotificationType.WECOM`.
:param str, optional webhookUrl: Notification option for ``type`` :attr:`~.NotificationType.ZOHOCLIQ`.
"""
def proxy_docstring(mode) -> str:
return f"""
:param ProxyProtocol{", optional" if mode == "edit" else ""} protocol: Proxy Protocol
:param str{", optional" if mode == "edit" else ""} host: Proxy Server
:param str{", optional" if mode == "edit" else ""} port: Port
:param bool, optional auth: Proxy server has authentication, defaults to False
:param str, optional username: User, defaults to None
:param str, optional password: Password, defaults to None
:param bool, optional active: Enabled. This proxy will not effect on monitor requests until it is activated. You can control temporarily disable the proxy from all monitors by activation status., defaults to True
:param bool, optional default: Set As Default. This proxy will be enabled by default for new monitors. You can still disable the proxy separately for each monitor., , defaults to False
:param bool, optional applyExisting: Apply on all existing monitors, defaults to False
"""
def docker_host_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} name: Friendly Name
:param DockerType{", optional" if mode == "edit" else ""} dockerType: Connection Type
:param str, optional dockerDaemon: Docker Daemon, defaults to None
"""
def maintenance_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} title: Title
:param MaintenanceStrategy{", optional" if mode == "edit" else ""} strategy: Strategy
:param bool, optional active: True if maintenance is active, defaults to ``True``
:param str, optional description: Description, defaults to ``""``
:param list, optional dateRange: DateTime Range, defaults to ``["<current date>"]``
:param int, optional intervalDay: Interval (Run once every day), defaults to ``1``
:param list, optional weekdays: List that contains the days of the week on which the maintenance is enabled (Sun = ``0``, Mon = ``1``, ..., Sat = ``6``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_WEEKDAY`., defaults to ``[]``.
:param list, optional daysOfMonth: List that contains the days of the month on which the maintenance is enabled (Day 1 = ``1``, Day 2 = ``2``, ..., Day 31 = ``31``) and the last day of the month (``"lastDay1"``). Required for ``strategy`` :attr:`~.MaintenanceStrategy.RECURRING_DAY_OF_MONTH`., defaults to ``[]``.
:param list, optional timeRange: Maintenance Time Window of a Day, defaults to ``[{{"hours": 2, "minutes": 0}}, {{"hours": 3, "minutes": 0}}]``.
:param str, optional cron: Cron Schedule. Required for ``strategy`` :attr:`~.MaintenanceStrategy.CRON`., defaults to ``"30 3 * * *"``
:param int, optional durationMinutes: Duration (Minutes). Required for ``strategy`` :attr:`~.MaintenanceStrategy.CRON`., defaults to ``60``
:param str, optional timezone: Timezone, defaults to ``None`` (Server Timezone)
"""
def tag_docstring(mode) -> str:
return f"""
:param str{", optional" if mode == "edit" else ""} name: Tag name
:param str{", optional" if mode == "edit" else ""} color: Tag color
"""

View file

@ -16,7 +16,3 @@ class Event(str, Enum):
INFO = "info"
CERT_INFO = "certInfo"
DOCKER_HOST_LIST = "dockerHostList"
AUTO_LOGIN = "autoLogin"
INIT_SERVER_TIMEZONE = "initServerTimezone"
MAINTENANCE_LIST = "maintenanceList"
API_KEY_LIST = "apiKeyList"

View file

@ -1,10 +1,2 @@
class UptimeKumaException(Exception):
"""
There was an exception that occurred while communicating with Uptime Kuma.
"""
class Timeout(UptimeKumaException):
"""
A timeout has occurred while communicating with Uptime Kuma.
"""
pass

View file

@ -2,22 +2,9 @@ from enum import Enum
class IncidentStyle(str, Enum):
"""Enumerate incident styles."""
INFO = "info"
"""Info"""
WARNING = "warning"
"""Warning"""
DANGER = "danger"
"""Danger"""
PRIMARY = "primary"
"""Primary"""
LIGHT = "light"
"""Light"""
DARK = "dark"
"""Dark"""

View file

@ -1,23 +0,0 @@
from enum import Enum
class MaintenanceStrategy(str, Enum):
"""Enumerate maintenance strategies."""
MANUAL = "manual"
"""Active/Inactive Manually"""
SINGLE = "single"
"""Single Maintenance Window"""
RECURRING_INTERVAL = "recurring-interval"
"""Recurring - Interval"""
RECURRING_WEEKDAY = "recurring-weekday"
"""Recurring - Day of Week"""
RECURRING_DAY_OF_MONTH = "recurring-day-of-month"
"""Recurring - Day of Month"""
CRON = "cron"
"""Cron Expression"""

View file

@ -1,17 +0,0 @@
from enum import Enum
class MonitorStatus(int, Enum):
"""Enumerate monitor statuses."""
DOWN = 0
"""DOWN"""
UP = 1
"""UP"""
PENDING = 2
"""PENDING"""
MAINTENANCE = 3
"""MAINTENANCE"""

View file

@ -2,70 +2,15 @@ from enum import Enum
class MonitorType(str, Enum):
"""Enumerate monitor types."""
GROUP = "group"
"""Group"""
HTTP = "http"
"""HTTP(s)"""
PORT = "port"
"""TCP Port"""
PING = "ping"
"""Ping"""
KEYWORD = "keyword"
"""HTTP(s) - Keyword"""
JSON_QUERY = "json-query"
"""HTTP(s) - Json Query"""
GRPC_KEYWORD = "grpc-keyword"
"""gRPC(s) - Keyword"""
DNS = "dns"
"""DNS"""
DOCKER = "docker"
"""Docker Container"""
REAL_BROWSER = "real-browser"
"""HTTP(s) - Browser Engine (Chrome/Chromium)"""
PUSH = "push"
"""Push"""
STEAM = "steam"
"""Steam Game Server"""
GAMEDIG = "gamedig"
"""GameDig"""
MQTT = "mqtt"
"""MQTT"""
KAFKA_PRODUCER = "kafka-producer"
"""Kafka Producer"""
SQLSERVER = "sqlserver"
"""Microsoft SQL Server"""
POSTGRES = "postgres"
"""PostgreSQL"""
MYSQL = "mysql"
"""MySQL/MariaDB"""
MONGODB = "mongodb"
"""MongoDB"""
RADIUS = "radius"
"""Radius"""
REDIS = "redis"
"""Redis"""
TAILSCALE_PING = "tailscale-ping"
"""Tailscale Ping"""

View file

@ -2,481 +2,267 @@ from enum import Enum
class NotificationType(str, Enum):
"""Enumerate notification types."""
ALERTA = "alerta"
"""Alerta"""
ALERTNOW = "AlertNow"
"""AlertNow"""
ALIYUNSMS = "AliyunSMS"
"""AliyunSMS"""
APPRISE = "apprise"
"""Apprise (Support 50+ Notification services)"""
BARK = "Bark"
"""Bark"""
CLICKSENDSMS = "clicksendsms"
"""ClickSend SMS"""
DINGDING = "DingDing"
"""DingDing"""
DISCORD = "discord"
"""Discord"""
FEISHU = "Feishu"
"""Feishu"""
FLASHDUTY = "FlashDuty"
"""FlashDuty"""
FREEMOBILE = "FreeMobile"
"""FreeMobile (mobile.free.fr)"""
GOALERT = "GoAlert"
"""GoAlert"""
GOOGLECHAT = "GoogleChat"
"""Google Chat (Google Workspace)"""
GORUSH = "gorush"
"""Gorush"""
GOTIFY = "gotify"
"""Gotify"""
HOMEASSISTANT = "HomeAssistant"
"""Home Assistant"""
KOOK = "Kook"
"""Kook"""
LINE = "line"
"""LINE Messenger"""
LINENOTIFY = "LineNotify"
"""LINE Notify"""
LUNASEA = "lunasea"
"""LunaSea"""
MATRIX = "matrix"
"""Matrix"""
MATTERMOST = "mattermost"
"""Mattermost"""
NOSTR = "nostr"
"""Nostr"""
NTFY = "ntfy"
"""Ntfy"""
OCTOPUSH = "octopush"
"""Octopush"""
ONEBOT = "OneBot"
"""OneBot"""
OPSGENIE = "Opsgenie"
"""Opsgenie"""
PAGERDUTY = "PagerDuty"
"""PagerDuty"""
PAGERTREE = "PagerTree"
"""PagerTree"""
PROMOSMS = "promosms"
"""PromoSMS"""
PUSHBULLET = "pushbullet"
"""Pushbullet"""
PUSHDEER = "PushDeer"
"""PushDeer"""
PUSHOVER = "pushover"
"""Pushover"""
PUSHY = "pushy"
"""Pushy"""
ROCKET_CHAT = "rocket.chat"
"""Rocket.Chat"""
SERVERCHAN = "ServerChan"
"""ServerChan"""
SERWERSMS = "serwersms"
"""SerwerSMS.pl"""
SIGNAL = "signal"
"""Signal"""
SLACK = "slack"
"""Slack"""
SMSC = "smsc"
"""SMSC"""
SMSEAGLE = "SMSEagle"
"""SMSEagle"""
SMSMANAGER = "SMSManager"
"""SmsManager (smsmanager.cz)"""
SMTP = "smtp"
"""Email (SMTP)"""
SPLUNK = "Splunk"
"""Splunk"""
SQUADCAST = "squadcast"
"""SquadCast"""
STACKFIELD = "stackfield"
"""Stackfield"""
TEAMS = "teams"
"""Microsoft Teams"""
PUSHBYTECHULUS = "PushByTechulus"
"""Push by Techulus"""
TELEGRAM = "telegram"
"""Telegram"""
TWILIO = "twilio"
"""Twilio"""
WEBHOOK = "webhook"
"""Webhook"""
WECOM = "WeCom"
"""WeCom"""
ZOHOCLIQ = "ZohoCliq"
"""ZohoCliq"""
ALERTNOW = "AlertNow"
BARK = "Bark"
HOMEASSISTANT = "HomeAssistant"
LINENOTIFY = "LineNotify"
notification_provider_options = {
NotificationType.ALERTA: dict(
alertaApiEndpoint=dict(type="str", required=True),
alertaApiKey=dict(type="str", required=True),
alertaEnvironment=dict(type="str", required=True),
alertaAlertState=dict(type="str", required=True),
alertaRecoverState=dict(type="str", required=True),
),
NotificationType.ALERTNOW: dict(
alertNowWebhookURL=dict(type="str", required=True),
),
NotificationType.ALIYUNSMS: dict(
phonenumber=dict(type="str", required=True),
templateCode=dict(type="str", required=True),
signName=dict(type="str", required=True),
accessKeyId=dict(type="str", required=True),
secretAccessKey=dict(type="str", required=True),
),
NotificationType.APPRISE: dict(
appriseURL=dict(type="str", required=True),
title=dict(type="str", required=False),
),
NotificationType.BARK: dict(
barkEndpoint=dict(type="str", required=True),
barkGroup=dict(type="str", required=True),
barkSound=dict(type="str", required=True),
),
NotificationType.CLICKSENDSMS: dict(
clicksendsmsLogin=dict(type="str", required=True),
clicksendsmsPassword=dict(type="str", required=True),
clicksendsmsToNumber=dict(type="str", required=True),
clicksendsmsSenderName=dict(type="str", required=False),
),
NotificationType.DINGDING: dict(
webHookUrl=dict(type="str", required=True),
secretKey=dict(type="str", required=True),
),
NotificationType.DISCORD: dict(
discordUsername=dict(type="str", required=False),
discordWebhookUrl=dict(type="str", required=True),
discordPrefixMessage=dict(type="str", required=False),
),
NotificationType.FEISHU: dict(
feishuWebHookUrl=dict(type="str", required=True),
),
NotificationType.FLASHDUTY: dict(
flashdutySeverity=dict(type="str", required=True),
flashdutyIntegrationKey=dict(type="str", required=False),
),
NotificationType.FREEMOBILE: dict(
freemobileUser=dict(type="str", required=True),
freemobilePass=dict(type="str", required=True),
),
NotificationType.GOALERT: dict(
goAlertBaseURL=dict(type="str", required=True),
goAlertToken=dict(type="str", required=True),
),
NotificationType.GOOGLECHAT: dict(
googleChatWebhookURL=dict(type="str", required=True),
),
NotificationType.GORUSH: dict(
gorushDeviceToken=dict(type="str", required=True),
gorushPlatform=dict(type="str", required=False),
gorushTitle=dict(type="str", required=False),
gorushPriority=dict(type="str", required=False),
gorushRetry=dict(type="int", required=False),
gorushTopic=dict(type="str", required=False),
gorushServerURL=dict(type="str", required=True),
),
NotificationType.GOTIFY: dict(
gotifyserverurl=dict(type="str", required=True),
gotifyapplicationToken=dict(type="str", required=True),
gotifyPriority=dict(type="int", required=True),
),
NotificationType.HOMEASSISTANT: dict(
notificationService=dict(type="str", required=False),
homeAssistantUrl=dict(type="str", required=True),
longLivedAccessToken=dict(type="str", required=True),
),
NotificationType.KOOK: dict(
kookGuildID=dict(type="str", required=True),
kookBotToken=dict(type="str", required=True),
),
NotificationType.LINE: dict(
lineChannelAccessToken=dict(type="str", required=True),
lineUserID=dict(type="str", required=True),
),
NotificationType.LINENOTIFY: dict(
lineNotifyAccessToken=dict(type="str", required=True),
),
NotificationType.LUNASEA: dict(
lunaseaTarget=dict(type="str", required=True),
lunaseaUserID=dict(type="str", required=False),
lunaseaDevice=dict(type="str", required=False),
),
NotificationType.MATRIX: dict(
internalRoomId=dict(type="str", required=True),
accessToken=dict(type="str", required=True),
homeserverUrl=dict(type="str", required=True),
),
NotificationType.MATTERMOST: dict(
mattermostusername=dict(type="str", required=False),
mattermostWebhookUrl=dict(type="str", required=True),
mattermostchannel=dict(type="str", required=False),
mattermosticonemo=dict(type="str", required=False),
mattermosticonurl=dict(type="str", required=False),
),
NotificationType.NOSTR: dict(
sender=dict(type="str", required=True),
recipients=dict(type="str", required=True),
relays=dict(type="str", required=True),
),
NotificationType.NTFY: dict(
ntfyAuthenticationMethod=dict(type="str", required=False),
ntfyusername=dict(type="str", required=False),
ntfypassword=dict(type="str", required=False),
ntfyaccesstoken=dict(type="str", required=False),
ntfytopic=dict(type="str", required=True),
ntfyPriority=dict(type="int", required=True),
ntfyserverurl=dict(type="str", required=True),
ntfyIcon=dict(type="str", required=False),
),
NotificationType.OCTOPUSH: dict(
octopushVersion=dict(type="str", required=False),
octopushAPIKey=dict(type="str", required=True),
octopushLogin=dict(type="str", required=True),
octopushPhoneNumber=dict(type="str", required=True),
octopushSMSType=dict(type="str", required=False),
octopushSenderName=dict(type="str", required=False),
),
NotificationType.ONEBOT: dict(
httpAddr=dict(type="str", required=True),
accessToken=dict(type="str", required=True),
msgType=dict(type="str", required=False),
recieverId=dict(type="str", required=True),
),
NotificationType.OPSGENIE: dict(
opsgeniePriority=dict(type="int", required=False),
opsgenieRegion=dict(type="str", required=True),
opsgenieApiKey=dict(type="str", required=True),
),
NotificationType.PAGERDUTY: dict(
pagerdutyAutoResolve=dict(type="str", required=False),
pagerdutyIntegrationUrl=dict(type="str", required=False),
pagerdutyPriority=dict(type="str", required=False),
pagerdutyIntegrationKey=dict(type="str", required=True),
),
NotificationType.PAGERTREE: dict(
pagertreeAutoResolve=dict(type="str", required=False),
pagertreeIntegrationUrl=dict(type="str", required=False),
pagertreeUrgency=dict(type="str", required=False),
),
NotificationType.PROMOSMS: dict(
promosmsAllowLongSMS=dict(type="bool", required=False),
promosmsLogin=dict(type="str", required=True),
promosmsPassword=dict(type="str", required=True),
promosmsPhoneNumber=dict(type="str", required=True),
promosmsSMSType=dict(type="str", required=False),
promosmsSenderName=dict(type="str", required=False),
),
NotificationType.PUSHBULLET: dict(
pushbulletAccessToken=dict(type="str", required=True),
),
NotificationType.PUSHDEER: dict(
pushdeerServer=dict(type="str", required=False),
pushdeerKey=dict(type="str", required=True),
),
NotificationType.PUSHOVER: dict(
pushoveruserkey=dict(type="str", required=True),
pushoverapptoken=dict(type="str", required=True),
pushoversounds=dict(type="str", required=False),
pushoverpriority=dict(type="str", required=False),
pushovertitle=dict(type="str", required=False),
pushoverdevice=dict(type="str", required=False),
pushoverttl=dict(type="int", required=False),
),
NotificationType.PUSHY: dict(
pushyAPIKey=dict(type="str", required=True),
pushyToken=dict(type="str", required=True),
),
NotificationType.ROCKET_CHAT: dict(
rocketchannel=dict(type="str", required=False),
rocketusername=dict(type="str", required=False),
rocketiconemo=dict(type="str", required=False),
rocketwebhookURL=dict(type="str", required=True),
),
NotificationType.SERVERCHAN: dict(
serverChanSendKey=dict(type="str", required=True),
),
NotificationType.SERWERSMS: dict(
serwersmsUsername=dict(type="str", required=True),
serwersmsPassword=dict(type="str", required=True),
serwersmsPhoneNumber=dict(type="str", required=True),
serwersmsSenderName=dict(type="str", required=False),
),
NotificationType.SIGNAL: dict(
signalNumber=dict(type="str", required=True),
signalRecipients=dict(type="str", required=True),
signalURL=dict(type="str", required=True),
),
NotificationType.SLACK: dict(
slackchannelnotify=dict(type="bool", required=False),
slackchannel=dict(type="str", required=False),
slackusername=dict(type="str", required=False),
slackiconemo=dict(type="str", required=False),
slackwebhookURL=dict(type="str", required=True),
),
NotificationType.SMSC: dict(
smscTranslit=dict(type="str", required=False),
smscLogin=dict(type="str", required=True),
smscPassword=dict(type="str", required=True),
smscToNumber=dict(type="str", required=True),
smscSenderName=dict(type="str", required=False),
),
NotificationType.SMSEAGLE: dict(
smseagleEncoding=dict(type="bool", required=False),
smseaglePriority=dict(type="int", required=False),
smseagleRecipientType=dict(type="str", required=False),
smseagleToken=dict(type="str", required=True),
smseagleRecipient=dict(type="str", required=True),
smseagleUrl=dict(type="str", required=True),
),
NotificationType.SMSMANAGER: dict(
smsmanagerApiKey=dict(type="str", required=False),
numbers=dict(type="str", required=False),
messageType=dict(type="str", required=False),
),
NotificationType.SMTP: dict(
smtpHost=dict(type="str", required=True),
smtpPort=dict(type="int", required=True),
smtpSecure=dict(type="str", required=False),
smtpIgnoreTLSError=dict(type="bool", required=False),
smtpDkimDomain=dict(type="str", required=False),
smtpDkimKeySelector=dict(type="str", required=False),
smtpDkimPrivateKey=dict(type="str", required=False),
smtpDkimHashAlgo=dict(type="str", required=False),
smtpDkimheaderFieldNames=dict(type="str", required=False),
smtpDkimskipFields=dict(type="str", required=False),
smtpUsername=dict(type="str", required=False),
smtpPassword=dict(type="str", required=False),
customSubject=dict(type="str", required=False),
smtpFrom=dict(type="str", required=True),
smtpCC=dict(type="str", required=False),
smtpBCC=dict(type="str", required=False),
smtpTo=dict(type="str", required=False),
),
NotificationType.SPLUNK: dict(
splunkAutoResolve=dict(type="str", required=False),
splunkSeverity=dict(type="str", required=False),
splunkRestURL=dict(type="str", required=True),
),
NotificationType.SQUADCAST: dict(
squadcastWebhookURL=dict(type="str", required=True),
),
NotificationType.STACKFIELD: dict(
stackfieldwebhookURL=dict(type="str", required=True),
),
NotificationType.TEAMS: dict(
webhookUrl=dict(type="str", required=True),
),
NotificationType.PUSHBYTECHULUS: dict(
pushAPIKey=dict(type="str", required=True),
),
NotificationType.TELEGRAM: dict(
telegramChatID=dict(type="str", required=True),
telegramSendSilently=dict(type="bool", required=False),
telegramProtectContent=dict(type="bool", required=False),
telegramMessageThreadID=dict(type="str", required=False),
telegramBotToken=dict(type="str", required=True),
),
NotificationType.TWILIO: dict(
twilioAccountSID=dict(type="str", required=True),
twilioApiKey=dict(type="str", required=False),
twilioAuthToken=dict(type="str", required=True),
twilioToNumber=dict(type="str", required=True),
twilioFromNumber=dict(type="str", required=True),
),
NotificationType.WEBHOOK: dict(
webhookContentType=dict(type="str", required=True),
webhookCustomBody=dict(type="str", required=False),
webhookAdditionalHeaders=dict(type="str", required=False),
webhookURL=dict(type="str", required=True),
),
NotificationType.WECOM: dict(
weComBotKey=dict(type="str", required=True),
),
NotificationType.ZOHOCLIQ: dict(
webhookUrl=dict(type="str", required=True),
),
NotificationType.ALERTA: [
"alertaApiEndpoint",
"alertaApiKey",
"alertaEnvironment",
"alertaAlertState",
"alertaRecoverState",
],
NotificationType.ALIYUNSMS: [
"phonenumber",
"templateCode",
"signName",
"accessKeyId",
"secretAccessKey",
],
NotificationType.APPRISE: [
"appriseURL",
"title",
],
NotificationType.CLICKSENDSMS: [
"clicksendsmsLogin",
"clicksendsmsPassword",
"clicksendsmsToNumber",
"clicksendsmsSenderName",
],
NotificationType.DINGDING: [
"webHookUrl",
"secretKey",
],
NotificationType.DISCORD: [
"discordUsername",
"discordWebhookUrl",
"discordPrefixMessage",
],
NotificationType.FEISHU: [
"feishuWebHookUrl",
],
NotificationType.GOOGLECHAT: [
"googleChatWebhookURL",
],
NotificationType.GORUSH: [
"gorushDeviceToken",
"gorushPlatform",
"gorushTitle",
"gorushPriority",
"gorushRetry",
"gorushTopic",
"gorushServerURL",
],
NotificationType.GOTIFY: [
"gotifyserverurl",
"gotifyapplicationToken",
"gotifyPriority",
],
NotificationType.LINE: [
"lineChannelAccessToken",
"lineUserID",
],
NotificationType.LUNASEA: [
"lunaseaDevice",
],
NotificationType.MATRIX: [
"internalRoomId",
"accessToken",
"homeserverUrl",
],
NotificationType.MATTERMOST: [
"mattermostusername",
"mattermostWebhookUrl",
"mattermostchannel",
"mattermosticonemo",
"mattermosticonurl",
],
NotificationType.NTFY: [
"ntfyserverurl",
"ntfytopic",
"ntfyPriority",
],
NotificationType.OCTOPUSH: [
"octopushVersion",
"octopushAPIKey",
"octopushLogin",
"octopushPhoneNumber",
"octopushSMSType",
"octopushSenderName",
"octopushDMLogin",
"octopushDMAPIKey",
"octopushDMPhoneNumber",
"octopushDMSenderName",
"octopushDMSMSType",
],
NotificationType.ONEBOT: [
"httpAddr",
"accessToken",
"msgType",
"recieverId",
],
NotificationType.PAGERDUTY: [
"pagerdutyAutoResolve",
"pagerdutyIntegrationUrl",
"pagerdutyPriority",
"pagerdutyIntegrationKey",
],
NotificationType.PROMOSMS: [
"promosmsLogin",
"promosmsPassword",
"promosmsPhoneNumber",
"promosmsSMSType",
"promosmsSenderName",
],
NotificationType.PUSHBULLET: [
"pushbulletAccessToken",
],
NotificationType.PUSHDEER: [
"pushdeerKey",
],
NotificationType.PUSHOVER: [
"pushoveruserkey",
"pushoverapptoken",
"pushoversounds",
"pushoverpriority",
"pushovertitle",
"pushoverdevice",
],
NotificationType.PUSHY: [
"pushyAPIKey",
"pushyToken",
],
NotificationType.ROCKET_CHAT: [
"rocketchannel",
"rocketusername",
"rocketiconemo",
"rocketwebhookURL",
"rocketbutton",
],
NotificationType.SERWERSMS: [
"serwersmsUsername",
"serwersmsPassword",
"serwersmsPhoneNumber",
"serwersmsSenderName",
],
NotificationType.SIGNAL: [
"signalNumber",
"signalRecipients",
"signalURL",
],
NotificationType.SLACK: [
"slackbutton",
"slackchannel",
"slackusername",
"slackiconemo",
"slackwebhookURL",
"slackbutton",
],
NotificationType.SMTP: [
"smtpHost",
"smtpPort",
"smtpSecure",
"smtpIgnoreTLSError",
"smtpDkimDomain",
"smtpDkimKeySelector",
"smtpDkimPrivateKey",
"smtpDkimHashAlgo",
"smtpDkimheaderFieldNames",
"smtpDkimskipFields",
"smtpUsername",
"smtpPassword",
"customSubject",
"smtpFrom",
"smtpCC",
"smtpBCC",
"smtpTo",
],
NotificationType.STACKFIELD: [
"stackfieldwebhookURL",
],
NotificationType.TEAMS: [
"webhookUrl",
],
NotificationType.PUSHBYTECHULUS: [
"pushAPIKey",
],
NotificationType.TELEGRAM: [
"telegramBotToken",
"telegramChatID",
],
NotificationType.WEBHOOK: [
"webhookContentType",
"webhookURL",
],
NotificationType.WECOM: [
"weComBotKey",
],
NotificationType.ALERTNOW: [
"alertNowWebhookURL",
],
NotificationType.BARK: [
"barkEndpoint",
"barkGroup",
"barkSound",
],
NotificationType.HOMEASSISTANT: [
"homeAssistantUrl",
"longLivedAccessToken",
],
NotificationType.LINENOTIFY: [
"lineNotifyAccessToken",
],
}
notification_provider_conditions = dict(
gotifyPriority=dict(
min=0,
max=10,
),
ntfyPriority=dict(
min=1,
max=5,
),
opsgeniePriority=dict(
min=1,
max=5,
),
pushoverttl=dict(
min=0,
),
smseaglePriority=dict(
min=0,
max=9,
),
smtpPort=dict(
min=0,
max=65535,
),
)
notification_provider_conditions = {
"gotifyPriority": {
"min": 0,
"max": 10,
},
"ntfyPriority": {
"min": 1,
"max": 5,
},
"smtpPort": {
"min": 0,
"max": 65535,
},
}

View file

@ -2,22 +2,8 @@ from enum import Enum
class ProxyProtocol(str, Enum):
"""Enumerate proxy protocols."""
HTTPS = "https"
"""HTTPS"""
HTTP = "http"
"""HTTP"""
SOCKS = "socks"
"""SOCKS"""
SOCKS5 = "socks5"
"""SOCKS v5"""
SOCKS5H = "socks5h"
"""SOCKS v5 (+DNS)"""
SOCKS4 = "socks4"
"""SOCKS v4"""