From 0c6f529c36ad77e383ad60ef27f7f74678025306 Mon Sep 17 00:00:00 2001 From: Catalin Date: Tue, 27 Aug 2024 14:21:31 +0300 Subject: [PATCH] HDX-9987 & HDX-9990 add tests --- .../ckanext/hdx_package/actions/authorize.py | 2 +- .../test_contact_contributor_view.py | 180 ++++++++++++++ .../test_pages/test_request_access_view.py | 233 ++++++++++++++++++ .../ckanext/hdx_package/views/dataset.py | 12 +- .../package/contact_contributor.html | 3 +- 5 files changed, 422 insertions(+), 8 deletions(-) create mode 100644 ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_contact_contributor_view.py create mode 100644 ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_request_access_view.py diff --git a/ckanext-hdx_package/ckanext/hdx_package/actions/authorize.py b/ckanext-hdx_package/ckanext/hdx_package/actions/authorize.py index f9f0b92b91..fbfe3658f2 100644 --- a/ckanext-hdx_package/ckanext/hdx_package/actions/authorize.py +++ b/ckanext-hdx_package/ckanext/hdx_package/actions/authorize.py @@ -171,7 +171,7 @@ def hdx_mark_resource_in_hapi(context: Context, data_dict: DataDict): return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_IN_HAPI_FLAG) -def hdx_request_access(context: Context): +def hdx_request_access(context: Context, data_dict: DataDict): """ Only a logged-in user can request data access. """ diff --git a/ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_contact_contributor_view.py b/ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_contact_contributor_view.py new file mode 100644 index 0000000000..4be45fbef6 --- /dev/null +++ b/ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_contact_contributor_view.py @@ -0,0 +1,180 @@ +import mock +import pytest +import ckan.model as model +import ckan.plugins.toolkit as tk +import ckan.tests.factories as factories +from ckanext.hdx_org_group.helpers.static_lists import ORGANIZATION_TYPE_LIST + +get_action = tk.get_action +config = tk.config +h = tk.h + +USER_NAME = 'test_user' +USER_EMAIL = 'test_user@test.org' +SYSADMIN_NAME = 'sysadmin_user' +SYSADMIN_EMAIL = 'sysadmin_user@test.org' +ORGANIZATION_NAME = 'test_organization' +LOCATION_NAME = 'test_location' +PUBLIC_DATASET_NAME = 'public_test_dataset' +REQUEST_DATA_DATASET_NAME = 'contact_contributor_test_dataset' +CONST_CONTACT_CONTRIBUTOR = h.HDX_CONST('UI_CONSTANTS')['CONTACT_CONTRIBUTOR'] +CONST_SIGNIN = h.HDX_CONST('UI_CONSTANTS')['SIGNIN'] + + +@pytest.fixture() +def setup_data(): + factories.Sysadmin(name=SYSADMIN_NAME, email=SYSADMIN_EMAIL, fullname='Sysadmin User') + factories.User(name=USER_NAME, email=USER_EMAIL, fullname='Test User') + factories.Organization( + name=ORGANIZATION_NAME, + title='Test Organization', + hdx_org_type=ORGANIZATION_TYPE_LIST[0][1], + org_url='https://hdx.hdxtest.org/' + ) + factories.Group(name=LOCATION_NAME) + + +def get_sysadmin_context(): + return {'model': model, 'user': SYSADMIN_NAME} + + +def get_user_context(): + return {'model': model, 'user': USER_NAME} + + +@pytest.mark.usefixtures('clean_db', 'clean_index', 'setup_data') +class TestContactContributorView(object): + @mock.patch('ckanext.hdx_package.actions.get.hdx_mailer.mail_recipient') + def test_contact_contributor(self, mock_mail_recipient, app): + user_token = factories.APIToken(user=USER_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': user_token} + + package = { + 'package_creator': 'test function', + 'private': False, + 'dataset_date': '01/01/1960-12/31/2012', + 'caveats': 'These are the caveats', + 'license_other': 'TEST OTHER LICENSE', + 'methodology': 'This is a test methodology', + 'dataset_source': 'Test data', + 'license_id': 'hdx-other', + 'name': PUBLIC_DATASET_NAME, + 'notes': 'This is a public test dataset', + 'title': 'Public Test Dataset', + 'groups': [{'name': LOCATION_NAME}], + 'owner_org': ORGANIZATION_NAME, + 'maintainer': SYSADMIN_NAME, + 'data_update_frequency': '-1', + 'resources': [ + { + 'url': config.get('ckan.site_url', '') + '/storage/f/test_folder/test.csv', + 'resource_type': 'file.upload', + 'format': 'CSV', + 'name': 'test.csv' + } + ] + } + pkg_dict = get_action('package_create')(get_sysadmin_context(), package) + + contact_contributor_url = h.url_for('hdx_dataset.contact_contributor', id=pkg_dict.get('name') or pkg_dict.get('id')) + + # GET request without user (anonymous access) + response = app.get(contact_contributor_url, headers={}) + assert '/sign-in/?info_message_type=contact-contributor' in response.request.url, \ + 'Anonymous users should be redirected to the sign-in page with the contact-contributor parameter' + assert CONST_SIGNIN['contact-contributor'] in response.body, 'Anonymous users should see the contact contributor info message' + + # GET request with regular user + response = app.get(contact_contributor_url, headers=auth_headers) + assert response.status_code == 200 + assert CONST_CONTACT_CONTRIBUTOR['PAGE_TITLE'] in response.body + assert CONST_CONTACT_CONTRIBUTOR['BODY_MAIN_TEXT'] in response.body + assert 'id="contact-contributor-form"' in response.body, 'The contact contributor form should be visible' + + assert f"'datasetName': '{PUBLIC_DATASET_NAME}'," in response.body, \ + 'The analytics info dict should contain dataset name' + + # POST request without user (anonymous access) + response = app.post(contact_contributor_url, data={}) + assert '/sign-in/?info_message_type=contact-contributor' in response.request.url, \ + 'Anonymous users should be redirected to the sign-in page with the contact-contributor parameter' + assert CONST_SIGNIN['contact-contributor'] in response.body, 'Anonymous users should see the contact contributor info message' + + # POST request with invalid email address & missing fields + response = app.post(contact_contributor_url, data={'email': 'invalid'}, headers=auth_headers) + assert response.status_code == 200 + assert 'Email invalid is not a valid format' in response.body, \ + 'There should be an error about invalid email format' + assert 'Missing value' in response.body, 'There should be errors about missing required values' + assert 'id="contact-contributor-form"' in response.body, 'The contact contributor form should be visible' + + # POST request with valid data + data_dict = { + 'topic': 'suggested edits', + 'fullname': 'John Doe', + 'email': 'hdx.feedback@gmail.com', + 'msg': 'my message', + 'pkg_id': pkg_dict.get('id'), + 'pkg_owner_org': pkg_dict.get('owner_org'), + 'pkg_title': pkg_dict.get('title'), + } + response = app.post(contact_contributor_url, data=data_dict, headers=auth_headers) + assert response.status_code == 200 + assert '

{0}

'.format(CONST_CONTACT_CONTRIBUTOR['PAGE_TITLE']) not in response.body + assert '

{0}

'.format( + CONST_CONTACT_CONTRIBUTOR['PAGE_TITLE_MESSAGE_SENT']) in response.body + assert '' in response.body, \ + 'The hidden input indicates analytics tracking for request submission' + assert (''.format(data_dict.get('topic')) in + response.body), 'The hidden input indicates analytics tracking message subject for request submission' + + def test_contact_contributor_request_data_dataset(self, app): + user_token = factories.APIToken(user=USER_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': user_token} + + package = { + 'package_creator': 'test function', + 'private': False, + 'dataset_date': '[1960-01-01 TO 2012-12-31]', + 'indicator': '0', + 'caveats': 'These are the caveats', + 'license_other': 'TEST OTHER LICENSE', + 'methodology': 'This is a test methodology', + 'dataset_source': 'Test data', + 'license_id': 'hdx-other', + 'name': REQUEST_DATA_DATASET_NAME, + 'notes': 'This is a contact contributor test', + 'title': 'Request Access Test Dataset', + 'groups': [{'name': LOCATION_NAME}], + 'owner_org': ORGANIZATION_NAME, + 'maintainer': SYSADMIN_NAME, + 'is_requestdata_type': True, + 'file_types': ['csv'], + 'field_names': ['field1', 'field2'] + } + pkg_dict = get_action('package_create')(get_sysadmin_context(), package) + + contact_contributor_url = h.url_for('hdx_dataset.contact_contributor', id=pkg_dict.get('name') or pkg_dict.get('id')) + + # GET request on HDX Connect dataset + response = app.get(contact_contributor_url, headers=auth_headers) + assert response.status_code == 404, ('A 404 error should be returned when accessing a HDX Connect dataset') + + # POST request on HDX Connect dataset + response = app.post(contact_contributor_url, data={}, headers=auth_headers) + assert response.status_code == 404, ('A 404 error should be returned when doing a POST to a HDX Connect dataset') + + + def test_contact_contributor_non_existent_dataset(self, app): + user_token = factories.APIToken(user=USER_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': user_token} + + invalid_url = h.url_for('hdx_dataset.contact_contributor', id='invalid-dataset') + + # GET request on non-existent dataset + response = app.get(invalid_url, headers=auth_headers) + assert response.status_code == 404, 'A 404 error should be returned when accessing a non-existent dataset' + + # POST request on non-existent dataset + response = app.post(invalid_url, data={}, headers=auth_headers) + assert response.status_code == 404, 'A 404 error should be returned when doing a POST to a non-existent dataset' diff --git a/ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_request_access_view.py b/ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_request_access_view.py new file mode 100644 index 0000000000..79d8d1ebae --- /dev/null +++ b/ckanext-hdx_package/ckanext/hdx_package/tests/test_pages/test_request_access_view.py @@ -0,0 +1,233 @@ +import mock +import pytest +import ckan.model as model +import ckan.plugins.toolkit as tk +import ckan.tests.factories as factories +from ckanext.hdx_org_group.helpers.static_lists import ORGANIZATION_TYPE_LIST + +get_action = tk.get_action +config = tk.config +h = tk.h + +USER_NAME = 'test_user' +USER_EMAIL = 'test_user@test.org' +SYSADMIN_NAME = 'sysadmin_user' +SYSADMIN_EMAIL = 'sysadmin_user@test.org' +ORGANIZATION_NAME = 'test_organization' +LOCATION_NAME = 'test_location' +REQUEST_ACCESS_DATASET_NAME = 'request_access_test_dataset' +PUBLIC_DATASET_NAME = 'public_test_dataset' +PRIVATE_DATASET_NAME = 'private_test_dataset' +CONST_REQUEST_ACCESS = h.HDX_CONST('UI_CONSTANTS')['REQUEST_ACCESS'] +CONST_SIGNIN = h.HDX_CONST('UI_CONSTANTS')['SIGNIN'] + + +@pytest.fixture() +def setup_data(): + factories.Sysadmin(name=SYSADMIN_NAME, email=SYSADMIN_EMAIL, fullname='Sysadmin User') + factories.User(name=USER_NAME, email=USER_EMAIL, fullname='Test User') + factories.Organization( + name=ORGANIZATION_NAME, + title='Test Organization', + hdx_org_type=ORGANIZATION_TYPE_LIST[0][1], + org_url='https://hdx.hdxtest.org/' + ) + factories.Group(name=LOCATION_NAME) + + +def get_sysadmin_context(): + return {'model': model, 'user': SYSADMIN_NAME} + + +def get_user_context(): + return {'model': model, 'user': USER_NAME} + + +@pytest.mark.usefixtures('clean_db', 'clean_index', 'setup_data') +class TestRequestAccessView(object): + @mock.patch('ckanext.hdx_package.actions.get.hdx_mailer.mail_recipient') + def test_request_access(self, mock_mail_recipient, app): + user_token = factories.APIToken(user=USER_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': user_token} + + package = { + 'package_creator': 'test function', + 'private': False, + 'dataset_date': '[1960-01-01 TO 2012-12-31]', + 'indicator': '0', + 'caveats': 'These are the caveats', + 'license_other': 'TEST OTHER LICENSE', + 'methodology': 'This is a test methodology', + 'dataset_source': 'Test data', + 'license_id': 'hdx-other', + 'name': REQUEST_ACCESS_DATASET_NAME, + 'notes': 'This is a request access test', + 'title': 'Request Access Test Dataset', + 'groups': [{'name': LOCATION_NAME}], + 'owner_org': ORGANIZATION_NAME, + 'maintainer': SYSADMIN_NAME, + 'is_requestdata_type': True, + 'file_types': ['csv'], + 'field_names': ['field1', 'field2'] + } + pkg_dict = get_action('package_create')(get_sysadmin_context(), package) + + request_access_url = h.url_for('hdx_dataset.request_access', id=pkg_dict.get('name') or pkg_dict.get('id')) + + # GET request without user (anonymous access) + response = app.get(request_access_url, headers={}) + assert '/sign-in/?info_message_type=hdx-connect' in response.request.url, \ + 'Anonymous users should be redirected to the sign-in page with the hdx-connect parameter' + assert CONST_SIGNIN['hdx-connect'] in response.body, 'Anonymous users should see the HDX Connect info message' + + # GET request with regular user + response = app.get(request_access_url, headers=auth_headers) + assert response.status_code == 200 + assert CONST_REQUEST_ACCESS['PAGE_TITLE'] in response.body + assert CONST_REQUEST_ACCESS['BODY_MAIN_TEXT'] in response.body + assert 'id="request-access-form"' in response.body, 'The request access form should be visible' + + assert f"'datasetName': '{REQUEST_ACCESS_DATASET_NAME}'," in response.body, ('The analytics info dict should ' + 'contain dataset name') + assert "'datasetAvailability': 'metadata only'," in response.body, ('The analytics info dict should contain ' + 'dataset availability') + + # POST request without user (anonymous access) + response = app.post(request_access_url, data={}) + assert '/sign-in/?info_message_type=hdx-connect' in response.request.url, \ + 'Anonymous users should be redirected to the sign-in page with the hdx-connect parameter' + assert CONST_SIGNIN['hdx-connect'] in response.body, 'Anonymous users should see the HDX Connect info message' + + # POST request with invalid email address & missing fields + response = app.post(request_access_url, data={'email_address': 'invalid'}, headers=auth_headers) + assert response.status_code == 200 + assert 'Please provide a valid email address' in response.body, ('There should be an error about invalid email ' + 'format') + assert 'Missing value' in response.body, 'There should be errors about missing required values' + assert 'id="request-access-form"' in response.body, 'The request access form should be visible' + + # POST request with valid data + data_dict = { + 'package_id': pkg_dict.get('id'), + 'sender_name': 'John Doe', + 'message_content': 'I want to access additional data.', + 'organization': 'test_organization', + 'email_address': 'test@test.com', + 'sender_country': 'Sweden', + 'sender_organization_id': 'test_organization', + 'sender_organization_type': 'NGO', + 'sender_intend': 'other', + 'sender_intend_other': 'Testing Purposes', + } + response = app.post(request_access_url, data=data_dict, headers=auth_headers) + assert response.status_code == 200 + assert '

{0}

'.format(CONST_REQUEST_ACCESS['PAGE_TITLE']) not in response.body + assert '

{0}

'.format( + CONST_REQUEST_ACCESS['PAGE_TITLE_REQUEST_SENT']) in response.body + assert '' in response.body, \ + 'The hidden input indicates analytics tracking for request submission' + + # POST request with a pending request + response = app.post(request_access_url, data=data_dict, headers=auth_headers) + assert response.status_code == 200 + assert 'You already have a pending request. Please wait for the reply.' in response.body, \ + 'The same user cannot send another request when there is a pending one' + assert 'id="request-access-form"' not in response.body, ('The request access form should not be visible for ' + 'pending requests') + + def test_request_access_public_dataset(self, app): + user_token = factories.APIToken(user=USER_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': user_token} + + package = { + 'package_creator': 'test function', + 'private': False, + 'dataset_date': '01/01/1960-12/31/2012', + 'caveats': 'These are the caveats', + 'license_other': 'TEST OTHER LICENSE', + 'methodology': 'This is a test methodology', + 'dataset_source': 'Test data', + 'license_id': 'hdx-other', + 'name': PUBLIC_DATASET_NAME, + 'notes': 'This is a public test dataset', + 'title': 'Public Test Dataset', + 'groups': [{'name': LOCATION_NAME}], + 'owner_org': ORGANIZATION_NAME, + 'maintainer': SYSADMIN_NAME, + 'data_update_frequency': '-1', + 'resources': [ + { + 'url': config.get('ckan.site_url', '') + '/storage/f/test_folder/test.csv', + 'resource_type': 'file.upload', + 'format': 'CSV', + 'name': 'test.csv' + } + ] + } + pkg_dict = get_action('package_create')(get_sysadmin_context(), package) + + request_access_url = h.url_for('hdx_dataset.request_access', id=pkg_dict.get('name') or pkg_dict.get('id')) + + # GET request on public dataset + response = app.get(request_access_url, headers=auth_headers) + assert response.status_code == 404, ('A 404 error should be returned when accessing a public dataset') + + # POST request on public dataset + response = app.post(request_access_url, data={}, headers=auth_headers) + assert response.status_code == 404, ('A 404 error should be returned when doing a POST to a public dataset') + + def test_request_access_private_dataset(self, app): + sysadmin_token = factories.APIToken(user=SYSADMIN_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': sysadmin_token} + + package = { + 'package_creator': 'test function', + 'private': True, + 'dataset_date': '01/01/1960-12/31/2012', + 'caveats': 'These are the caveats', + 'license_other': 'TEST OTHER LICENSE', + 'methodology': 'This is a test methodology', + 'dataset_source': 'Test data', + 'license_id': 'hdx-other', + 'name': PRIVATE_DATASET_NAME, + 'notes': 'This is a private test dataset', + 'title': 'Private Test Dataset', + 'groups': [{'name': LOCATION_NAME}], + 'owner_org': ORGANIZATION_NAME, + 'maintainer': SYSADMIN_NAME, + 'data_update_frequency': '-1', + 'resources': [ + { + 'url': config.get('ckan.site_url', '') + '/storage/f/test_folder/test.csv', + 'resource_type': 'file.upload', + 'format': 'CSV', + 'name': 'test.csv' + } + ] + } + pkg_dict = get_action('package_create')(get_sysadmin_context(), package) + + request_access_url = h.url_for('hdx_dataset.request_access', id=pkg_dict.get('name') or pkg_dict.get('id')) + + # GET request on private dataset + response = app.get(request_access_url, headers=auth_headers) + assert response.status_code == 404, ('A 404 error should be returned when accessing a private dataset') + + # POST request on private dataset + response = app.post(request_access_url, data={}, headers=auth_headers) + assert response.status_code == 404, ('A 404 error should be returned when doing a POST to a private dataset') + + + def test_request_access_non_existent_dataset(self, app): + user_token = factories.APIToken(user=USER_NAME, expires_in=2, unit=60 * 60)['token'] + auth_headers = {'Authorization': user_token} + + invalid_url = h.url_for('hdx_dataset.request_access', id='invalid-dataset') + + # GET request on non-existent dataset + response = app.get(invalid_url, headers=auth_headers) + assert response.status_code == 404, 'A 404 error should be returned when accessing a non-existent dataset' + + # POST request on non-existent dataset + response = app.post(invalid_url, data={}, headers=auth_headers) + assert response.status_code == 404, 'A 404 error should be returned when doing a POST to a non-existent dataset' diff --git a/ckanext-hdx_package/ckanext/hdx_package/views/dataset.py b/ckanext-hdx_package/ckanext/hdx_package/views/dataset.py index 8155fff08e..b33a592174 100644 --- a/ckanext-hdx_package/ckanext/hdx_package/views/dataset.py +++ b/ckanext-hdx_package/ckanext/hdx_package/views/dataset.py @@ -561,6 +561,7 @@ def post(self, id: str) -> Union[Response, str]: u'auth_user_obj': g.userobj, } + data = errors = {} try: pkg_dict = get_action('package_show')(context, {'id': id}) @@ -599,7 +600,7 @@ def post(self, id: str) -> Union[Response, str]: return abort(404, _('Dataset not found')) except NotAuthorized: - came_from = h.url_for('hdx_dataset.contact_contributor', id=pkg_dict.get('name')) + came_from = h.url_for('hdx_dataset.contact_contributor', id=id) return redirect(h.url_for('hdx_signin.login', info_message_type='contact-contributor', came_from=came_from)) except captcha.CaptchaError: @@ -657,7 +658,7 @@ def get(self, id: str, return abort(404, _('Dataset not found')) except NotAuthorized: - came_from = h.url_for('hdx_dataset.contact_contributor', id=pkg_dict.get('name')) + came_from = h.url_for('hdx_dataset.contact_contributor', id=id) return redirect(h.url_for('hdx_signin.login', info_message_type='contact-contributor', came_from=came_from)) @@ -671,6 +672,7 @@ def post(self, id: str) -> Union[Response, str]: u'auth_user_obj': g.userobj, } + data = errors = {} try: pkg_dict = get_action('package_show')(context, {'id': id}) @@ -681,7 +683,7 @@ def post(self, id: str) -> Union[Response, str]: pending_request = h.hdx_pending_request_data(g.userobj.id, pkg_dict.get('id')) if len(pending_request) > 0: - return redirect('hdx_dataset.request_access', id=pkg_dict.get('name')) + return redirect('hdx_dataset.request_access', id=id) dataset_request_access_logic = DatasetRequestAccessLogic(context, request) @@ -715,7 +717,7 @@ def post(self, id: str) -> Union[Response, str]: return abort(404, _('Dataset not found')) except NotAuthorized: - came_from = h.url_for('hdx_dataset.request_access', id=pkg_dict.get('name')) + came_from = h.url_for('hdx_dataset.request_access', id=id) return redirect(h.url_for('hdx_signin.login', info_message_type='hdx-connect', came_from=came_from)) except MailerException as e: @@ -777,7 +779,7 @@ def get(self, id: str, return abort(404, _('Dataset not found')) except NotAuthorized: - came_from = h.url_for('hdx_dataset.request_access', id=pkg_dict.get('name')) + came_from = h.url_for('hdx_dataset.request_access', id=id) return redirect(h.url_for('hdx_signin.login', info_message_type='hdx-connect', came_from=came_from)) diff --git a/ckanext-hdx_theme/ckanext/hdx_theme/templates/package/contact_contributor.html b/ckanext-hdx_theme/ckanext/hdx_theme/templates/package/contact_contributor.html index 27b37443bd..82bb733899 100644 --- a/ckanext-hdx_theme/ckanext/hdx_theme/templates/package/contact_contributor.html +++ b/ckanext-hdx_theme/ckanext/hdx_theme/templates/package/contact_contributor.html @@ -59,8 +59,7 @@ {% endif %} -
+ {{ h.csrf_input() }}