Commit 275618dd authored by parciak's avatar parciak
Browse files

Added unit tests for Check and clean up of Receive unit tests. Also, changed...


Added unit tests for Check and clean up of Receive unit tests. Also, changed credentials to a list instead of a dict.

Signed-off-by: parciak's avatarMarcel Parciak <marcel.parciak@gmail.com>
parent 89b7823e
......@@ -123,6 +123,7 @@ def receive(payload: awmodels.RequestReceive, background_tasks: BackgroundTasks)
metafile = utils.create_file_for_archive(archive_id)
if not metafile:
# TODO: real error please
raise errors.ActiveWorkflowException(
"Another Error needed here", "Something does not work"
)
......
......@@ -21,7 +21,7 @@ class ParamsCommon(BaseModel, abc.ABC):
message: Optional[Dict[str, Any]] = {}
options: Optional[Dict[str, Any]] = {}
memory: Optional[Dict[str, Any]] = {}
credentials: Optional[Dict[str, Any]] = {}
credentials: Optional[List[Dict[str, str]]] = []
class ResultCommon(BaseModel, abc.ABC):
......
from typing import Any, Dict
from fastapi.testclient import TestClient
from annotator.main import app, endpoint_name
from annotator import config
from annotator import test_utils
client = TestClient(app)
def test_register_compliance_01():
"""
Tests if a register request yields a valid AW response.
"""
register_request = {"method": "register", "params": {}}
response = client.post(f"/{endpoint_name}", json=register_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert "result" in response_data.keys()
for key in ["name", "display_name", "description", "default_options"]:
assert key in response_data["result"]
assert response_data["result"][
"name"
] == config.BasicSettings().application_name.replace(" ", "")
assert type(response_data["result"]["name"]) == str
assert type(response_data["result"]["display_name"]) == str
assert type(response_data["result"]["description"]) == str
assert type(response_data["result"]["default_options"]) == dict
def disabled_test_receive_compliance_01():
"""
Tests if a receive requests without data yields a valid AW response.
"""
receive_request = {
"method": "receive",
"params": {
"message": {"payload": {}},
"options": {},
"memory": {},
"credentials": [],
},
}
response = client.post(f"/{endpoint_name}", json=receive_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert test_utils.is_valid_response(response_data)
def disabled_test_receive_compliance_02():
"""
Tests if a receive request with data yields a valid AW response.
"""
receive_request = {
"method": "receive",
"params": {
"message": {"payload": {"some": "message"}},
"options": {"some": "option"},
"memory": {"some": "memory"},
"credentials": [{"some": "token"}],
},
}
response = client.post(f"/{endpoint_name}", json=receive_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert test_utils.is_valid_response(response_data)
def disabled_test_check_compliance_01():
"""
Tests if a check request without data yields a valid AW response.
"""
check_request = {
"method": "check",
"params": {"message": None, "options": {}, "memory": {}, "credentials": [],},
}
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert test_utils.is_valid_response(response_data)
def disabled_test_check_compliance_02():
"""
Tests if a check request with data yields a valid AW response.
"""
check_request = {
"method": "check",
"params": {
"message": None,
"options": {"some": "option"},
"memory": {"some": "memory"},
"credentials": [{"some": "credential"}],
},
}
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert test_utils.is_valid_response(response_data)
import os
import random
import string
import tempfile
# from typing import Any, Dict, Tuple
import pytest
from fastapi.testclient import TestClient
from annotator.main import app, endpoint_name
from annotator import test_utils
client = TestClient(app)
@pytest.fixture(scope="module")
def cdstar_archive():
"""
Fakes an annotated archive id, yielding the archive_id and cleaning it up afterwards if necessary.
"""
archive_id = "".join(random.choices(string.ascii_lowercase + string.digits, k=12))
filepath = os.path.join(tempfile.gettempdir(), f"meta_{archive_id}")
with open(filepath, "w") as metafile:
metafile.write("NONCE")
yield archive_id
# check should delete the filepath usually. This is a failure cleanup
if os.path.exists(filepath):
os.unlink(filepath)
def test_check_01(cdstar_archive):
"""
Tests if a valid annotation upload yields a valid check result afterwards.
"""
# checks should be performed after a valid receive using the archive id from memory
check_request = {
"method": "check",
"params": {
"message": None,
"options": {},
"memory": {"annotations": [cdstar_archive]},
"credentials": [],
},
}
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response_json = response.json()
assert response_json
assert test_utils.is_valid_response(response_json)
# on success, respond with any kind of message to indicate a annotation run is done
assert len(response_json["result"]["messages"]) > 0
# make sure that the written state file is erased
assert not os.path.exists(
os.path.join(tempfile.gettempdir(), f"meta_{cdstar_archive}.json")
)
def test_check_02():
"""
Tests if an empty check request has no unnecessary outputs (errors, logs, messages).
"""
check_request = {
"method": "check",
"params": {
"message": None,
"options": {},
"memory": {"annotations": []},
"credentials": [],
},
}
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert test_utils.is_valid_response(response_data)
assert len(response_data["result"]["errors"]) == 0
assert len(response_data["result"]["messages"]) == 0
assert len(response_data["result"]["logs"]) == 0
def test_check_03(cdstar_archive):
"""
Tests if a retrieved annotation upload is not responded twice.
"""
check_request = {
"method": "check",
"params": {
"message": None,
"options": {},
"memory": {"annotations": [cdstar_archive]},
"credentials": [],
},
}
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
assert test_utils.is_valid_response(response_data)
assert len(response_data["result"]["errors"]) == 0
assert len(response_data["result"]["logs"]) == 0
assert len(response_data["result"]["messages"]) == 0
if "archives" in response_data["result"]["memory"].keys():
assert len(response_data["result"]["memory"]["archives"]) == 0
def test_check_04():
"""
Tests if any gibberish archives suppplied in the memory gets dropped.
"""
check_request = {
"method": "check",
"params": {
"message": None,
"options": {},
"memory": {"annotations": ["someReallyCoolStuffInHere"]},
"credentials": [],
},
}
response = client.post(f"/{endpoint_name}", json=check_request)
assert response.status_code >= 200 and response.status_code <= 299
response_data = response.json()
assert response_data
if "archives" in response_data["result"]["memory"].keys():
assert len(response_data["result"]["memory"]["archives"]) == 0
......@@ -3,18 +3,18 @@ import io
import json
import os
import tempfile
from typing import Any, Dict, Tuple
from typing import Any, Dict
import cloudant
import cloudant.database as CloudantDatabase
import cloudant.document as CloudantDocument
import pycdstar3
import pytest
import requests
from fastapi.testclient import TestClient
from annotator.main import app, endpoint_name
from annotator import config
from annotator import test_utils
client = TestClient(app)
......@@ -62,43 +62,6 @@ def delete_metadata(archive_id: str):
archive.delete()
def is_schemaorg_jsonld(json: Dict[str, Any]) -> bool:
"""
Return true if `json` contains `@context`, `@id` and `@type` as well as `@context` refers to
`http://schema.org`.
"""
keys = json.keys()
return (
"@context" in keys
and "@id" in keys
and "@type" in keys
and json["@context"].startswith("http://schema.org")
)
def is_schemaorg_reference(reference: Dict[str, Any]) -> bool:
"""
Returns true if `reference` contains a valid JSON-LD reference, i.e. contains keys `@type`
and `@id`.
"""
keys = reference.keys()
return "@type" in keys and "@id" in keys
def get_json_of_uri(uri: str, auth: Tuple[str, str] = None) -> Dict[str, Any]:
"""
Retrieves the JSON response of `uri` under BasicAuth `auth` if supplied.
"""
# Make the request and assert it was successful
r = requests.get(uri, auth=auth)
assert r.status_code >= 200 and r.status_code <= 299
# Get the JSON response and assert it is a dictionary (no application used in testing responds with lists afaik)
r_json = r.json()
assert r_json
assert isinstance(r_json, dict)
return r_json
def test_invalid_request_01():
"""
Tests if an invalid request yields a 405 response.
......@@ -116,27 +79,9 @@ def test_invalid_request_02():
assert response.status_code >= 400 and response.status_code <= 499
def test_register_01():
"""
Tests if a register request yields a valid AW response.
"""
register_request = {"method": "register", "params": {}}
response = client.post(f"/{endpoint_name}", json=register_request)
assert response.status_code == 200
response_data = response.json()
assert response_data
assert "result" in response_data.keys()
for key in ["name", "display_name", "description", "default_options"]:
assert key in response_data["result"]
assert response_data["result"][
"name"
] == config.BasicSettings().application_name.replace(" ", "")
assert type(response_data["result"]["default_options"]) == dict
def post_valid_receive_request(cdstar_archive: str) -> Dict[str, Any]:
"""
Tests if a valid Receive request to the agent yields a valid AW response.
Posts an existing cdstar archive to the agent and assumes a valid AW response as well as return the resulting json.
"""
register_request = {
"method": "receive",
......@@ -144,16 +89,14 @@ def post_valid_receive_request(cdstar_archive: str) -> Dict[str, Any]:
"message": {"payload": {"archive_id": cdstar_archive}},
"options": {"annotation": {}},
"memory": {},
"credentials": {},
"credentials": [],
},
}
response = client.post(f"/{endpoint_name}", json=register_request)
assert response.status_code >= 200 and response.status_code <= 299
json = response.json()
assert json
assert "result" in json.keys()
for key in ["errors", "logs", "memory", "messages"]:
assert key in json["result"].keys()
assert test_utils.is_valid_response(json)
return json
......@@ -167,16 +110,18 @@ def test_receive_01(cdstar_archive):
)
# external request: use requests directly instead of the TestClient
meta_json = get_json_of_uri(
meta_json = test_utils.get_json_of_uri(
f"{config.BasicSettings().couch_uri}/{config.BasicSettings().couch_db}/{cdstar_archive}",
auth=(config.BasicSettings().couch_user, config.BasicSettings().couch_pass),
)
# assert that the response is a JSON-LD of schema.org/Dataset
assert is_schemaorg_jsonld(meta_json)
assert test_utils.is_schemaorg_jsonld(meta_json)
assert meta_json["@type"] == "Dataset"
assert meta_json["identifier"] == cdstar_archive
test_utils.cleanup_metafile(cdstar_archive)
def test_receive_02(cdstar_archive):
"""
......@@ -185,19 +130,19 @@ def test_receive_02(cdstar_archive):
post_valid_receive_request(cdstar_archive)
# external request: use requests directly instead of the TestClient
cdstar_json = get_json_of_uri(
cdstar_json = test_utils.get_json_of_uri(
f"{config.BasicSettings().cdstar_uri}/{config.BasicSettings().cdstar_vault}/{cdstar_archive}",
auth=(config.BasicSettings().cdstar_user, config.BasicSettings().cdstar_pass),
)
# external request: use requests directly instead of the TestClient
meta_json = get_json_of_uri(
meta_json = test_utils.get_json_of_uri(
f"{config.BasicSettings().couch_uri}/{config.BasicSettings().couch_db}/{cdstar_archive}",
auth=(config.BasicSettings().couch_user, config.BasicSettings().couch_pass),
)
# Compare both JSON replies and the data they contain
assert is_schemaorg_jsonld(meta_json)
assert test_utils.is_schemaorg_jsonld(meta_json)
assert meta_json["identifier"] == cdstar_archive
assert datetime.datetime.strptime(
cdstar_json["created"], "%Y-%m-%dT%H:%M:%S.%f%z"
......@@ -208,6 +153,8 @@ def test_receive_02(cdstar_archive):
assert int(cdstar_json["file_count"]) == int(meta_json["size"]["value"])
assert int(cdstar_json["file_count"]) == len(meta_json["hasPart"])
test_utils.cleanup_metafile(cdstar_archive)
def test_receive_03(cdstar_archive):
"""
......@@ -218,7 +165,7 @@ def test_receive_03(cdstar_archive):
# Check if metadata is available for each file of CDSTAR
for file_offset in range(0, generated_files_count, file_list_limit):
# external request: use requests directly instead of the TestClient
cdstar_json = get_json_of_uri(
cdstar_json = test_utils.get_json_of_uri(
f"{config.BasicSettings().cdstar_uri}/{config.BasicSettings().cdstar_vault}/{cdstar_archive}?files&limit={file_list_limit}&offset={file_offset}",
auth=(
config.BasicSettings().cdstar_user,
......@@ -230,14 +177,14 @@ def test_receive_03(cdstar_archive):
assert cdstar_json["files"]
for file_info in cdstar_json["files"]:
# external request: use requests directly instead of the TestClient
meta_json = get_json_of_uri(
meta_json = test_utils.get_json_of_uri(
f"{config.BasicSettings().couch_uri}/{config.BasicSettings().couch_db}/{file_info['id']}",
auth=(
config.BasicSettings().couch_user,
config.BasicSettings().couch_pass,
),
)
assert is_schemaorg_jsonld(meta_json)
assert test_utils.is_schemaorg_jsonld(meta_json)
# compare CDSTAR and CouchDB metadata values
assert meta_json["identifier"] == file_info["id"]
......@@ -250,6 +197,8 @@ def test_receive_03(cdstar_archive):
file_info["modified"], "%Y-%m-%dT%H:%M:%S.%f%z"
) == datetime.datetime.fromisoformat(meta_json["dateModified"])
test_utils.cleanup_metafile(cdstar_archive)
def test_receive_04(cdstar_archive):
"""
......@@ -257,7 +206,7 @@ def test_receive_04(cdstar_archive):
"""
post_valid_receive_request(cdstar_archive)
meta_json = get_json_of_uri(
meta_json = test_utils.get_json_of_uri(
f"{config.BasicSettings().couch_uri}/{config.BasicSettings().couch_db}/{cdstar_archive}",
auth=(config.BasicSettings().couch_user, config.BasicSettings().couch_pass),
)
......@@ -265,24 +214,37 @@ def test_receive_04(cdstar_archive):
# iterate through the parts of a archive metadata
assert meta_json["hasPart"]
for partOf in meta_json["hasPart"]:
assert is_schemaorg_reference(partOf)
assert test_utils.is_schemaorg_reference(partOf)
# follow the reference and check validity
meta_file_json = get_json_of_uri(
meta_file_json = test_utils.get_json_of_uri(
partOf["@id"],
auth=(config.BasicSettings().couch_user, config.BasicSettings().couch_pass),
)
assert is_schemaorg_jsonld(meta_file_json)
assert test_utils.is_schemaorg_jsonld(meta_file_json)
assert meta_file_json["isPartOf"]
assert is_schemaorg_reference(meta_file_json["isPartOf"])
assert test_utils.is_schemaorg_reference(meta_file_json["isPartOf"])
# check the backreference, does hasPart <-> isPartOf work?
assert meta_file_json["isPartOf"]["@id"] == meta_json["@id"]
backref_json = get_json_of_uri(
backref_json = test_utils.get_json_of_uri(
meta_file_json["isPartOf"]["@id"],
auth=(config.BasicSettings().couch_user, config.BasicSettings().couch_pass),
)
assert is_schemaorg_jsonld(backref_json)
assert test_utils.is_schemaorg_jsonld(backref_json)
test_utils.cleanup_metafile(cdstar_archive)
def test_receive_05(cdstar_archive):
"""
Tests if a receive request yields a memory for checking afterwards.
"""
response_data = post_valid_receive_request(cdstar_archive)
assert test_utils.has_memory_archive(response_data)
test_utils.cleanup_metafile(cdstar_archive)
def test_receive_invalid_archive_01():
......@@ -296,12 +258,3 @@ def test_receive_invalid_archive_01():
assert len(response_data["result"]["errors"]) >= 1
assert len(response_data["result"]["logs"]) == 0
assert len(response_data["result"]["messages"]) == 0
def test_check_01():
register_request = {
"method": "check",
"params": {"message": None, "options": {}, "memory": {}, "credentials": {},},
}
response = client.post(f"/{endpoint_name}", json=register_request)
assert response.status_code >= 200 and response.status_code <= 299
import tempfile
import os
from typing import Any, Dict, Tuple
import requests
def is_valid_response(json: Dict[str, Any]) -> bool:
# see: https://github.com/automaticmode/active_workflow/blob/master/docs/remote_agent_api.md#receive-method
assert "result" in json.keys()
# assert each key is present in `json`
for key in ["errors", "logs", "memory", "messages"]:
assert key in json["result"]
# assert that these keys are lists
for key in ["errors", "logs", "messages"]:
assert type(json["result"][key]) == list
# assert that these keys are lists holding strings
for key in ["errors", "logs"]:
if len(json["result"][key]) > 0:
for entry in json["result"][key]:
assert type(entry) == str
# assert that this key is a list holding dictionaries
if len(json["result"]["messages"]) > 0:
for entry in json["result"]["messages"]:
assert type(entry) == dict
# assert that this key is a dictionary
assert type(json["result"]["memory"]) == dict
return True
def cleanup_metafile(archive_id: str):
"""
Deletes the meta file that is written by receive.
"""
os.unlink(os.path.join(tempfile.gettempdir(), f"meta_{archive_id}.json"))
def is_schemaorg_jsonld(json: Dict[str, Any]) -> bool:
"""
Return true if `json` contains `@context`, `@id` and `@type` as well as `@context` refers to
`http://schema.org`.
"""
keys = json.keys()
return (
"@context" in keys
and "@id" in keys
and "@type" in keys
and json["@context"].startswith("http://schema.org")
)
def is_schemaorg_reference(reference: Dict[str, Any]) -> bool:
"""
Returns true if `reference` contains a valid JSON-LD reference, i.e. contains keys `@type`
and `@id`.
"""
keys = reference.keys()
return "@type" in keys and "@id" in keys
def get_json_of_uri(uri: str, auth: Tuple[str, str] = None) -> Dict[str, Any]:
"""
Retrieves the JSON response of `uri` under BasicAuth `auth` if supplied.
"""
# Make the request and assert it was successful
r = requests.get(uri, auth=auth)
assert r.status_code >= 200 and r.status_code <= 299
# Get the JSON response and assert it is a dictionary (no application used in testing responds with lists afaik)
r_json = r.json()
assert r_json
assert isinstance(r_json, dict)
return r_json
def has_memory_archive(json: Dict[str, Any]) -> bool:
"""
Checks if a response JSON contains a valid archive in its memory.
"""
assert "archives" in json["result"]["memory"].keys()
assert type(json["result"]["memory"]["archives"]) == list
assert len(json["result"]["memory"]["archives"]) > 0
return True
Supports Markdown