ArtNet-Webservice/tests/createAPI/create_delete_tag.py

import requests
import json
from typing import List, Tuple

from tests.createAPI.create_delete_tag_category import test_tag_category_entries, list_tag_categories, \
    create_tag_categories, delete_category_entries, delete_tag_category

test_tag_entries = [{"name": "tag1", "description": "description1", "category_id": 0, "id": None},
                    {"name": "tag2", "description": "description2", "category_id": 0, "id": None},
                    {"name": "tag3", "description": "description3", "category_id": 0, "id": None},
                    {"name": "tag4", "description": "description4", "category_id": 2, "id": None}]


def list_tags(url: str, port: int):
    r = requests.get(f"http://{url}:{port}/artnet/metadata/tag")
    if r.status_code != 200:
        raise Exception("Failed querying for tags!")
    tags = json.loads(r.text)
    return tags


def create_tag_entries(url: str, port: int):
    """
    Create many tags for testing purposes
    :param url:
    :param port:
    :return:
    """
    for i in range(len(test_tag_entries)):
        r = create_tag(url, port, name=test_tag_entries[i]["name"], description=test_tag_entries[i]["description"],
                       category_id=test_tag_category_entries[test_tag_entries[i]["category_id"]]["id"])

        if r.status_code != 200:
            print(f"Create Tag Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
            raise Exception("Create Tag Entry Test: FAILED")
        else:
            test_tag_entries[i]["id"] = json.loads(r.text)["id"]


def update_tag_entries(url: str, port: int):
    """
    Update all tags in their fields and check the result
    :param url:
    :param port:
    :return:
    """
    for i in range(len(test_tag_entries) - 2):
        new_name = test_tag_entries[i]["name"] + "_updated"
        new_desc = test_tag_entries[i]["description"] + "_updated"
        new_categ = test_tag_category_entries[1]["id"]

        r = update_tag(url, port, tag_id=test_tag_entries[i]["id"], name=new_name, description=new_desc,
                       category_id=new_categ)

        if r.status_code != 200:
            print(f"Updating Tag Entry test Nr.{i} failed with {r.status_code} and reason {r.text}")
            raise Exception("Update Tag Entry Test: FAILED")

        new_tag = get_tag_by_ID(url, port, test_tag_entries[i]["id"])

        if new_tag["name"] != new_name:
            print(f"Tag name is not matching the update name! Current: {new_tag['name']} Expected: {new_name}")
            raise Exception("Update Tag Entry Test: FAILED")
        if new_tag["description"] != new_desc:
            print(f"Tag description is not matching the update description! "
                  f"Current: {new_tag['description']} Expected: {new_desc}")
            raise Exception("Update Tag Entry Test: FAILED")
        if new_tag["category_id"] != new_categ:
            print(f"Tag category is not matching the update category! "
                  f"Current: {new_tag['category_id']} Expected: {new_categ}")
            raise Exception("Update Tag Entry Test: FAILED")

def delete_some_tag_entries_via_category(url: str, port: int, tag_index: int):
    """
    Delete the tags by deleting their category and expecting that to delete the tags too.
    :param url:
    :param port:
    :param tag_index: an index of test_tag_entries indicating which tag to target
    :return:
    """

    r = delete_tag_category(url, port,
                            category_id=test_tag_category_entries[test_tag_entries[tag_index]["category_id"]]["id"])

    if r.status_code != 200:
        print(f"Deleting a tag category Nr.{2} failed with {r.status_code} and reason {r.text}")
        raise Exception("Delete Tag via Category Cascade Test: ERROR")

    tag = get_tag_by_ID(url, port, tag_id=test_tag_entries[tag_index]["id"])
    if 'detail' not in tag.keys() and tag["detail"] == 'No matching tag found!':
        print(f"Failed to delete tag via deleting their category with status {r.status_code} and reason {r.text}")
        raise Exception("Deleting Tag via Category Cascade Test: FAILED")


def delete_tag_entries(url: str, port: int):
    """
    Delete all tag entries specified by test_tag_entries
    :param url:
    :param port:
    :return:
    """
    for i in range(len(test_tag_entries)):
        r = delete_tag(url, port, test_tag_entries[i]["id"])

        if r.status_code != 200:
            print(f"Delete Tag Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
            raise Exception("Delete Tag Entry Test: FAILED")


def get_tag_by_ID(url: str, port: int, tag_id: int):
    """
    Fetch a tag specified by tag_id
    :param url:
    :param port:
    :param tag_id:
    :return:
    """
    r = requests.get(f"http://{url}:{port}/artnet/metadata/tag?id={tag_id}")

    if r.status_code != 200 and r.status_code != 404:
        raise Exception("Failed querying for tag!")

    return json.loads(r.text)


def create_tag(url: str, port: int, name: str, category_id: int, description: str = None):
    """
    Create a tag
    :param url:
    :param port:
    :param name:
    :param category_id:
    :param description:
    :return:
    """
    r = requests.post(f"http://{url}:{port}/artnet/metadata/tag",
                      json={"name": name, "category_id": category_id, "description": description})

    return r


def update_tag(url: str, port: int, tag_id: int, name: str = None, description: str = None, category_id: int = None,
               implications: List[int] = None, alias: List[int] = None):
    """
    Update a tag with the specified fields. If avoided they are not updated.
    :param url:
    :param port:
    :param tag_id:
    :param name:
    :param description:
    :param category_id:
    :param implications:
    :param alias:
    :return:
    """
    r = requests.post(f"http://{url}:{port}/artnet/metadata/tag?id={tag_id}",
                      json={"name": name, "description": description, "category_id": category_id,
                            "implications": implications, "alias": alias})
    return r


def delete_tag(url: str, port: int, tag_id: int):
    """
    Delete a tag specified by its id
    :param url:
    :param port:
    :param tag_id:
    :return:
    """
    r = requests.delete(f"http://{url}:{port}/artnet/metadata/tag?id={tag_id}")

    return r


def run_tag_test(url: str, port: int):
    print()
    print("----------------")
    print(f"Starting tag test with ({len(list_tags(url, port))}) tags and "
          f"({len(list_tag_categories(url, port))}) tag categories!")
    print(f"Creating {len(test_tag_category_entries)} tag category entries for the tests ...")
    create_tag_categories(url, port)
    print(f"Creating {len(test_tag_entries)} tag entries ...")
    create_tag_entries(url, port)
    create_tag_result = False if not len(list_tags(url, port)) == len(test_tag_entries) else True
    print(f"Found {len(list_tags(url, port))} tag entries!")
    print()

    # Update test
    print("Trying to update the tag fields ...")
    update_tag_entries(url, port)
    update_tag_result = True
    print("Finished updating the tags!")
    print()

    # Delete test
    print(f"Found {len(list_tags(url, port))} tag entries!")
    print("Deleting the tag entries ...")
    delete_tag_entries(url, port)  # would throw an exception if an error occurred
    delete_tag_result = False if not len(list_tags(url, port)) == 0 else True
    print(f"Found {len(list_tags(url, port))} tag entries!")
    print()

    # Creating tags for cascade deleting test
    print("Preparing tag category cascade deletion test ...")
    create_tag_entries(url, port)

    # Delete via Category test
    delete_some_tag_entries_via_category(url, port, tag_index=0)  # should delete 3 tags in total, leaving 1 behind
    delete_tag_category_cascade_result = False if not len(list_tags(url, port)) == 1 else True
    print()

    # Clean up
    print("Cleaning tag categories up ...")
    delete_category_entries(url, port)
    print(f"Tag test complete with {len(list_tags(url, port))} tags and "
          f"{len(list_tag_categories(url, port))} categories!")

    return create_tag_result, update_tag_result, delete_tag_result, delete_tag_category_cascade_result
    # create, update, delete, delete-category-cascade


if __name__ == "__main__":
    url, port = "127.0.0.1", 8000
    l = len(list_tag_categories(url, port))
    if l > 0:
        print(f"Deleting leftover category entries ({l}) ...")
        delete_category_entries(url, port)
    l = len(list_tags(url, port))
    if l > 0:
        print(f"Deleting leftover tag entries ({l}) ...")
        delete_tag_entries(url, port)

    run_tag_test(url, port)