Skip to content

Commit

Permalink
Runner test script improperly generating survey metadata (#1302)
Browse files Browse the repository at this point in the history
  • Loading branch information
@VirajP1002
VirajP1002 authored Feb 28, 2024
1 parent 2f539a5 commit c5fdfe3
Show file tree
Hide file tree
Showing 3 changed files with 246 additions and 18 deletions.
19 changes: 13 additions & 6 deletions tests/integration/create_token.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import time
from copy import deepcopy
from time import time
from uuid import uuid4

from sdc.crypto.encrypter import encrypt
Expand Down Expand Up @@ -94,6 +94,11 @@
}


def populate_with_extra_payload_items(extra_payload, payload):
for key, value in extra_payload.items():
payload[key] = value


class TokenGenerator:
def __init__(self, key_store, upstream_kid, sr_public_kid):
self._key_store = key_store
Expand All @@ -109,7 +114,7 @@ def _get_payload_with_params(
payload=PAYLOAD,
**extra_payload,
): # pylint: disable=dangerous-default-value
payload_vars = payload.copy()
payload_vars = deepcopy(payload)
payload_vars["tx_id"] = str(uuid4())
if schema_name:
payload_vars["schema_name"] = schema_name
Expand All @@ -118,14 +123,16 @@ def _get_payload_with_params(
if cir_instrument_id:
payload_vars["cir_instrument_id"] = cir_instrument_id

payload_vars["iat"] = time.time()
payload_vars["iat"] = time()
payload_vars["exp"] = payload_vars["iat"] + float(3600) # one hour from now
payload_vars["jti"] = str(uuid4())
payload_vars["case_id"] = str(uuid4())
payload_vars["response_expires_at"] = get_response_expires_at()

for key, value in extra_payload.items():
payload_vars[key] = value
if data := payload_vars.get("survey_metadata"):
populate_with_extra_payload_items(extra_payload, data.get("data"))
else:
populate_with_extra_payload_items(extra_payload, payload_vars)

return payload_vars

Expand All @@ -147,7 +154,7 @@ def create_token_v2(self, schema_name, theme="default", **extra_payload):
return self.generate_token(payload)

def create_supplementary_data_token(self, schema_name, **extra_payload):
payload = deepcopy(PAYLOAD_V2_SUPPLEMENTARY_DATA)
payload = PAYLOAD_V2_SUPPLEMENTARY_DATA

# iterate over a copy so items can be deleted
for key, value in list(extra_payload.items()):
Expand Down
44 changes: 32 additions & 12 deletions tests/integration/session/test_login.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,16 @@ def test_login_with_invalid_questionnaire_claims_should_be_forbidden(self):

self.assertStatusForbidden()

def test_login_with_invalid_questionnaire_claims_should_be_forbidden_v2_get(self):
# flag_1 should be a boolean
token = self.token_generator.create_token_v2(
"test_metadata_routing", flag_1=123
)

self.get(url=f"/session?token={token}")

self.assertStatusForbidden()

def test_login_with_invalid_version_should_be_forbidden(self):
token = self.token_generator.create_token_invalid_version("test_checkbox")

Expand All @@ -167,7 +177,7 @@ def test_login_token_with_schema_url_should_redirect_to_survey(self):
)

# When
with HTTMock(self.schema_url_mock):
with HTTMock(self._schema_url_mock):
self.get(url=f"/session?token={token}")

self.assertStatusOK()
Expand All @@ -182,23 +192,23 @@ def test_login_token_with_incorrect_schema_url_results_in_500(self):
)

# When
with HTTMock(self.schema_url_mock_500):
with HTTMock(self._schema_url_mock_500):
self.get(url=f"/session?token={token}")

# Then
self.assertException()

@staticmethod
@urlmatch(netloc=r"eq-survey-register", path=r"\/my-test-schema")
def schema_url_mock(_url, _request):
def _schema_url_mock(_url, _request):
schema_path = SCHEMA_PATH_MAP["test"]["en"]["test_textarea"]

with open(schema_path, encoding="utf8") as json_data:
return json_data.read()

@staticmethod
@urlmatch(netloc=r"eq-survey-register", path=r"\/my-test-schema-not-found")
def schema_url_mock_500(_url, _request):
def _schema_url_mock_500(_url, _request):
return response(500)


Expand Down Expand Up @@ -287,6 +297,16 @@ def test_login_with_invalid_questionnaire_claims_should_be_forbidden(self):

self.assertStatusForbidden()

def test_login_with_invalid_questionnaire_claims_should_be_forbidden_v2_post(self):
# flag_1 should be a boolean
token = self.token_generator.create_token_v2(
"test_metadata_routing", flag_1=123
)

self.get(url=f"/session?token={token}")

self.assertStatusForbidden()

def test_v2_business_login_with_invalid_questionnaire_claims_should_be_forbidden(
self,
):
Expand Down Expand Up @@ -330,7 +350,7 @@ def test_login_token_with_schema_url_should_redirect_to_survey(self):
)

# When
with HTTMock(self.schema_url_mock):
with HTTMock(self._schema_url_mock):
self.post(url=f"/session?token={token}")

self.assertStatusOK()
Expand All @@ -345,7 +365,7 @@ def test_login_token_with_incorrect_schema_url_results_in_404(self):
)

# When
with HTTMock(self.schema_url_mock_500):
with HTTMock(self._schema_url_mock_500):
self.post(url=f"/session?token={token}")

# Then
Expand All @@ -368,7 +388,7 @@ def test_login_token_with_cir_instrument_id_should_redirect_to_survey(self):
)

# When
with HTTMock(self.cir_url_mock):
with HTTMock(self._cir_url_mock):
self.post(url=f"/session?token={token}")

# Then
Expand All @@ -384,15 +404,15 @@ def test_login_token_with_invalid_cir_instrument_id_results_in_500(self):
)

# When
with HTTMock(self.cir_url_mock_500):
with HTTMock(self._cir_url_mock_500):
self.post(url=f"/session?token={token}")

# Then
self.assertException()

@staticmethod
@urlmatch(netloc=r"eq-survey-register", path=r"\/my-test-schema")
def schema_url_mock(_url, _request):
def _schema_url_mock(_url, _request):
schema_path = SCHEMA_PATH_MAP["test"]["en"]["test_textarea"]

with open(schema_path, encoding="utf8") as json_data:
Expand All @@ -403,7 +423,7 @@ def schema_url_mock(_url, _request):
path=CIR_RETRIEVE_COLLECTION_INSTRUMENT_URL,
query="guid=f0519981-426c-8b93-75c0-bfc40c66fe25",
)
def cir_url_mock(_url, _request):
def _cir_url_mock(_url, _request):
schema_path = SCHEMA_PATH_MAP["test"]["en"]["test_textarea"]

with open(schema_path, encoding="utf8") as json_data:
Expand All @@ -414,10 +434,10 @@ def cir_url_mock(_url, _request):
path=CIR_RETRIEVE_COLLECTION_INSTRUMENT_URL,
query="guid=a0df1208-dff5-4a3d-b35d-f9620c4a48ef",
)
def cir_url_mock_500(_url, _request):
def _cir_url_mock_500(_url, _request):
return response(500)

@staticmethod
@urlmatch(netloc=r"eq-survey-register", path=r"\/my-test-schema-not-found")
def schema_url_mock_500(_url, _request):
def _schema_url_mock_500(_url, _request):
return response(500)
201 changes: 201 additions & 0 deletions tests/integration/test_create_token.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,201 @@
from unittest import mock

from app.authentication.authenticator import decrypt_token
from tests.integration.app_context_test_case import AppContextTestCase
from tests.integration.create_token import (
PAYLOAD_V2_BUSINESS,
PAYLOAD_V2_SOCIAL,
PAYLOAD_V2_SUPPLEMENTARY_DATA,
)
from tests.integration.integration_test_case import IntegrationTestCase

EXPECTED_TOKEN_BUSINESS = {
"account_service_url": "http://upstream.url",
"case_id": "1001",
"collection_exercise_sid": "789",
"exp": 1709058195.091798,
"iat": 1709054595.091798,
"jti": "1001",
"language_code": "en",
"response_expires_at": "2024-02-28T09:59:43.109276+00:00",
"response_id": "1234567890123456",
"roles": [],
"schema_name": "test_metadata_routing.json",
"survey_metadata": {
"data": {
"display_address": "68 Abingdon Road, Goathill",
"employment_date": "1983-06-02",
"flag_1": 123,
"link": "https://example.com",
"period_id": "202402",
"period_str": "April 2016",
"ref_p_end_date": "2016-04-30",
"ref_p_start_date": "2016-04-01",
"ru_name": "Integration Testing",
"ru_ref": "12345678901A",
"trad_as": "Integration Tests",
"user_id": "integration-test",
}
},
"tx_id": "1001",
"version": "v2",
}
EXPECTED_TOKEN_SOCIAL = {
"account_service_url": "http://upstream.url",
"case_id": "1001",
"collection_exercise_sid": "789",
"exp": 1709058195.091798,
"iat": 1709054595.091798,
"jti": "1001",
"language_code": "en",
"response_expires_at": "2024-02-28T09:59:43.109276+00:00",
"response_id": "1234567890123456",
"roles": [],
"schema_name": "social_demo.json",
"survey_metadata": {
"data": {
"case_ref": "1000000000000001",
"date": "2016-05-12",
"flag_1": True,
"qid": PAYLOAD_V2_SOCIAL["survey_metadata"]["data"]["qid"],
"user_id": "64389274239",
},
"receipting_keys": ["qid"],
},
"tx_id": "1001",
"version": "v2",
}


class TestCreateToken(IntegrationTestCase, AppContextTestCase):
"""
The purpose of this test class is to test the creation of a token (from create_token.py) to ensure
metadata in a decrypted token is nested/found in the correct level.
"""

# Patches are used since the values "uuid4", "time" and "get_response_expires_at" return are dynamic
@mock.patch("tests.integration.create_token.uuid4")
@mock.patch(
"tests.integration.create_token.time",
)
@mock.patch(
"tests.integration.create_token.get_response_expires_at",
)
def test_payload_content_and_structure_from_token(
self, mock_response_expiry_time, mock_time, mock_uuid
):
mock_uuid.return_value = 1001
mock_time.return_value = 1709054595.091798
mock_response_expiry_time.return_value = "2024-02-28T09:59:43.109276+00:00"
test_parameters = [
{
"schema": "test_metadata_routing.json",
"theme": "default",
"additional_payload": {
"flag_1": 123,
"period_id": "202402",
"link": "https://example.com",
},
"payload": PAYLOAD_V2_BUSINESS,
"expected_token": EXPECTED_TOKEN_BUSINESS,
},
{
"schema": "social_demo.json",
"theme": "social",
"additional_payload": {
"flag_1": True,
"user_id": "64389274239",
"date": "2016-05-12",
},
"payload": PAYLOAD_V2_SOCIAL,
"expected_token": EXPECTED_TOKEN_SOCIAL,
},
]
for value in test_parameters:
with self.subTest():
additional_payload = value["additional_payload"]
token = self.token_generator.create_token_v2(
schema_name=value["schema"],
theme=value["theme"],
**additional_payload,
)

with self.test_app.app_context():
decrypted_token = decrypt_token(token)
self.assertEqual(value["expected_token"], decrypted_token)

def test_uuid_consistent_after_decryption(self):
token = self.token_generator.create_token_v2(
"test_checkbox.json", theme="social", value="Dummy Text"
)
with self.test_app.app_context():
decrypted_token = decrypt_token(token)
assert decrypted_token["survey_metadata"] == {
"data": {
"case_ref": "1000000000000001",
"qid": PAYLOAD_V2_SOCIAL["survey_metadata"]["data"]["qid"],
"value": "Dummy Text",
},
"receipting_keys": ["qid"],
}

def test_sds_metadata_included_in_token(self):
token = self.token_generator.create_supplementary_data_token(
"test_checkbox.json"
)
with self.test_app.app_context():
decrypted_token = decrypt_token(token)
self.assertEqual(
decrypted_token, PAYLOAD_V2_SUPPLEMENTARY_DATA | decrypted_token
)

def test_additional_payload_added_in_token(self):
token = self.token_generator.create_supplementary_data_token(
"test_checkbox.json", flag_1=True
)
with self.test_app.app_context():
decrypted_token = decrypt_token(token)
assert decrypted_token["survey_metadata"] == {
"data": {
"display_address": "68 Abingdon Road, Goathill",
"employment_date": "1983-06-02",
"flag_1": True,
"period_id": "201604",
"period_str": "April 2016",
"ref_p_end_date": "2016-04-30",
"ref_p_start_date": "2016-04-01",
"ru_name": "Integration Testing",
"ru_ref": "12345678901A",
"sds_dataset_id": "44f1b432-9421-49e5-bd26-e63e18a30b69",
"survey_id": "123",
"trad_as": "Integration Tests",
"user_id": "integration-test",
}
}

def test_metadata_is_removed_from_token(self):
metadata_tokens = [
{
"token": self.token_generator.create_token_without_jti(
"test_number.json"
),
"removed_metadata": "jti",
},
{
"token": self.token_generator.create_token_without_case_id(
"test_numbers.json"
),
"removed_metadata": "case_id",
},
{
"token": self.token_generator.create_token_without_trad_as(
"test_numbers.json"
),
"removed_metadata": "trad_as",
},
]
for values in metadata_tokens:
with self.subTest():
with self.test_app.app_context():
decrypted_token = decrypt_token(values["token"])
self.assertNotIn(values["removed_metadata"], decrypted_token)

0 comments on commit c5fdfe3

Please sign in to comment.