Added Unit Tests to test the WebService
Added scripts to test various API calls of the webservice by creating, modifying and deleting entries from a test database of ArtNet. Do NOT use those on production.master
parent
f2821c561e
commit
5a8b88202e
@ -0,0 +1,180 @@
|
||||
import requests
|
||||
import json
|
||||
from typing import List, Tuple
|
||||
|
||||
from tests.createAPI.create_delete_artist import list_artists, create_artist_entries, delete_artist_entries
|
||||
from tests.createAPI.create_delete_presence import test_presence_entries, list_presences, create_presence_entries
|
||||
|
||||
|
||||
test_art_entries = [
|
||||
{"hash": "hash1", "path": "artist1/image1", "title": "Image Title 1", "link": "http://localhost/artist1/image1.png",
|
||||
"presences": [(test_presence_entries[0]["name"], test_presence_entries[0]["domain"])]},
|
||||
{"hash": "hash2", "path": "artist1/image2", "title": "Image Title 2", "link": "http://localhost/artist1/image2.png",
|
||||
"presences": [(test_presence_entries[1]["name"], test_presence_entries[1]["domain"])]},
|
||||
{"hash": "hash3", "path": "artist2/image1", "title": "Image Title 3", "link": "http://localhost/artist2/image3.png",
|
||||
"presences": [(test_presence_entries[0]["name"], test_presence_entries[0]["domain"]),
|
||||
(test_presence_entries[1]["name"], test_presence_entries[1]["domain"])]},
|
||||
]
|
||||
|
||||
|
||||
def list_art(url: str, port: int):
|
||||
"""
|
||||
Return a list of all art known to the database
|
||||
:param url:
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
r = requests.get(f"http://{url}:{port}/artnet/metadata/art")
|
||||
if r.status_code != 200:
|
||||
raise Exception("Failed querying for art!")
|
||||
art = json.loads(r.text)
|
||||
return art
|
||||
|
||||
|
||||
def create_art_entries(url: str, port: int):
|
||||
"""
|
||||
Creates many art entries for testing purposes
|
||||
:param url:
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
for i in range(len(test_art_entries)):
|
||||
r = create_art(url, port, md5_hash=test_art_entries[i]["hash"], path=test_art_entries[i]["path"],
|
||||
title=test_art_entries[i]["title"], link=test_art_entries[i]["link"],
|
||||
presences=test_art_entries[i]["presences"])
|
||||
if r.status_code != 200:
|
||||
print(f"Create Art Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
|
||||
raise Exception("Create Art Entry Test: FAILED")
|
||||
else:
|
||||
test_art_entries[i]["ID"] = json.loads(r.text)["id"]
|
||||
|
||||
print("Created art entries: SUCCESS")
|
||||
|
||||
|
||||
def update_art_entries(url: str, port: int):
|
||||
for i in range(len(test_art_entries)):
|
||||
r = update_art(url, port, md5_hash=test_art_entries[i]["hash"]+"moreHash",
|
||||
path=test_art_entries[i]["path"]+"evenBetterPath",
|
||||
title=test_art_entries[i]["title"]+" updated",
|
||||
presences=test_art_entries[i]["presences"], art_id=test_art_entries[i]["ID"])
|
||||
|
||||
if r.status_code != 200:
|
||||
print(f"Update Art Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
|
||||
raise Exception("Update Art Entry Test: FAILED")
|
||||
|
||||
for i in range(len(test_art_entries)):
|
||||
r = requests.get(f"http://{url}:{port}/artnet/metadata/art?id={test_art_entries[i]['ID']}")
|
||||
if json.loads(r.text)["title"].split(" ")[-1] != "updated":
|
||||
raise Exception("Update Art Entry Test: Failed (unexpected or no updated title)")
|
||||
if json.loads(r.text)["link"] != test_art_entries[i]["link"]:
|
||||
raise Exception("Update Art Entry Test: Failed (unexpected link)")
|
||||
|
||||
print("Updated art entries: SUCCESS")
|
||||
|
||||
|
||||
def delete_art_entries(url: str, port: int):
|
||||
arts = list_art(url, port)
|
||||
|
||||
for i in range(len(arts)):
|
||||
if "id" in arts[i].keys():
|
||||
r = delete_art_by_id(url, port, art_id=arts[i]["id"])
|
||||
else:
|
||||
try:
|
||||
r = delete_art_by_hash(url, port, md5_hash=arts[i]["hash"])
|
||||
except ValueError:
|
||||
r = None
|
||||
|
||||
if r is not None and r.status_code != 200:
|
||||
raise Exception(f"Could not delete the art! {r.status_code}, {r.text}")
|
||||
|
||||
print("Deleted art entries: SUCCESS")
|
||||
|
||||
|
||||
def create_art(url: str, port: int, md5_hash: str, path: str, title: str, link: str, presences: List[Tuple[str, str]]):
|
||||
r = requests.post(f"http://{url}:{port}/artnet/metadata/art",
|
||||
json={"hash": md5_hash, "path": path, "title": title, "link": link,
|
||||
"presences": [{"name": presence[0], "domain": presence[1]} for presence in presences]})
|
||||
|
||||
return r
|
||||
|
||||
|
||||
def update_art(url: str, port: int, art_id: int, md5_hash: str = None, path: str = None, title: str = None,
|
||||
link: str = None, presences: List[Tuple[str, str]] = None):
|
||||
body = {}
|
||||
if md5_hash is not None:
|
||||
body["hash"] = md5_hash
|
||||
if path is not None:
|
||||
body["path"] = path
|
||||
if title is not None:
|
||||
body["title"] = title
|
||||
if link is not None:
|
||||
body["link"] = link
|
||||
if presences is not None:
|
||||
body["presences"] = [{"name": presence[0], "domain": presence[1]} for presence in presences]
|
||||
|
||||
r = requests.post(f"http://{url}:{port}/artnet/metadata/art?id={art_id}",
|
||||
json=body)
|
||||
|
||||
return r
|
||||
|
||||
|
||||
def delete_art_by_id(url: str, port: int, art_id: int):
|
||||
r = requests.delete(f"http://{url}:{port}/artnet/metadata/art?id={art_id}")
|
||||
|
||||
return r
|
||||
|
||||
|
||||
def delete_art_by_hash(url: str, port: int, md5_hash: str):
|
||||
r1 = requests.get(f"http://{url}:{port}/artnet/metadata/art?hash={md5_hash}")
|
||||
if r1.status_code != 200:
|
||||
raise ValueError("Could not delete art because it could not be found!")
|
||||
data = json.loads(str(r1.text))
|
||||
art_id = data['id']
|
||||
|
||||
r = requests.delete(f"http://{url}:{port}/artnet/metadata/art?id={art_id}")
|
||||
|
||||
return r
|
||||
|
||||
|
||||
def run_art_test(url: str, port: int):
|
||||
print()
|
||||
print("----------------")
|
||||
|
||||
print(f"Starting presence test with "
|
||||
f"({len(list_artists(url, port))}) artists, ({len(list_presences(url, port))}) presences, "
|
||||
f"({len(list_art(url, port))}) art!")
|
||||
print("Setting up artist and presences for the art test ...")
|
||||
create_artist_entries(url, port)
|
||||
create_presence_entries(url, port)
|
||||
|
||||
print("Starting the art test ...")
|
||||
if len(list_art(url, port)) > 0:
|
||||
print("Deleting previous art entries of tests ...")
|
||||
delete_art_entries(url, port)
|
||||
|
||||
print(f"Creating {len(test_art_entries)} art entries ...")
|
||||
create_art_entries(url, port)
|
||||
create_art_result = False if not len(list_art(url, port)) == 3 else True
|
||||
print(f"Found {len(list_art(url, port))} art entries!")
|
||||
if not create_art_result:
|
||||
print(f"Failure! Unexpected number of art entries!")
|
||||
|
||||
print("Updating all art entries with new data ...")
|
||||
update_art_entries(url, port)
|
||||
update_art_result = True # given, the update function would raise an exception on error
|
||||
print(f"Found {len(list_art(url, port))} art entries!")
|
||||
|
||||
# TODO write test for updating presence-art relation
|
||||
|
||||
print("Deleting all art entries ...")
|
||||
delete_art_entries(url, port)
|
||||
delete_art_result = False if not len(list_art(url, port)) == 0 else True
|
||||
print(f"Found {len(list_art(url, port))} art entries!")
|
||||
|
||||
print("Art test complete! Cleaning up setup ...")
|
||||
delete_artist_entries(url, port)
|
||||
print("----------------")
|
||||
print()
|
||||
|
||||
return create_art_result, update_art_result, delete_art_result
|
||||
|
||||
@ -0,0 +1,81 @@
|
||||
import requests
|
||||
from typing import List
|
||||
import json
|
||||
|
||||
# DB Schema Version 2.0 | 2021-10-10
|
||||
|
||||
test_artist_entries = [
|
||||
{"id": None, "name": "Artist1", "topics": []},
|
||||
{"id": None, "name": "Artist2", "topics": []},
|
||||
{"id": None, "name": "Artist3", "topics": []}
|
||||
]
|
||||
|
||||
|
||||
def list_artists(url: str, port: int):
|
||||
"""
|
||||
Return a list of all artist known to the database
|
||||
:param url:
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
r = requests.get(f"http://{url}:{port}/artnet/metadata/artist")
|
||||
if r.status_code != 200:
|
||||
print(r.text)
|
||||
raise Exception("Failed querying for artists!")
|
||||
artists = json.loads(r.text)
|
||||
return artists
|
||||
|
||||
|
||||
def create_artist_entries(url: str, port: int):
|
||||
for i in range(len(test_artist_entries)):
|
||||
r = create_artist(url, port, test_artist_entries[i]["name"], test_artist_entries[i]["topics"])
|
||||
|
||||
if r.status_code != 200:
|
||||
print(f"Status:{r.status_code} Text:{r.text}")
|
||||
raise Exception("Artist Entry creation failed!")
|
||||
|
||||
artist_id = int(r.text)
|
||||
test_artist_entries[i]["id"] = artist_id
|
||||
|
||||
|
||||
def delete_artist_entries(url: str, port: int):
|
||||
artists = list_artists(url, port)
|
||||
for i in range(len(artists)):
|
||||
r = delete_artist(url, port, artists[i]["id"])
|
||||
if r.status_code != 200:
|
||||
print(f"Deleted artist entry #{artists[i]['id']}, {artists[i]['name']}: FAILURE")
|
||||
else:
|
||||
print(f"Deleted artist entry #{artists[i]['id']}, {artists[i]['name']}: SUCCESS")
|
||||
|
||||
|
||||
def create_artist(url: str, port: int, name: str, topics: List[int]):
|
||||
r = requests.post(f"http://{url}:{port}/artnet/metadata/artist",
|
||||
json={"name": name, "topics": topics})
|
||||
|
||||
return r
|
||||
|
||||
|
||||
def delete_artist(url: str, port: int, id: int):
|
||||
r = requests.delete(f"http://{url}:{port}/artnet/metadata/artist?id={id}")
|
||||
return r
|
||||
|
||||
|
||||
def run_artist_test(url: str, port: int):
|
||||
print()
|
||||
print("----------------")
|
||||
print(f"Starting the artist test with ({len(list_artists(url, port))}) artists!")
|
||||
print("Starting the artist test ...")
|
||||
print(f"Creating {len(test_artist_entries)} artists ...")
|
||||
create_artist_entries(url, port)
|
||||
print(f"Found {len(list_artists(url, port))} artist entries")
|
||||
create_artist_result = False if not len(list_artists(url, port)) == 3 else True
|
||||
|
||||
print("Deleting all artists ...")
|
||||
delete_artist_entries(url, port)
|
||||
delete_artist_result = False if not len(list_artists(url, port)) == 0 else True
|
||||
print(f"Found {len(list_artists(url, port))} artist entries")
|
||||
print("Completed the artist test!")
|
||||
print("----------------")
|
||||
print()
|
||||
|
||||
return create_artist_result, delete_artist_result
|
||||
@ -0,0 +1,85 @@
|
||||
import requests
|
||||
import json
|
||||
from typing import List, Tuple
|
||||
|
||||
from tests.createAPI.create_delete_artist import test_artist_entries, list_artists, create_artist_entries, delete_artist_entries
|
||||
|
||||
test_presence_entries = [
|
||||
{"name": "presence_name1", "domain": "presence_domain1", "artist_id": 0},
|
||||
{"name": "presence_name2", "domain": "presence_domain1", "artist_id": 1},
|
||||
{"name": "presence_name3", "domain": "presence_domain2", "artist_id": 1},
|
||||
]
|
||||
|
||||
|
||||
def create_presence_entries(url: str, port: int):
|
||||
for i in range(len(test_presence_entries)):
|
||||
r = create_presence(url, port, name=test_presence_entries[i]["name"], domain=test_presence_entries[i]["domain"],
|
||||
artist_id=test_artist_entries[test_presence_entries[i]["artist_id"]]["id"])
|
||||
if r.status_code != 200:
|
||||
print(f"Tried to create presence with name:{test_presence_entries[i]['name']} "
|
||||
f"domain:{test_presence_entries[i]['domain']} artist_id:{test_presence_entries[i]['artist_id']}")
|
||||
print(f"Create Presence failed with {r.status_code} and reason {r.text}!")
|
||||
raise Exception("Create Presence Entry Test: Failed")
|
||||
|
||||
|
||||
def create_presence(url: str, port: int, name: str, domain: str, artist_id: int):
|
||||
r = requests.post(f"http://{url}:{port}/artnet/metadata/presence",
|
||||
json={"name": name, "domain": domain, "artist_id": artist_id})
|
||||
|
||||
return r
|
||||
|
||||
|
||||
def delete_presence(url: str, port: int, name: str, domain: str):
|
||||
r = requests.delete(f"http://{url}:{port}/artnet/metadata/presence?name={name}&domain={domain}")
|
||||
|
||||
if r.status_code != 200:
|
||||
print(f"Deleting Presence entry {name}@{domain}: FAILED")
|
||||
return False
|
||||
else:
|
||||
print(f"Deleting Presence entry {name}@{domain}: SUCCESS")
|
||||
return True
|
||||
|
||||
|
||||
def list_presences(url: str, port: int):
|
||||
"""
|
||||
Return a list of all presences known to the database
|
||||
:param url:
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
r = requests.get(f"http://{url}:{port}/artnet/metadata/presence")
|
||||
if r.status_code != 200:
|
||||
raise Exception("Failed querying for presences!")
|
||||
presences = json.loads(r.text)
|
||||
return presences
|
||||
|
||||
|
||||
def run_presence_test(url: str, port: int):
|
||||
print()
|
||||
print("----------------")
|
||||
print(f"Starting presence test with "
|
||||
f"({len(list_artists(url, port))}) artists, ({len(list_presences(url, port))}) presences!")
|
||||
print("Setting up artists for the presence test ...")
|
||||
create_artist_entries(url, port)
|
||||
|
||||
print("Starting the presence test ...")
|
||||
print(f"Creating {len(test_presence_entries)} presences ...")
|
||||
create_presence_entries(url, port)
|
||||
create_presence_result = False if not len(list_presences(url, port)) == 3 else True
|
||||
print(f"Found {len(list_presences(url, port))} presence entries!")
|
||||
|
||||
print(f"Deleting (1) specific presence entry ({test_presence_entries[0]['name']}, "
|
||||
f"{test_presence_entries[0]['domain']}) ...")
|
||||
delete_presence_result = delete_presence(url, port, test_presence_entries[0]["name"],
|
||||
test_presence_entries[0]["domain"])
|
||||
print(f"Found {len(list_presences(url, port))} presence entries!")
|
||||
|
||||
print(f"Deleting leftover presences ({len(list_presences(url, port))}) via cascading ...")
|
||||
delete_artist_entries(url, port) # artist deletion should cascade into the presences
|
||||
print(f"Found {len(list_presences(url, port))} presence entries!")
|
||||
cascade_delete_presence_result = False if not len(list_presences(url, port)) == 0 else True
|
||||
print("Completed the presence test!")
|
||||
print("----------------")
|
||||
print()
|
||||
|
||||
return create_presence_result, delete_presence_result, cascade_delete_presence_result
|
||||
@ -0,0 +1,114 @@
|
||||
import requests
|
||||
import json
|
||||
from typing import List, Tuple
|
||||
|
||||
from tests.createAPI.create_delete_artist import list_artists, create_artist_entries, delete_artist_entries
|
||||
from tests.createAPI.create_delete_presence import test_presence_entries, list_presences, create_presence_entries
|
||||
from tests.createAPI.create_delete_art import test_art_entries, list_art, create_art_entries, delete_art_entries
|
||||
|
||||
test_art_presence_relations = [
|
||||
{"art_hash": test_art_entries[0]["hash"], "presence_name": test_presence_entries[0]["name"], "presence_domain": test_presence_entries[0]["domain"]},
|
||||
{"art_hash": test_art_entries[1]["hash"], "presence_name": test_presence_entries[0]["name"], "presence_domain": test_presence_entries[0]["domain"]},
|
||||
{"art_hash": test_art_entries[2]["hash"], "presence_name": test_presence_entries[0]["name"], "presence_domain": test_presence_entries[0]["domain"]},
|
||||
{"art_hash": test_art_entries[2]["hash"], "presence_name": test_presence_entries[1]["name"], "presence_domain": test_presence_entries[1]["domain"]},
|
||||
]
|
||||
|
||||
|
||||
def list_art_to_presence_relations(url: str, port: int):
|
||||
"""
|
||||
List all relations between art and presence entries
|
||||
:param url:
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
results = []
|
||||
|
||||
arts = list_art(url, port)
|
||||
for art in arts:
|
||||
r = requests.get(f"http://{url}:{port}/artnet/metadata/presence?art_id={art['id']}")
|
||||
if r.status_code != 200:
|
||||
raise Exception("Failed to query presences with art id!")
|
||||
|
||||
result = json.loads(r.text)
|
||||
results.append(result)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def art_hash_to_id(arts: List[dict]):
|
||||
"""
|
||||
Converts a list of arts into the format of {hash: id, hash1: id1}
|
||||
:param arts:
|
||||
:return:
|
||||
"""
|
||||
result = {}
|
||||
for i in range(len(arts)):
|
||||
result[arts[i]["md5_hash"].strip()] = arts[i]["id"]
|
||||
return result
|
||||
|
||||
|
||||
def create_art_presence_relations(url: str, port: int):
|
||||
arts = list_art(url, port)
|
||||
art_hash_id = art_hash_to_id(arts)
|
||||
for i in range(len(test_art_presence_relations)):
|
||||
print(f"Creating art-presence relation with id:{art_hash_id[test_art_presence_relations[i]['art_hash']]} "
|
||||
f"presence_name:{test_art_presence_relations[i]['presence_name']} "
|
||||
f"presence_domain:{test_art_presence_relations[i]['presence_domain']}")
|
||||
r = create_art_presence_relation(url, port, art_id=art_hash_id[test_art_presence_relations[i]["art_hash"]],
|
||||
presence_name=test_art_presence_relations[i]["presence_name"],
|
||||
presence_domain=test_art_presence_relations[i]["presence_domain"])
|
||||
if r.status_code != 200:
|
||||
print(f"Tried to create art-presence relation with name:{test_art_presence_relations[i]['presence_name']} "
|
||||
f"domain:{test_art_presence_relations[i]['presence_domain']}"
|
||||
f" art_id:{test_art_presence_relations[i]['art_id']}")
|
||||
print(f"Create Relation failed with {r.status_code} and reason {r.text}!")
|
||||
raise Exception("Couldn't create art-presence relation!")
|
||||
|
||||
l = list_art_to_presence_relations(url, port)
|
||||
if len(l) == 3 and len(l[-1]) == 2:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def create_art_presence_relation(url: str, port: int, presence_name: str, presence_domain: str, art_id: int):
|
||||
r = requests.post(f"http://{url}:{port}/artnet/metadata/art?id={art_id}",
|
||||
json={"presences": [{"name": presence_name, "domain": presence_domain}]})
|
||||
return r
|
||||
|
||||
|
||||
def run_art_presence_relation_test(url, port):
|
||||
print()
|
||||
print("----------------")
|
||||
if len(list_presences(url, port)) > 0:
|
||||
print("Deleting leftover presence entries via cascading from artists ...")
|
||||
delete_artist_entries(url, port) # deleting presences via cascading
|
||||
if len(list_art(url, port)) > 0:
|
||||
print("Deleting leftover art entries ...")
|
||||
delete_art_entries(url, port)
|
||||
|
||||
print(f"Starting presence test with "
|
||||
f"({len(list_presences(url, port))}) presences, "
|
||||
f"({len(list_art(url, port))}) art!")
|
||||
|
||||
print("Setting up art and presences for the art-presence-relation test ...")
|
||||
create_artist_entries(url, port)
|
||||
create_presence_entries(url, port)
|
||||
create_art_entries(url, port)
|
||||
|
||||
art = list_art(url, port) # debug
|
||||
presences = list_presences(url, port) # debug
|
||||
|
||||
print()
|
||||
print("Starting the art-presence test ...")
|
||||
if sum([len(x) for x in list_art_to_presence_relations(url, port)]) != 4: # 3 expected from create_art_entries()
|
||||
print("Anomalous amount of presence-art relations ...")
|
||||
exit(1)
|
||||
|
||||
print("Creating art-presence relations ...")
|
||||
create_result = create_art_presence_relations(url, port)
|
||||
print(f"Found {sum([len(x) for x in list_art_to_presence_relations(url, port)])} art-presence relations!")
|
||||
|
||||
# TODO delete art-presence relation test
|
||||
|
||||
return create_result
|
||||
@ -0,0 +1,47 @@
|
||||
from tests.createAPI.create_delete_art import run_art_test, list_art, delete_art_entries
|
||||
from tests.createAPI.create_delete_presence import run_presence_test, list_presences
|
||||
from tests.createAPI.create_delete_artist import delete_artist_entries, list_artists, run_artist_test
|
||||
|
||||
from tests.relationAPI.relate_art import run_art_presence_relation_test
|
||||
|
||||
|
||||
def run_tests(url: str, port: int):
|
||||
"""
|
||||
Run unit tests against all API interfaces
|
||||
:param url:
|
||||
:param port:
|
||||
:return:
|
||||
"""
|
||||
if len(list_artists(url, port)) > 0:
|
||||
print("Deleting leftover artists ...")
|
||||
delete_artist_entries(url, port)
|
||||
if len(list_art(url, port)) > 0:
|
||||
print("Deleting leftover art ...")
|
||||
delete_art_entries(url, port)
|
||||
|
||||
create_artist_result, delete_artist_result = run_artist_test(url, port)
|
||||
create_presence_result, delete_presence_result, cascade_delete_presence_result = run_presence_test(url, port)
|
||||
create_art_result, update_art_result, delete_art_result = run_art_test(url, port)
|
||||
|
||||
print()
|
||||
print("-------- Test Results ---------")
|
||||
print("Leftover DB Objects:")
|
||||
r = list_artists(url, port)
|
||||
print(f"Artists: {len(r)}, {r}")
|
||||
r = list_presences(url, port)
|
||||
print(f"Presences: {len(r)}, {r}")
|
||||
r = list_art(url, port)
|
||||
print(f"Art: {len(r)}, {r}")
|
||||
print()
|
||||
print("Functions:")
|
||||
print(f"Artists: \tCreate: {create_artist_result} \tUpdate: {'N/A'} \tDelete: {delete_artist_result} \t(Direct)")
|
||||
print(f"Presences: \tCreate: {create_presence_result} \tUpdate: {'N/A'} "
|
||||
f"\tDelete: {delete_presence_result} \t(Direct), {cascade_delete_presence_result} \t(cascade@artist)")
|
||||
print(f"Art: \t\tCreate: {create_art_result} \tUpdate: {update_art_result} "
|
||||
f"\tDelete: {delete_art_result} \t(Direct)")
|
||||
print("-------------------------------")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
#run_tests("127.0.0.1", 8000)
|
||||
run_art_presence_relation_test("127.0.0.1", 8000)
|
||||
Loading…
Reference in New Issue