borgmatic-collective/borgmatic
borgmatic-collective/borgmatic
12
63
Fork 16
Code Issues 50 Pull Requests 1 Actions Packages Releases 133 Activity

fixes and first unit test attempt

Browse Source
This commit is contained in:
Tony Fernandez 2024-10-28 09:36:30 -04:00
parent ffb431e3ab
commit 385ef2d012
3 changed files with 219 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
borgmatic/config/schema.yaml
View File

@ -1628,7 +1628,6 @@ properties:
type: string
description: |
The address of your zabbix instance.
Not specifying a server will default to the Zabbix cloud.
example: https://zabbix.your-domain.com
username:
type: string

33
borgmatic/hooks/zabbix.py
View File

@ -31,11 +31,11 @@ def ping_monitor(hook_config, config, config_filename, state, monitoring_log_lev
state_config = hook_config.get(
state.name.lower(),
{
'value': f'invalid',
'value': state.name.lower(),
},
)
base_url = hook_config.get('server', 'https://cloud.zabbix.com/zabbix/api_jsonrpc.php')
server = hook_config.get('server')
username = hook_config.get('username')
password = hook_config.get('password')
api_key = hook_config.get('api_key')
@ -46,7 +46,11 @@ def ping_monitor(hook_config, config, config_filename, state, monitoring_log_lev
headers = {'Content-Type': 'application/json-rpc'}
logger.info(f'{config_filename}: Updating Zabbix {dry_run_label}')
logger.debug(f'{config_filename}: Using Zabbix URL: {base_url}')
logger.debug(f'{config_filename}: Using Zabbix URL: {server}')
if server is None:
logger.warning(f'{config_filename}: Server missing for Zabbix')
return
# Determine the zabbix method used to store the value: itemid or host/key
if itemid is not None:
@ -68,11 +72,14 @@ def ping_monitor(hook_config, config, config_filename, state, monitoring_log_lev
}
elif host is not None:
logger.warning(f'{config_filename}: Key missing for Zabbix authentication')
logger.warning(f'{config_filename}: Key missing for Zabbix')
return
elif key is not None:
logger.warning(f'{config_filename}: Host missing for Zabbix authentication')
logger.warning(f'{config_filename}: Host missing for Zabbix.')
return
else:
logger.warning(f'{config_filename}: No zabbix itemid or host/key provided.')
return
# Determine the authentication method: API key or username/password
@ -92,8 +99,14 @@ def ping_monitor(hook_config, config, config_filename, state, monitoring_log_lev
"id": 1
}
if not dry_run:
response = requests.post(base_url, headers=headers, json=auth_data)
data['auth'] = response.json().get('result')
logging.getLogger('urllib3').setLevel(logging.ERROR)
try:
response = requests.post(server, headers=headers, json=auth_data)
data['auth'] = response.json().get('result')
if not response.ok:
response.raise_for_status()
except requests.exceptions.RequestException as error:
logger.warning(f'{config_filename}: Zabbix error: {error}')
elif username is not None:
logger.warning(f'{config_filename}: Password missing for Zabbix authentication')
@ -102,11 +115,15 @@ def ping_monitor(hook_config, config, config_filename, state, monitoring_log_lev
elif password is not None:
logger.warning(f'{config_filename}: Username missing for Zabbix authentication')
return
else:
logger.warning(f'{config_filename}: Authentication data missing for Zabbix')
return
if not dry_run:
logging.getLogger('urllib3').setLevel(logging.ERROR)
try:
response = requests.post(base_url, headers=headers, json=data)
response = requests.post(server, headers=headers, json=data)
if not response.ok:
response.raise_for_status()
except requests.exceptions.RequestException as error:

194
tests/unit/hooks/test_zabbix.py Normal file
View File

@ -0,0 +1,194 @@
from enum import Enum
from flexmock import flexmock
import borgmatic.hooks.monitor
from borgmatic.hooks import zabbix as module
server = 'https://zabbix.com/zabbix/api_jsonrpc.php'
itemid = 55105
username = 'testuser'
password = 'fakepassword'
api_key = 'fakekey'
host = 'borg-server'
key = 'borg.status'
value = 'fail'
data_host_key = {
"jsonrpc": "2.0",
"method": "history.push",
"params": {"host": host, "key": key, "value": value},
"id": 1,
}
data_itemid = {
"jsonrpc": "2.0",
"method": "history.push",
"params": {"itemid": itemid, "value": value},
"id": 1,
}
data_user_login = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {"username": username, "password": password},
"id": 1,
}
auth_headers_api_key = {
'Content-Type': 'application/json-rpc',
'Authorization': f'Bearer {api_key}'
}
auth_headers_username_password = {
'Content-Type': 'application/json-rpc'
}
def test_ping_monitor_config_with_api_key_only():
hook_config = {
'api_key': api_key
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_with_host_only():
hook_config = {
'host': host
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_with_key_only():
hook_config = {
'key': key
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_with_server_only():
hook_config = {
'server': server
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_user_password_no_zabbix_data():
hook_config = {
'server': server,
'username': username,
'password': password
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_api_key_no_zabbix_data():
hook_config = {
'server': server,
'api_key': api_key
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_itemid_no_auth_data():
hook_config = {
'server': server,
'itemid': itemid
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_host_and_key_no_auth_data():
hook_config = {
'server': server,
'host': host,
'key': key
}
flexmock(module.logger).should_receive('warning').once()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)
def test_ping_monitor_config_host_and_key_with_api_key_auth_data():
hook_config = {
'server': server,
'host': host,
'key': key,
'api_key': api_key
}
flexmock(module.requests).should_receive('post').with_args(
f'{server}',
headers=auth_headers_api_key,
json=data_host_key,
).and_return(flexmock(ok=True)).once()
flexmock(module.logger).should_receive('warning').never()
module.ping_monitor(
hook_config,
{},
'config.yaml',
borgmatic.hooks.monitor.State.FAIL,
monitoring_log_level=1,
dry_run=False,
)