basic sensu api tests

brian_polling_manager/sensu.py

@@ -9,9 +9,7 @@ logger = logging.getLogger(__name__)
 
 
 def load_all_checks(params, namespace='default'):
-
     url = random.choice(params['api-base'])
-    # url = params['api-base'][0]
     r = requests.get(
         f'{url}/api/core/v2/namespaces/{namespace}/checks',
         headers={
@@ -25,30 +23,30 @@ def load_all_checks(params, namespace='default'):
 
 
 def create_check(params, check, namespace='default'):
-    logger.error(f'creating missing check: {check["metadata"]["name"]}')
-    # url = random.choice(params['api-base'])
-    # r = requests.post(
-    #     f'{url}/api/core/v2/namespaces/{namespace}/checks',
-    #     headers={
-    #         'Authorization': f'Key {params["token"]}',
-    #         'Content-Type': 'application/json',
-    #     },
-    #     json=check)
-    # r.raise_for_status()
+    logger.debug(f'creating missing check: {check["metadata"]["name"]}')
+    url = random.choice(params['api-base'])
+    r = requests.post(
+        f'{url}/api/core/v2/namespaces/{namespace}/checks',
+        headers={
+            'Authorization': f'Key {params["token"]}',
+            'Content-Type': 'application/json',
+        },
+        json=check)
+    r.raise_for_status()
 
 
 def update_check(params, check, namespace='default'):
     name = check["metadata"]["name"]
-    logger.error(f'updating existing check: {name}')
-    # url = random.choice(params['api-base'])
-    # r = requests.post(
-    #     f'{url}/api/core/v2/namespaces/{namespace}/checks/{name}',
-    #     headers={
-    #         'Authorization': f'Key {params["token"]}',
-    #         'Content-Type': 'application/json',
-    #     },
-    #     json=check)
-    # r.raise_for_status()
+    logger.debug(f'updating existing check: {name}')
+    url = random.choice(params['api-base'])
+    r = requests.put(
+        f'{url}/api/core/v2/namespaces/{namespace}/checks/{name}',
+        headers={
+            'Authorization': f'Key {params["token"]}',
+            'Content-Type': 'application/json',
+        },
+        json=check)
+    r.raise_for_status()
 
 
 def delete_check(params, check, namespace='default'):
@@ -56,9 +54,9 @@ def delete_check(params, check, namespace='default'):
         name = check
     else:
         name = check["metadata"]["name"]
-    logger.error(f'deleting unwanted check: {name}')
-    # url = random.choice(params['api-base'])
-    # r = requests.delete(
-    #     f'{url}/api/core/v2/namespaces/{namespace}/checks',
-    #     headers={'Authorization': f'Key {params["token"]}'})
-    # r.raise_for_status()
+    logger.debug(f'deleting unwanted check: {name}')
+    url = random.choice(params['api-base'])
+    r = requests.delete(
+        f'{url}/api/core/v2/namespaces/{namespace}/checks/{name}',
+        headers={'Authorization': f'Key {params["token"]}'})
+    r.raise_for_status()

requirements.txt

 click
 jsonschema
 requests
+
+pytest
+responses

test/conftest.py

+import json
+import os
+import re
+import tempfile
+
+import jsonschema
+import pytest
+import responses
+
+
+def _load_test_data(filename):
+    full_path = os.path.join(
+        os.path.dirname(__file__),
+        'data', filename)
+    with open(full_path) as f:
+        return f.read()
+
+
+@pytest.fixture
+def config():
+    with tempfile.TemporaryDirectory() as state_dir_name:
+        yield {
+            'inventory': [
+                'http://bogus-inventory01.xxx.yyy:12345',
+                'http://bogus-inventory02.xxx.yyy:12345',
+                'http://bogus-inventory03.xxx.yyy:12345'
+            ],
+            'sensu': {
+                'api-base': [
+                    'https://bogus-sensu01.xxx.yyy:12345',
+                    'https://bogus-sensu02.xxx.yyy:12345',
+                    'https://bogus-sensu03.xxx.yyy:12345'
+                ],
+                'token': 'abc-token-blah-blah',
+                'interface-check': {
+                    'script': '/var/lib/sensu/bin/counter2influx.sh',
+                    'measurement': 'counters',
+                    'interval': 300,
+                    'subscriptions': ['interfacecounters'],
+                    'output_metric_handlers': ['influx-db-handler'],
+                    'namespace': 'default',
+                    'round_robin': True,
+                    'command': '{script} {measurement} {community} {hostname} {interface} {ifIndex}',
+                }
+            },
+            'statedir': state_dir_name
+        }
+
+
+@pytest.fixture
+def mocked_sensu():
+
+    saved_sensu_checks = {
+        c['metadata']['name']: c
+        for c in json.loads(_load_test_data('checks.json'))}
+
+    _check_schema = {
+        '$schema': 'http://json-schema.org/draft-07/schema#',
+
+        'type': 'object',
+        'properties': {
+            'command': {'type': 'string'},
+            'interval': {'type': 'integer'},
+            'subscriptions': {'type': 'array', 'items': {'type': 'string'}},
+            'proxy_entity_name': {'type': 'string'},
+            'round_robin': {'type': 'boolean'},
+            'output_metric_format': {'enum': ['influxdb_line']},
+            'output_metric_handlers': {'type': 'array', 'items': {'type': 'string'}},
+            'metadata': {
+                'type': 'object',
+                'properties': {
+                    'name': {'type': 'string'},
+                    'namespace': {'type': 'string'}
+                },
+                'required': ['name', 'namespace'],
+                'additionalProperties': True
+            },
+        },
+        'required': [
+            'command', 'interval', 'subscriptions',
+            'output_metric_format', 'output_metric_handlers',
+            'round_robin', 'metadata'],
+        'additionalProperties': True
+    }
+
+    # mocked api for returning all checks
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*sensu.+/api/core/v2/namespaces/[^\/]+/checks$'),
+        body=json.dumps(list(saved_sensu_checks.items()))
+    )
+
+    def new_check_callback(request):
+        check = json.loads(request.body)
+        jsonschema.validate(check, _check_schema)
+        path_elems = request.path_url.split('/')
+        assert len(path_elems) == 7  # sanity
+        assert path_elems[5] == check['metadata']['namespace']
+        assert check['metadata']['name'] not in saved_sensu_checks
+        saved_sensu_checks[check['metadata']['name']] = check
+        return (201, {}, '')
+
+    # mocked api for creating a check
+    responses.add_callback(
+        method=responses.POST,
+        url=re.compile(r'.*sensu.+/api/core/v2/namespaces/[^\/]+/checks$'),
+        callback=new_check_callback)
+
+    def update_check_callback(request):
+        check = json.loads(request.body)
+        jsonschema.validate(check, _check_schema)
+        path_elems = request.path_url.split('/')
+        assert len(path_elems) == 8  # sanity
+        assert path_elems[5] == check['metadata']['namespace']
+        assert path_elems[-1] == check['metadata']['name']
+        assert check['metadata']['name'] in saved_sensu_checks, \
+            'we only intend to call this method for updating existing checks'
+        saved_sensu_checks[check['metadata']['name']] = check
+        return 201, {}, ''
+
+    # mocked api for updating a check
+    responses.add_callback(
+        method=responses.PUT,
+        url=re.compile(r'.*sensu.+/api/core/v2/namespaces/[^\/]+/checks/[^\/]+$'),
+        callback=update_check_callback)
+
+    def delete_check_callback(request):
+        path_elems = request.path_url.split('/')
+        assert len(path_elems) == 8  # sanity
+        del saved_sensu_checks[path_elems[-1]]
+        return 204, {}, ''
+
+    # mocked api for deleting a check
+    responses.add_callback(
+        method=responses.DELETE,
+        url=re.compile(r'.*sensu.+/api/core/v2/namespaces/[^\/]+/checks/[^\/]+$'),
+        callback=delete_check_callback)
+
+    yield saved_sensu_checks
+
+@pytest.fixture
+def mocked_inventory():
+
+    # mocked api for returning all checks
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*inventory.+/poller/interfaces.*'),
+        body=_load_test_data('interfaces.json'))
+

test/test_sensu_checks.py

+import copy
+import random
+
+import responses
+
+from brian_polling_manager import sensu, inventory, interfaces
+
+
+@responses.activate
+def test_load_checks(config, mocked_sensu):
+    checks = list(sensu.load_all_checks(config['sensu']))
+    assert len(checks) > 0  # test data has checks in it
+
+
+@responses.activate
+def test_check_lifecycle(config, mocked_sensu, mocked_inventory):
+
+    test_interface = random.choice(inventory.load_interfaces(config['inventory']))
+    test_interface['name'] = 'xyz'
+
+    new_check = interfaces._make_check(
+        config['sensu']['interface-check'],
+        test_interface)
+
+    # create the new check
+    check_name = new_check['metadata']['name']
+    assert check_name not in mocked_sensu
+    sensu.create_check(config['sensu'], new_check)
+    assert check_name in mocked_sensu
+
+    # modify interval & update, then verify the correct call was made
+    updated_check = copy.copy(new_check)
+    updated_check['interval'] += 1
+    sensu.update_check(config['sensu'], updated_check)
+    assert mocked_sensu[check_name]['interval'] == updated_check['interval']
+
+    # delete the check and confirm the correct call was made
+    sensu.delete_check(config['sensu'], check_name)
+    assert check_name not in mocked_sensu