Compare commits

...

87 commits

Author SHA1 Message Date
lucasheld
22ca1813ff bump version to 1.2.1 2023-09-26 22:32:04 +02:00
lucasheld
370b7e3e18 fix: drop first info event without a version
closes #55
2023-09-26 22:29:33 +02:00
Lucas Held
75a0b57eea
Update README.md 2023-09-23 13:59:27 +02:00
lucasheld
ba047114c9 bump version to 1.2.0 2023-08-29 18:41:23 +02:00
lucasheld
7902213ddb feat: add support for uptime kuma 1.23.0 and 1.23.1 2023-08-29 18:37:27 +02:00
lucasheld
0d49e97fe5 fix: validate accepted status codes types
closes #42
2023-08-12 18:20:59 +02:00
lucasheld
18107848f8 fix: convert sendUrl from bool to int 2023-08-12 16:40:25 +02:00
lucasheld
3543f09a5f fix: rstip url globally 2023-08-12 16:39:44 +02:00
lucasheld
2611b344f1 fix: remove name from maintenance monitors and status pages 2023-08-12 16:37:28 +02:00
lucasheld
be97a4fd8f bump version to 1.1.0 2023-07-07 22:28:25 +02:00
lucasheld
06f1173569 feat: add support for uptime kuma 1.22.0 and 1.22.1 2023-07-07 22:28:20 +02:00
lucasheld
934ab15457 bump version to 1.0.1 2023-05-25 23:53:19 +02:00
lucasheld
ce6f25d604 fix: ValueError if monitor authMethod is None 2023-05-25 23:51:09 +02:00
Lucas Held
144e426ed9
Update CHANGELOG.md 2023-05-25 21:39:12 +02:00
lucasheld
a5a00aec9f bump version to 1.0.0 2023-05-25 21:35:10 +02:00
lucasheld
84d4009d6a feat: replace raw return values with enum values
BREAKING CHANGE:
Types of return values changed to enum values:
  - monitor: `type` (str -> MonitorType), `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)
2023-05-25 21:26:54 +02:00
lucasheld
33b8ffc476 refactor: reformat imports 2023-05-25 18:37:49 +02:00
lucasheld
712cd85aae test: fix notification test 2023-05-25 18:35:56 +02:00
lucasheld
8a0ad53753 fix: check only for required notification arguments 2023-05-20 22:06:05 +02:00
lucasheld
ce1cc12740 fix: adjust get_monitor_status method to previous changes 2023-05-20 20:35:04 +02:00
lucasheld
762dd4a657 fix: process the HEARTBEAT event correctly
BREAKING CHANGE: Removed `get_heartbeat` method. This method was never intended to retrieve information. Use `get_heartbeats` or `get_important_heartbeats` instead.
2023-05-20 20:31:39 +02:00
lucasheld
24b6d367de docs: update notification docstring 2023-05-20 15:42:40 +02:00
lucasheld
27d8e63f7b test: decrease timeout and wait_events 2023-05-20 14:23:01 +02:00
lucasheld
a34b45df8e chore: create dev-requirements.txt 2023-05-20 14:20:38 +02:00
Lucas Held
1359576413
feat: raise exception when deleting an element that does not exist (#37) 2023-05-20 14:09:09 +02:00
Lucas Held
e7693e6081
feat: check for required notification arguments (#36) 2023-05-20 13:10:12 +02:00
lucasheld
f0c5f2ba9d feat: drop support for Uptime Kuma versions < 1.21.3
BREAKING CHANGE: Uptime Kuma versions < 1.21.3 are not supported in uptime-kuma-api 1.x.x
2023-05-20 12:43:57 +02:00
Lucas Held
77630e96b7
fix: memory leak (#29)
* fix: dictionary changed size during iteration during deepcopy

* fix: memory leak in event data

BREAKING CHANGE: changed return values of get_heartbeats, get_important_heartbeats, avg_ping, uptime, get_heartbeat, cert_info

* fix: int to bool conversion

* remove todos

* update examples and adjust types
2023-05-19 14:07:34 +02:00
Lucas Held
6c4342a39d
Merge pull request #32 from lucasheld/feature/monitor-status
fix monitor status values and get current monitor status
2023-05-19 13:51:23 +02:00
Lucas Held
9728cfdb34
feat: implement timeouts for all methods (#34)
BREAKING CHANGE: 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.
2023-05-19 13:50:39 +02:00
lucasheld
8e841cd324 feat: add support for uptime kuma 1.21.3
BREAKING CHANGE: maintenance parameter `timezone` renamed to `timezoneOption`
2023-05-19 13:49:36 +02:00
lucasheld
d2cfc6652d refactor: use square brackets for tuple type hint 2023-05-06 13:36:26 +02:00
Lucas Held
a576ed9f3a
Update README.md 2023-05-02 20:43:52 +02:00
lucasheld
a9f2b6d894 feat: implement get_monitor_status helper method 2023-05-02 20:36:49 +02:00
lucasheld
b87eed2597 fix: adjust monitor status type to allow all used values
BREAKING CHANGE: monitor `status` type changed from `bool` to `MonitorStatus`
2023-05-02 20:34:26 +02:00
lucasheld
7ef61f8ce1 feat: drop python 3.6 support
BREAKING CHANGE: Python 3.7+ required
2023-05-02 17:57:32 +02:00
lucasheld
50ff8f1219 feat: drop python 3.6 support
BREAKING CHANGE: Python 3.7+ required
2023-05-01 19:06:48 +02:00
Vinalti
19bd8aecfa
Clean up code and implement best practices (#27)
* clean up code and implement best practices: 
 - `type(a) == list` replace with `isinstance(a, list)`
 - `adict['key']` replaced with `adict.get('key')`
 - annotation `-> list` replace by more accurate `-> list[dict]`

* improve compatibility with previous python versions with Typing

* little fix
2023-05-01 18:57:55 +02:00
lucasheld
e42f6461c0 feat: implement context manager for UptimeKumaApi class 2023-05-01 18:41:45 +02:00
lucasheld
391e5a3077 feat: add param wait_events 2023-04-30 16:48:36 +02:00
Lucas Held
31fe4466a2
Merge pull request #26 from Vinalti/add-ssl_verify-param
Add ssl_verify param
2023-04-30 16:03:11 +02:00
Vinalti
341d320549
Add ssl_verify param 2023-04-28 00:07:01 +02:00
lucasheld
e176b6b613 bump version to 0.13.0 2023-04-07 21:14:13 +02:00
Lucas Held
bdd441dd84
Merge pull request #22 from lucasheld/feature/custom-headers
Feature: Implement custom SocketIO Headers
2023-04-07 21:08:13 +02:00
lucasheld
d7f033030e feat: add support for uptime kuma 1.21.2 2023-04-07 21:03:36 +02:00
lucasheld
be537a14d2 fix: do not wait for events that have already arrived 2023-04-07 20:07:52 +02:00
lucasheld
307b25249c test: delete maintenances and api keys in setUp 2023-04-07 20:03:30 +02:00
lucasheld
e1fd3b7f03 feat: implement custom socketio headers 2023-03-30 12:19:31 +02:00
lucasheld
cc3588cf4d bump version to 0.12.0 2023-03-30 11:50:02 +02:00
lucasheld
2773d02ee6 feat: add support for uptime kuma 1.21.1 2023-03-30 11:47:47 +02:00
Lucas Held
b33d2323ca
Update README.md 2023-03-20 15:30:22 +01:00
lucasheld
0492018e6d bump version to 0.11.0 2023-03-20 15:16:06 +01:00
lucasheld
42c040f451 feat: add support for uptime kuma 1.21.0 2023-03-20 15:14:39 +01:00
Lucas Held
550be17817
Update README.md 2023-02-27 15:28:28 +01:00
Lucas Held
481eb67ff7
Update README.md 2023-02-15 13:06:38 +01:00
lucasheld
5bfd1044f6 Update README.md 2023-02-13 23:26:01 +01:00
lucasheld
5ce6cf2068 bump version to 0.10.0 2023-02-13 22:51:50 +01:00
lucasheld
14e9f47406 feat: add support for uptime kuma 1.20.0 2023-02-13 22:51:21 +01:00
Lucas Held
8427deffe0
Update README.md 2023-01-20 12:33:50 +01:00
lucasheld
03781e2399 bump version to 0.9.0 2023-01-17 21:29:51 +01:00
lucasheld
5e5c327b20 chore: update year 2023-01-17 21:06:18 +01:00
lucasheld
3e56459fb2 feat: add support for uptime kuma 1.19.5 2023-01-17 21:01:55 +01:00
Lucas Held
cc48b688da
Update README.md 2023-01-09 14:30:14 +01:00
lucasheld
a909c75542 bump version to 0.8.0 2023-01-04 22:49:01 +01:00
lucasheld
982c37045a feat: add support for uptime kuma 1.19.3 2023-01-04 22:46:34 +01:00
lucasheld
4c9d017f62 bump version to 0.7.1 2023-01-01 13:33:05 +01:00
lucasheld
c1d941a200 docs: complete values for uptime kuma 1.19.2 2022-12-30 21:37:13 +01:00
lucasheld
4538901cea docs: fix list type hints 2022-12-29 01:17:48 +01:00
lucasheld
9e3cbe7d59 docs: replace list[dict] with list type hint and add missing type hints 2022-12-29 01:09:44 +01:00
Lucas Held
10a22c8bc2
Update README.md 2022-12-29 00:44:28 +01:00
lucasheld
fd9411c57e bump version to 0.7.0 2022-12-29 00:25:54 +01:00
lucasheld
d01ff6d80e feat: add support for uptime kuma 1.19.2 2022-12-29 00:22:53 +01:00
lucasheld
1e4be04ad7 fix: skip condition check for None values 2022-12-29 00:17:57 +01:00
lucasheld
bc53116d9f bump version to 0.6.0 2022-12-23 14:09:13 +01:00
lucasheld
e0c4207984 feat: add parameter wait_timeout to adjust connection timeout 2022-12-23 14:07:46 +01:00
lucasheld
d68168b769 docs: add return types and exceptions 2022-12-17 15:31:47 +01:00
lucasheld
748d2b191a docs: write param and type in the same line and add optional to type 2022-12-17 15:31:36 +01:00
Lucas Held
6021e9ebc4
Update README.md 2022-12-16 22:10:52 +01:00
lucasheld
ce5ba2d943 docs: add docstrings and sphinx, readthedocs configuration 2022-12-16 21:43:01 +01:00
lucasheld
face6c731d bump version to 0.5.2 2022-11-05 18:50:57 +01:00
lucasheld
cb62b6c814 test: adjust uptime kuma versions 2022-11-05 18:48:59 +01:00
lucasheld
dac368e2a5 fix: add type to notification provider options 2022-11-05 18:48:31 +01:00
Lucas Held
3c17b2bf16
Update README.md 2022-10-11 11:05:45 +02:00
lucasheld
2c9bf42be5 bump version to 0.5.1 2022-10-07 14:34:04 +02:00
lucasheld
52d33d8751 fix: remove required notification provider args check 2022-10-07 14:32:52 +02:00
lucasheld
3d9a2a40b0 bump version to 0.5.0 2022-10-06 19:03:06 +02:00
lucasheld
c8529d65c3 feat: add support for uptime kuma 1.18.3 2022-10-06 19:01:51 +02:00
55 changed files with 5630 additions and 800 deletions

1
.gitignore vendored
View file

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

6
.readthedocs.yaml Normal file
View file

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

View file

@ -1,5 +1,132 @@
## Changelog ## 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 ### Release 0.4.0
#### Feature #### Feature
@ -8,10 +135,6 @@
#### Bugfix #### Bugfix
- update event list data after changes - update event list data after changes
#### Test
- ignore ResourceWarning
- add script to run tests for all supported uptime kuma versions
### Release 0.3.0 ### Release 0.3.0
#### Feature #### Feature
@ -27,9 +150,6 @@
- remove `tags` from monitor input - remove `tags` from monitor input
- convert monitor notificationIDList only once - convert monitor notificationIDList only once
#### Test
- create test objects with available arguments
### Release 0.2.1 ### Release 0.2.1
#### Bugfix #### Bugfix

View file

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

View file

@ -6,9 +6,14 @@ 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. 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.6+ is required. Python version 3.7+ is required.
Supported Uptime Kuma versions: 1.17.0 - 1.18.2 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 |
Installation Installation
--- ---
@ -20,7 +25,11 @@ You can install it using pip:
pip install uptime-kuma-api pip install uptime-kuma-api
``` ```
Examples Documentation
---
The API Reference is available on [Read the Docs](https://uptime-kuma-api.readthedocs.io).
Example
--- ---
Once you have installed the python package, you can use it to communicate with an Uptime Kuma instance. Once you have installed the python package, you can use it to communicate with an Uptime Kuma instance.
@ -45,3 +54,17 @@ At the end, the connection to the API must be disconnected so that the program d
```python ```python
>>> api.disconnect() >>> 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"
)
```

4
dev-requirements.txt Normal file
View file

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

20
docs/Makefile Normal file
View file

@ -0,0 +1,20 @@
# 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)

46
docs/api.rst Normal file
View file

@ -0,0 +1,46 @@
.. _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

53
docs/conf.py Normal file
View file

@ -0,0 +1,53 @@
# 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"
}

32
docs/index.rst Normal file
View file

@ -0,0 +1,32 @@
.. 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

8
docs/install.rst Normal file
View file

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

35
docs/make.bat Normal file
View file

@ -0,0 +1,35 @@
@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,13 +1,21 @@
#!/bin/sh #!/bin/sh
for version in 1.17.0 1.17.1 1.18.0 1.18.2 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 do
echo "Starting uptime kuma $version..." echo "Starting uptime kuma $version..."
docker run -d -it --rm -p 3001:3001 --name uptimekuma "louislam/uptime-kuma:$version" > /dev/null 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" ]] while [[ "$(curl -s -L -o /dev/null -w ''%{http_code}'' 127.0.0.1:3001)" != "200" ]]
do do
sleep 1 sleep 0.5
done done
echo "Running tests..." echo "Running tests..."
@ -15,6 +23,7 @@ do
echo "Stopping uptime kuma..." echo "Stopping uptime kuma..."
docker stop uptimekuma > /dev/null docker stop uptimekuma > /dev/null
sleep 1
echo '' echo ''
done done

1
scripts/.gitignore vendored
View file

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

View file

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

View file

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

View file

@ -3,6 +3,32 @@ from bs4 import BeautifulSoup
from utils import parse_vue_template, write_to_file 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(): def parse_monitor_types():
content = parse_vue_template("uptime-kuma/src/pages/EditMonitor.vue") content = parse_vue_template("uptime-kuma/src/pages/EditMonitor.vue")
@ -10,10 +36,13 @@ def parse_monitor_types():
select = soup.find("select", id="type") select = soup.find("select", id="type")
options = select.find_all("option") options = select.find_all("option")
types = [] types = {}
for o in options: for o in options:
type_ = o.attrs["value"] type_ = o.attrs["value"]
types.append(type_) types[type_] = {
"value": type_,
"title": titles[type_]
}
return types return types

View file

@ -0,0 +1,109 @@
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,12 +3,96 @@ import re
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from utils import deduplicate_list, write_to_file 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",
}
def build_notification_providers(): def build_notification_providers():
providers = [] root = "uptime-kuma"
for path in glob.glob('uptime-kuma/server/notification-providers/*'): providers = {}
# get providers and input names
for path in sorted(glob.glob(f'{root}/server/notification-providers/*')):
with open(path) as f: with open(path) as f:
content = f.read() content = f.read()
match = re.search(r'class [^ ]+ extends NotificationProvider {', content) match = re.search(r'class [^ ]+ extends NotificationProvider {', content)
@ -16,20 +100,20 @@ def build_notification_providers():
match = re.search(r'name = "([^"]+)";', content) match = re.search(r'name = "([^"]+)";', content)
name = match.group(1) name = match.group(1)
inputs = re.findall(r'notification\.([^ ,.;})\]]+)', content) inputs = re.findall(r'notification\??\.([^ ,.;})\]]+)', content)
inputs = deduplicate_list(inputs) inputs = deduplicate_list(inputs)
inputs = [i.strip() for i in inputs] inputs = [i.strip() for i in inputs]
providers.append({ providers[name] = {
"name": name, "title": titles[name],
"inputs": inputs, "inputs": {},
}) }
return providers for input_ in inputs:
if input_ not in ignored_inputs.get(name, []):
providers[name]["inputs"][input_] = {}
# get inputs
def build_notification_provider_conditions(): for path in glob.glob(f'{root}/src/components/notifications/*'):
conditions = {}
for path in glob.glob('uptime-kuma/src/components/notifications/*'):
if path.endswith("index.js"): if path.endswith("index.js"):
continue continue
with open(path) as f: with open(path) as f:
@ -37,25 +121,59 @@ def build_notification_provider_conditions():
match = re.search(r'<template>[\s\S]+</template>', content, re.MULTILINE) match = re.search(r'<template>[\s\S]+</template>', content, re.MULTILINE)
html = match.group(0) html = match.group(0)
soup = BeautifulSoup(html, "html.parser") soup = BeautifulSoup(html, "html.parser")
inputs = soup.find_all("input") inputs = soup.find_all(attrs={"v-model": True})
for input in inputs: for input_ in inputs:
condition = {} conditions = {}
attrs = input.attrs attrs = input_.attrs
v_model = attrs.get("v-model") 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") min_ = attrs.get("min")
max_ = attrs.get("max")
if min_: if min_:
condition["min"] = int(min_) conditions["min"] = int(min_)
max_ = attrs.get("max")
if max_: if max_:
condition["max"] = int(max_) conditions["max"] = int(max_)
param_name = re.match(r'\$parent.notification.(.*)$', v_model).group(1)
if condition: # find provider inputs dict
conditions[param_name] = condition input_found = False
return conditions 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
notification_providers = build_notification_providers() notification_providers = build_notification_providers()
notification_provider_conditions = build_notification_provider_conditions()
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"]
write_to_file( write_to_file(
"notification_providers.py.j2", "./../uptime_kuma_api/notification_providers.py", "notification_providers.py.j2", "./../uptime_kuma_api/notification_providers.py",

View file

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

View file

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

View file

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

View file

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

View file

@ -24,3 +24,23 @@ def write_to_file(template, destination, **kwargs):
rendered = template.render(**kwargs) rendered = template.render(**kwargs)
with open(destination, "w") as f: with open(destination, "w") as f:
f.write(rendered) 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", author_email="lucasheld@hotmail.de",
license=info["__license__"], license=info["__license__"],
packages=["uptime_kuma_api"], packages=["uptime_kuma_api"],
python_requires=">=3.6, <4", python_requires=">=3.7, <4",
install_requires=[ install_requires=[
"python-socketio[client]>=5.0.0", "python-socketio[client]>=5.0.0",
"packaging" "packaging"
], ],
classifiers=[ classifiers=[
"Development Status :: 1 - Planning", "Development Status :: 5 - Production/Stable",
"Environment :: Web Environment", "Environment :: Web Environment",
"Intended Audience :: Developers", "Intended Audience :: Developers",
"License :: OSI Approved :: MIT License", "License :: OSI Approved :: MIT License",
@ -43,7 +43,6 @@ setup(
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Programming Language :: Python", "Programming Language :: Python",
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7", "Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.9",

View file

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

59
tests/test_api_key.py Normal file
View file

@ -0,0 +1,59 @@
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,12 +6,6 @@ from uptime_kuma_test_case import UptimeKumaTestCase
class TestDockerHost(UptimeKumaTestCase): class TestDockerHost(UptimeKumaTestCase):
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): def test_docker_host(self):
# get empty list to make sure that future accesses will also work # get empty list to make sure that future accesses will also work
self.api.get_docker_hosts() self.api.get_docker_hosts()
@ -23,8 +17,10 @@ class TestDockerHost(UptimeKumaTestCase):
} }
# test docker host # test docker host
with self.assertRaisesRegex(UptimeKumaException, r'connect ENOENT /var/run/docker.sock'): if parse_version(self.api.version) != parse_version("1.23.0"):
self.api.test_docker_host(**expected_docker_host) # 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)
# add docker host # add docker host
r = self.api.add_docker_host(**expected_docker_host) r = self.api.add_docker_host(**expected_docker_host)
@ -37,6 +33,7 @@ class TestDockerHost(UptimeKumaTestCase):
# get docker hosts # get docker hosts
docker_hosts = self.api.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) docker_host = self.find_by_id(docker_hosts, docker_host_id)
self.assertIsNotNone(docker_host) self.assertIsNotNone(docker_host)
self.compare(docker_host, expected_docker_host) self.compare(docker_host, expected_docker_host)
@ -54,6 +51,10 @@ class TestDockerHost(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException): with self.assertRaises(UptimeKumaException):
self.api.get_docker_host(docker_host_id) 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__': if __name__ == '__main__':
unittest.main() unittest.main()

13
tests/test_game_list.py Normal file
View file

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

View file

@ -0,0 +1,15 @@
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,5 +1,6 @@
import unittest import unittest
from uptime_kuma_api import UptimeKumaApi
from uptime_kuma_test_case import UptimeKumaTestCase from uptime_kuma_test_case import UptimeKumaTestCase
@ -9,6 +10,16 @@ class TestInfo(UptimeKumaTestCase):
self.assertIn("version", info) self.assertIn("version", info)
self.assertIn("latestVersion", 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__': if __name__ == '__main__':
unittest.main() unittest.main()

269
tests/test_maintenance.py Normal file
View file

@ -0,0 +1,269 @@
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,7 +1,7 @@
import unittest import unittest
from packaging.version import parse as parse_version from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod from uptime_kuma_api import UptimeKumaException, MonitorType, AuthMethod, MonitorStatus
from uptime_kuma_test_case import UptimeKumaTestCase from uptime_kuma_test_case import UptimeKumaTestCase
@ -21,12 +21,9 @@ class TestMonitor(UptimeKumaTestCase):
"maxretries": 0, "maxretries": 0,
"notificationIDList": [notification_id_1, notification_id_2], "notificationIDList": [notification_id_1, notification_id_2],
"upsideDown": False, "upsideDown": False,
"url": "http://127.0.0.1" "url": "http://127.0.0.1",
"resendInterval": 0
} }
if parse_version(self.api.version) >= parse_version("1.18"):
expected_monitor.update({
"resendInterval": 0
})
# add monitor # add monitor
r = self.api.add_monitor(**expected_monitor) r = self.api.add_monitor(**expected_monitor)
@ -39,7 +36,11 @@ class TestMonitor(UptimeKumaTestCase):
# get monitors # get monitors
monitors = self.api.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) monitor = self.find_by_id(monitors, monitor_id)
self.assertTrue(type(monitor["type"]) == MonitorType)
self.assertTrue(type(monitor["authMethod"]) == AuthMethod)
self.assertIsNotNone(monitor) self.assertIsNotNone(monitor)
self.compare(monitor, expected_monitor) self.compare(monitor, expected_monitor)
@ -62,7 +63,8 @@ class TestMonitor(UptimeKumaTestCase):
self.assertEqual(r["msg"], "Resumed Successfully.") self.assertEqual(r["msg"], "Resumed Successfully.")
# get monitor beats # get monitor beats
self.api.get_monitor_beats(monitor_id, 6) r = self.api.get_monitor_beats(monitor_id, 6)
self.assertTrue(type(r[0]["status"]) == MonitorStatus)
# delete monitor # delete monitor
r = self.api.delete_monitor(monitor_id) r = self.api.delete_monitor(monitor_id)
@ -132,6 +134,17 @@ class TestMonitor(UptimeKumaTestCase):
} }
self.do_test_monitor_type(expected_monitor) 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",
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_port(self): def test_monitor_type_port(self):
expected_monitor = { expected_monitor = {
"type": MonitorType.PORT, "type": MonitorType.PORT,
@ -145,7 +158,8 @@ class TestMonitor(UptimeKumaTestCase):
expected_monitor = { expected_monitor = {
"type": MonitorType.PING, "type": MonitorType.PING,
"name": "monitor 1", "name": "monitor 1",
"hostname": "127.0.0.1" "hostname": "127.0.0.1",
"packetSize": 56
} }
self.do_test_monitor_type(expected_monitor) self.do_test_monitor_type(expected_monitor)
@ -158,6 +172,17 @@ class TestMonitor(UptimeKumaTestCase):
} }
self.do_test_monitor_type(expected_monitor) 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): def test_monitor_type_dns(self):
expected_monitor = { expected_monitor = {
"type": MonitorType.DNS, "type": MonitorType.DNS,
@ -169,6 +194,16 @@ class TestMonitor(UptimeKumaTestCase):
} }
self.do_test_monitor_type(expected_monitor) 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
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_push(self): def test_monitor_type_push(self):
expected_monitor = { expected_monitor = {
"type": MonitorType.PUSH, "type": MonitorType.PUSH,
@ -188,6 +223,18 @@ class TestMonitor(UptimeKumaTestCase):
} }
self.do_test_monitor_type(expected_monitor) 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): def test_monitor_type_mqtt(self):
expected_monitor = { expected_monitor = {
"type": MonitorType.MQTT, "type": MonitorType.MQTT,
@ -212,9 +259,6 @@ class TestMonitor(UptimeKumaTestCase):
self.do_test_monitor_type(expected_monitor) self.do_test_monitor_type(expected_monitor)
def test_monitor_type_postgres(self): 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 = { expected_monitor = {
"type": MonitorType.POSTGRES, "type": MonitorType.POSTGRES,
"name": "monitor 1", "name": "monitor 1",
@ -223,23 +267,24 @@ class TestMonitor(UptimeKumaTestCase):
} }
self.do_test_monitor_type(expected_monitor) self.do_test_monitor_type(expected_monitor)
def test_monitor_type_docker(self): def test_monitor_type_mysql(self):
if parse_version(self.api.version) < parse_version("1.18"):
self.skipTest("Unsupported in this Uptime Kuma version")
docker_host_id = self.add_docker_host()
expected_monitor = { expected_monitor = {
"type": MonitorType.DOCKER, "type": MonitorType.MYSQL,
"name": "monitor 1", "name": "monitor 1",
"docker_container": "test", "databaseConnectionString": "mysql://username:password@host:port/database",
"docker_host": docker_host_id "databaseQuery": "select getdate()"
}
self.do_test_monitor_type(expected_monitor)
def test_monitor_type_mongodb(self):
expected_monitor = {
"type": MonitorType.MONGODB,
"name": "monitor 1",
"databaseConnectionString": "mongodb://username:password@host:port/database"
} }
self.do_test_monitor_type(expected_monitor) self.do_test_monitor_type(expected_monitor)
def test_monitor_type_radius(self): 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 = { expected_monitor = {
"type": MonitorType.RADIUS, "type": MonitorType.RADIUS,
"name": "monitor 1", "name": "monitor 1",
@ -251,6 +296,85 @@ class TestMonitor(UptimeKumaTestCase):
} }
self.do_test_monitor_type(expected_monitor) 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_monitor_type_group(self):
if parse_version(self.api.version) < parse_version("1.22"):
self.skipTest("Unsupported in this Uptime Kuma version")
# 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"]
# 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)
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)
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -1,5 +1,6 @@
import unittest import unittest
from uptime_kuma_api import UptimeKumaException
from uptime_kuma_test_case import UptimeKumaTestCase from uptime_kuma_test_case import UptimeKumaTestCase
@ -32,6 +33,10 @@ class TestMonitorTag(UptimeKumaTestCase):
monitor = self.find_by_id(monitors, monitor_id) monitor = self.find_by_id(monitors, monitor_id)
self.assertEqual(monitor["tags"], []) 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__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -13,12 +13,13 @@ class TestNotification(UptimeKumaTestCase):
"name": "notification 1", "name": "notification 1",
"isDefault": True, "isDefault": True,
"applyExisting": True, "applyExisting": True,
"type": NotificationType.PUSHBYTECHULUS, "type": NotificationType.TELEGRAM,
"pushAPIKey": "123456789" "telegramChatID": "123456789",
"telegramBotToken": "987654321"
} }
# test notification # test notification
with self.assertRaisesRegex(UptimeKumaException, r'Invalid API key'): with self.assertRaisesRegex(UptimeKumaException, r'Not Found'):
self.api.test_notification(**expected_notification) self.api.test_notification(**expected_notification)
# add notification # add notification
@ -32,7 +33,9 @@ class TestNotification(UptimeKumaTestCase):
# get notifications # get notifications
notifications = self.api.get_notifications() notifications = self.api.get_notifications()
self.assertTrue(type(notifications[0]["type"]) == NotificationType)
notification = self.find_by_id(notifications, notification_id) notification = self.find_by_id(notifications, notification_id)
self.assertTrue(type(notification["type"]) == NotificationType)
self.assertIsNotNone(notification) self.assertIsNotNone(notification)
self.compare(notification, expected_notification) self.compare(notification, expected_notification)
@ -42,7 +45,8 @@ class TestNotification(UptimeKumaTestCase):
expected_notification["applyExisting"] = False expected_notification["applyExisting"] = False
expected_notification["type"] = NotificationType.PUSHDEER expected_notification["type"] = NotificationType.PUSHDEER
expected_notification["pushdeerKey"] = "987654321" expected_notification["pushdeerKey"] = "987654321"
del expected_notification["pushAPIKey"] del expected_notification["telegramChatID"]
del expected_notification["telegramBotToken"]
r = self.api.edit_notification(notification_id, **expected_notification) r = self.api.edit_notification(notification_id, **expected_notification)
self.assertEqual(r["msg"], "Saved") self.assertEqual(r["msg"], "Saved")
notification = self.api.get_notification(notification_id) notification = self.api.get_notification(notification_id)
@ -55,6 +59,10 @@ class TestNotification(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException): with self.assertRaises(UptimeKumaException):
self.api.delete_notification(notification_id) 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__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -31,6 +31,7 @@ class TestProxy(UptimeKumaTestCase):
# get proxies # get proxies
proxies = self.api.get_proxies() proxies = self.api.get_proxies()
self.assertTrue(type(proxies[0]["protocol"]) == ProxyProtocol)
proxy = self.find_by_id(proxies, proxy_id) proxy = self.find_by_id(proxies, proxy_id)
self.assertIsNotNone(proxy) self.assertIsNotNone(proxy)
self.compare(proxy, expected_proxy) self.compare(proxy, expected_proxy)
@ -50,6 +51,10 @@ class TestProxy(UptimeKumaTestCase):
with self.assertRaises(UptimeKumaException): with self.assertRaises(UptimeKumaException):
self.api.get_proxy(proxy_id) 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__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -1,6 +1,5 @@
import json import json
import unittest import unittest
from packaging.version import parse as parse_version
from uptime_kuma_test_case import UptimeKumaTestCase from uptime_kuma_test_case import UptimeKumaTestCase
@ -16,14 +15,12 @@ class TestSettings(UptimeKumaTestCase):
"primaryBaseURL": "", "primaryBaseURL": "",
"steamAPIKey": "", "steamAPIKey": "",
"tlsExpiryNotifyDays": [7, 14, 21], "tlsExpiryNotifyDays": [7, 14, 21],
"disableAuth": False "disableAuth": False,
"trustProxy": False,
"serverTimezone": "Europe/Berlin",
"dnsCache": True
} }
if parse_version(self.api.version) >= parse_version("1.18"):
expected_settings.update({
"trustProxy": False
})
# set settings # set settings
r = self.api.set_settings(self.password, **expected_settings) r = self.api.set_settings(self.password, **expected_settings)
self.assertEqual(r["msg"], "Saved") self.assertEqual(r["msg"], "Saved")

View file

@ -34,7 +34,8 @@ class TestStatusPage(UptimeKumaTestCase):
} }
] ]
} }
] ],
"googleAnalyticsId": ""
} }
# add status page # add status page
@ -70,9 +71,11 @@ class TestStatusPage(UptimeKumaTestCase):
"style": IncidentStyle.DANGER "style": IncidentStyle.DANGER
} }
incident = self.api.post_incident(slug, **incident_expected) incident = self.api.post_incident(slug, **incident_expected)
self.assertTrue(type(incident["style"]) == IncidentStyle)
self.compare(incident, incident_expected) self.compare(incident, incident_expected)
status_page = self.api.get_status_page(slug) status_page = self.api.get_status_page(slug)
self.compare(status_page["incident"], incident) self.compare(status_page["incident"], incident)
self.assertTrue(type(status_page["incident"]["style"]) == IncidentStyle)
# unpin incident # unpin incident
self.api.unpin_incident(slug) self.api.unpin_incident(slug)
@ -88,6 +91,10 @@ class TestStatusPage(UptimeKumaTestCase):
status_page = self.find_by_id(status_pages, slug, "slug") status_page = self.find_by_id(status_pages, slug, "slug")
self.assertIsNone(status_page) 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__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -26,12 +26,24 @@ class TestTag(UptimeKumaTestCase):
self.assertIsNotNone(tag) self.assertIsNotNone(tag)
self.compare(tag, expected_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 # delete tag
r = self.api.delete_tag(tag_id) r = self.api.delete_tag(tag_id)
self.assertEqual(r["msg"], "Deleted Successfully.") self.assertEqual(r["msg"], "Deleted Successfully.")
with self.assertRaises(UptimeKumaException): with self.assertRaises(UptimeKumaException):
self.api.get_tag(tag_id) 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__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -1,6 +1,5 @@
import unittest import unittest
import warnings import warnings
from packaging.version import parse as parse_version
from uptime_kuma_api import UptimeKumaApi, MonitorType, DockerType from uptime_kuma_api import UptimeKumaApi, MonitorType, DockerType
@ -38,7 +37,7 @@ class UptimeKumaTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
warnings.simplefilter("ignore", ResourceWarning) warnings.simplefilter("ignore", ResourceWarning)
self.api = UptimeKumaApi(self.url) self.api = UptimeKumaApi(self.url, timeout=1, wait_events=0.01)
global token global token
if not token: if not token:
@ -74,11 +73,20 @@ class UptimeKumaTestCase(unittest.TestCase):
for status_page in status_pages: for status_page in status_pages:
self.api.delete_status_page(status_page["slug"]) self.api.delete_status_page(status_page["slug"])
if parse_version(self.api.version) >= parse_version("1.18"): # delete docker hosts
# delete docker hosts docker_hosts = self.api.get_docker_hosts()
docker_hosts = self.api.get_docker_hosts() for docker_host in docker_hosts:
for docker_host in docker_hosts: self.api.delete_docker_host(docker_host["id"])
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 # login again to receive initial messages
self.api.disconnect() self.api.disconnect()
@ -96,10 +104,10 @@ class UptimeKumaTestCase(unittest.TestCase):
if obj[key] == value: if obj[key] == value:
return obj return obj
def add_monitor(self): def add_monitor(self, name="monitor 1"):
r = self.api.add_monitor( r = self.api.add_monitor(
type=MonitorType.HTTP, type=MonitorType.HTTP,
name="monitor 1", name=name,
url="http://127.0.0.1" url="http://127.0.0.1"
) )
monitor_id = r["monitorID"] monitor_id = r["monitorID"]
@ -139,3 +147,9 @@ class UptimeKumaTestCase(unittest.TestCase):
) )
docker_host_id = r["id"] docker_host_id = r["id"]
return docker_host_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,10 +1,12 @@
from .__version__ import __title__, __version__, __author__, __copyright__ from .__version__ import __title__, __version__, __author__, __copyright__
from .auth_method import AuthMethod from .auth_method import AuthMethod
from .monitor_status import MonitorStatus
from .monitor_type import MonitorType from .monitor_type import MonitorType
from .notification_providers import NotificationType, notification_provider_options, notification_provider_conditions from .notification_providers import NotificationType, notification_provider_options, notification_provider_conditions
from .proxy_protocol import ProxyProtocol from .proxy_protocol import ProxyProtocol
from .incident_style import IncidentStyle from .incident_style import IncidentStyle
from .docker_type import DockerType from .docker_type import DockerType
from .exceptions import UptimeKumaException from .maintenance_strategy import MaintenanceStrategy
from .exceptions import UptimeKumaException, Timeout
from .event import Event from .event import Event
from .api import UptimeKumaApi from .api import UptimeKumaApi

View file

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

File diff suppressed because one or more lines are too long

View file

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

View file

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

View file

@ -0,0 +1,387 @@
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

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

View file

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

View file

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

View file

@ -0,0 +1,23 @@
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

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

View file

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

View file

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