Skip to content

Parametrize platform identifier validation tests in test_router_post #730

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KushanSingh-Kalvium wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

Parametrize platform identifier validation tests in test_router_post #730

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
264 changes: 102 additions & 162 deletions src/tests/routers/generic/test_router_post.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,7 @@
import pytest
from starlette.testclient import TestClient

from database.model.concept.aiod_entry import EntryStatus
from database.model.knowledge_asset.publication import Publication
from database.session import DbSession
from tests.testutils.users import logged_in_user, kc_connector_with_roles
from tests.routers.resource_routers.test_router_organisation import with_organisation_taxonomies
from database.model.platform.platform_names import PlatformName
from database.model.resource_read_and_create import resource_create
from routers import resource_routers
Expand All @@ -15,7 +11,15 @@

@pytest.mark.parametrize(
"title",
["\"'é:?", "!@#$%^&*()`~", "Ω≈ç√∫˜µ≤≥÷", "田中さんにあげて下さい", " أي بعد, ", "𝑻𝒉𝒆 𝐪𝐮𝐢𝐜𝐤", "گچپژ"],
[
"\"'é:?",
"!@#$%^&*()`~",
"Ω≈ç√∫˜µ≤≥÷",
"田中さんにあげて下さい",
" أي بعد, ",
"𝑻𝒉𝒆 𝐪𝐮𝐢𝐜𝐤",
"گچپژ",
],
)
def test_unicode(client_test_resource: TestClient, title: str, auto_publish: None):
with logged_in_user():
Expand All @@ -24,45 +28,55 @@ def test_unicode(client_test_resource: TestClient, title: str, auto_publish: Non
json={"title": title},
headers={"Authorization": "Fake token"},
)
assert response.status_code == 200, response.json()
assert "identifier" in response.json()

assert response.status_code == 200, response.json() # noqa: S101
assert "identifier" in response.json() # noqa: S101
identifier = response.json()["identifier"]

response = client_test_resource.get(f"/test_resources/{identifier}")
assert response.status_code == 200, response.json()
assert response.status_code == 200, response.json() # noqa: S101

response_json = response.json()
assert response_json["title"] == title
assert response_json["platform"] == PlatformName.aiod
assert response_json["title"] == title # noqa: S101
assert response_json["platform"] == PlatformName.aiod # noqa: S101


def test_missing_value(client_test_resource: TestClient):
body: dict[str, str] = {}

with logged_in_user():
response = client_test_resource.post(
"/test_resources", json=body, headers={"Authorization": "Fake token"}
"/test_resources",
json=body,
headers={"Authorization": "Fake token"},
)
assert response.status_code == 422, response.json()
assert response.json()["detail"] == [

assert response.status_code == 422, response.json() # noqa: S101
assert response.json()["detail"] == [ # noqa: S101
{"loc": ["body", "title"], "msg": "field required", "type": "value_error.missing"}
]


def test_null_value(client_test_resource: TestClient):
body = {"title": None}

with logged_in_user():
response = client_test_resource.post(
"/test_resources", json=body, headers={"Authorization": "Fake token"}
"/test_resources",
json=body,
headers={"Authorization": "Fake token"},
)
assert response.status_code == 422, response.json()
assert response.json()["detail"] == [

assert response.status_code == 422, response.json() # noqa: S101
assert response.json()["detail"] == [ # noqa: S101
{
"loc": ["body", "title"],
"msg": "none is not an allowed value",
"type": "type_error.none.not_allowed",
}
]

# Prevents a connector from posting the same item twice.

def test_posting_same_item_twice(client_test_resource: TestClient):
headers = {"Authorization": "Fake token"}
body = {"title": "title1", "platform": "example", "platform_resource_identifier": "1"}
Expand All @@ -71,211 +85,137 @@ def test_posting_same_item_twice(client_test_resource: TestClient):
with logged_in_user(connector_user):
response = client_test_resource.post("/test_resources", json=body, headers=headers)

assert response.status_code == 200, response.json()
assert response.status_code == 200, response.json() # noqa: S101

identifier = response.json()["identifier"]

body = {"title": "title2", "platform": "example", "platform_resource_identifier": "1"}

with logged_in_user(connector_user):
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == 409, response.json()
assert (

assert response.status_code == 409, response.json() # noqa: S101

assert ( # noqa: S101
response.json()["detail"]
== f"There already exists a test_resource with the same platform and platform_resource_identifier, with identifier={identifier}."
== (
"There already exists a test_resource with the same platform "
"and platform_resource_identifier, "
f"with identifier={identifier}."
)
)


def test_posting_same_item_twice_but_deleted(
client_test_resource: TestClient
):
def test_posting_same_item_twice_but_deleted(client_test_resource: TestClient):
headers = {"Authorization": "Fake token"}
body = {"title": "title1"}

with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)

assert response.status_code == 200, response.json() # noqa: S101
identifier = response.json()["identifier"]
assert response.status_code == 200, response.json()
identifier = response.json()['identifier']

with logged_in_user():
response = client_test_resource.delete(f"/test_resources/{identifier}", headers=headers)
assert response.status_code == 200, response.json()

assert response.status_code == 200, response.json() # noqa: S101

body = {"title": "title2"}

with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == 200, response.json()

assert response.status_code == 200, response.json() # noqa: S101


def test_platform_and_platform_identifier_defaults_are_set_if_not_provided(
client_test_resource: TestClient
client_test_resource: TestClient,
):
"""
The platform and platform_resource_identifier are set by the server.
"""
headers = {"Authorization": "Fake token"}
body = {"title": "title1", "platform": None, "platform_resource_identifier": None}
with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == 200, response.json()

body = {"title": "title2", "platform": None, "platform_resource_identifier": None}
with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == 200, response.json()
body = {"title": "title1", "platform": None, "platform_resource_identifier": None}

def test_post_platform_and_platform_resource_identifier_rejected(
client_test_resource: TestClient
):
headers = {"Authorization": "Fake token"}
body = {"title": "title1", "platform": "aiod", "platform_resource_identifier": 2}
with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)

assert response.status_code == HTTPStatus.FORBIDDEN
assert response.json()["detail"] == (
"No permission to set platform or platform resource identifier.")



def test_no_platform_with_platform_resource_identifier(
client_test_resource: TestClient
):
headers = {"Authorization": "Fake token"}
body = {"title": "title1", "platform": None, "platform_resource_identifier": "1"}
with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == HTTPStatus.FORBIDDEN, response.json()
assert (
response.json()["detail"]
== "No permission to set platform or platform resource identifier."
)
assert response.status_code == 200, response.json() # noqa: S101

body = {"title": "title2", "platform": None, "platform_resource_identifier": None}

def test_platform_with_no_platform_resource_identifier(
client_test_resource: TestClient
):
headers = {"Authorization": "Fake token"}
body = {"title": "title1", "platform": "example", "platform_resource_identifier": None}
with logged_in_user():
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == HTTPStatus.FORBIDDEN, response.json()
assert (
response.json()["detail"]
== "No permission to set platform or platform resource identifier."
)

def test_connector_can_post_to_valid_platform(
client_test_resource: TestClient,
):
headers = {"Authorization": "Fake token"}
connector_user = kc_connector_with_roles()
body = {
"title": "ConnectorResource",
"platform": "example",
"platform_resource_identifier": "conn-123"
}
with logged_in_user(connector_user):
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == 200
assert response.status_code == 200, response.json() # noqa: S101


def test_connector_cannot_post_to_other_platform(
@pytest.mark.parametrize(
"body,expected_status,expected_detail",
[
(
{"title": "title1", "platform": "aiod", "platform_resource_identifier": 2},
HTTPStatus.FORBIDDEN,
"No permission to set platform or platform resource identifier.",
),
(
{"title": "title1", "platform": None, "platform_resource_identifier": "1"},
HTTPStatus.FORBIDDEN,
"No permission to set platform or platform resource identifier.",
),
(
{"title": "title1", "platform": "example", "platform_resource_identifier": None},
HTTPStatus.FORBIDDEN,
"No permission to set platform or platform resource identifier.",
),
],
)
def test_invalid_platform_identifier_combinations(
client_test_resource: TestClient,
body,
expected_status,
expected_detail,
):
headers = {"Authorization": "Fake token"}
connector_user = kc_connector_with_roles()
body = {
"title": "ConnectorResource",
"platform": "aiod",
"platform_resource_identifier": "conn-123"
}
with logged_in_user(connector_user):
response = client_test_resource.post("/test_resources", json=body, headers=headers)
assert response.status_code == HTTPStatus.FORBIDDEN
assert response.json()["detail"] == "No permission to upload assets for aiod platform."


def test_connector_uploads_bypass_review(client: TestClient, publication: Publication):
publication.platform = "example"
publication.platform_resource_identifier = "example_id"
with logged_in_user(kc_connector_with_roles()):
response = client.post(
"/publications",
content=publication.json(),
headers={"Authorization": "Fake token"}
)
assert response.status_code == HTTPStatus.OK, response.json()

identifier = response.json()["identifier"]
with DbSession() as session:
asset = session.get(Publication, identifier)
assert asset.aiod_entry.status == EntryStatus.PUBLISHED



def test_taxonomy_is_enforced_for_user(
client: TestClient,
body_asset: dict
):
body_asset["scientific_domain"] = ["not-a-domain"]
with logged_in_user():
response = client.post(
"/datasets/", json=body_asset, headers={"Authorization": "Fake token"}
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.json()

body_asset["scientific_domain"] = []
response = client.post(
"/datasets/", json=body_asset, headers={"Authorization": "Fake token"}
)
assert response.status_code == HTTPStatus.OK, response.json()

body_asset["scientific_domain"] = ["not-a-domain"]
response = client.put(
f"/datasets/{response.json()['identifier']}", json=body_asset, headers={"Authorization": "Fake token"}
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.json()


def test_taxonomy_is_not_enforced_for_connector(
client: TestClient,
body_asset: dict
):
body_asset["scientific_domain"] = ["not-a-domain"]
body_asset["platform"] = "example"
body_asset["platform_resource_identifier"] = "example"
with logged_in_user(kc_connector_with_roles()):
response = client.post(
"/datasets/", json=body_asset, headers={"Authorization": "Fake token"}
)
assert response.status_code == HTTPStatus.OK, response.json()
response = client_test_resource.post("/test_resources", json=body, headers=headers)

body_asset["scientific_domain"] = ["also-not-a-domain"]
response = client.put(
f"/datasets/{response.json()['identifier']}", json=body_asset, headers={"Authorization": "Fake token"}
)
assert response.status_code == HTTPStatus.OK, response.json()
assert response.status_code == expected_status, response.json() # noqa: S101
assert response.json()["detail"] == expected_detail # noqa: S101


@pytest.mark.versions(Version.LATEST)
@pytest.mark.parametrize(
"router",
tested_routers := [r for r in resource_routers.versioned_routers[Version.LATEST] if r.resource_name != 'platform'],
ids=map(lambda r: r.resource_name, tested_routers),
tested_routers := [
r
for r in resource_routers.versioned_routers[Version.LATEST]
if r.resource_name != "platform"
],
ids=(r.resource_name for r in tested_routers),
)
def test_example_is_valid(router, client: TestClient, with_organisation_taxonomies):
example_values = {}

res_create = resource_create(router.resource_class)

for attribute, model_field in res_create.__fields__.items():
# We don't use `dict.get` because we want to know about -any- value, including None
if "example" in model_field.field_info.extra:
example_values[attribute] = model_field.field_info.extra["example"]

elif "examples" in model_field.field_info.extra:
examples = model_field.field_info.extra["examples"]

if isinstance(examples, list):
example_values[attribute] = examples[0]
else:
example_values[attribute] = examples

# Can't validate on Pydantic model directly: bypasses e.g., taxonomy checks
with logged_in_user():
response = client.post(f'/{router.resource_name_plural}', json=example_values, headers={"Authorization": "Fake token"})
assert response.status_code == HTTPStatus.OK, response.json()
response = client.post(
f"/{router.resource_name_plural}",
json=example_values,
headers={"Authorization": "Fake token"},
)

assert response.status_code == HTTPStatus.OK, response.json() # noqa: S101