Skip to content
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.

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

EM-1514 Uncomment product ID tests #263

Merged
merged 6 commits into from
Jan 29, 2025
Merged
Changes from all commits
Commits
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
164 changes: 65 additions & 99 deletions tests/test_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,66 +29,27 @@ def test_api_not_accessible_by_api_key(proxy_path, nhsd_apim_proxy_url, nhsd_api
assert resp.status_code == 401


@pytest.mark.nhsd_apim_authorization({"access": "application", "level": "level3"})
def test_post_event(
nhsd_apim_proxy_url,
nhsd_apim_auth_headers,
mns_test_signal_event,
_apigee_app_base_url,
_create_test_app
):
# Update client app via Apigee with necessary custom attributes
created_test_app_name = _create_test_app["name"]
apigee_update_url = f"{_apigee_app_base_url}/{created_test_app_name}"
attributes_to_update = {
"attributes": [
{
"name": "permissions",
"value": "events:create:mns-test-signal-1"
},
{
"name": "product-id",
"value": "P-MNS-TST"
},
{
"name": "product-device-id",
"value": str(uuid.uuid4())
},
]
}

update_response = requests.put(
f"{apigee_update_url}",
json=attributes_to_update,
headers={"Authorization": f"Bearer {os.environ['APIGEE_ACCESS_TOKEN']}"}
)
update_response.raise_for_status()

# POST event to /events endpoint
nhsd_apim_auth_headers["X-Correlation-ID"] = f"apim-smoketests-{uuid.uuid4()}"
nhsd_apim_auth_headers["Content-Type"] = "application/cloudevents+json"
retries = 0

while retries < 5:
resp = requests.post(
f"{nhsd_apim_proxy_url}/events",
headers=nhsd_apim_auth_headers,
json=mns_test_signal_event
)

if resp.status_code == 403:
retries = retries + 1
continue

break

assert resp.status_code == 200
assert resp.json() == {"id": mns_test_signal_event["id"]}


@pytest.mark.parametrize(
"attributes_to_update, expected_status_code, expected_error_message",
"attributes_to_update, expected_status_code, expected_error",
[
(
[
{
"name": "permissions",
"value": "events:create:mns-test-signal-1"
},
{
"name": "product-id",
"value": "P-MNS-TST"
},
{
"name": "product-device-id",
"value": str(uuid.uuid4())
},
],
200,
None,
),
(
[
{
Expand All @@ -101,49 +62,48 @@ def test_post_event(
},
],
403,
"Unauthorised",
{"errors": "Unauthorised"},
),
(
[
{
"name": "permissions",
"value": "events:create:mns-test-signal-1",
},
{
"name": "product-device-id",
"value": str(uuid.uuid4()),
},
],
502,
{"message": "Internal server error"},
),
(
[
{
"name": "permissions",
"value": "events:create:mns-test-signal-1",
},
{
"name": "product-id",
"value": "P-MNS-TST",
},
],
502,
{"message": "Internal server error"},
),
# UNCOMMENT AFTER BACKEND CHECKS FOR NHSE-Product-ID HEADER
# (
# [
# {
# "name": "permissions",
# "value": "events:create:mns-test-signal-1",
# },
# {
# "name": "product-device-id",
# "value": str(uuid.uuid4()),
# },
# ],
# 500,
# "Missing required attribute 'product-id' on calling client application",
# ),
# (
# [
# {
# "name": "permissions",
# "value": "events:create:mns-test-signal-1",
# },
# {
# "name": "product-id",
# "value": "P-MNS-TST",
# },
# ],
# 500,
# "Missing required attribute 'product-device-id' on calling client application",
# ),
]
)
@pytest.mark.nhsd_apim_authorization({"access": "application", "level": "level3"})
def test_missing_required_target_attributes(
def test_post_event(
nhsd_apim_proxy_url,
nhsd_apim_auth_headers,
mns_test_signal_event,
_apigee_app_base_url,
_create_test_app,
attributes_to_update,
expected_status_code,
expected_error_message
expected_error
):
# Update client app via Apigee with necessary custom attributes
created_test_app_name = _create_test_app["name"]
Expand All @@ -160,14 +120,20 @@ def test_missing_required_target_attributes(
nhsd_apim_auth_headers["X-Correlation-ID"] = f"apim-smoketests-{uuid.uuid4()}"
nhsd_apim_auth_headers["Content-Type"] = "application/cloudevents+json"

# Wait 5 seconds as apigee can take time to register update
sleep(5)
retries = 0
while retries < 5:
resp = requests.post(
f"{nhsd_apim_proxy_url}/events",
headers=nhsd_apim_auth_headers,
json=mns_test_signal_event
)

if resp.status_code != expected_status_code:
retries = retries + 1
sleep(1)
continue

resp = requests.post(
f"{nhsd_apim_proxy_url}/events",
headers=nhsd_apim_auth_headers,
json=mns_test_signal_event
)
break

assert resp.status_code == expected_status_code
assert resp.json() == {"errors": expected_error_message}
assert resp.status_code == expected_status_code, resp.text
assert resp.json() == expected_error if expected_error else {"id": mns_test_signal_event["id"]}
Loading