tmss_test_environment_unittest_setup.py

#!/usr/bin/env python3

# Copyright (C) 2018    ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.    See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.

'''
By importing this helper module in your unittest module you get a TMSSTestDatabaseInstance
which is automatically destroyed at the end of the unittest session.
'''

import logging
logger = logging.getLogger(__name__)

# before we import any django modules the DJANGO_SETTINGS_MODULE, TMSS_LDAPCREDENTIALS and TMSS_DBCREDENTIALS need to be known/set.
# import and start an isolated TMSSTestEnvironment (with fresh database and attached django and ldap server on free ports)
# this automagically sets the required  DJANGO_SETTINGS_MODULE, TMSS_LDAPCREDENTIALS and TMSS_DBCREDENTIALS envvars.
from lofar.sas.tmss.test.test_utils import TMSSTestEnvironment
tmss_test_env = TMSSTestEnvironment()
try:
    tmss_test_env.start()
except:
    tmss_test_env.stop()
    exit(1)

# tell unittest to stop (and automagically cleanup) the test database once all testing is done.
def tearDownModule():
    tmss_test_env.stop()

################################################################################################
# the methods below can be used to to HTTP REST calls to the django server and check the results
################################################################################################

import json
import requests
AUTH = requests.auth.HTTPBasicAuth(tmss_test_env.ldap_server.dbcreds.user, tmss_test_env.ldap_server.dbcreds.password)
BASE_URL = tmss_test_env.django_server.url
OIDC_URL = tmss_test_env.django_server.oidc_url
from lofar.sas.tmss.test.test_utils import assertDataWithUrls
import lofar.sas.tmss.tmss.settings as TMSS_SETTINGS


def _call_API_and_assert_expected_response(test_instance, url, call, data, expected_code, expected_content):
    """
    Call API method on the provided url and assert the expected code is returned and the expected content is in the response content
    :return: response as dict. This either contains the data of an entry or error details. If JSON cannot be parsed, return string.
    """
    if call == 'PUT':
        response = requests.put(url, json=data, auth=AUTH)
    elif call == 'POST':
        response = requests.post(url, json=data, auth=AUTH)
    elif call == 'GET':
        response = requests.get(url, auth=AUTH)
    elif call == 'PATCH':
        response = requests.patch(url, json=data, auth=AUTH)
    elif call == 'DELETE':
        response = requests.delete(url, auth=AUTH)
    else:
        raise ValueError("The provided call '%s' is not a valid API method choice" % call)

    if response.status_code != expected_code:
        logger.error("!!! Unexpected: [%s] - %s %s: %s", test_instance.id(), call, url, response.content.decode('utf-8').strip())
    test_instance.assertEqual(response.status_code, expected_code)

    content = response.content.decode('utf-8')

    from django.db import models

    if response.status_code in range(200, 300) and expected_content is not None:
        r_dict = json.loads(content)
        for key, value in expected_content.items():
            if key not in r_dict.keys():
                logger.error('!!! Missing key: %s in %s', key, r_dict.keys())
            test_instance.assertTrue(key in r_dict.keys())
            if isinstance(value, models.Model):
                value = str(value.pk)
                value = value.replace(' ', '%20')
                test_instance.assertTrue(str(value) in r_dict[key])
            elif type(value) is list:
                test_instance.assertEqual(sorted(value), sorted(r_dict[key]), msg="lists differ for key=%s"%key) # compare lists independent of ordering
            else:
                test_instance.assertEqual(value, r_dict[key])
        return r_dict

    try:
        return json.loads(content)
    except:
        return content


def PUT_and_assert_expected_response(test_instance, url, data, expected_code, expected_content):
    """
    PUT data on url and assert the expected code is returned and the expected content is in the response content
    """
    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'PUT', data, expected_code, expected_content)
    return r_dict


def POST_and_assert_expected_response(test_instance, url, data, expected_code, expected_content):
    """
    POST data on url and assert the expected code is returned and the expected content is in the response content
    :return: response dict
    """
    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'POST', data, expected_code, expected_content)
    return r_dict


def GET_and_assert_equal_expected_code(test_instance, url, expected_code):
    """
    GET from url and assert the expected code is returned and the expected content is in the response content
    """
    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'GET', {}, expected_code, None)
    return r_dict


def GET_and_assert_in_expected_response_result_list(test_instance, url, expected_content, expected_nbr_results):
    """
    GET from url and assert the expected code is returned and the expected content is in the response content
    Use this check when multiple results (list) are returned
    """
    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'GET', {}, 200, None)
    page_size = TMSS_SETTINGS.REST_FRAMEWORK.get('PAGE_SIZE')
    if page_size is not None and expected_nbr_results > page_size:
        logger.warning("Limited result length due to pagination setting (%d)", page_size)
        test_instance.assertEqual(page_size, len(r_dict["results"]))
        test_instance.assertEqual(page_size, len(r_dict["results"]))
        test_instance.assertEqual(expected_nbr_results, r_dict["count"])
        test_instance.assertNotEqual(None, r_dict['next'])
        url_check = False
    else:
        test_instance.assertEqual(expected_nbr_results, len(r_dict["results"]))
        test_instance.assertEqual(r_dict["count"], len(r_dict["results"]))
        test_instance.assertEqual(None, r_dict['next'])
        url_check = True

    for item in r_dict["results"]:
        for key in expected_content.keys():
            test_instance.assertIn(key, item.keys())

    if url_check:
        assertDataWithUrls(test_instance, r_dict['results'][expected_nbr_results-1], expected_content)
    return r_dict


def GET_OK_and_assert_equal_expected_response(test_instance, url, expected_content):
    """
    GET from url and assert the expected code is returned and the expected content is equal the response content
    assertDataWithUrls is already checked in _call_API_and_assert_expected_response
    """
    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'GET', {}, 200, expected_content)
    #     assertDataWithUrls(test_instance, r_dict, expected_content)
    return r_dict


def PATCH_and_assert_expected_response(test_instance, url, data, expected_code, expected_content):
    """
    POST data on url and assert the provided values have changed based on the server response.
    :return: url for new item
    """
    r_dict = _call_API_and_assert_expected_response(test_instance, url, 'PATCH', data, expected_code, expected_content)
    return r_dict


def DELETE_and_assert_gone(test_instance, url):
    """
    DELETE item at provided url and assert that the request was accepted by the server
    :return: url for new item
    """
    response = requests.delete(url, auth=AUTH)
    if response.status_code != 204:
        logger.error("!!! Unexpected: [%s] - %s %s: %s", test_instance.id(), 'DELETE', url, response.content)
    test_instance.assertEqual(response.status_code, 204)

    response = requests.get(url, auth=AUTH)
    if response.status_code != 404:
        logger.error("!!! Unexpected: [%s] - %s %s: %s", test_instance.id(), 'GET', url, response.content)
    test_instance.assertEqual(response.status_code, 404)