Merge pull request #175 from ferozsalam/migrate-alerter-tests

Migrate all the alerter tests into individual files

tests/alerters/chatwork_test.py

@@ -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

tests/alerters/command_test.py

@@ -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

tests/alerters/datadog_test.py

@@ -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