path: root/moon_manager/tests/func_tests/features/steps/meta_data.py

# Software Name: MOON

# Version: 5.4

# SPDX-FileCopyrightText: Copyright (c) 2018-2020 Orange and its contributors
# SPDX-License-Identifier: Apache-2.0

# This software is distributed under the 'Apache License 2.0',
# the text of which is available at 'http://www.apache.org/licenses/LICENSE-2.0.txt'
# or see the "LICENSE" file for more details.


from behave import *
from Static_Variables import GeneralVariables
from astropy.table import Table, Column
from common_functions import *
import requests
import json
import logging

apis_urls = GeneralVariables()
api_subjectcategory = {'name': "", 'description': ""}
api_objectcategory = {'name': "", 'description': ""}
api_actioncategory = {'name': "", 'description': ""}

logger = logging.getLogger(__name__)


# Step Definition Implementation:
# 1) Get all the existing subject meta data in the system
# 2) Loop by id and delete them
@Given('the system has no subject categories')
def step_impl(context):
    logger.info("Given the system has no subject categories")
    api_responseflag = {'value': False}
    headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

    response = requests.get(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI, headers=apis_urls.auth_headers)
    if len(response.json()[apis_urls.metadatasubjectcategoryAPI]) != 0:
        for ids in dict(response.json()[apis_urls.metadatasubjectcategoryAPI]).keys():
            response = requests.delete(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI + "/" + ids,
                                       headers=headers)

# Step Definition Implementation:
# 1) Get all the existing action meta data in the system
# 2) Loop by id and delete them
@Given('the system has no action categories')
def step_impl(context):
    logger.info("Given the system has no action categories")
    api_responseflag = {'value': False}
    headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

    response = requests.get(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI, headers=apis_urls.auth_headers)
    if len(response.json()[apis_urls.metadataactioncategoryAPI]) != 0:
        for ids in dict(response.json()[apis_urls.metadataactioncategoryAPI]).keys():
            response = requests.delete(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI + "/" + ids,
                                       headers=headers)


# Step Definition Implementation:
# 1) Get all the existing object meta data in the system
# 2) Loop by id and delete them
@Given('the system has no object categories')
def step_impl(context):
    logger.info("Given the system has no object categories")
    api_responseflag = {'value': False}
    headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

    response = requests.get(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI, headers=apis_urls.auth_headers)
    if len(response.json()[apis_urls.metadataobjectcategoryAPI]) != 0:
        for ids in dict(response.json()[apis_urls.metadataobjectcategoryAPI]).keys():
            response = requests.delete(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI + "/" + ids,
                                       headers=headers)



# Step Definition Implementation:
# 1) Insert subject meta data using the post request
@Given('the following meta data subject category exists')
def step_impl(context):
    logger.info("Given the following meta data subject category exists")
    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

        data = {
            'name': row["subjectmetadataname"],
            'description': row["subjectmetadatadescription"],
        }
        logger.info(
            "subject category name: '" + row["subjectmetadataname"] + "' and subject category description: '" + row[
                "subjectmetadatadescription"] + "'")
        response = requests.post(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI, headers=headers,
                                 data=json.dumps(data))

# Step Definition Implementation:
# 1) Insert object meta data using the post request
@Given('the following meta data object category exists')
def step_impl(context):
    logger.info("Given the following meta data object category exists")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

        data = {
            'name': row["objectmetadataname"],
            'description': row["objectmetadatadescription"],
        }
        logger.info(
            "object category name: '" + row["objectmetadataname"] + "' and object category description: '" + row[
                "objectmetadatadescription"] + "'")
        response = requests.post(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI, headers=headers,
                                 data=json.dumps(data))

# Step Definition Implementation:
# 1) Insert action meta data using the post request
@Given('the following meta data action category exists')
def step_impl(context):
    logger.info("Given the following meta data action category exists")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}
        data = {
            'name': row["actionmetadataname"],
            'description': row["actionmetadatadescription"],
        }
        logger.info(
            "action category name: '" + row["actionmetadataname"] + "' and action category description: '" + row[
                "actionmetadatadescription"] + "'")
        response = requests.post(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI, headers=headers,
                                 data=json.dumps(data))

# Step Definition Implementation:
# 1) Add subject meta data using the post request
# 2) If the request code was 200 set the api response flag to true else false
@When('the user sets to add the following meta data subject category')
def step_impl(context):
    logger.info("When the user sets to add the following meta data subject category")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}
        data = {
            'name': row["subjectmetadataname"],
            'description': row["subjectmetadatadescription"],
        }
        logger.info(
            "subject category name: '" + row["subjectmetadataname"] + "' and subject category description: '" + row[
                "subjectmetadatadescription"] + "'")

        response = requests.post(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI, headers=headers,
                                 data=json.dumps(data))

    if response.status_code == 200:
        GeneralVariables.api_responseflag['value'] = 'True'
    else:
        GeneralVariables.api_responseflag['value'] = 'False'


# Step Definition Implementation:
# 1) Get all the subject meta data by get request
# 2) Loop by ids and search for the matching subject meta data by name and delete it
# 3) If the request code was 200 set the api response flag to true else false
@When('the user sets to delete the following meta data subject category')
def step_impl(context):
    logger.info("When the user sets to delete the following meta data subject category")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}
        logger.info("subject category name: '" + row["subjectmetadataname"] + "'")

        response = requests.get(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI,
                                headers=apis_urls.auth_headers)
        for ids in dict(response.json()[apis_urls.metadatasubjectcategoryAPI]).keys():
            if (response.json()[apis_urls.metadatasubjectcategoryAPI][ids]['name'] == row["subjectmetadataname"]):
                response = requests.delete(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI + "/" + ids,
                                           headers=headers)
    if response.status_code == 200:
        GeneralVariables.api_responseflag['value'] = 'True'
    else:
        GeneralVariables.api_responseflag['value'] = 'False'

# Step Definition Implementation:
# 1) Add object meta data using the post request
# 2) If the request code was 200 set the api response flag to true else false
@When('the user sets to add the following meta data object category')
def step_impl(context):
    logger.info("When the user sets to add the following meta data object category")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}
        data = {
            'name': row["objectmetadataname"],
            'description': row["objectmetadatadescription"],
        }
        logger.info(
            "object category Name: '" + row["objectmetadataname"] + "' and object category description: '" + row[
                "objectmetadatadescription"] + "''")
        response = requests.post(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI, headers=headers,
                                 data=json.dumps(data))
    if response.status_code == 200:
        GeneralVariables.api_responseflag['value'] = 'True'
    else:
        GeneralVariables.api_responseflag['value'] = 'False'

# Step Definition Implementation:
# 1) Get all the object meta data by get request
# 2) Loop by ids and search for the matching object meta data by name and delete it
# 3) If the request code was 200 set the api response flag to true else false
@When('the user sets to delete the following meta data object category')
def step_impl(context):
    logger.info("When the user sets to delete the following meta data object category")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}
        logger.info("object category name: '" + row["objectmetadataname"] + "'")

        response = requests.get(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI,
                                headers=apis_urls.auth_headers)
        for ids in dict(response.json()[apis_urls.metadataobjectcategoryAPI]).keys():
            if (response.json()[apis_urls.metadataobjectcategoryAPI][ids]['name'] == row["objectmetadataname"]):
                response = requests.delete(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI + "/" + ids,
                                           headers=headers)
    if response.status_code == 200:
        GeneralVariables.api_responseflag['value'] = 'True'
    else:
        GeneralVariables.api_responseflag['value'] = 'False'

# Step Definition Implementation:
# 1) Add subject meta data using the post request
# 2) If the request code was 200 set the api response flag to true else false
@When('the user sets to add the following meta data action category')
def step_impl(context):
    logger.info("When the user sets to add the following meta data action category")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

        data = {
            'name': row["actionmetadataname"],
            'description': row["actionmetadatadescription"],
        }
        logger.info(
            "action category name: '" + row["actionmetadataname"] + "' and action category description: '" + row[
                "actionmetadatadescription"] + "'")

        response = requests.post(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI, headers=headers,
                                 data=json.dumps(data))
    if response.status_code == 200:
        GeneralVariables.api_responseflag['value'] = 'True'
    else:
        GeneralVariables.api_responseflag['value'] = 'False'

# Step Definition Implementation:
# 1) Get all the action meta data by get request
# 2) Loop by ids and search for the matching action meta data by name and delete it
# 3) If the request code was 200 set the api response flag to true else false
@When('the user sets to delete the following meta data action category')
def step_impl(context):
    logger.info("When the user sets to delete the following meta data action category")

    model = getattr(context, "model", None)
    for row in context.table:
        headers = {"Content-Type": "application/json", "X-Api-Key": apis_urls.token}

        logger.info("action category name: '" + row["actionmetadataname"] + "'")

        response = requests.get(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI,
                                headers=apis_urls.auth_headers)
        for ids in dict(response.json()[apis_urls.metadataactioncategoryAPI]).keys():
            # logger.info(ids)
            if (response.json()[apis_urls.metadataactioncategoryAPI][ids]['name'] == row["actionmetadataname"]):
                response = requests.delete(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI + "/" + ids,
                                           headers=headers)
                # logger.info(response.status_code)
    if response.status_code == 200:
        GeneralVariables.api_responseflag['value'] = 'True'
    else:
        GeneralVariables.api_responseflag['value'] = 'False'

# Step Definition Implementation:
# 1) Get all the existing subject meta data by get request and put them into a table
# 2) Loop using both the expected and actual tables and assert the data row by row
@Then('the following meta data subject category should be existed in the system')
def step_impl(context):
    logger.info("Then the following meta data subject category should be existed in the system")

    model = getattr(context, "model", None)
    response = requests.get(apis_urls.serverURL + apis_urls.metadatasubjectcategoryAPI, headers=apis_urls.auth_headers)
    apiresult = Table(names=('subjectcategoryname', 'subjectcategorydescription'), dtype=('S100', 'S100'))
    if len(response.json()[apis_urls.metadatasubjectcategoryAPI]) != 0:
        for ids in dict(response.json()[apis_urls.metadatasubjectcategoryAPI]).keys():
            apisubjectcategoryname = response.json()[apis_urls.metadatasubjectcategoryAPI][ids]['name']
            apisubjectcategorydescription = response.json()[apis_urls.metadatasubjectcategoryAPI][ids]['description']
            apiresult.add_row(vals=(apisubjectcategoryname, apisubjectcategorydescription))
    else:
        apiresult.add_row(vals=("", ""))
    for row1, row2 in zip(context.table, apiresult):
        logger.info("asserting the expected subject category name: '" + str(
            row1["subjectmetadataname"]) + "' is the same as the actual existing '" + str(
            row2["subjectcategoryname"]) + "'")
        assert str(row1["subjectmetadataname"]) == str(
            row2["subjectcategoryname"]), "subject category name is not correct!"
        logger.info("assertion passed!")
        logger.info("asserting the expected subject category description: '" + str(
            row1["subjectmetadatadescription"]) + "' is the same as the actual existing '" + str(
            row2["subjectcategorydescription"]) + "'")
        assert str(row1["subjectmetadatadescription"]) == str(
            row2["subjectcategorydescription"]), "Subject meta-data category description is not correct!"
        logger.info("assertion passed!")

# Step Definition Implementation:
# 1) Get all the existing object meta data by get request and put them into a table
# 2) Loop using both the expected and actual tables and assert the data row by row
@Then('the following meta data object category should be existed in the system')
def step_impl(context):
    model = getattr(context, "model", None)
    logger.info("Then the following meta data object category should be existed in the system")
    response = requests.get(apis_urls.serverURL + apis_urls.metadataobjectcategoryAPI, headers=apis_urls.auth_headers)
    apiresult = Table(names=('objectcategoryname', 'objectcategorydescription'), dtype=('S100', 'S100'))

    if len(response.json()[apis_urls.metadataobjectcategoryAPI]) != 0:
        for ids in dict(response.json()[apis_urls.metadataobjectcategoryAPI]).keys():
            apiobjectcategoryname = response.json()[apis_urls.metadataobjectcategoryAPI][ids]['name']
            apiobjectcategorydescription = response.json()[apis_urls.metadataobjectcategoryAPI][ids]['description']
            apiresult.add_row(vals=(apiobjectcategoryname, apiobjectcategorydescription))
    else:
        apiresult.add_row(vals=("", ""))
    for row1, row2 in zip(context.table, apiresult):
        logger.info("asserting the expected object category description: '" + str(
            row1["objectmetadataname"]) + "' is the same as the actual existing '" + str(
            row2["objectcategoryname"]) + "'")
        assert str(row1["objectmetadataname"]) == str(
            row2["objectcategoryname"]), "object category name is not correct!"
        logger.info("assertion passed!")
        logger.info("asserting the expected object category description: '" + str(
            row1["objectmetadatadescription"]) + "' is the same as the actual existing '" + str(
            row2["objectcategorydescription"]) + "'")
        assert str(row1["objectmetadatadescription"]) == str(
            row2["objectcategorydescription"]), "object meta-data category description is not correct!"
        logger.info("assertion passed!")

# Step Definition Implementation:
# 1) Get all the existing action meta data by get request and put them into a table
# 2) Loop using both the expected and actual tables and assert the data row by row
@Then('the following meta data action category should be existed in the system')
def step_impl(context):
    logger.info("Then the following meta data action category should be existed in the system")

    model = getattr(context, "model", None)
    response = requests.get(apis_urls.serverURL + apis_urls.metadataactioncategoryAPI, headers=apis_urls.auth_headers)
    apiresult = Table(names=('actioncategoryname', 'actioncategorydescription'), dtype=('S100', 'S100'))
    if len(response.json()[apis_urls.metadataactioncategoryAPI]) != 0:
        for ids in dict(response.json()[apis_urls.metadataactioncategoryAPI]).keys():
            apiactioncategoryname = response.json()[apis_urls.metadataactioncategoryAPI][ids]['name']
            apiactioncategorydescription = response.json()[apis_urls.metadataactioncategoryAPI][ids]['description']
            apiresult.add_row(vals=(apiactioncategoryname, apiactioncategorydescription))
    else:
        apiresult.add_row(vals=("", ""))
    for row1, row2 in zip(context.table, apiresult):
        logger.info("asserting the expected action category description: '" + str(
            row1["actionmetadataname"]) + "' is the same as the actual existing '" + str(
            row2["actioncategoryname"]) + "'")

        assert str(row1["actionmetadataname"]) == str(
            row2["actioncategoryname"]), "action category name is not correct!"
        logger.info("assertion passed!")

        logger.info("asserting the expected action category description: '" + str(
            row1["actionmetadatadescription"]) + "' is the same as the actual existing '" + str(
            row2["actioncategorydescription"]) + "'")

        assert str(row1["actionmetadatadescription"]) == str(
            row2["actioncategorydescription"]), "action meta-data category description is not correct!"
        logger.info("assertion passed!")

# Step Definition Implementation:
# Assert the saved api response flag with the expected flag
@Then('the system should reply the following')
def step_impl(context):
    logger.info("Then the system should reply the following:")
    model = getattr(context, "model", None)
    for row in context.table:
        logger.info("asserting the expected api response: '" + row["flag"] + "' and the actual response: '" +
                    GeneralVariables.api_responseflag['value'] + "'")
        assert row["flag"] == GeneralVariables.api_responseflag['value'], "Validation is not correct, Expected: " + row[
            "flag"] + " but the API response was: " + GeneralVariables.api_responseflag['value']
        logger.info("assertion passed!")