ody_migr_mmth_test.py

#!/usr/bin/env python
"""Test Mammoth connection for all data categories and CU options.

This test module contains test cases for testing connection
to Mammoth and validating response for all CUs (eg. SCRUBCU)
and all data categories (eg. settings, members, etc.).

Note: Each test method runs the same test case on multiple
CUs and multiple applicable data categories.

Usage Example:
      $ export ODY_MIGR_SECRET_KEY=**** &&\
      python3 opt/odyssey/tools/bin/ody_migr_mmth_test.py
"""


import unittest
import os

from ody_migr_mmth_endpoint import MammothMigration
from ody_migr_utils import generate_hash, get_valid_json
from ody_migr_config import (ENV_MIGR_SECRET_KEY,
                             TEST_USERNAME,
                             TEST_PASSWORD,
                             RESPONSE_KEYS,
                             RESPONSE_HIST_KEYS,
                             EXPECTED_DATA_KEYS,
                             EXPECTED_HASH_DICT,
                             DEFAULT_ERROR_STR,
                             TESTING_SERVER,
                             SUPPORTED_CU_TEST,
                             DATA_CHOICES_FOR_TEST,
                             DATA_OPT_SETTINGS,
                             DATA_OPT_MEMDATA,
                             DATA_OPT_CREATE_HIST,
                             DATA_OPT_GET_HIST,
                             KEY_MEMBERS)
import logging
# disable logging during testing
logging.disable(logging.ERROR)

# verify that necessary env variables are exported
assert os.getenv(ENV_MIGR_SECRET_KEY) is not None,\
    "Environment variable: `{}` is not set!".format(ENV_MIGR_SECRET_KEY)


class TestMammothMigration(unittest.TestCase):
    """Test cases for Mammoth connection and received data response."""

    def setUp(self):
        """Set up test environment.

        Prepare essential Mammoth connectors to test connections and
        data responses.
        """

        # Each of the following list stores objects and data for
        # each combination of data category and supported CU for testing
        # list of MammothMigration object
        self.list_of_migrators = []
        # list of expected pre-generated hashes
        self.list_of_expected_hashes = []
        # list of expected urls
        self.list_of_expected_urls = []
        # list of http post parameters for memdata
        self.list_of_http_requests = []

        # setting up
        for data_category in DATA_CHOICES_FOR_TEST:
            for cu in SUPPORTED_CU_TEST:
                memdata_params = ""

                if data_category == DATA_OPT_MEMDATA:
                    if cu == "SCRUBCU":
                        memdata_params = {
                            KEY_MEMBERS: get_valid_json([6680, 779999])}

                    elif (cu == "CRUISECU" and
                            data_category == DATA_OPT_MEMDATA):
                        memdata_params = {KEY_MEMBERS: get_valid_json([25198])}

                    this_migrator = MammothMigration(
                        cu,
                        data_category,
                        TESTING_SERVER,
                        TEST_USERNAME,
                        TEST_PASSWORD,
                        params=memdata_params
                    )

                elif data_category == DATA_OPT_CREATE_HIST:
                    this_migrator = MammothMigration(
                        cu,
                        data_category,
                        TESTING_SERVER,
                        TEST_USERNAME,
                        TEST_PASSWORD,
                        params={"restart": "Y"}
                    )

                else:
                    this_migrator = MammothMigration(
                        cu,
                        data_category,
                        TESTING_SERVER,
                        TEST_USERNAME,
                        TEST_PASSWORD,
                    )

                self.list_of_http_requests.append(memdata_params)

                self.list_of_migrators.append(this_migrator)

                this_hash = EXPECTED_HASH_DICT[data_category][cu]
                self.list_of_expected_hashes.append(this_hash)

                this_url = ("https://{}.homecu.net/hcuadm/mOdysseyMigrExp.prg"
                            "?cu={}&passphrase={}&action={}").format(
                    this_migrator.server,
                    this_migrator.cu,
                    this_hash,
                    this_migrator.data_category)
                self.list_of_expected_urls.append(this_url)

        self.default_err_msg = DEFAULT_ERROR_STR
        self.wrong_username = "sharma_ashish"
        self.invalid_secret = "some!CrappyS3cR3TF0rT3$tin9"

        self.expected_response_keys = RESPONSE_KEYS
        self.expected_response_hist_keys = RESPONSE_HIST_KEYS
        self.extected_data_keys = EXPECTED_DATA_KEYS

    def test_hash_generation(self):
        """Test authentication code generation.

        Obtained authentication code must match the expected code
        generated for valid cu and action when generated with correct
        SECRET KEY"""
        for ind, this_expected_hash in enumerate(self.list_of_expected_hashes):
            this_migrator = self.list_of_migrators[ind]
            obtained_hash = generate_hash(
                this_migrator.data_category,
                this_migrator.cu,
                os.environ.get(ENV_MIGR_SECRET_KEY)
            )
            self.assertEqual(this_expected_hash, obtained_hash)

    def test_request_url_correctness(self):
        """Test url destination correctness given cu and data_category."""
        for ind, this_expected_url in enumerate(self.list_of_expected_urls):
            this_migrator = self.list_of_migrators[ind]
            obtained_url = this_migrator._dest_url()
            self.assertEqual(this_expected_url, obtained_url)

    def test_invalid_authentication(self):
        """Invalid credential should not allow access the Mammoth resources.

        Expect SystemExit exception raised when tried to authenticate
        with incorrect credentials
        """

        for this_migrator in self.list_of_migrators:
            this_migrator.username = self.wrong_username
            with self.assertRaises(SystemExit):
                this_migrator._initiate_http_request()

    def test_response_with_error(self):
        """Test connection with invalid secret key."""

        for this_migrator in self.list_of_migrators:
            original_secret = os.environ.get(ENV_MIGR_SECRET_KEY)
            os.environ[ENV_MIGR_SECRET_KEY] = self.invalid_secret
            # populates response dictionary after successful connection
            this_migrator._initiate_http_request()
            # but returns with error response message: Invalid passphrase
            with self.assertRaises((SystemExit)):
                this_migrator._validate_response()
            os.environ[ENV_MIGR_SECRET_KEY] = original_secret

    def test_file_not_found_memhist(self):
        """Test file not exist for gethistoryfile executed before creation.

        SOCU, CU that we do not use for testing and is expected not to have
        memhist.gz and memhist lock file in /home/SOCU/tmp directory in
        Mammoth. If such files exist, delete first.
        """
        this_migrator = MammothMigration(
            "SOCU",
            DATA_OPT_GET_HIST,
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD,
        )

        this_migrator._initiate_http_request()
        # must raise SystemExit on error (file doesn't exist)
        with self.assertRaises(SystemExit):
            this_migrator._validate_response()

    def test_successful_connection_validate_data(self):
        """Test successful data (all categories) pull from Mammoth."""

        for ind, this_migrator in enumerate(self.list_of_migrators):
            # we only have few cases for memdata and memhist to test
            if (this_migrator.data_category == DATA_OPT_MEMDATA and
                    self.list_of_http_requests[ind] == ""):
                continue

            this_migrator._initiate_http_request()

            resp = this_migrator.response

            if this_migrator.data_category == DATA_OPT_CREATE_HIST:
                self.assertIsInstance(resp, dict)

                # check expected response keys for createhistoryfile exist
                for expected_resp_key in self.expected_response_hist_keys:
                    self.assertIn(expected_resp_key, resp.keys())

                # verify usual conditions of error and file response values
                chist_error = resp["error"]
                data_fname = resp["file"]
                assert (data_fname == "" and chist_error != "") \
                    or (data_fname != "" and chist_error == "")

            elif this_migrator.data_category == DATA_OPT_GET_HIST:
                self.assertIsInstance(resp.content, bytes)
                # history response is a stream of binary content, so will
                # raise ValueError if tried to decode into json
                with self.assertRaises((ValueError)):
                    resp.json()

            # all other data category
            else:
                self.assertIsInstance(resp, dict)
                self.assertEqual(resp["error"], self.default_err_msg)

                # check that all the expected response keys exist
                for expected_resp_key in self.expected_response_keys:
                    self.assertIn(expected_resp_key, resp.keys())

                data_dict = resp["data"]

                # data keys membership check for memdata
                if this_migrator.data_category == DATA_OPT_MEMDATA:
                    memdata_true_dict_struct = self.extected_data_keys[
                        this_migrator.data_category]

                    self.assertEqual(len(data_dict.keys()), 1)
                    self.assertEqual(list(data_dict.keys())[0], "account")
                    # verify data keys exist for all the member accounts
                    for member_response in data_dict["account"].keys():
                        self.assertEqual(sorted(
                            data_dict["account"][member_response].keys()),
                            sorted(memdata_true_dict_struct))

                # simple membership assertion for other data categories
                else:
                    for expected_data_key in self.extected_data_keys[
                            this_migrator.data_category]:
                        self.assertIn(expected_data_key, data_dict.keys())

                # validate some obvious data
                if this_migrator.data_category == DATA_OPT_SETTINGS:
                    # validate exported data for `cuadmin`
                    self.assertEqual(len(data_dict["cuadmin"]), 1)
                    self.assertEqual(data_dict["cuadmin"][
                        0]["cu"].strip(), this_migrator.cu)

    def tearDown(self):
        pass


if __name__ == "__main__":
    unittest.main()