diff --git a/test/test_grafana_dashboard.py b/test/test_grafana_dashboard.py
new file mode 100644
index 0000000000000000000000000000000000000000..ed5eee8fca16a8dde0f221ef48aa5b7261dd7541
--- /dev/null
+++ b/test/test_grafana_dashboard.py
@@ -0,0 +1,195 @@
+
+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
diff --git a/test/test_grafana_datasource.py b/test/test_grafana_datasource.py
new file mode 100644
index 0000000000000000000000000000000000000000..f614eec8060396261c71ecba80e21f284f0fa6d2
--- /dev/null
+++ b/test/test_grafana_datasource.py
@@ -0,0 +1,140 @@
+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
diff --git a/test/test_grafana_organization.py b/test/test_grafana_organization.py
new file mode 100644
index 0000000000000000000000000000000000000000..4528594eec9c679d1e43193a0124144b8593f83d
--- /dev/null
+++ b/test/test_grafana_organization.py
@@ -0,0 +1,111 @@
+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
diff --git a/test/test_grafana_request.py b/test/test_grafana_request.py
new file mode 100644
index 0000000000000000000000000000000000000000..f4787827bc72eb058bc770881389a0d1aa8b170a
--- /dev/null
+++ b/test/test_grafana_request.py
@@ -0,0 +1,102 @@
+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')
diff --git a/test/test_update.py b/test/test_update.py
new file mode 100644
index 0000000000000000000000000000000000000000..1ffcf9d887c29fce2780f3b4057a67c7332b7b6f
--- /dev/null
+++ b/test/test_update.py
@@ -0,0 +1,87 @@
+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]