Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Migrate all the alerter tests into individual files #175

Merged
merged 1 commit into from
May 22, 2021
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file added tests/alerters/__init__.py
Empty file.
653 changes: 653 additions & 0 deletions tests/alerters/alerta_test.py

Large diffs are not rendered by default.

103 changes: 103 additions & 0 deletions tests/alerters/chatwork_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
import mock
import pytest
from requests import RequestException
from requests.auth import HTTPProxyAuth

from elastalert.alerters.chatwork import ChatworkAlerter
from elastalert.loaders import FileRulesLoader
from elastalert.util import EAException


def test_chatwork():
rule = {
'name': 'Test Chatwork Rule',
'type': 'any',
'chatwork_apikey': 'xxxx1',
'chatwork_room_id': 'xxxx2',
'alert': []
}
rules_loader = FileRulesLoader({})
rules_loader.load_modules(rule)
alert = ChatworkAlerter(rule)
match = {
'@timestamp': '2021-01-01T00:00:00',
'somefield': 'foobarbaz'
}
with mock.patch('requests.post') as mock_post_request:
alert.alert([match])
expected_data = {
'body': 'Test Chatwork Rule\n\n@timestamp: 2021-01-01T00:00:00\nsomefield: foobarbaz\n',
}

mock_post_request.assert_called_once_with(
'https://api.chatwork.com/v2/rooms/xxxx2/messages',
params=mock.ANY,
headers={'X-ChatWorkToken': 'xxxx1'},
proxies=None,
auth=None
)

actual_data = mock_post_request.call_args_list[0][1]['params']
assert expected_data == actual_data


def test_chatwork_proxy():
rule = {
'name': 'Test Chatwork Rule',
'type': 'any',
'chatwork_apikey': 'xxxx1',
'chatwork_room_id': 'xxxx2',
'chatwork_proxy': 'http://proxy.url',
'chatwork_proxy_login': 'admin',
'chatwork_proxy_pass': 'password',
'alert': []
}
rules_loader = FileRulesLoader({})
rules_loader.load_modules(rule)
alert = ChatworkAlerter(rule)
match = {
'@timestamp': '2021-01-01T00:00:00',
'somefield': 'foobarbaz'
}
with mock.patch('requests.post') as mock_post_request:
alert.alert([match])
expected_data = {
'body': 'Test Chatwork Rule\n\n@timestamp: 2021-01-01T00:00:00\nsomefield: foobarbaz\n',
}

mock_post_request.assert_called_once_with(
'https://api.chatwork.com/v2/rooms/xxxx2/messages',
params=mock.ANY,
headers={'X-ChatWorkToken': 'xxxx1'},
proxies={'https': 'http://proxy.url'},
auth=HTTPProxyAuth('admin', 'password')
)

actual_data = mock_post_request.call_args_list[0][1]['params']
assert expected_data == actual_data


def test_chatwork_ea_exception():
try:
rule = {
'name': 'Test Chatwork Rule',
'type': 'any',
'chatwork_apikey': 'xxxx1',
'chatwork_room_id': 'xxxx2',
'chatwork_proxy': 'http://proxy.url',
'chatwork_proxy_login': 'admin',
'chatwork_proxy_pass': 'password',
'alert': []
}
rules_loader = FileRulesLoader({})
rules_loader.load_modules(rule)
alert = ChatworkAlerter(rule)
match = {
'@timestamp': '2021-01-01T00:00:00',
'somefield': 'foobarbaz'
}
mock_run = mock.MagicMock(side_effect=RequestException)
with mock.patch('requests.post', mock_run), pytest.raises(RequestException):
alert.alert([match])
except EAException:
assert True
97 changes: 97 additions & 0 deletions tests/alerters/command_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import json
import subprocess

import mock
import pytest

from elastalert.alerters.command import CommandAlerter
from elastalert.alerts import BasicMatchString
from elastalert.util import EAException
from tests.alerts_test import mock_rule


def test_command():
# Test command as list with a formatted arg
rule = {'command': ['/bin/test/', '--arg', '%(somefield)s']}
alert = CommandAlerter(rule)
match = {'@timestamp': '2014-01-01T00:00:00',
'somefield': 'foobarbaz',
'nested': {'field': 1}}
with mock.patch("elastalert.alerters.command.subprocess.Popen") as mock_popen:
alert.alert([match])
assert mock_popen.called_with(['/bin/test', '--arg', 'foobarbaz'], stdin=subprocess.PIPE, shell=False)

# Test command as string with formatted arg (old-style string format)
rule = {'command': '/bin/test/ --arg %(somefield)s'}
alert = CommandAlerter(rule)
with mock.patch("elastalert.alerters.command.subprocess.Popen") as mock_popen:
alert.alert([match])
assert mock_popen.called_with('/bin/test --arg foobarbaz', stdin=subprocess.PIPE, shell=False)

# Test command as string without formatted arg (old-style string format)
rule = {'command': '/bin/test/foo.sh'}
alert = CommandAlerter(rule)
with mock.patch("elastalert.alerters.command.subprocess.Popen") as mock_popen:
alert.alert([match])
assert mock_popen.called_with('/bin/test/foo.sh', stdin=subprocess.PIPE, shell=True)

# Test command with pipe_match_json
rule = {'command': ['/bin/test/', '--arg', '%(somefield)s'],
'pipe_match_json': True}
alert = CommandAlerter(rule)
match = {'@timestamp': '2014-01-01T00:00:00',
'somefield': 'foobarbaz'}
with mock.patch("elastalert.alerters.command.subprocess.Popen") as mock_popen:
mock_subprocess = mock.Mock()
mock_popen.return_value = mock_subprocess
mock_subprocess.communicate.return_value = (None, None)
alert.alert([match])
assert mock_popen.called_with(['/bin/test', '--arg', 'foobarbaz'], stdin=subprocess.PIPE, shell=False)
assert mock_subprocess.communicate.called_with(input=json.dumps(match))

# Test command with pipe_alert_text
rule = {'command': ['/bin/test/', '--arg', '%(somefield)s'],
'pipe_alert_text': True, 'type': mock_rule(), 'name': 'Test'}
alert = CommandAlerter(rule)
match = {'@timestamp': '2014-01-01T00:00:00',
'somefield': 'foobarbaz'}
alert_text = str(BasicMatchString(rule, match))
with mock.patch("elastalert.alerters.command.subprocess.Popen") as mock_popen:
mock_subprocess = mock.Mock()
mock_popen.return_value = mock_subprocess
mock_subprocess.communicate.return_value = (None, None)
alert.alert([match])
assert mock_popen.called_with(['/bin/test', '--arg', 'foobarbaz'], stdin=subprocess.PIPE, shell=False)
assert mock_subprocess.communicate.called_with(input=alert_text.encode())

# Test command with fail_on_non_zero_exit
rule = {'command': ['/bin/test/', '--arg', '%(somefield)s'],
'fail_on_non_zero_exit': True}
alert = CommandAlerter(rule)
match = {'@timestamp': '2014-01-01T00:00:00',
'somefield': 'foobarbaz'}
with pytest.raises(Exception) as exception:
with mock.patch("elastalert.alerters.command.subprocess.Popen") as mock_popen:
mock_subprocess = mock.Mock()
mock_popen.return_value = mock_subprocess
mock_subprocess.wait.return_value = 1
alert.alert([match])
assert mock_popen.called_with(['/bin/test', '--arg', 'foobarbaz'], stdin=subprocess.PIPE, shell=False)
assert "Non-zero exit code while running command" in str(exception)

# Test OSError
try:
rule = {'command': ['/bin/test/', '--arg', '%(somefield)s'],
'pipe_alert_text': True, 'type': mock_rule(), 'name': 'Test'}
alert = CommandAlerter(rule)
match = {'@timestamp': '2014-01-01T00:00:00',
'somefield': 'foobarbaz'}
alert_text = str(BasicMatchString(rule, match))
mock_run = mock.MagicMock(side_effect=OSError)
with mock.patch("elastalert.alerters.command.subprocess.Popen", mock_run), pytest.raises(OSError) as mock_popen:
mock_subprocess = mock.Mock()
mock_popen.return_value = mock_subprocess
mock_subprocess.communicate.return_value = (None, None)
alert.alert([match])
except EAException:
assert True
69 changes: 69 additions & 0 deletions tests/alerters/datadog_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
import json

import mock
import pytest
from requests import RequestException

from elastalert.alerters.datadog import DatadogAlerter
from elastalert.loaders import FileRulesLoader
from elastalert.util import EAException


def test_datadog_alerter():
rule = {
'name': 'Test Datadog Event Alerter',
'type': 'any',
'datadog_api_key': 'test-api-key',
'datadog_app_key': 'test-app-key',
'alert': [],
'alert_subject': 'Test Datadog Event Alert'
}
rules_loader = FileRulesLoader({})
rules_loader.load_modules(rule)
alert = DatadogAlerter(rule)
match = {
'@timestamp': '2021-01-01T00:00:00',
'name': 'datadog-test-name'
}
with mock.patch('requests.post') as mock_post_request:
alert.alert([match])

expected_data = {
'title': rule['alert_subject'],
'text': "Test Datadog Event Alerter\n\n@timestamp: 2021-01-01T00:00:00\nname: datadog-test-name\n"
}
mock_post_request.assert_called_once_with(
"https://api.datadoghq.com/api/v1/events",
data=mock.ANY,
headers={
'Content-Type': 'application/json',
'DD-API-KEY': rule['datadog_api_key'],
'DD-APPLICATION-KEY': rule['datadog_app_key']
}
)
actual_data = json.loads(mock_post_request.call_args_list[0][1]['data'])
assert expected_data == actual_data


def test_datadog_alerterea_exception():
try:
rule = {
'name': 'Test Datadog Event Alerter',
'type': 'any',
'datadog_api_key': 'test-api-key',
'datadog_app_key': 'test-app-key',
'alert': [],
'alert_subject': 'Test Datadog Event Alert'
}
rules_loader = FileRulesLoader({})
rules_loader.load_modules(rule)
alert = DatadogAlerter(rule)
match = {
'@timestamp': '2021-01-01T00:00:00',
'name': 'datadog-test-name'
}
mock_run = mock.MagicMock(side_effect=RequestException)
with mock.patch('requests.post', mock_run), pytest.raises(RequestException):
alert.alert([match])
except EAException:
assert True
Loading