generated from Luis/nextjs-python-web-template
284 lines
9.1 KiB
Python
284 lines
9.1 KiB
Python
import urllib.parse as url
|
|
from typing import Any
|
|
|
|
from api import TestClient
|
|
|
|
default_entity_did_url = "entity_did=did%3Asov%3Atest%3A1234"
|
|
default_entity_did = "did:sov:test:1234"
|
|
default_entity_did2 = "did:sov:test:1235"
|
|
default_entity_did3 = "did:sov:test:1236"
|
|
default_entity_did4 = "did:sov:test:1237"
|
|
default_entity_did5 = "did:sov:test:1238"
|
|
|
|
|
|
def assert_extra_info(
|
|
infos: list[str],
|
|
request_body: dict[str, Any],
|
|
response: dict[str, str],
|
|
) -> None:
|
|
# print(type())
|
|
for info in infos:
|
|
assert info in response.keys()
|
|
# TODO maybe check the content of the extra info ...
|
|
response.pop(info)
|
|
assert response == request_body
|
|
|
|
|
|
def make_test_post_and_get(
|
|
api: TestClient,
|
|
request_body: dict[str, Any],
|
|
paramter: str,
|
|
get_request: str = default_entity_did_url,
|
|
apiversion: str = "v1",
|
|
) -> None:
|
|
# test post
|
|
response = api.post(
|
|
f"/api/{apiversion}/create_{paramter}",
|
|
json=request_body,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
assert response.status_code == 200
|
|
if paramter == "repository":
|
|
assert_extra_info(["time_created"], request_body, response.json())
|
|
elif paramter == "consumer":
|
|
assert_extra_info(["id"], request_body, response.json())
|
|
elif paramter == "entity":
|
|
assert_extra_info(
|
|
["consumers", "producers", "repository"], request_body, response.json()
|
|
)
|
|
else:
|
|
assert response.json() == request_body
|
|
# test get
|
|
response = api.get(
|
|
f"api/{apiversion}/get_{paramter}?{get_request}&skip=0&limit=100"
|
|
)
|
|
assert response.status_code == 200
|
|
if paramter == "repository":
|
|
assert_extra_info(["time_created"], request_body, response.json()[0])
|
|
elif paramter == "consumer":
|
|
assert_extra_info(["id"], request_body, response.json()[0])
|
|
elif paramter == "entity":
|
|
assert_extra_info(
|
|
["consumers", "producers", "repository"], request_body, response.json()
|
|
)
|
|
else:
|
|
assert response.json() == [request_body]
|
|
|
|
|
|
#########################
|
|
# #
|
|
# Producer #
|
|
# #
|
|
#########################
|
|
def test_producer(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30df",
|
|
"service_name": "Carlo'''s Printing",
|
|
"service_type": "3D Printing",
|
|
"endpoint_url": "http://127.0.0.1:8000",
|
|
"status": "unknown",
|
|
"other": {"test": "test"},
|
|
"entity_did": default_entity_did,
|
|
}
|
|
paramter = "producer"
|
|
# get_request = "entity_did=did%3Asov%3Atest%3A1234"
|
|
make_test_post_and_get(api, request_body, paramter)
|
|
|
|
|
|
def test_producer2(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d1",
|
|
"service_name": "Luis'''s Fax",
|
|
"service_type": "Fax",
|
|
"endpoint_url": "http://127.0.0.1:8001",
|
|
"status": "unknown",
|
|
"other": {"faxen": "dicke"},
|
|
"entity_did": default_entity_did2,
|
|
}
|
|
paramter = "producer"
|
|
get_request = "entity_did=" + url.quote(default_entity_did2)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
def test_producer3(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d2",
|
|
"service_name": "Erdem'''s VR-Stream",
|
|
"service_type": "VR-Stream",
|
|
"endpoint_url": "http://127.0.0.1:8002",
|
|
"status": "unknown",
|
|
"other": {"oculos": "rift"},
|
|
"entity_did": default_entity_did3,
|
|
}
|
|
paramter = "producer"
|
|
get_request = "entity_did=" + url.quote(default_entity_did3)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
def test_producer4(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d3",
|
|
"service_name": "Onur'''s gallary",
|
|
"service_type": "gallary",
|
|
"endpoint_url": "http://127.0.0.1:8003",
|
|
"status": "unknown",
|
|
"other": {"nice": "pics"},
|
|
"entity_did": default_entity_did4,
|
|
}
|
|
paramter = "producer"
|
|
get_request = "entity_did=" + url.quote(default_entity_did4)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
def test_producer5(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d4",
|
|
"service_name": "Sara'''s Game-Shop",
|
|
"service_type": "Game-Shop",
|
|
"endpoint_url": "http://127.0.0.1:8004",
|
|
"status": "unknown",
|
|
"other": {"war": "games"},
|
|
"entity_did": default_entity_did5,
|
|
}
|
|
paramter = "producer"
|
|
get_request = "entity_did=" + url.quote(default_entity_did5)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
#########################
|
|
# #
|
|
# Consumer #
|
|
# #
|
|
#########################
|
|
def test_consumer(api: TestClient) -> None:
|
|
request_body = {
|
|
"entity_did": default_entity_did,
|
|
"producer_uuid": "8e285c0c-4e40-430a-a477-26b3b81e30df",
|
|
"other": {"test": "test"},
|
|
}
|
|
paramter = "consumer"
|
|
# get_request = "entity_did=did%3Asov%3Atest%3A1234"
|
|
make_test_post_and_get(api, request_body, paramter)
|
|
|
|
|
|
def test_consumer2(api: TestClient) -> None:
|
|
request_body = {
|
|
"entity_did": default_entity_did2,
|
|
"producer_uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d4",
|
|
"other": {"war": "games"},
|
|
}
|
|
paramter = "consumer"
|
|
get_request = "entity_did=" + url.quote(default_entity_did2)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
#########################
|
|
# #
|
|
# REPOSITORY #
|
|
# #
|
|
#########################
|
|
def test_repository(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30df",
|
|
"service_name": "Carlo'''s Printing",
|
|
"service_type": "3D Printing",
|
|
"endpoint_url": "http://127.0.0.1:8000",
|
|
"status": "unknown",
|
|
"other": {"test": "test"},
|
|
"entity_did": default_entity_did,
|
|
}
|
|
paramter = "repository"
|
|
# get_request = "entity_did=did%3Asov%3Atest%3A1234"
|
|
make_test_post_and_get(api, request_body, paramter)
|
|
|
|
|
|
def test_repository2(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d1",
|
|
"service_name": "Luis'''s Fax",
|
|
"service_type": "Fax",
|
|
"endpoint_url": "http://127.0.0.1:8001",
|
|
"status": "unknown",
|
|
"other": {"faxen": "dicke"},
|
|
"entity_did": default_entity_did2,
|
|
}
|
|
paramter = "repository"
|
|
get_request = "entity_did=" + url.quote(default_entity_did2)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
def test_repository3(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d2",
|
|
"service_name": "Erdem'''s VR-Stream",
|
|
"service_type": "VR-Stream",
|
|
"endpoint_url": "http://127.0.0.1:8002",
|
|
"status": "unknown",
|
|
"other": {"oculos": "rift"},
|
|
"entity_did": default_entity_did3,
|
|
}
|
|
paramter = "repository"
|
|
get_request = "entity_did=" + url.quote(default_entity_did3)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
def test_repository4(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d3",
|
|
"service_name": "Onur'''s gallary",
|
|
"service_type": "gallary",
|
|
"endpoint_url": "http://127.0.0.1:8003",
|
|
"status": "unknown",
|
|
"other": {"nice": "pics"},
|
|
"entity_did": default_entity_did4,
|
|
}
|
|
paramter = "repository"
|
|
get_request = "entity_did=" + url.quote(default_entity_did4)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
def test_repository5(api: TestClient) -> None:
|
|
request_body = {
|
|
"uuid": "8e285c0c-4e40-430a-a477-26b3b81e30d4",
|
|
"service_name": "Sara'''s Game-Shop",
|
|
"service_type": "Game-Shop",
|
|
"endpoint_url": "http://127.0.0.2:8004",
|
|
"status": "unknown",
|
|
"other": {"war": "games"},
|
|
"entity_did": default_entity_did5,
|
|
}
|
|
paramter = "repository"
|
|
get_request = "entity_did=" + url.quote(default_entity_did5)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|
|
|
|
|
|
#########################
|
|
# #
|
|
# Entity #
|
|
# #
|
|
#########################
|
|
def test_entity(api: TestClient) -> None:
|
|
request_body = {
|
|
"did": default_entity_did,
|
|
"name": "C1",
|
|
"ip": "127.0.0.1",
|
|
"attached": False,
|
|
"other": {"test": "test"},
|
|
}
|
|
paramter = "entity"
|
|
# get_request = "entity_did=did%3Asov%3Atest%3A1234"
|
|
make_test_post_and_get(api, request_body, paramter)
|
|
|
|
|
|
def test_entity2(api: TestClient) -> None:
|
|
request_body = {
|
|
"did": default_entity_did2,
|
|
"name": "C2",
|
|
"ip": "127.0.0.2",
|
|
"attached": False,
|
|
"other": {"test": "test"},
|
|
}
|
|
paramter = "entity"
|
|
get_request = "entity_did=" + url.quote(default_entity_did2)
|
|
make_test_post_and_get(api, request_body, paramter, get_request)
|