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