From 91d11742f435193287f8ca541f3f83813f5a429c Mon Sep 17 00:00:00 2001 From: Marc Sommerhalder Date: Tue, 28 Jan 2025 11:19:41 +0100 Subject: [PATCH] PB-1406: Make cognito authenticated users superuser --- app/middleware/api_gateway_middleware.py | 6 +- .../tests_09/test_geoadmin_header_auth.py | 11 +- .../tests_10/test_asset_upload_endpoint.py | 167 ++++++++++++- app/tests/tests_10/test_assets_endpoint.py | 153 ++++++++++-- .../test_collection_asset_upload_endpoint.py | 164 ++++++++++++- .../test_collection_assets_endpoint.py | 129 ++++++++-- .../tests_10/test_collections_endpoint.py | 110 ++++++++- .../tests_10/test_external_assets_endpoint.py | 39 ++- app/tests/tests_10/test_generic_api.py | 40 ++- .../tests_10/test_geoadmin_header_auth.py | 11 +- .../test_item_to_collection_endpoint.py | 27 ++- app/tests/tests_10/test_items_endpoint.py | 227 +++++++++++++++--- 12 files changed, 953 insertions(+), 131 deletions(-) diff --git a/app/middleware/api_gateway_middleware.py b/app/middleware/api_gateway_middleware.py index 18854c72..727680e7 100644 --- a/app/middleware/api_gateway_middleware.py +++ b/app/middleware/api_gateway_middleware.py @@ -29,4 +29,8 @@ def authenticate(self, request, remote_user): if not settings.FEATURE_AUTH_ENABLE_APIGW: return None - return super().authenticate(request, remote_user) + user = super().authenticate(request, remote_user) + if user: + user.is_superuser = True + user.save() + return user diff --git a/app/tests/tests_09/test_geoadmin_header_auth.py b/app/tests/tests_09/test_geoadmin_header_auth.py index 5b81aed2..03615abc 100644 --- a/app/tests/tests_09/test_geoadmin_header_auth.py +++ b/app/tests/tests_09/test_geoadmin_header_auth.py @@ -2,7 +2,6 @@ from parameterized import parameterized -from django.contrib.auth import get_user_model from django.test import Client from django.test import override_settings @@ -15,19 +14,19 @@ @override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class GeoadminHeadersAuthForPutEndpointTestCase(StacBaseTestCase): - valid_username = "another_test_user" def setUp(self): # pylint: disable=invalid-name self.client = Client(enforce_csrf_checks=True) self.factory = Factory() self.collection = self.factory.create_collection_sample() - get_user_model().objects.create_superuser(self.valid_username) @parameterized.expand([ - (valid_username, "true", 201), + ("another_test_user", "true", 201), + ("another_test_user", "false", 401), + ("another_test_user", "", 401), (None, None, 401), - (valid_username, "false", 401), - ("wronguser", "true", 403), + (None, "false", 401), + (None, "true", 401), ]) def test_collection_upsert_create_with_geoadmin_header_auth( self, username_header, authenticated_header, expected_response_code diff --git a/app/tests/tests_10/test_asset_upload_endpoint.py b/app/tests/tests_10/test_asset_upload_endpoint.py index d0bee3c6..245407db 100644 --- a/app/tests/tests_10/test_asset_upload_endpoint.py +++ b/app/tests/tests_10/test_asset_upload_endpoint.py @@ -26,7 +26,6 @@ from tests.tests_10.data_factory import Factory from tests.tests_10.utils import reverse_version from tests.utils import S3TestMixin -from tests.utils import client_login from tests.utils import get_file_like_object from tests.utils import mock_s3_asset_file @@ -52,7 +51,6 @@ class AssetUploadBaseTest(StacBaseTestCase, S3TestMixin): @mock_s3_asset_file def setUp(self): # pylint: disable=invalid-name self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample().model self.item = self.factory.create_item_sample(collection=self.collection).model @@ -182,6 +180,7 @@ def check_aborted_response(self, json_response): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUploadCreateEndpointTestCase(AssetUploadBaseTest): def test_asset_upload_create_abort_multipart(self): @@ -193,6 +192,9 @@ def test_asset_upload_create_abort_multipart(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -208,6 +210,9 @@ def test_asset_upload_create_abort_multipart(self): response = self.client.post( self.get_abort_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={}, content_type="application/json" ) @@ -238,6 +243,9 @@ def test_asset_upload_create_multipart_duplicate(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -253,6 +261,9 @@ def test_asset_upload_create_multipart_duplicate(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -288,6 +299,7 @@ def test_asset_upload_create_multipart_duplicate(self): self.assertEqual(len(response['Uploads']), 1, msg='More or less uploads found on S3') +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUploadCreateRaceConditionTest(StacBaseTransactionTestCase, S3TestMixin): @mock_s3_asset_file @@ -320,6 +332,9 @@ def asset_upload_atomic_create_test(worker): client.login(username=self.username, password=self.password) return client.post( path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -342,6 +357,7 @@ def asset_upload_atomic_create_test(worker): self.assertEqual(response.json()['description'], "Upload already in progress") +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUpload1PartEndpointTestCase(AssetUploadBaseTest): def upload_asset_with_dyn_cache(self, update_interval=None): @@ -353,6 +369,9 @@ def upload_asset_with_dyn_cache(self, update_interval=None): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -371,6 +390,9 @@ def upload_asset_with_dyn_cache(self, update_interval=None): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size, number_parts) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -387,6 +409,9 @@ def test_asset_upload_1_part_md5_integrity(self): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -404,6 +429,9 @@ def test_asset_upload_1_part_md5_integrity(self): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size, number_parts) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -437,6 +465,9 @@ def test_asset_upload_no_content_encoding(self): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -454,6 +485,9 @@ def test_asset_upload_no_content_encoding(self): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size, number_parts) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -476,6 +510,9 @@ def test_asset_upload_gzip(self): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like_compress)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -496,6 +533,9 @@ def test_asset_upload_gzip(self): ) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -508,6 +548,7 @@ def test_asset_upload_gzip(self): self.assertEqual(size_compress, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUpload2PartEndpointTestCase(AssetUploadBaseTest): def test_asset_upload_2_parts_md5_integrity(self): @@ -522,6 +563,9 @@ def test_asset_upload_2_parts_md5_integrity(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -538,6 +582,9 @@ def test_asset_upload_2_parts_md5_integrity(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -548,6 +595,7 @@ def test_asset_upload_2_parts_md5_integrity(self): self.assertEqual(size, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUploadInvalidEndpointTestCase(AssetUploadBaseTest): def test_asset_upload_invalid_content_encoding(self): @@ -561,6 +609,9 @@ def test_asset_upload_invalid_content_encoding(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -584,6 +635,9 @@ def test_asset_upload_1_part_no_md5(self): file_like, checksum_multihash = get_file_like_object(size) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash }, @@ -601,6 +655,9 @@ def test_asset_upload_2_parts_no_md5(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash }, @@ -611,7 +668,12 @@ def test_asset_upload_2_parts_no_md5(self): def test_asset_upload_create_empty_payload(self): response = self.client.post( - self.get_create_multipart_upload_path(), data={}, content_type="application/json" + self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data={}, + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual( @@ -627,6 +689,9 @@ def test_asset_upload_create_invalid_data(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 0, "file:checksum": 'abcdef', @@ -652,6 +717,9 @@ def test_asset_upload_create_too_many_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 101, "file:checksum": 'abcdef', 'md5_parts': md5_parts }, @@ -670,6 +738,9 @@ def test_asset_upload_create_empty_md5_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 2, "md5_parts": [], @@ -693,6 +764,9 @@ def test_asset_upload_create_duplicate_md5_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 3, "md5_parts": [{ @@ -722,6 +796,9 @@ def test_asset_upload_create_too_many_md5_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 2, "md5_parts": [{ @@ -751,6 +828,9 @@ def test_asset_upload_create_md5_parts_missing_part_number(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 2, "md5_parts": [ @@ -782,6 +862,9 @@ def test_asset_upload_2_parts_too_small(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -797,6 +880,9 @@ def test_asset_upload_2_parts_too_small(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -823,6 +909,9 @@ def test_asset_upload_1_parts_invalid_etag(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -838,6 +927,9 @@ def test_asset_upload_1_parts_invalid_etag(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': [{ 'etag': 'dummy', 'part_number': 1 }]}, @@ -868,6 +960,9 @@ def test_asset_upload_1_parts_too_many_parts_in_complete(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -884,6 +979,9 @@ def test_asset_upload_1_parts_too_many_parts_in_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -902,6 +1000,9 @@ def test_asset_upload_2_parts_incomplete_upload(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -916,6 +1017,9 @@ def test_asset_upload_2_parts_incomplete_upload(self): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size // 2, 1) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -935,6 +1039,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -950,6 +1057,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={}, content_type="application/json" ) @@ -958,6 +1068,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': []}, content_type="application/json" ) @@ -966,6 +1079,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': ["dummy-etag"]}, content_type="application/json" ) @@ -997,6 +1113,9 @@ def test_asset_upload_1_parts_duplicate_complete(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -1012,6 +1131,9 @@ def test_asset_upload_1_parts_duplicate_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -1019,6 +1141,9 @@ def test_asset_upload_1_parts_duplicate_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -1029,6 +1154,7 @@ def test_asset_upload_1_parts_duplicate_complete(self): self.assertEqual(size, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUploadDeleteInProgressEndpointTestCase(AssetUploadBaseTest): def test_delete_asset_upload_in_progress(self): @@ -1040,6 +1166,9 @@ def test_delete_asset_upload_in_progress(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -1050,7 +1179,12 @@ def test_delete_asset_upload_in_progress(self): self.assertStatusCode(201, response) upload_id = response.json()['upload_id'] - response = self.client.delete(self.get_delete_asset_path()) + response = self.client.delete( + self.get_delete_asset_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(400, response) self.assertEqual( response.json()['description'], ['Asset asset-1.tiff has still an upload in progress'] @@ -1065,10 +1199,20 @@ def test_delete_asset_upload_in_progress(self): msg='Asset has been deleted' ) - response = self.client.post(self.get_abort_multipart_upload_path(upload_id)) + response = self.client.post( + self.get_abort_multipart_upload_path(upload_id), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) - response = self.client.delete(self.get_delete_asset_path()) + response = self.client.delete( + self.get_delete_asset_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) self.assertFalse( @@ -1212,6 +1356,7 @@ def test_get_asset_uploads_status_query(self): self.assertEqual(upload['status'], AssetUpload.Status.ABORTED) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetUploadListPartsEndpointTestCase(AssetUploadBaseTest): def test_asset_upload_list_parts(self): @@ -1224,6 +1369,9 @@ def test_asset_upload_list_parts(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -1278,6 +1426,9 @@ def test_asset_upload_list_parts(self): # Complete the upload response = self.client.post( self.get_complete_multipart_upload_path(upload_id), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -1287,6 +1438,7 @@ def test_asset_upload_list_parts(self): self.assertEqual(size, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ExternalAssetUploadtestCase(AssetUploadBaseTest): def test_create_multipart_upload_on_external_asset(self): @@ -1304,6 +1456,9 @@ def test_create_multipart_upload_on_external_asset(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, diff --git a/app/tests/tests_10/test_assets_endpoint.py b/app/tests/tests_10/test_assets_endpoint.py index b376de87..d84f341b 100644 --- a/app/tests/tests_10/test_assets_endpoint.py +++ b/app/tests/tests_10/test_assets_endpoint.py @@ -1,3 +1,4 @@ +# pylint: disable=too-many-lines import logging from base64 import b64encode from datetime import datetime @@ -24,7 +25,6 @@ from tests.tests_10.data_factory import Factory from tests.tests_10.utils import reverse_version from tests.utils import S3TestMixin -from tests.utils import client_login from tests.utils import disableLogger from tests.utils import mock_s3_asset_file @@ -139,6 +139,7 @@ def test_single_assets_endpoint_item_expired(self): self.assertStatusCode(404, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsUnimplementedEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -147,7 +148,6 @@ def setUp(self): # pylint: disable=invalid-name self.collection = self.factory.create_collection_sample().model self.item = self.factory.create_item_sample(collection=self.collection).model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_unimplemented_post(self): @@ -156,12 +156,16 @@ def test_asset_unimplemented_post(self): asset = self.factory.create_asset_sample(item=self.item, required_only=True) response = self.client.post( f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets', + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset.get_json('post'), content_type="application/json" ) self.assertStatusCode(405, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsCreateEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -170,7 +174,6 @@ def setUp(self): # pylint: disable=invalid-name self.collection = self.factory.create_collection_sample().model self.item = self.factory.create_item_sample(collection=self.collection).model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_upsert_create_only_required(self): @@ -182,7 +185,14 @@ def test_asset_upsert_create_only_required(self): json_to_send = asset.get_json('put') # Send a non normalized form of the type to see if it is also accepted json_to_send['type'] = 'image/TIFF;application=geotiff; Profile=cloud-optimized' - response = self.client.put(path, data=json_to_send, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=json_to_send, + content_type="application/json" + ) json_data = response.json() self.assertStatusCode(201, response) self.assertLocationHeader(f"{path}", response) @@ -223,6 +233,9 @@ def test_asset_upsert_create(self): # Now use upsert to create the new asset response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_name]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset.get_json('put'), content_type="application/json" ) @@ -273,7 +286,12 @@ def test_asset_upsert_create_non_existing_parent_item_in_path(self): # Now use upsert to create the new asset response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(404, response) @@ -298,7 +316,12 @@ def test_asset_upsert_create_non_existing_parent_collection_in_path(self): # Now use upsert to create the new asset response = self.client.put( - path, data=asset.get_json('post'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('post'), + content_type="application/json" ) self.assertStatusCode(404, response) @@ -317,7 +340,12 @@ def test_asset_upsert_create_empty_string(self): path = \ f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset["name"]}' response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) json_data = response.json() @@ -332,7 +360,12 @@ def invalid_request_wrapper(self, sample_name, expected_error_messages, **extra_ path = \ f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset["name"]}' response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual( @@ -389,7 +422,12 @@ def test_asset_upsert_create_characters_geoadmin_variant(self): path = \ f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset["name"]}' response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(201, response) @@ -401,7 +439,12 @@ def test_asset_upsert_create_characters_geoadmin_variant(self): path = \ f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset["name"]}' response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual( @@ -417,6 +460,7 @@ def test_asset_upsert_create_characters_geoadmin_variant(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsWriteEndpointAssetFileTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -425,7 +469,6 @@ def setUp(self): # pylint: disable=invalid-name self.collection = self.factory.create_collection_sample().model self.item = self.factory.create_item_sample(collection=self.collection).model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name # NOTE: Unfortunately this test cannot be done with the moto mocking. @@ -452,6 +495,7 @@ def setUp(self): # pylint: disable=invalid-name # ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsUpdateEndpointAssetFileTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -463,7 +507,6 @@ def setUp(self): # pylint: disable=invalid-name ) self.asset = self.factory.create_asset_sample(item=self.item.model, db_create=True) self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_endpoint_patch_put_href(self): @@ -477,7 +520,14 @@ def test_asset_endpoint_patch_put_href(self): patch_payload = {'href': 'https://testserver/non-existing-asset'} path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' - response = self.client.patch(path, data=patch_payload, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=patch_payload, + content_type="application/json" + ) self.assertStatusCode(400, response) description = response.json()['description'] self.assertIn('href', description, msg=f'Unexpected field error {description}') @@ -487,7 +537,14 @@ def test_asset_endpoint_patch_put_href(self): msg="Unexpected error message" ) - response = self.client.put(path, data=put_payload, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=put_payload, + content_type="application/json" + ) self.assertStatusCode(400, response) description = response.json()['description'] self.assertIn('href', description, msg=f'Unexpected field error {description}') @@ -498,6 +555,7 @@ def test_asset_endpoint_patch_put_href(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsUpdateEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -509,7 +567,6 @@ def setUp(self): # pylint: disable=invalid-name ) self.asset = self.factory.create_asset_sample(item=self.item.model, db_create=True) self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_endpoint_put(self): @@ -527,7 +584,12 @@ def test_asset_endpoint_put(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.put( - path, data=changed_asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(200, response) @@ -555,7 +617,12 @@ def test_asset_endpoint_put_extra_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.put( - path, data=changed_asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'extra_attribute': ['Unexpected property in payload']}, @@ -579,6 +646,9 @@ def test_asset_endpoint_put_read_only_in_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.put( path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=changed_asset.get_json('put', keep_read_only=True), content_type="application/json" ) @@ -605,7 +675,12 @@ def test_asset_endpoint_put_rename_asset(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.put( - path, data=changed_asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'id': 'Renaming is not allowed'}, @@ -643,7 +718,12 @@ def test_asset_endpoint_patch_rename_asset(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.patch( - path, data=changed_asset.get_json('patch'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('patch'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'id': 'Renaming is not allowed'}, @@ -683,7 +763,12 @@ def test_asset_endpoint_patch_extra_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.patch( - path, data=changed_asset.get_json('patch'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('patch'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'extra_payload': ['Unexpected property in payload']}, @@ -705,6 +790,9 @@ def test_asset_endpoint_patch_read_only_in_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' response = self.client.patch( path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=changed_asset.get_json('patch', keep_read_only=True), content_type="application/json" ) @@ -723,6 +811,9 @@ def test_asset_atomic_upsert_create_500(self): 'test-asset-detail-http-500', args=[self.collection['name'], self.item['name'], sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -754,6 +845,9 @@ def test_asset_atomic_upsert_update_500(self): 'test-asset-detail-http-500', args=[self.collection['name'], self.item['name'], sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -812,6 +906,9 @@ def asset_atomic_upsert_test(worker): asset_sample['name'] ] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_sample.get_json('put'), content_type='application/json' ) @@ -837,6 +934,7 @@ def asset_atomic_upsert_test(worker): self.assertEqual(status_201, 1, msg="Not only one upsert did a create !") +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsDeleteEndpointTestCase(StacBaseTestCase, S3TestMixin): @mock_s3_asset_file @@ -846,7 +944,6 @@ def setUp(self): # pylint: disable=invalid-name self.item = self.factory.create_item_sample(collection=self.collection).model self.asset = self.factory.create_asset_sample(item=self.item).model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_endpoint_delete_asset(self): @@ -856,7 +953,12 @@ def test_asset_endpoint_delete_asset(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/items/{item_name}/assets/{asset_name}' s3_path = get_asset_path(self.item, asset_name) self.assertS3ObjectExists(s3_path) - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) # Check that is has really been deleted @@ -877,7 +979,12 @@ def test_asset_endpoint_delete_asset_invalid_name(self): f"/{STAC_BASE_V}/collections/{collection_name}" f"/items/{item_name}/assets/non-existent-asset" ) - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(404, response) diff --git a/app/tests/tests_10/test_collection_asset_upload_endpoint.py b/app/tests/tests_10/test_collection_asset_upload_endpoint.py index fcd69a6e..271dfd15 100644 --- a/app/tests/tests_10/test_collection_asset_upload_endpoint.py +++ b/app/tests/tests_10/test_collection_asset_upload_endpoint.py @@ -26,7 +26,6 @@ from tests.tests_10.data_factory import Factory from tests.tests_10.utils import reverse_version from tests.utils import S3TestMixin -from tests.utils import client_login from tests.utils import get_file_like_object from tests.utils import mock_s3_asset_file @@ -52,7 +51,6 @@ class CollectionAssetUploadBaseTest(StacBaseTestCase, S3TestMixin): @mock_s3_asset_file def setUp(self): # pylint: disable=invalid-name self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample().model self.asset = self.factory.create_collection_asset_sample( @@ -182,6 +180,7 @@ def check_aborted_response(self, json_response): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUploadCreateEndpointTestCase(CollectionAssetUploadBaseTest): def test_asset_upload_create_abort_multipart(self): @@ -193,6 +192,9 @@ def test_asset_upload_create_abort_multipart(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -208,6 +210,9 @@ def test_asset_upload_create_abort_multipart(self): response = self.client.post( self.get_abort_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={}, content_type="application/json" ) @@ -241,6 +246,9 @@ def test_asset_upload_create_multipart_duplicate(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -256,6 +264,9 @@ def test_asset_upload_create_multipart_duplicate(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -293,6 +304,7 @@ def test_asset_upload_create_multipart_duplicate(self): self.assertEqual(len(response['Uploads']), 1, msg='More or less uploads found on S3') +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUploadCreateRaceConditionTest(StacBaseTransactionTestCase, S3TestMixin): @mock_s3_asset_file @@ -326,6 +338,9 @@ def asset_upload_atomic_create_test(worker): client.login(username=self.username, password=self.password) return client.post( path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -348,6 +363,7 @@ def asset_upload_atomic_create_test(worker): self.assertEqual(response.json()['description'], "Upload already in progress") +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUpload1PartEndpointTestCase(CollectionAssetUploadBaseTest): def upload_asset_with_dyn_cache(self, update_interval=None): @@ -359,6 +375,9 @@ def upload_asset_with_dyn_cache(self, update_interval=None): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -377,6 +396,9 @@ def upload_asset_with_dyn_cache(self, update_interval=None): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size, number_parts) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -393,6 +415,9 @@ def test_asset_upload_1_part_md5_integrity(self): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -410,6 +435,9 @@ def test_asset_upload_1_part_md5_integrity(self): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size, number_parts) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -443,6 +471,9 @@ def test_asset_upload_no_content_encoding(self): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -460,6 +491,9 @@ def test_asset_upload_no_content_encoding(self): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size, number_parts) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -482,6 +516,9 @@ def test_asset_upload_gzip(self): md5_parts = [{'part_number': 1, 'md5': base64_md5(file_like_compress)}] response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -502,6 +539,9 @@ def test_asset_upload_gzip(self): ) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -514,6 +554,7 @@ def test_asset_upload_gzip(self): self.assertEqual(size_compress, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUpload2PartEndpointTestCase(CollectionAssetUploadBaseTest): def test_asset_upload_2_parts_md5_integrity(self): @@ -528,6 +569,9 @@ def test_asset_upload_2_parts_md5_integrity(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -544,6 +588,9 @@ def test_asset_upload_2_parts_md5_integrity(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -554,6 +601,7 @@ def test_asset_upload_2_parts_md5_integrity(self): self.assertEqual(size, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUploadInvalidEndpointTestCase(CollectionAssetUploadBaseTest): def test_asset_upload_invalid_content_encoding(self): @@ -567,6 +615,9 @@ def test_asset_upload_invalid_content_encoding(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'md5_parts': md5_parts, @@ -590,6 +641,9 @@ def test_asset_upload_1_part_no_md5(self): file_like, checksum_multihash = get_file_like_object(size) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash }, @@ -607,6 +661,9 @@ def test_asset_upload_2_parts_no_md5(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash }, @@ -617,7 +674,12 @@ def test_asset_upload_2_parts_no_md5(self): def test_asset_upload_create_empty_payload(self): response = self.client.post( - self.get_create_multipart_upload_path(), data={}, content_type="application/json" + self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data={}, + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual( @@ -633,6 +695,9 @@ def test_asset_upload_create_invalid_data(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 0, "file:checksum": 'abcdef', @@ -658,6 +723,9 @@ def test_asset_upload_create_too_many_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 101, "file:checksum": 'abcdef', 'md5_parts': md5_parts }, @@ -676,6 +744,9 @@ def test_asset_upload_create_empty_md5_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 2, "md5_parts": [], @@ -699,6 +770,9 @@ def test_asset_upload_create_duplicate_md5_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 3, "md5_parts": [{ @@ -728,6 +802,9 @@ def test_asset_upload_create_too_many_md5_parts(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 2, "md5_parts": [{ @@ -757,6 +834,9 @@ def test_asset_upload_create_md5_parts_missing_part_number(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': 2, "md5_parts": [ @@ -788,6 +868,9 @@ def test_asset_upload_2_parts_too_small(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -803,6 +886,9 @@ def test_asset_upload_2_parts_too_small(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -829,6 +915,9 @@ def test_asset_upload_1_parts_invalid_etag(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -844,6 +933,9 @@ def test_asset_upload_1_parts_invalid_etag(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': [{ 'etag': 'dummy', 'part_number': 1 }]}, @@ -874,6 +966,9 @@ def test_asset_upload_1_parts_too_many_parts_in_complete(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -890,6 +985,9 @@ def test_asset_upload_1_parts_too_many_parts_in_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -908,6 +1006,9 @@ def test_asset_upload_2_parts_incomplete_upload(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -922,6 +1023,9 @@ def test_asset_upload_2_parts_incomplete_upload(self): parts = self.s3_upload_parts(json_data['upload_id'], file_like, size // 2, 1) response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -941,6 +1045,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -956,6 +1063,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={}, content_type="application/json" ) @@ -964,6 +1074,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': []}, content_type="application/json" ) @@ -972,6 +1085,9 @@ def test_asset_upload_1_parts_invalid_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': ["dummy-etag"]}, content_type="application/json" ) @@ -1003,6 +1119,9 @@ def test_asset_upload_1_parts_duplicate_complete(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -1018,6 +1137,9 @@ def test_asset_upload_1_parts_duplicate_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -1025,6 +1147,9 @@ def test_asset_upload_1_parts_duplicate_complete(self): response = self.client.post( self.get_complete_multipart_upload_path(json_data['upload_id']), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) @@ -1035,6 +1160,7 @@ def test_asset_upload_1_parts_duplicate_complete(self): self.assertEqual(size, self.asset.file_size) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUploadDeleteInProgressEndpointTestCase(CollectionAssetUploadBaseTest): def test_delete_asset_upload_in_progress(self): @@ -1046,6 +1172,9 @@ def test_delete_asset_upload_in_progress(self): response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -1056,7 +1185,12 @@ def test_delete_asset_upload_in_progress(self): self.assertStatusCode(201, response) upload_id = response.json()['upload_id'] - response = self.client.delete(self.get_delete_asset_path()) + response = self.client.delete( + self.get_delete_asset_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(400, response) self.assertEqual( response.json()['description'], @@ -1070,10 +1204,20 @@ def test_delete_asset_upload_in_progress(self): msg='Collection Asset has been deleted' ) - response = self.client.post(self.get_abort_multipart_upload_path(upload_id)) + response = self.client.post( + self.get_abort_multipart_upload_path(upload_id), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) - response = self.client.delete(self.get_delete_asset_path()) + response = self.client.delete( + self.get_delete_asset_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) self.assertFalse( @@ -1084,6 +1228,7 @@ def test_delete_asset_upload_in_progress(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class GetCollectionAssetUploadsEndpointTestCase(CollectionAssetUploadBaseTest): def create_dummies_uploads(self): @@ -1216,6 +1361,7 @@ def test_get_asset_uploads_status_query(self): self.assertEqual(upload['status'], CollectionAssetUpload.Status.ABORTED) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetUploadListPartsEndpointTestCase(CollectionAssetUploadBaseTest): def test_asset_upload_list_parts(self): @@ -1228,6 +1374,9 @@ def test_asset_upload_list_parts(self): md5_parts = create_md5_parts(number_parts, offset, file_like) response = self.client.post( self.get_create_multipart_upload_path(), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={ 'number_parts': number_parts, 'file:checksum': checksum_multihash, @@ -1282,6 +1431,9 @@ def test_asset_upload_list_parts(self): # Complete the upload response = self.client.post( self.get_complete_multipart_upload_path(upload_id), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'parts': parts}, content_type="application/json" ) diff --git a/app/tests/tests_10/test_collection_assets_endpoint.py b/app/tests/tests_10/test_collection_assets_endpoint.py index 040740d9..83316949 100644 --- a/app/tests/tests_10/test_collection_assets_endpoint.py +++ b/app/tests/tests_10/test_collection_assets_endpoint.py @@ -22,7 +22,6 @@ from tests.tests_10.data_factory import Factory from tests.tests_10.utils import reverse_version from tests.utils import S3TestMixin -from tests.utils import client_login from tests.utils import disableLogger from tests.utils import mock_s3_asset_file @@ -95,6 +94,7 @@ def test_single_asset_endpoint(self): self.assertEtagHeader(None, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetsUnimplementedEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -102,7 +102,6 @@ def setUp(self): # pylint: disable=invalid-name self.factory = Factory() self.collection = self.factory.create_collection_sample().model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_unimplemented_post(self): @@ -112,12 +111,16 @@ def test_asset_unimplemented_post(self): ) response = self.client.post( f'/{STAC_BASE_V}/collections/{collection_name}/assets', + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset.get_json('post'), content_type="application/json" ) self.assertStatusCode(405, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetsCreateEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -125,7 +128,6 @@ def setUp(self): # pylint: disable=invalid-name self.factory = Factory() self.collection = self.factory.create_collection_sample().model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_upsert_create_only_required(self): @@ -138,7 +140,14 @@ def test_asset_upsert_create_only_required(self): json_to_send = asset.get_json('put') # Send a non normalized form of the type to see if it is also accepted json_to_send['type'] = 'image/TIFF;application=geotiff; Profile=cloud-optimized' - response = self.client.put(path, data=json_to_send, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=json_to_send, + content_type="application/json" + ) json_data = response.json() self.assertStatusCode(201, response) self.assertLocationHeader(f"{path}", response) @@ -178,6 +187,9 @@ def test_asset_upsert_create(self): # Now use upsert to create the new asset response = self.client.put( reverse_version('collection-asset-detail', args=[collection.name, asset_name]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset.get_json('put'), content_type="application/json" ) @@ -224,7 +236,12 @@ def test_asset_upsert_create_non_existing_parent_collection_in_path(self): # Now use upsert to create the new asset response = self.client.put( - path, data=asset.get_json('post'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('post'), + content_type="application/json" ) self.assertStatusCode(404, response) @@ -237,7 +254,12 @@ def test_asset_upsert_create_empty_string(self): path = \ f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset["name"]}' response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) json_data = response.json() @@ -253,7 +275,12 @@ def invalid_request_wrapper(self, sample_name, expected_error_messages, **extra_ path = \ f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset["name"]}' response = self.client.put( - path, data=asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual( @@ -297,6 +324,7 @@ def test_asset_upsert_create_type_extension_mismatch(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetsUpdateEndpointAssetFileTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -307,7 +335,6 @@ def setUp(self): # pylint: disable=invalid-name collection=self.collection.model, db_create=True ) self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_endpoint_patch_put_href(self): @@ -320,7 +347,14 @@ def test_asset_endpoint_patch_put_href(self): patch_payload = {'href': 'https://testserver/non-existing-asset'} path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' - response = self.client.patch(path, data=patch_payload, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=patch_payload, + content_type="application/json" + ) self.assertStatusCode(400, response) description = response.json()['description'] self.assertIn('href', description, msg=f'Unexpected field error {description}') @@ -330,7 +364,14 @@ def test_asset_endpoint_patch_put_href(self): msg="Unexpected error message" ) - response = self.client.put(path, data=put_payload, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=put_payload, + content_type="application/json" + ) self.assertStatusCode(400, response) description = response.json()['description'] self.assertIn('href', description, msg=f'Unexpected field error {description}') @@ -341,6 +382,7 @@ def test_asset_endpoint_patch_put_href(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetsUpdateEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -351,7 +393,6 @@ def setUp(self): # pylint: disable=invalid-name collection=self.collection.model, db_create=True ) self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_endpoint_put(self): @@ -368,7 +409,12 @@ def test_asset_endpoint_put(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.put( - path, data=changed_asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(200, response) @@ -395,7 +441,12 @@ def test_asset_endpoint_put_extra_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.put( - path, data=changed_asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'extra_attribute': ['Unexpected property in payload']}, @@ -418,6 +469,9 @@ def test_asset_endpoint_put_read_only_in_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.put( path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=changed_asset.get_json('put', keep_read_only=True), content_type="application/json" ) @@ -443,7 +497,12 @@ def test_asset_endpoint_put_rename_asset(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.put( - path, data=changed_asset.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'id': 'Renaming is not allowed'}, @@ -478,7 +537,12 @@ def test_asset_endpoint_patch_rename_asset(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.patch( - path, data=changed_asset.get_json('patch'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('patch'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'id': 'Renaming is not allowed'}, @@ -515,7 +579,12 @@ def test_asset_endpoint_patch_extra_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.patch( - path, data=changed_asset.get_json('patch'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=changed_asset.get_json('patch'), + content_type="application/json" ) self.assertStatusCode(400, response) self.assertEqual({'extra_payload': ['Unexpected property in payload']}, @@ -536,6 +605,9 @@ def test_asset_endpoint_patch_read_only_in_payload(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' response = self.client.patch( path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=changed_asset.get_json('patch', keep_read_only=True), content_type="application/json" ) @@ -556,6 +628,9 @@ def test_asset_atomic_upsert_create_500(self): 'test-collection-asset-detail-http-500', args=[self.collection['name'], sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -587,6 +662,9 @@ def test_asset_atomic_upsert_update_500(self): 'test-collection-asset-detail-http-500', args=[self.collection['name'], sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -634,6 +712,9 @@ def asset_atomic_upsert_test(worker): 'collection-asset-detail', args=[self.collection_sample['name'], asset_sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_sample.get_json('put'), content_type='application/json' ) @@ -655,6 +736,7 @@ def asset_atomic_upsert_test(worker): self.assertEqual(status_201, 1, msg="Not only one upsert did a create !") +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionAssetsDeleteEndpointTestCase(StacBaseTestCase, S3TestMixin): @mock_s3_asset_file @@ -663,7 +745,6 @@ def setUp(self): # pylint: disable=invalid-name self.collection = self.factory.create_collection_sample().model self.asset = self.factory.create_collection_asset_sample(collection=self.collection).model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_asset_endpoint_delete_asset(self): @@ -672,7 +753,12 @@ def test_asset_endpoint_delete_asset(self): path = f'/{STAC_BASE_V}/collections/{collection_name}/assets/{asset_name}' s3_path = get_collection_asset_path(self.collection, asset_name) self.assertS3ObjectExists(s3_path) - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) # Check that is has really been deleted @@ -689,7 +775,12 @@ def test_asset_endpoint_delete_asset(self): def test_asset_endpoint_delete_asset_invalid_name(self): collection_name = self.collection.name path = f"/{STAC_BASE_V}/collections/{collection_name}/assets/non-existent-asset" - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(404, response) diff --git a/app/tests/tests_10/test_collections_endpoint.py b/app/tests/tests_10/test_collections_endpoint.py index d23e9e1f..0e6b6ae4 100644 --- a/app/tests/tests_10/test_collections_endpoint.py +++ b/app/tests/tests_10/test_collections_endpoint.py @@ -23,7 +23,6 @@ from tests.tests_10.data_factory import Factory from tests.tests_10.data_factory import SampleData from tests.tests_10.utils import reverse_version -from tests.utils import client_login from tests.utils import disableLogger from tests.utils import mock_s3_asset_file @@ -117,11 +116,11 @@ def test_single_collection_assets_endpoint(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionsUnImplementedEndpointTestCase(StacBaseTestCase): def setUp(self): # pylint: disable=invalid-name self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample() self.maxDiff = None # pylint: disable=invalid-name @@ -129,17 +128,20 @@ def setUp(self): # pylint: disable=invalid-name def test_collections_post_unimplemented(self): response = self.client.post( f"/{STAC_BASE_V}/collections", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=self.collection.get_json('post'), content_type='application/json' ) self.assertStatusCode(405, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionsCreateEndpointTestCase(StacBaseTestCase): def setUp(self): # pylint: disable=invalid-name self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample() self.maxDiff = None # pylint: disable=invalid-name @@ -150,6 +152,9 @@ def test_collection_upsert_create(self): # the dataset to update does not exist yet response = self.client.put( f"/{STAC_BASE_V}/collections/{sample['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -163,6 +168,9 @@ def test_invalid_collections_create(self): response = self.client.put( f"/{STAC_BASE_V}/collections/{collection['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=collection.get_json('put'), content_type='application/json' ) @@ -177,7 +185,12 @@ def test_collections_min_mandatory_create(self): path = f"/{STAC_BASE_V}/collections/{collection['name']}" response = self.client.put( - path, data=collection.get_json('put'), content_type='application/json' + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=collection.get_json('put'), + content_type='application/json' ) response_json = response.json() logger.debug(response_json) @@ -195,6 +208,9 @@ def test_collections_less_than_mandatory_create(self): response = self.client.put( f"/{STAC_BASE_V}/collections/{collection['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=collection.get_json('put'), content_type='application/json' ) @@ -218,7 +234,12 @@ def test_collections_create_unpublished(self): path = f"/{STAC_BASE_V}/collections/{collection_sample['name']}" response = self.client.put( - path, data=collection_sample.get_json('put'), content_type='application/json' + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=collection_sample.get_json('put'), + content_type='application/json' ) self.assertStatusCode(201, response) self.assertLocationHeader(f'{path}', response) @@ -259,6 +280,9 @@ def test_collections_create_unpublished(self): # Publish the collection response = self.client.patch( f"/{STAC_BASE_V}/collections/{collection.name}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data={'published': True}, content_type='application/json' ) @@ -283,6 +307,9 @@ def test_collection_atomic_upsert_create_500(self): with self.settings(DEBUG_PROPAGATE_API_EXCEPTIONS=True), disableLogger('stac_api.apps'): response = self.client.put( reverse('test-collection-detail-http-500', args=[sample['name']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -294,11 +321,11 @@ def test_collection_atomic_upsert_create_500(self): self.assertStatusCode(404, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionsUpdateEndpointTestCase(StacBaseTestCase): def setUp(self): # pylint: disable=invalid-name self.client = Client() - client_login(self.client) self.collection_factory = CollectionFactory() self.collection = self.collection_factory.create_sample(db_create=True) self.maxDiff = None # pylint: disable=invalid-name @@ -310,6 +337,9 @@ def test_collections_put(self): response = self.client.put( f"/{STAC_BASE_V}/collections/{sample['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -329,6 +359,9 @@ def test_collections_put_extra_payload(self): response = self.client.put( f"/{STAC_BASE_V}/collections/{sample['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -346,6 +379,9 @@ def test_collections_put_read_only_in_payload(self): response = self.client.put( f"/{STAC_BASE_V}/collections/{sample['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put', keep_read_only=True), content_type='application/json' ) @@ -364,6 +400,9 @@ def test_collection_put_change_id(self): self.assertNotEqual(self.collection['name'], sample['name']) response = self.client.put( f"/{STAC_BASE_V}/collections/{self.collection['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -390,6 +429,9 @@ def test_collection_put_remove_optional_fields(self): self.assertNotEqual('', f'{self.collection["title"]}') response = self.client.put( f"/{STAC_BASE_V}/collections/{sample['name']}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -408,6 +450,9 @@ def test_collection_patch(self): response = self.client.patch( f"/{STAC_BASE_V}/collections/{collection_name}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=payload_json, content_type='application/json' ) @@ -427,6 +472,9 @@ def test_collection_patch_extra_payload(self): # for start the payload has no description response = self.client.patch( f"/{STAC_BASE_V}/collections/{collection_name}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=payload_json, content_type='application/json' ) @@ -442,6 +490,9 @@ def test_collection_patch_read_only_in_payload(self): self.assertNotEqual(self.collection['license'], payload_json['license']) response = self.client.patch( f"/{STAC_BASE_V}/collections/{collection_name}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=payload_json, content_type='application/json' ) @@ -464,6 +515,9 @@ def test_collection_atomic_upsert_update_500(self): # console therefore disable it. response = self.client.put( reverse('test-collection-detail-http-500', args=[sample['name']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -476,11 +530,11 @@ def test_collection_atomic_upsert_update_500(self): self.check_stac_collection(self.collection.json, response.json()) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionsDeleteEndpointTestCase(StacBaseTestCase): def setUp(self): # pylint: disable=invalid-name self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample(db_create=True) self.item = self.factory.create_item_sample(self.collection.model, db_create=True) @@ -489,7 +543,12 @@ def setUp(self): # pylint: disable=invalid-name def test_authorized_collection_delete(self): path = reverse_version('collection-detail', args=[self.collection["name"]]) - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(400, response) self.assertEqual( @@ -500,7 +559,11 @@ def test_authorized_collection_delete(self): item_path = reverse_version( 'item-detail', args=[self.collection["name"], self.item['name']] ) - response = self.client.delete(item_path) + response = self.client.delete( + item_path, headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + } + ) self.assertStatusCode(200, response) # try the collection delete again @@ -669,11 +732,11 @@ def test_disabled_base_authentication(self): self.run_test(401, headers=headers) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class CollectionLinksEndpointTestCase(StacBaseTestCase): def setUp(self): self.client = Client() - client_login(self.client) @classmethod def setUpTestData(cls) -> None: @@ -686,7 +749,14 @@ def test_create_collection_link_with_simple_link(self): data = self.collection_data.get_json('put') path = f'/{STAC_BASE_V}/collections/{self.collection.name}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertEqual(response.status_code, 200) @@ -706,7 +776,14 @@ def test_create_collection_link_with_hreflang(self): }] path = f'/{STAC_BASE_V}/collections/{self.collection.name}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertEqual(response.status_code, 200) @@ -747,7 +824,14 @@ def test_create_collection_link_with_invalid_hreflang(self): }] path = f'/{STAC_BASE_V}/collections/{self.collection.name}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertEqual(response.status_code, 400) content = response.json() diff --git a/app/tests/tests_10/test_external_assets_endpoint.py b/app/tests/tests_10/test_external_assets_endpoint.py index 66cab8e5..18aa7b1d 100644 --- a/app/tests/tests_10/test_external_assets_endpoint.py +++ b/app/tests/tests_10/test_external_assets_endpoint.py @@ -2,16 +2,17 @@ from django.conf import settings from django.test import Client +from django.test import override_settings from stac_api.models.item import Asset from tests.tests_10.base_test import StacBaseTestCase from tests.tests_10.data_factory import Factory from tests.tests_10.utils import reverse_version -from tests.utils import client_login from tests.utils import mock_s3_asset_file +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class AssetsExternalAssetEndpointTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -20,7 +21,6 @@ def setUp(self): # pylint: disable=invalid-name self.collection = self.factory.create_collection_sample().model self.item = self.factory.create_item_sample(collection=self.collection).model self.client = Client() - client_login(self.client) self.maxDiff = None # pylint: disable=invalid-name def test_create_asset_with_external_url(self): @@ -37,6 +37,9 @@ def test_create_asset_with_external_url(self): # create the asset, which isn't allowed response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -50,6 +53,9 @@ def test_create_asset_with_external_url(self): # create the asset, now it's allowed response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -94,6 +100,9 @@ def test_create_asset_validate_external_url(self): # create the asset response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -137,6 +146,9 @@ def test_create_asset_validate_external_url_not_found(self): # create the asset response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -180,6 +192,9 @@ def test_create_asset_validate_external_url_bad_content(self): # create the asset response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -208,6 +223,9 @@ def test_update_asset_with_external_url(self): response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset.attr_name]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type='application/json' ) @@ -254,6 +272,9 @@ def test_create_asset_with_invalid_external_url(self): # create the asset response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -286,6 +307,9 @@ def test_create_asset_with_inexistent_external_url(self): # create the asset with an existing one response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -296,6 +320,9 @@ def test_create_asset_with_inexistent_external_url(self): # create the asset with an existing one response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -327,6 +354,9 @@ def test_create_asset_with_inexistent_domain(self): # create the asset with an existing one response = self.client.put( reverse_version('asset-detail', args=[collection.name, item.name, asset_data['id']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=asset_data, content_type="application/json" ) @@ -348,7 +378,10 @@ def test_delete_asset_with_external_url(self): ) response = self.client.delete( - reverse_version('asset-detail', args=[collection.name, item.name, asset.attr_name]) + reverse_version('asset-detail', args=[collection.name, item.name, asset.attr_name]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + } ) self.assertStatusCode(200, response) diff --git a/app/tests/tests_10/test_generic_api.py b/app/tests/tests_10/test_generic_api.py index d0d42efd..981768f0 100644 --- a/app/tests/tests_10/test_generic_api.py +++ b/app/tests/tests_10/test_generic_api.py @@ -15,7 +15,6 @@ from tests.tests_10.base_test import StacBaseTestCase from tests.tests_10.data_factory import Factory from tests.utils import S3TestMixin -from tests.utils import client_login from tests.utils import disableLogger from tests.utils import get_http_error_description from tests.utils import mock_s3_asset_file @@ -193,6 +192,7 @@ def test_pagination(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ApiETagPreconditionTestCase(StacBaseTestCase): @mock_s3_asset_file @@ -252,7 +252,6 @@ def test_get_precondition(self): self.assertStatusCode(412, response4) def test_put_precondition(self): - client_login(self.client) for (endpoint, sample) in [ ( f'collections/{self.collection["name"]}', @@ -285,7 +284,10 @@ def test_put_precondition(self): response = self.client.put( f"/{STAC_BASE_V}/{endpoint}", - sample.get_json('put'), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), content_type="application/json", HTTP_IF_MATCH='"abc"' ) @@ -293,15 +295,16 @@ def test_put_precondition(self): response = self.client.put( f"/{STAC_BASE_V}/{endpoint}", - sample.get_json('put'), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), content_type="application/json", HTTP_IF_MATCH=self.get_etag(endpoint) ) self.assertStatusCode(200, response) def test_wrong_media_type(self): - client_login(self.client) - for (request_methods, endpoint, data) in [ ( ['put', 'patch'], @@ -325,12 +328,16 @@ def test_wrong_media_type(self): client_requests = [getattr(self.client, method) for method in request_methods] for client_request in client_requests: response = client_request( - f"/{STAC_BASE_V}/{endpoint}", data=data, content_type="plain/text" + f"/{STAC_BASE_V}/{endpoint}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="plain/text" ) self.assertStatusCode(415, response) def test_patch_precondition(self): - client_login(self.client) for (endpoint, data) in [ ( f'collections/{self.collection["name"]}', @@ -358,7 +365,10 @@ def test_patch_precondition(self): response = self.client.patch( f"/{STAC_BASE_V}/{endpoint}", - data, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, content_type="application/json", HTTP_IF_MATCH='"abc"' ) @@ -366,14 +376,16 @@ def test_patch_precondition(self): response = self.client.patch( f"/{STAC_BASE_V}/{endpoint}", - data, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, content_type="application/json", HTTP_IF_MATCH=self.get_etag(endpoint) ) self.assertStatusCode(200, response) def test_delete_precondition(self): - client_login(self.client) for endpoint in [ f'collections/{self.collection["name"]}/items/{self.item["name"]}' f'/assets/{self.asset["name"]}', @@ -385,6 +397,9 @@ def test_delete_precondition(self): response = self.client.delete( f"/{STAC_BASE_V}/{endpoint}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, content_type="application/json", HTTP_IF_MATCH='"abc"' ) @@ -394,6 +409,9 @@ def test_delete_precondition(self): response = self.client.delete( f"/{STAC_BASE_V}/{endpoint}", + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, content_type="application/json", HTTP_IF_MATCH=etag1 ) diff --git a/app/tests/tests_10/test_geoadmin_header_auth.py b/app/tests/tests_10/test_geoadmin_header_auth.py index 93e0478d..2d753cd3 100644 --- a/app/tests/tests_10/test_geoadmin_header_auth.py +++ b/app/tests/tests_10/test_geoadmin_header_auth.py @@ -2,7 +2,6 @@ from parameterized import parameterized -from django.contrib.auth import get_user_model from django.test import Client from django.test import override_settings @@ -15,19 +14,19 @@ @override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class GeoadminHeadersAuthForPutEndpointTestCase(StacBaseTestCase): - valid_username = "another_test_user" def setUp(self): # pylint: disable=invalid-name self.client = Client(enforce_csrf_checks=True) self.factory = Factory() self.collection = self.factory.create_collection_sample() - get_user_model().objects.create_superuser(self.valid_username) @parameterized.expand([ - (valid_username, "true", 201), + ("another_test_user", "true", 201), + ("another_test_user", "false", 401), + ("another_test_user", "", 401), (None, None, 401), - (valid_username, "false", 401), - ("wronguser", "true", 403), + (None, "false", 401), + (None, "true", 401), ]) def test_collection_upsert_create_with_geoadmin_header_auth( self, username_header, authenticated_header, expected_response_code diff --git a/app/tests/tests_10/test_item_to_collection_endpoint.py b/app/tests/tests_10/test_item_to_collection_endpoint.py index 61e588f5..314363df 100644 --- a/app/tests/tests_10/test_item_to_collection_endpoint.py +++ b/app/tests/tests_10/test_item_to_collection_endpoint.py @@ -2,23 +2,23 @@ from django.contrib.gis.geos import Polygon from django.test import Client +from django.test import override_settings from tests.tests_10.base_test import STAC_BASE_V from tests.tests_10.base_test import StacBaseTransactionTestCase from tests.tests_10.data_factory import Factory from tests.tests_10.utils import calculate_extent -from tests.utils import client_login logger = logging.getLogger(__name__) # Here we need to use TransactionTestCase due to the pgtrigger, in a normal # test case we cannot test effect of pgtrigger. +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class OneItemSpatialTestCase(StacBaseTransactionTestCase): def setUp(self): self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample().model self.items = self.factory.create_item_samples(['item-switzerland-west'], @@ -46,7 +46,12 @@ def test_no_items(self): item_name = self.items[0].model.name # delete the item path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{item_name}' - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) response_collection = self.client.get(f"/{STAC_BASE_V}/collections/{collection_name}") @@ -57,11 +62,11 @@ def test_no_items(self): self.assertEqual(bbox_collection, [0, 0, 0, 0]) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class TwoItemsSpatialTestCase(StacBaseTransactionTestCase): def setUp(self): self.client = Client() - client_login(self.client) self.factory = Factory() self.collection = self.factory.create_collection_sample().model self.items = self.factory.create_item_samples( @@ -125,7 +130,12 @@ def test_one_left_item(self): # delete the eastern item path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{item_east}' - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) calculate_extent() @@ -151,7 +161,12 @@ def test_update_covering_item(self): ) path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{item_name}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) calculate_extent() diff --git a/app/tests/tests_10/test_items_endpoint.py b/app/tests/tests_10/test_items_endpoint.py index b53f3d74..b83762de 100644 --- a/app/tests/tests_10/test_items_endpoint.py +++ b/app/tests/tests_10/test_items_endpoint.py @@ -28,7 +28,6 @@ from tests.tests_10.data_factory import Factory from tests.tests_10.data_factory import ItemFactory from tests.tests_10.utils import reverse_version -from tests.utils import client_login from tests.utils import disableLogger from tests.utils import mock_s3_asset_file @@ -447,6 +446,7 @@ def test_items_endpoint_datetime_open_start_range_query(self): self._navigate_to_previous_items(['item-yesterday-1', 'item-2', 'item-1'], json_response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ItemsUnImplementedEndpointTestCase(StacBaseTestCase): @classmethod @@ -456,18 +456,21 @@ def setUpTestData(cls): def setUp(self): self.client = Client() - client_login(self.client) def test_item_post_unimplemented(self): sample = self.factory.create_item_sample(self.collection) response = self.client.post( f'/{STAC_BASE_V}/collections/{self.collection.name}/items', + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('post'), content_type="application/json" ) self.assertStatusCode(405, response) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ItemsCreateEndpointTestCase(StacBaseTestCase): @classmethod @@ -477,13 +480,17 @@ def setUpTestData(cls): def setUp(self): self.client = Client() - client_login(self.client) def test_item_upsert_create(self): sample = self.factory.create_item_sample(self.collection) path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{sample.json["id"]}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(201, response) @@ -493,7 +500,12 @@ def test_item_endpoint_create_only_required(self): sample = self.factory.create_item_sample(self.collection, required_only=True) path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{sample["name"]}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(201, response) @@ -512,6 +524,9 @@ def test_item_upsert_create_non_existing_parent_collection_in_path(self): sample = self.factory.create_item_sample(self.collection, required_only=True) response = self.client.put( f'/{STAC_BASE_V}/collections/non-existing-collection/items/{sample.json["id"]}', + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type="application/json" ) @@ -524,6 +539,9 @@ def test_item_atomic_upsert_create_500(self): with self.settings(DEBUG_PROPAGATE_API_EXCEPTIONS=True), disableLogger('stac_api.apps'): response = self.client.put( reverse('test-item-detail-http-500', args=[self.collection.name, sample['name']]), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -540,7 +558,14 @@ def test_item_endpoint_create_invalid_data(self): data = self.factory.create_item_sample(self.collection, sample='item-invalid').get_json('put') path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{data["id"]}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) # Make sure that the item is not found in DB @@ -557,7 +582,14 @@ def test_item_endpoint_create_missing_datetime(self): properties_end_datetime=None ).get_json('put') path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{data["id"]}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) # Make sure that the item is not found in DB @@ -567,6 +599,7 @@ def test_item_endpoint_create_missing_datetime(self): ) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ItemsUpdateEndpointTestCase(StacBaseTestCase): @classmethod @@ -579,7 +612,6 @@ def setUpTestData(cls): def setUp(self): self.client = Client() - client_login(self.client) def test_item_endpoint_put(self): sample = self.factory.create_item_sample( @@ -587,7 +619,12 @@ def test_item_endpoint_put(self): ) path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(200, response) @@ -605,7 +642,12 @@ def test_item_endpoint_put_extra_payload(self): ) path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) self.assertStatusCode(400, response) @@ -614,7 +656,14 @@ def test_item_endpoint_put_read_only_in_payload(self): self.collection.model, sample='item-2', name=self.item['name'], created=datetime.now() ).get_json('put') path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) def test_item_endpoint_put_update_to_datetime_range(self): @@ -629,7 +678,12 @@ def test_item_endpoint_put_update_to_datetime_range(self): ) path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(200, response) @@ -656,7 +710,12 @@ def test_item_endpoint_put_remove_properties_title(self): } ) response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(200, response) @@ -674,7 +733,12 @@ def test_item_endpoint_put_remove_properties_title(self): } ) response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(200, response) @@ -689,7 +753,12 @@ def test_item_endpoint_put_rename_item(self): ) path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' response = self.client.put( - path, data=sample.get_json('put'), content_type="application/json" + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=sample.get_json('put'), + content_type="application/json" ) json_data = response.json() self.assertStatusCode(400, response) @@ -714,7 +783,14 @@ def test_item_endpoint_put_rename_item(self): def test_item_endpoint_patch(self): data = {"properties": {"title": "patched title"}} path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) json_data = response.json() self.assertStatusCode(200, response) self.assertEqual(self.item['name'], json_data['id']) @@ -740,7 +816,14 @@ def test_item_endpoint_patch_remove_properties_title(self): # Remove properties_title data = {"properties": {"title": None}} - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) json_data = response.json() self.assertStatusCode(200, response) self.assertEqual(self.item['name'], json_data['id']) @@ -756,23 +839,51 @@ def test_item_endpoint_patch_remove_properties_title(self): def test_item_endpoint_patch_extra_payload(self): data = {"crazy:stuff": "not allowed"} path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) def test_item_endpoint_patch_read_only_in_payload(self): data = {"created": utc_aware(datetime.utcnow())} path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) def test_item_endpoint_patch_invalid_datetimes(self): data = {"properties": {"datetime": "patched title",}} path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) data = {"properties": {"start_datetime": "2020-10-28T13:05:10Z",}} - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertStatusCode(400, response) def test_item_endpoint_patch_rename_item(self): @@ -780,7 +891,14 @@ def test_item_endpoint_patch_rename_item(self): "id": f'new-{self.item["name"]}', } path = f'/{STAC_BASE_V}/collections/{self.collection["name"]}/items/{self.item["name"]}' - response = self.client.patch(path, data=data, content_type="application/json") + response = self.client.patch( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) json_data = response.json() self.assertStatusCode(400, response) self.assertEqual(json_data['description'], {'id': 'Renaming is not allowed'}) @@ -815,6 +933,9 @@ def test_item_atomic_upsert_update_500(self): reverse( 'test-item-detail-http-500', args=[self.collection['name'], sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=sample.get_json('put'), content_type='application/json' ) @@ -851,6 +972,9 @@ def item_atomic_upsert_test(worker): reverse_version( 'item-detail', args=[collection_sample['name'], item_sample['name']] ), + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, data=item_sample.get_json('put'), content_type='application/json' ) @@ -870,6 +994,7 @@ def item_atomic_upsert_test(worker): self.assertEqual(status_201, 1, msg="Not only one upsert did a create !") +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ItemsDeleteEndpointTestCase(StacBaseTestCase): @classmethod @@ -879,7 +1004,6 @@ def setUpTestData(cls): @mock_s3_asset_file def setUp(self): self.client = Client() - client_login(self.client) self.collection = self.factory.create_collection_sample().model self.item = self.factory.create_item_sample(self.collection, sample='item-1').model self.asset = self.factory.create_asset_sample(self.item, sample='asset-1').model @@ -887,19 +1011,34 @@ def setUp(self): def test_item_endpoint_delete_item(self): # Check that deleting, while assets are present, is not allowed path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{self.item.name}' - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(400, response) self.assertEqual(response.json()['description'], ['Deleting Item with assets not allowed']) # delete asset first asset_path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{self.item.name}' \ f'/assets/{self.asset.name}' - response = self.client.delete(asset_path) + response = self.client.delete( + asset_path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) # try item delete again path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{self.item.name}' - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(200, response) # Check that is has really been deleted @@ -913,7 +1052,12 @@ def test_item_endpoint_delete_item(self): def test_item_endpoint_delete_item_invalid_name(self): path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/unknown-item' - response = self.client.delete(path) + response = self.client.delete( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + ) self.assertStatusCode(404, response) @@ -1020,11 +1164,11 @@ def test_disabled_base_authentication(self): self.run_test(401, headers=headers) +@override_settings(FEATURE_AUTH_ENABLE_APIGW=True) class ItemsLinksEndpointTestCase(StacBaseTestCase): def setUp(self): self.client = Client() - client_login(self.client) @classmethod def setUpTestData(cls) -> None: @@ -1041,7 +1185,14 @@ def test_create_item_link_with_simple_link(self): data = self.item_data.get_json('put') path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{self.item.name}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertEqual(response.status_code, 200) @@ -1061,7 +1212,14 @@ def test_create_item_link_with_hreflang(self): }] path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{self.item.name}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertEqual(response.status_code, 200) @@ -1101,7 +1259,14 @@ def test_update_item_link_with_invalid_hreflang(self): }] path = f'/{STAC_BASE_V}/collections/{self.collection.name}/items/{self.item.name}' - response = self.client.put(path, data=data, content_type="application/json") + response = self.client.put( + path, + headers={ + "Geoadmin-Username": "apiuser", "Geoadmin-Authenticated": "true" + }, + data=data, + content_type="application/json" + ) self.assertEqual(response.status_code, 400) content = response.json()