From f6bb4f1378c21de2e68d3d959cfdce2ec399019c Mon Sep 17 00:00:00 2001 From: index-git Date: Tue, 19 Dec 2023 09:13:20 +0100 Subject: [PATCH 1/4] Let test_layman_users_and_roles_to_geoserver_roles deal also with roles --- src/layman/common/geoserver/__init__.py | 9 ++++++--- src/layman/common/geoserver/geoserver_test.py | 15 +++++++++++++++ src/layman/layer/geoserver/__init__.py | 4 ++-- src/layman/layer/geoserver/wfs.py | 4 ++-- src/layman/layer/geoserver/wms.py | 4 ++-- .../static_data/single_publication/layers_test.py | 6 +++--- 6 files changed, 30 insertions(+), 12 deletions(-) create mode 100644 src/layman/common/geoserver/geoserver_test.py diff --git a/src/layman/common/geoserver/__init__.py b/src/layman/common/geoserver/__init__.py index 62cb6953d..d887085d8 100644 --- a/src/layman/common/geoserver/__init__.py +++ b/src/layman/common/geoserver/__init__.py @@ -1,13 +1,16 @@ from geoserver.util import username_to_rolename from layman import settings +from layman.authz import is_user -def layman_users_to_geoserver_roles(layman_users): +def layman_users_and_roles_to_geoserver_roles(layman_users_and_roles): geoserver_roles = set() - for layman_user in layman_users: + for layman_user in layman_users_and_roles: if layman_user == settings.RIGHTS_EVERYONE_ROLE: geoserver_roles.add('ROLE_ANONYMOUS') geoserver_roles.add('ROLE_AUTHENTICATED') - else: + elif is_user(layman_user): geoserver_roles.add(username_to_rolename(layman_user)) + else: + geoserver_roles.add(layman_user) return geoserver_roles diff --git a/src/layman/common/geoserver/geoserver_test.py b/src/layman/common/geoserver/geoserver_test.py new file mode 100644 index 000000000..94e089999 --- /dev/null +++ b/src/layman/common/geoserver/geoserver_test.py @@ -0,0 +1,15 @@ +import pytest + +from layman import settings +from . import layman_users_and_roles_to_geoserver_roles + + +@pytest.mark.parametrize('layman_users_and_roles, exp_geoserver_roles', [ + pytest.param({'username'}, {'USER_USERNAME'}, id='username'), + pytest.param({'ROLE'}, {'ROLE'}, id='rolename'), + pytest.param({settings.RIGHTS_EVERYONE_ROLE}, {'ROLE_ANONYMOUS', 'ROLE_AUTHENTICATED'}, id='everyone-role'), + pytest.param({f'username2', 'ROLE2', settings.RIGHTS_EVERYONE_ROLE}, {'USER_USERNAME2', 'ROLE2', 'ROLE_ANONYMOUS', 'ROLE_AUTHENTICATED'}, id='everything'), +]) +def test_layman_users_and_roles_to_geoserver_roles(layman_users_and_roles, exp_geoserver_roles): + result = layman_users_and_roles_to_geoserver_roles(layman_users_and_roles) + assert result == exp_geoserver_roles diff --git a/src/layman/layer/geoserver/__init__.py b/src/layman/layer/geoserver/__init__.py index 312112419..e96e9e53f 100644 --- a/src/layman/layer/geoserver/__init__.py +++ b/src/layman/layer/geoserver/__init__.py @@ -91,10 +91,10 @@ def set_security_rules(workspace, layer, access_rights, auth, geoserver_workspac read_roles = access_rights.get('read') if access_rights and access_rights.get('read') else layer_info['access_rights']['read'] write_roles = access_rights.get('write') if access_rights and access_rights.get('write') else layer_info['access_rights']['write'] - security_read_roles = gs_common.layman_users_to_geoserver_roles(read_roles) + security_read_roles = gs_common.layman_users_and_roles_to_geoserver_roles(read_roles) gs_util.ensure_layer_security_roles(geoserver_workspace, layer, security_read_roles, 'r', auth) - security_write_roles = gs_common.layman_users_to_geoserver_roles(write_roles) + security_write_roles = gs_common.layman_users_and_roles_to_geoserver_roles(write_roles) gs_util.ensure_layer_security_roles(geoserver_workspace, layer, security_write_roles, 'w', auth) diff --git a/src/layman/layer/geoserver/wfs.py b/src/layman/layer/geoserver/wfs.py index 6dca91afd..fc0baf4e9 100644 --- a/src/layman/layer/geoserver/wfs.py +++ b/src/layman/layer/geoserver/wfs.py @@ -39,11 +39,11 @@ def patch_layer(workspace, layername, title, description, original_data_source, clear_cache(workspace) if access_rights and access_rights.get('read'): - security_read_roles = gs_common.layman_users_to_geoserver_roles(access_rights['read']) + security_read_roles = gs_common.layman_users_and_roles_to_geoserver_roles(access_rights['read']) gs_util.ensure_layer_security_roles(workspace, layername, security_read_roles, 'r', settings.LAYMAN_GS_AUTH) if access_rights and access_rights.get('write'): - security_write_roles = gs_common.layman_users_to_geoserver_roles(access_rights['write']) + security_write_roles = gs_common.layman_users_and_roles_to_geoserver_roles(access_rights['write']) gs_util.ensure_layer_security_roles(workspace, layername, security_write_roles, 'w', settings.LAYMAN_GS_AUTH) diff --git a/src/layman/layer/geoserver/wms.py b/src/layman/layer/geoserver/wms.py index b33b908fb..270e3eda2 100644 --- a/src/layman/layer/geoserver/wms.py +++ b/src/layman/layer/geoserver/wms.py @@ -58,11 +58,11 @@ def patch_layer(workspace, layername, original_data_source, title, description, clear_cache(workspace) if access_rights and access_rights.get('read'): - security_read_roles = gs_common.layman_users_to_geoserver_roles(access_rights['read']) + security_read_roles = gs_common.layman_users_and_roles_to_geoserver_roles(access_rights['read']) gs_util.ensure_layer_security_roles(geoserver_workspace, layername, security_read_roles, 'r', settings.LAYMAN_GS_AUTH) if access_rights and access_rights.get('write'): - security_write_roles = gs_common.layman_users_to_geoserver_roles(access_rights['write']) + security_write_roles = gs_common.layman_users_and_roles_to_geoserver_roles(access_rights['write']) gs_util.ensure_layer_security_roles(geoserver_workspace, layername, security_write_roles, 'w', settings.LAYMAN_GS_AUTH) diff --git a/tests/static_data/single_publication/layers_test.py b/tests/static_data/single_publication/layers_test.py index efe75313d..7c8f8a4ed 100644 --- a/tests/static_data/single_publication/layers_test.py +++ b/tests/static_data/single_publication/layers_test.py @@ -261,8 +261,8 @@ def test_gs_data_security(workspace, publ_type, publication): auth = settings.LAYMAN_GS_AUTH is_personal_workspace = workspace in data.USERS - owner_and_everyone_roles = gs_common.layman_users_to_geoserver_roles({workspace, settings.RIGHTS_EVERYONE_ROLE}) - owner_role_set = gs_common.layman_users_to_geoserver_roles({workspace}) + owner_and_everyone_roles = gs_common.layman_users_and_roles_to_geoserver_roles({workspace, settings.RIGHTS_EVERYONE_ROLE}) + owner_role_set = gs_common.layman_users_and_roles_to_geoserver_roles({workspace}) with app.app_context(): info = layman_util.get_publication_info(workspace, publ_type, publication, context={'keys': ['access_rights', 'wms']}) expected_roles = info['access_rights'] @@ -271,7 +271,7 @@ def test_gs_data_security(workspace, publ_type, publication): workspaces = [workspace, gs_workspace] if geodata_type != settings.GEODATA_TYPE_RASTER else [gs_workspace] for right_type in ['read', 'write']: for wspace in workspaces: - gs_expected_roles = gs_common.layman_users_to_geoserver_roles(expected_roles[right_type]) + gs_expected_roles = gs_common.layman_users_and_roles_to_geoserver_roles(expected_roles[right_type]) gs_roles = gs_util.get_security_roles(f'{wspace}.{publication}.{right_type[0]}', auth) assert gs_expected_roles == gs_roles\ or (is_personal_workspace From 8204cbf9f842fd227ebc8dedbe298aee2198d37d Mon Sep 17 00:00:00 2001 From: index-git Date: Tue, 19 Dec 2023 09:55:06 +0100 Subject: [PATCH 2/4] Test roles in GeoServer ACL rules --- .../publications/access_rights/test_role.py | 32 +++++++++++++------ .../single_publication/layers_test.py | 26 +-------------- 2 files changed, 24 insertions(+), 34 deletions(-) diff --git a/tests/dynamic_data/publications/access_rights/test_role.py b/tests/dynamic_data/publications/access_rights/test_role.py index 1475e0d92..ce07d76ac 100644 --- a/tests/dynamic_data/publications/access_rights/test_role.py +++ b/tests/dynamic_data/publications/access_rights/test_role.py @@ -1,19 +1,16 @@ import pytest +from geoserver import util as gs_util +from layman import app, settings, util as layman_util +from layman.common import geoserver as gs_common from test_tools import process_client, role_service from tests import EnumTestTypes, Publication from tests.asserts.final.publication import util as assert_util from tests.dynamic_data import base_test, base_test_classes -from tests.dynamic_data.publications import common_publications pytest_generate_tests = base_test.pytest_generate_tests -class PublicationTypes(base_test_classes.PublicationByDefinitionBase): - LAYER = (common_publications.LAYER_VECTOR_SLD, 'layer') - MAP = (common_publications.MAP_EMPTY, 'map') - - USERNAME = 'test_access_rights_role_user1' USER_ROLE1_ROLE3_EVERYONE = {USERNAME, 'ROLE1', 'ROLE3', 'EVERYONE'} USER_ROLE1 = {USERNAME, 'ROLE1'} @@ -27,7 +24,7 @@ class TestPublication(base_test.TestSingleRestPublication): publication_type = None rest_parametrization = [ - PublicationTypes, + base_test.PublicationByUsedServers, base_test_classes.RestMethod ] @@ -35,6 +32,8 @@ class TestPublication(base_test.TestSingleRestPublication): USERNAME, ] + external_tables_to_create = base_test_classes.EXTERNAL_TABLE_FOR_LAYERS_BY_USED_SERVERS + def before_class(self): for role in ROLES: role_service.ensure_role(role) @@ -74,5 +73,20 @@ def test_publication(self, publication, rest_method, rest_args): info = process_client.get_workspace_publication(publication.type, publication.workspace, publication.name, actor_name=USERNAME) - assert set(info['access_rights']['read']) == USER_ROLE1_ROLE2 - assert set(info['access_rights']['write']) == USER_ROLE1 + for right, exp_rights in [('read', USER_ROLE1_ROLE2), + ('write', USER_ROLE1), + ]: + assert set(info['access_rights'][right]) == exp_rights + + if publication.type == process_client.LAYER_TYPE: + with app.app_context(): + internal_info = layman_util.get_publication_info(publication.workspace, publication.type, publication.name, {'keys': ['geodata_type', 'wms']}) + + geodata_type = internal_info['geodata_type'] + gs_workspace = internal_info['_wms']['workspace'] + workspaces = [publication.workspace, gs_workspace] if geodata_type != settings.GEODATA_TYPE_RASTER else [gs_workspace] + for wspace in workspaces: + gs_expected_roles = gs_common.layman_users_and_roles_to_geoserver_roles(exp_rights) + rule = f'{wspace}.{publication.name}.{right[0]}' + gs_roles = gs_util.get_security_roles(rule, settings.LAYMAN_GS_AUTH) + assert gs_expected_roles == gs_roles, f'gs_expected_roles={gs_expected_roles}, gs_roles={gs_roles}, wspace={wspace}, rule={rule}' diff --git a/tests/static_data/single_publication/layers_test.py b/tests/static_data/single_publication/layers_test.py index 7c8f8a4ed..2bea82ea7 100644 --- a/tests/static_data/single_publication/layers_test.py +++ b/tests/static_data/single_publication/layers_test.py @@ -9,7 +9,7 @@ import crs as crs_def from geoserver import GS_REST_WORKSPACES, GS_REST, GS_AUTH, util as gs_util from layman import settings, app, util as layman_util -from layman.common import bbox as bbox_util, geoserver as gs_common +from layman.common import bbox as bbox_util from layman.common.micka import util as micka_common_util from layman.layer import util as layer_util, db as layer_db, get_layer_info_keys from layman.layer.geoserver.wms import DEFAULT_WMS_QGIS_STORE_PREFIX, VERSION @@ -254,30 +254,6 @@ def test_fill_project_template(workspace, publ_type, publication): assert excinfo.value.response.status_code == 500 -@pytest.mark.parametrize('workspace, publ_type, publication', data.LIST_LAYERS) -@pytest.mark.usefixtures('oauth2_provider_mock', 'ensure_layman') -def test_gs_data_security(workspace, publ_type, publication): - ensure_publication(workspace, publ_type, publication) - - auth = settings.LAYMAN_GS_AUTH - is_personal_workspace = workspace in data.USERS - owner_and_everyone_roles = gs_common.layman_users_and_roles_to_geoserver_roles({workspace, settings.RIGHTS_EVERYONE_ROLE}) - owner_role_set = gs_common.layman_users_and_roles_to_geoserver_roles({workspace}) - with app.app_context(): - info = layman_util.get_publication_info(workspace, publ_type, publication, context={'keys': ['access_rights', 'wms']}) - expected_roles = info['access_rights'] - gs_workspace = info['_wms']['workspace'] - geodata_type = data.PUBLICATIONS[(workspace, publ_type, publication)][data.TEST_DATA].get('geodata_type') - workspaces = [workspace, gs_workspace] if geodata_type != settings.GEODATA_TYPE_RASTER else [gs_workspace] - for right_type in ['read', 'write']: - for wspace in workspaces: - gs_expected_roles = gs_common.layman_users_and_roles_to_geoserver_roles(expected_roles[right_type]) - gs_roles = gs_util.get_security_roles(f'{wspace}.{publication}.{right_type[0]}', auth) - assert gs_expected_roles == gs_roles\ - or (is_personal_workspace - and gs_expected_roles == owner_and_everyone_roles == gs_roles.union(owner_role_set)), f'gs_expected_roles={gs_expected_roles}, gs_roles={gs_roles}, wspace={wspace}, is_personal_workspace={is_personal_workspace}' - - @pytest.mark.parametrize('workspace, publ_type, publication', [(wspace, ptype, pub) for wspace, ptype, pub in data.LIST_LAYERS if data.PUBLICATIONS[(wspace, ptype, pub)][data.TEST_DATA].get('micka_xml')]) From d8b2b01b725827dcf6c74afa40c335c062b7de45 Mon Sep 17 00:00:00 2001 From: index-git Date: Tue, 19 Dec 2023 11:48:59 +0100 Subject: [PATCH 3/4] Test access rights application on GeoServer --- CHANGELOG.md | 1 + .../final/publication/geoserver_proxy.py | 18 +++++++-- .../final/publication/geoserver_util.py | 2 +- .../test_access_rights_application.py | 39 +++++++++++++++++-- 4 files changed, 52 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5541b6c6d..996809686 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ - GET [Layers](doc/rest.md#get-layers)/[Maps](doc/rest.md#get-maps)/[Publications](doc/rest.md#get-publications) - DELETE Workspace [Layers](doc/rest.md#delete-workspace-layers)/[Maps](doc/rest.md#delete-workspace-maps) - POST Workspace [Layers](doc/rest.md#post-workspace-layers)/[Maps](doc/rest.md#post-workspace-maps) respects roles in [GRANT_CREATE_PUBLIC_WORKSPACE](doc/env-settings.md#grant_create_public_workspace) and [GRANT_PUBLISH_IN_PUBLIC_WORKSPACE](doc/env-settings.md#grant_publish_in_public_workspace) + - requests to [WMS](doc/endpoints.md#web-map-service) and [WFS](doc/endpoints.md#web-feature-service) endpoints - [#165](https://github.com/LayerManager/layman/issues/165) Many endpoints return previously associated [role names](doc/models.md#role) in `access_rights.read` and `access_rights.write` keys: - [GET](doc/rest.md#get-workspace-layer)/[PATCH](doc/rest.md#patch-workspace-layer) Workspace Layer - [GET](doc/rest.md#get-workspace-map)/[PATCH](doc/rest.md#patch-workspace-map) Workspace Map diff --git a/tests/asserts/final/publication/geoserver_proxy.py b/tests/asserts/final/publication/geoserver_proxy.py index d02244922..183f40634 100644 --- a/tests/asserts/final/publication/geoserver_proxy.py +++ b/tests/asserts/final/publication/geoserver_proxy.py @@ -15,12 +15,24 @@ def is_complete_in_workspace_wms(workspace, publ_type, name, *, version, headers geoserver_util.is_complete_in_workspace_wms_instance(wms_inst, name, validate_metadata_url=validate_metadata_url) -def is_complete_in_workspace_wms_1_3_0(workspace, publ_type, name, headers): +def is_complete_in_workspace_wms_1_3_0(workspace, publ_type, name, headers=None, *, actor_name=None): + headers = headers or {} + assert headers is not None or actor_name is not None + if actor_name: + assert process_client.TOKEN_HEADER not in headers + if actor_name and actor_name != settings.ANONYM_USER: + headers.update(process_client.get_authz_headers(actor_name)) assert publ_type == process_client.LAYER_TYPE is_complete_in_workspace_wms(workspace, publ_type, name, version='1.3.0', headers=headers) -def workspace_wfs_2_0_0_capabilities_available_if_vector(workspace, publ_type, name, headers): +def workspace_wfs_2_0_0_capabilities_available_if_vector(workspace, publ_type, name, headers=None, *, actor_name=None): + headers = headers or {} + assert headers is not None or actor_name is not None + if actor_name: + assert process_client.TOKEN_HEADER not in headers + if actor_name and actor_name != settings.ANONYM_USER: + headers.update(process_client.get_authz_headers(actor_name)) with app.app_context(): internal_wfs_url = test_util.url_for('geoserver_proxy_bp.proxy', subpath=workspace + '/wfs') @@ -32,7 +44,7 @@ def workspace_wfs_2_0_0_capabilities_available_if_vector(workspace, publ_type, n assert wfs_inst.contents wfs_name = f'{workspace}:{name}' - assert wfs_name in wfs_inst.contents + assert wfs_name in wfs_inst.contents, "Layer not found in Capabilities." wfs_layer = wfs_inst.contents[wfs_name] assert len(wfs_layer.metadataUrls) == 1 assert wfs_layer.metadataUrls[0]['url'].startswith('http://localhost:3080/record/xml/m-') diff --git a/tests/asserts/final/publication/geoserver_util.py b/tests/asserts/final/publication/geoserver_util.py index 82230362b..edab36ebd 100644 --- a/tests/asserts/final/publication/geoserver_util.py +++ b/tests/asserts/final/publication/geoserver_util.py @@ -1,6 +1,6 @@ def is_complete_in_workspace_wms_instance(wms_instance, name, *, validate_metadata_url): assert wms_instance.contents - assert name in wms_instance.contents + assert name in wms_instance.contents, "Layer not found in Capabilities." wms_layer = wms_instance.contents[name] for style_name, style_values in wms_layer.styles.items(): assert 'legend' in style_values, f'style_name={style_name}, style_values={style_values}' diff --git a/tests/dynamic_data/publications/access_rights/test_access_rights_application.py b/tests/dynamic_data/publications/access_rights/test_access_rights_application.py index e46e97279..5a282a1b7 100644 --- a/tests/dynamic_data/publications/access_rights/test_access_rights_application.py +++ b/tests/dynamic_data/publications/access_rights/test_access_rights_application.py @@ -5,6 +5,7 @@ from layman import settings, LaymanError from test_tools import process_client, role_service as role_service_util from tests import Publication, EnumTestTypes +from tests.asserts.final.publication import geoserver_proxy from tests.dynamic_data import base_test ENDPOINTS_TO_TEST = { @@ -31,6 +32,11 @@ (process_client.delete_workspace_publication, {}), ] +GEOSERVER_METHODS_TO_TEST = [ + (geoserver_proxy.is_complete_in_workspace_wms_1_3_0, {}), + (geoserver_proxy.workspace_wfs_2_0_0_capabilities_available_if_vector, {}), +] + def pytest_generate_tests(metafunc): # https://docs.pytest.org/en/6.2.x/parametrize.html#pytest-generate-tests @@ -73,8 +79,9 @@ def add_publication_test_cases_to_list(tc_list, publication, user, endpoints_to_ 'layer': publication.name, 'actor_name': user, 'publication_type': publication.type, + 'publ_type': publication.type, } - for method, args in endpoints_to_test[publication.type]: + for method, args in endpoints_to_test: pytest_id = f'{method.__name__}__{user.split("_")[-1]}__{publication.name[5:]}{("__" + next(iter(args.keys()))) if args else ""}' method_args = inspect.getfullargspec(method).args + inspect.getfullargspec(method).kwonlyargs @@ -93,7 +100,7 @@ def generate_positive_test_cases(publications_user_can_read): tc_list = [] for user, publications in publications_user_can_read.items(): for publication in publications: - add_publication_test_cases_to_list(tc_list, publication, user, ENDPOINTS_TO_TEST) + add_publication_test_cases_to_list(tc_list, publication, user, ENDPOINTS_TO_TEST[publication.type]) return tc_list @@ -104,7 +111,7 @@ def generate_negative_test_cases(publications_user_can_read, publication_all): if publication in available_publications: continue endpoints_to_test = {publ_type: endpoints + ENDPOINTS_TO_TEST_NEGATIVE_ONLY for publ_type, endpoints in ENDPOINTS_TO_TEST.items()} - add_publication_test_cases_to_list(tc_list, publication, user, endpoints_to_test) + add_publication_test_cases_to_list(tc_list, publication, user, endpoints_to_test[publication.type]) return tc_list @@ -139,6 +146,24 @@ def generate_multiendpoint_test_cases(publications_user_can_read, workspace, ): return tc_list +def generate_positive_geoserver_test_cases(publications_user_can_read): + tc_list = [] + for user, publications in publications_user_can_read.items(): + for publication in publications: + if publication.type == process_client.LAYER_TYPE: + add_publication_test_cases_to_list(tc_list, publication, user, GEOSERVER_METHODS_TO_TEST) + return tc_list + + +def generate_geoserver_negative_test_cases(publications_user_can_read, publication_all): + tc_list = [] + for user, available_publications in publications_user_can_read.items(): + for publication in publication_all: + if publication not in available_publications and publication.type == process_client.LAYER_TYPE: + add_publication_test_cases_to_list(tc_list, publication, user, GEOSERVER_METHODS_TO_TEST) + return tc_list + + @pytest.mark.timeout(60) @pytest.mark.usefixtures('ensure_layman_module', 'oauth2_provider_mock') class TestAccessRights: @@ -197,9 +222,10 @@ class TestAccessRights: } test_cases = { - 'test_single_positive': generate_positive_test_cases(PUBLICATIONS_BY_USER), + 'test_single_positive': generate_positive_test_cases(PUBLICATIONS_BY_USER) + generate_positive_geoserver_test_cases(PUBLICATIONS_BY_USER), 'test_single_negative': generate_negative_test_cases(PUBLICATIONS_BY_USER, PUBLICATIONS), 'test_multiendpoint': generate_multiendpoint_test_cases(PUBLICATIONS_BY_USER, OWNER), + 'test_geoserver_negative': generate_geoserver_negative_test_cases(PUBLICATIONS_BY_USER, PUBLICATIONS), } @pytest.fixture(scope='class', autouse=True) @@ -238,3 +264,8 @@ def test_multiendpoint(self, rest_method, rest_args, params): result = rest_method(**rest_args) result_publications = [(publ['workspace'], f"layman.{publ['publication_type']}", publ['name']) for publ in result] assert result_publications == params['exp_publications'] + + def test_geoserver_negative(self, rest_method, rest_args, ): + with pytest.raises(AssertionError) as exc_info: + rest_method(**rest_args) + assert exc_info.value.args[0].startswith('Layer not found in Capabilities.') From 714f97cf7a9efbf0e888c59e57de3ad121b11e52 Mon Sep 17 00:00:00 2001 From: index-git Date: Tue, 19 Dec 2023 14:08:35 +0100 Subject: [PATCH 4/4] Remove replaced tests --- .../single_publication/publications_test.py | 43 +------------------ 1 file changed, 1 insertion(+), 42 deletions(-) diff --git a/tests/static_data/single_publication/publications_test.py b/tests/static_data/single_publication/publications_test.py index 8bfb861b9..1c8a174eb 100644 --- a/tests/static_data/single_publication/publications_test.py +++ b/tests/static_data/single_publication/publications_test.py @@ -1,7 +1,7 @@ import pytest import crs as crs_def -from layman import app, util as layman_util, settings +from layman import app, settings from layman.common import bbox as bbox_util from layman.layer.filesystem import gdal from test_tools import assert_util, process_client @@ -44,47 +44,6 @@ def test_infos(workspace, publ_type, publication): assert rest_detail['geodata_type'] == exp_geodata_type -@pytest.mark.parametrize('workspace, publ_type, publication', data.LIST_ALL_PUBLICATIONS) -@pytest.mark.usefixtures('oauth2_provider_mock', 'ensure_layman') -def test_auth_get_publications(workspace, publ_type, publication): - ensure_publication(workspace, publ_type, publication) - - all_auth_info = util.get_users_and_headers_for_publication(workspace, publ_type, publication) - headers_list_in = all_auth_info['read'][util.KEY_AUTH][util.KEY_HEADERS] - headers_list_out = all_auth_info['read'][util.KEY_NOT_AUTH][util.KEY_HEADERS] - - for in_headers in headers_list_in: - infos = process_client.get_publications(publ_type, workspace=workspace, headers=in_headers) - publication_names = [li['name'] for li in infos] - assert publication in publication_names, in_headers - - for out_headers in headers_list_out: - infos = process_client.get_publications(publ_type, workspace=workspace, headers=out_headers) - publication_names = [li['name'] for li in infos] - assert publication not in publication_names, out_headers - - -@pytest.mark.parametrize('workspace, publ_type, publication', data.LIST_ALL_PUBLICATIONS) -@pytest.mark.usefixtures('oauth2_provider_mock', 'ensure_layman') -def test_auth_get_publication(workspace, publ_type, publication): - ensure_publication(workspace, publ_type, publication) - - all_auth_info = util.get_users_and_headers_for_publication(workspace, publ_type, publication) - readers = all_auth_info['read'][util.KEY_AUTH][util.KEY_USERS] - non_readers = all_auth_info['read'][util.KEY_NOT_AUTH][util.KEY_USERS] - - for user in readers: - with app.app_context(): - pub_info = layman_util.get_publication_info(workspace, publ_type, publication, {'actor_name': user}) - assert pub_info['name'] == publication, f'pub_info={pub_info}' - assert pub_info['type'] == publ_type, f'pub_info={pub_info}' - - for user in non_readers: - with app.app_context(): - pub_info = layman_util.get_publication_info(workspace, publ_type, publication, {'actor_name': user}) - assert not pub_info, pub_info - - @pytest.mark.parametrize('workspace, publ_type, publication', data.LIST_ALL_PUBLICATIONS) @pytest.mark.usefixtures('oauth2_provider_mock', 'ensure_layman') def test_internal_info(workspace, publ_type, publication):