add a bunch of tests

test/test_grafana_dashboard.py

+
+import json
+import responses
+from brian_dashboard_manager.grafana import dashboard, provision
+from brian_dashboard_manager.grafana.utils.request import TokenRequest
+
+
+@responses.activate
+def test_get_dashboard(data_config):
+
+    UID = 1
+
+    request = TokenRequest(**data_config, token='test')
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL +
+        f'api/dashboards/uid/{UID}',
+        callback=lambda f: (
+            404,
+            {},
+            ''))
+
+    data = dashboard._get_dashboard(request, UID)
+    assert data is None
+
+    responses.add_callback(method=responses.GET,
+                           url=request.BASE_URL +
+                           f'api/dashboards/uid/{UID+1}',
+                           callback=lambda f: (200,
+                                               {},
+                                               json.dumps({"uid": 1})))
+
+    data = dashboard._get_dashboard(request, UID + 1)
+    assert data['uid'] == 1
+
+
+@responses.activate
+def test_delete_dashboards(data_config):
+    UID = 1
+    dashboards = [{'uid': UID}]
+
+    request = TokenRequest(**data_config, token='test')
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL +
+        f'api/dashboards/uid/{UID}',
+        callback=lambda f: (
+            200,
+            {},
+            json.dumps(
+                dashboards[0])))
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL +
+        'api/search',
+        callback=lambda f: (
+            200,
+            {},
+            json.dumps(dashboards)))
+
+    def delete_callback(request):
+        uid = request.path_url.split('/')[-1]
+        assert int(uid) == UID
+        return 200, {}, json.dumps({'message': 'Dashboard has been deleted.'})
+
+    responses.add_callback(
+        method=responses.DELETE,
+        url=request.BASE_URL +
+        f'api/dashboards/uid/{UID}',
+        callback=delete_callback)
+
+    data = dashboard.delete_dashboards(request)
+    assert data is True
+
+    responses.add_callback(
+        method=responses.DELETE,
+        url=request.BASE_URL +
+        f'api/dashboards/uid/{UID+1}',
+        callback=lambda f: (
+            400,
+            {},
+            ''))
+
+    data = dashboard._delete_dashboard(request, UID + 1)
+    assert data is None
+
+
+@responses.activate
+def test_search_dashboard(data_config):
+    UID = 1
+    TITLE = 'testdashboard'
+    dashboards = [{'uid': UID, 'title': TITLE}]
+
+    request = TokenRequest(**data_config, token='test')
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL +
+        'api/search',
+        callback=lambda f: (
+            200,
+            {},
+            json.dumps(dashboards)))
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL +
+        f'api/dashboards/uid/{UID}',
+        callback=lambda f: (
+            200,
+            {},
+            json.dumps(
+                dashboards[0])))
+
+    data = dashboard._search_dashboard(
+        request, {'title': dashboards[0]['title']})
+    assert data['uid'] == UID
+
+    data = dashboard._search_dashboard(request, {'title': 'DoesNotExist'})
+    assert data is None
+
+
+@responses.activate
+def test_search_dashboard_error(data_config):
+    request = TokenRequest(**data_config, token='test')
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + 'api/search', callback=lambda f: (400, {}, ''))
+
+    data = dashboard._search_dashboard(request, {'title': 'DoesNotExist'})
+    assert data is None
+
+
+@responses.activate
+def test_create_dashboard(data_config):
+    UID = 1
+    ID = 1
+    VERSION = 1
+    TITLE = 'testdashboard'
+    dashboard = {'id': ID, 'uid': UID, 'title': TITLE, 'version': VERSION}
+    request = TokenRequest(**data_config, token='test')
+
+    responses.add_callback(method=responses.GET,
+                           url=request.BASE_URL + f'api/dashboards/uid/{UID}',
+                           callback=lambda f: (200,
+                                               {},
+                                               json.dumps({'dashboard': dashboard})))
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + 'api/search', callback=lambda f: (400, {}, ''))
+
+    def post_callback(request):
+        body = json.loads(request.body)
+        return 200, {}, json.dumps(body['dashboard'])
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/dashboards/db', callback=post_callback)
+
+    data = provision.create_dashboard(request, dashboard)
+    assert data == dashboard
+
+
+@responses.activate
+def test_create_dashboard_no_uid_error(data_config):
+    ID = 1
+    VERSION = 1
+    TITLE = 'testdashboard'
+    dashboard = {'id': ID, 'title': TITLE, 'version': VERSION}
+    request = TokenRequest(**data_config, token='test')
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + 'api/search', callback=lambda f: (400, {}, ''))
+
+    def post_callback(request):
+        body = json.loads(request.body)
+        # if a dashboard doesn't have an UID, the ID should not be sent to
+        # grafana.
+        assert 'id' not in body['dashboard']
+
+        # have already tested a successful response, respond with error here.
+        return 400, {}, ''
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/dashboards/db', callback=post_callback)
+
+    data = provision.create_dashboard(request, dashboard)
+    assert data is None

test/test_grafana_datasource.py

+import json
+import responses
+from brian_dashboard_manager.grafana import datasource, provision
+from brian_dashboard_manager.grafana.utils.request import AdminRequest
+
+
+@responses.activate
+def test_get_datasources(data_config):
+
+    BODY = []
+
+    request = AdminRequest(**data_config)
+
+    responses.add(
+        method=responses.GET,
+        url=request.BASE_URL + 'api/datasources', json=BODY)
+
+    data = datasource.get_datasources(request)
+    assert data == BODY
+
+
+@responses.activate
+def test_get_missing_datasource_definitions(data_config):
+    # this only retrieves data from the filesystem and checks against
+    # what's configured in grafana.. just make sure it fetches datasources
+
+    request = AdminRequest(**data_config)
+
+    responses.add(method=responses.GET, url=request.BASE_URL +
+                  'api/datasources', json=[])
+
+    # it returns a generator, so iterate :)
+    for data in provision.get_missing_datasource_definitions(
+            request, '/tmp/dirthatreallyshouldnotexistsousealonganduniquestring'):
+        pass
+
+
+def test_datasource_provisioned():
+    provisioned = datasource._datasource_provisioned({}, [])
+    assert provisioned
+
+    provisioned = datasource._datasource_provisioned({'id': 1}, [])
+    assert provisioned is False
+
+    provisioned = datasource._datasource_provisioned({'id': 1, "name": 'testcasetwo'},
+                                                     [{'id': -1, 'name': 'testcaseone'},
+                                                      {'id': 1, 'name': 'testcaseone'}])
+    assert provisioned is False
+
+    provisioned = datasource._datasource_provisioned({'id': 1},
+                                                     [{'id': -1, 'name': 'testcaseone'},
+                                                      {'id': 1, 'name': 'testcasetwo'}])
+    assert provisioned
+
+    provisioned = datasource._datasource_provisioned({'id': 2, "name": 'testcasetwo'},
+                                                     [{'id': -1, 'name': 'testcaseone'},
+                                                      {'id': 1, 'name': 'testcaseone'},
+                                                      {'id': 2, 'name': 'testcasetwo'}])
+    assert provisioned
+
+
+@responses.activate
+def test_create_prod_datasource(data_config):
+    ORG_ID = 1
+
+    BODY = {
+        "name": "brian-influx-datasource",
+        "type": "influxdb",
+        "access": "proxy",
+        "url": "http://test-brian-datasource.geant.org:8086",
+        "database": "test-db",
+        "basicAuth": False,
+        "isDefault": True,
+        "readOnly": False
+    }
+
+    request = AdminRequest(**data_config)
+
+    def post_callback(request):
+        body = json.loads(request.body)
+        # we are testing provisioning logic to prod, so the URL needs test ->
+        # prod translation.
+        assert body['url'] == 'http://prod-brian-datasource.geant.org:8086'
+        result = {
+            'datasource': {
+                'id': 1,
+                'orgId': ORG_ID,
+                'type': 'graphite',
+                'typeLogoUrl': '',
+                'password': '',
+                'user': '',
+                'basicAuthUser': '',
+                'basicAuthPassword': '',
+                'withCredentials': False,
+                'jsonData': {},
+                'secureJsonFields': {},
+                'version': 1
+            },
+            'id': 1,
+            'message': 'Datasource added',
+            'name': body['name']
+        }
+        result['datasource'].update(body)
+        return 200, {}, json.dumps(result)
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/datasources',
+        callback=post_callback)
+
+    data = provision.create_datasource(request, BODY, environment='prod')
+
+    assert data is not None
+
+
+@responses.activate
+def test_create_prod_datasource_fails(data_config):
+    BODY = {
+        "name": "brian-influx-datasource",
+        "type": "influxdb",
+        "access": "proxy",
+        "url": "http://test-brian-datasource.geant.org:8086",
+        "database": "test-db",
+        "basicAuth": False,
+        "isDefault": True,
+        "readOnly": False
+    }
+
+    request = AdminRequest(**data_config)
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/datasources',
+        callback=lambda f: (400, {}, ''))
+
+    data = provision.create_datasource(request, BODY, environment='prod')
+
+    # if an error occured when provisioning a datasource, we log the response
+    # but return None
+    assert data is None

test/test_grafana_organization.py

+import json
+import responses
+from datetime import datetime, timedelta
+from brian_dashboard_manager.grafana import provision
+from brian_dashboard_manager.grafana.utils.request import AdminRequest
+
+
+@responses.activate
+def test_get_organizations(data_config):
+    request = AdminRequest(**data_config)
+
+    responses.add(method=responses.GET,
+                  url=request.BASE_URL + 'api/orgs',
+                  json=[{'id': 91,
+                         'name': 'Testorg1'},
+                        {'id': 92,
+                         'name': 'GÉANT Testorg2'},
+                        {'id': 93,
+                         'name': 'NRENsTestorg3'},
+                        {'id': 94,
+                         'name': 'General Public'}])
+
+    data = provision.get_organizations(request)
+    assert data is not None
+
+
+@responses.activate
+def test_create_organization(data_config):
+    ORG_NAME = 'fakeorg123'
+
+    def post_callback(request):
+        body = json.loads(request.body)
+        assert body['name'] == ORG_NAME
+        return 200, {}, json.dumps(
+            {'orgId': 1, 'message': 'Organization created'})
+
+    request = AdminRequest(**data_config)
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/orgs',
+        callback=post_callback)
+
+    data = provision.create_organization(request, ORG_NAME)
+    assert data is not None
+
+
+@responses.activate
+def test_delete_expired_api_tokens(data_config):
+    ORG_ID = 1
+    KEY_ID = 1
+
+    def post_callback(request):
+        assert request.params['includeExpired'] == 'True'
+        time = (datetime.now() - timedelta(seconds=60)
+                ).strftime('%Y-%m-%dT%H:%M:%SZ')
+        return 200, {}, json.dumps([{'expiration': time, 'id': KEY_ID}])
+
+    request = AdminRequest(**data_config)
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + 'api/auth/keys',
+        callback=post_callback)
+
+    responses.add(
+        method=responses.POST,
+        url=request.BASE_URL +
+        f'api/user/using/{ORG_ID}',
+        json={
+            "message": "Active organization changed"})
+    responses.add(
+        method=responses.DELETE,
+        url=request.BASE_URL +
+        f'api/auth/keys/{KEY_ID}',
+        json={
+            "message": "API key deleted"})
+
+    provision.delete_expired_api_tokens(request, ORG_ID)
+
+
+@responses.activate
+def test_create_api_token(data_config):
+    ORG_ID = 1
+    TOKEN_ID = 1
+    BODY = {
+        'name': 'test-token',
+        'role': 'Admin',
+        'secondsToLive': 3600
+    }
+
+    request = AdminRequest(**data_config)
+
+    def post_callback(request):
+        body = json.loads(request.body)
+        assert body == BODY
+        return 200, {}, json.dumps({'id': TOKEN_ID})
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/auth/keys',
+        callback=post_callback)
+
+    responses.add(
+        method=responses.POST,
+        url=request.BASE_URL +
+        f'api/user/using/{ORG_ID}',
+        json={
+            "message": "Active organization changed"})
+
+    data = provision.create_api_token(request, ORG_ID, BODY)
+    assert data['id'] == TOKEN_ID

test/test_grafana_request.py

+import pytest
+import responses
+import requests
+import json
+from brian_dashboard_manager.grafana.utils.request import AdminRequest, TokenRequest
+
+
+def test_admin_request(data_config):
+    ENDPOINT = 'test/url/endpoint'
+    request = AdminRequest(**data_config)
+    assert request.BASE_URL == 'http://{admin_username}:{admin_password}@{hostname}:{grafana_port}/'.format(
+        **data_config)
+    assert request.username == data_config['admin_username']
+
+    def get_callback(request):
+        assert request.path_url[1:] == ENDPOINT
+        return 200, {}, ''
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + ENDPOINT,
+        callback=get_callback)
+
+    request.get(ENDPOINT)
+
+
+@responses.activate
+def test_token_request(data_config):
+    TOKEN = '123'
+    ENDPOINT = 'test/url/endpoint'
+    request = TokenRequest(**data_config, token=TOKEN)
+    assert request.BASE_URL == 'http://{hostname}:{grafana_port}/'.format(
+        **data_config)
+    assert request.token == TOKEN
+
+    def get_callback(request):
+        assert request.path_url[1:] == ENDPOINT
+        assert TOKEN in request.headers['authorization']
+        return 200, {}, ''
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + ENDPOINT,
+        callback=get_callback)
+
+    request.get(ENDPOINT)
+
+# document unimplemented handling of server-side errors
+
+
+@pytest.mark.xfail(raises=requests.exceptions.HTTPError)
+@responses.activate
+def test_POST_fails(data_config):
+    ORG_NAME = 'fakeorg123'
+
+    def post_callback(request):
+        body = json.loads(request.body)
+        assert body['name'] == ORG_NAME
+        return 500, {}, ''
+
+    request = AdminRequest(**data_config)
+
+    responses.add_callback(
+        method=responses.POST,
+        url=request.BASE_URL + 'api/orgs',
+        callback=post_callback)
+
+    request.post('api/orgs', json={'name': ORG_NAME})
+
+
+@pytest.mark.xfail(raises=requests.exceptions.HTTPError)
+@responses.activate
+def test_GET_fails(data_config):
+    ORG_NAME = 'fakeorg123'
+
+    def get_callback(request):
+        return 500, {}, ''
+
+    request = AdminRequest(**data_config)
+
+    responses.add_callback(
+        method=responses.GET,
+        url=request.BASE_URL + 'api/orgs',
+        callback=get_callback)
+
+    request.get('api/orgs', json={'name': ORG_NAME})
+
+
+@pytest.mark.xfail(raises=requests.exceptions.HTTPError)
+@responses.activate
+def test_DELETE_fails(data_config):
+    def delete_callback(request):
+        return 500, {}, ''
+
+    request = AdminRequest(**data_config)
+
+    responses.add_callback(
+        method=responses.DELETE,
+        url=request.BASE_URL + 'api/orgs/1',
+        callback=delete_callback)
+
+    request.delete('api/orgs/1')

test/test_update.py

+import responses
+import json
+
+DEFAULT_REQUEST_HEADERS = {
+    "Content-type": "application/json",
+    "Accept": ["application/json"]
+}
+
+
+@responses.activate
+def test_provision(data_config, mocker, client):
+
+    TEST_DATASOURCE = [{
+        "name": "brian-influx-datasource",
+        "type": "influxdb",
+        "access": "proxy",
+        "url": "http://test-brian-datasource.geant.org:8086",
+        "database": "test-db",
+        "basicAuth": False,
+        "isDefault": True,
+        "readOnly": False
+    }]
+
+    PROVISIONED_ORGANIZATION = {
+        'name': data_config['organizations'][0],
+        'id': 0
+    }
+
+    EXISTING_ORGS = [{'name': org, 'id': i + 1}
+                     for i, org in enumerate(data_config['organizations'][1:])]
+
+    _mocked_get_organizations = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.get_organizations')
+    # all organizations are provisioned except the first one.
+    _mocked_get_organizations.return_value = EXISTING_ORGS.copy()
+
+    _mocked_create_organization = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.create_organization')
+
+    # spoof creating first organization
+    _mocked_create_organization.return_value = PROVISIONED_ORGANIZATION
+
+    _mocked_delete_expired_api_tokens = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.delete_expired_api_tokens')
+    # we dont care about this, , tested separately
+    _mocked_delete_expired_api_tokens.return_value = None
+
+    _mocked_create_api_token = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.create_api_token')
+    _mocked_create_api_token.return_value = {
+        'key': 'testtoken', 'id': 0}  # api token
+
+    _mocked_get_missing_datasource_definitions = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.get_missing_datasource_definitions')
+    _mocked_get_missing_datasource_definitions.return_value = TEST_DATASOURCE  # test datasource
+
+    _mocked_create_datasource = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.create_datasource')
+    # we dont care about this, just mark it created
+    _mocked_create_datasource.return_value = True
+
+    _mocked_get_dashboard_definitions = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.get_dashboard_definitions')
+
+    UID = 1
+    ID = 1
+    VERSION = 1
+    TITLE = 'testdashboard'
+    dashboard = {'id': ID, 'uid': UID, 'title': TITLE, 'version': VERSION}
+    _mocked_get_dashboard_definitions.return_value = [
+        dashboard  # test dashboard
+    ]
+
+    _mocked_create_dashboard = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.create_dashboard')
+    # we dont care about this, just mark it created
+    # we dont care about this, tested separately
+    _mocked_create_dashboard.return_value = None
+
+    _mocked_delete_api_token = mocker.patch(
+        'brian_dashboard_manager.grafana.provision.delete_api_token')
+    # we dont care about this, tested separately
+    _mocked_delete_api_token.return_value = None
+    response = client.get('/update/', headers=DEFAULT_REQUEST_HEADERS)
+    assert response.status_code == 200
+    data = json.loads(response.data.decode('utf-8'))['data']
+    assert data == EXISTING_ORGS + [PROVISIONED_ORGANIZATION]