Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

769 join multiendpoints tests #890

Merged
merged 17 commits into from
Jul 24, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
237 changes: 0 additions & 237 deletions src/layman/common/prime_db_schema/publications_test.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,8 @@
import uuid
import pytest

import crs as crs_def
from layman import settings, app as app, LaymanError
from layman.layer import LAYER_TYPE
from layman.map import MAP_TYPE
from test_tools import prime_db_schema_client
from . import publications, workspaces, users

DB_SCHEMA = settings.LAYMAN_PRIME_SCHEMA
Expand All @@ -21,240 +18,6 @@
}


def test_publication_basic():
def publications_by_type(prefix,
publication_type,
style_type,
):
username = prefix + '_username'
publication_name = prefix + '_pub_name'
publication_title = prefix + '_pub_ Title'
publication_title2 = prefix + '_pub_ Title2'

with app.app_context():
workspaces.ensure_workspace(username)
uuid_orig = uuid.uuid4()
uuid_str = str(uuid_orig)
db_info = {"name": publication_name,
"title": publication_title,
"publ_type_name": publication_type,
"uuid": uuid_orig,
"actor_name": username,
'geodata_type': 'vector' if publication_type == LAYER_TYPE else None,
'style_type': style_type,
"access_rights": {"read": {settings.RIGHTS_EVERYONE_ROLE, },
"write": {settings.RIGHTS_EVERYONE_ROLE, },
},
'image_mosaic': False,
'wfs_wms_status': settings.EnumWfsWmsStatus.AVAILABLE.value if publication_type == LAYER_TYPE else None,
}
publications.insert_publication(username, db_info)
pubs = publications.get_publication_infos(username, publication_type)
assert pubs[(username, publication_type, publication_name)].get('name') == publication_name
assert pubs[(username, publication_type, publication_name)].get('title') == publication_title
assert pubs[(username, publication_type, publication_name)].get('uuid') == str(uuid_str)

publ_info = pubs[(username, publication_type, publication_name)]
assert 'geodata_type' in publ_info
assert publ_info['geodata_type'] == ('vector' if publication_type == LAYER_TYPE else None)

db_info = {"name": publication_name,
"title": publication_title2,
"actor_name": username,
"publ_type_name": publication_type,
"access_rights": {"read": {settings.RIGHTS_EVERYONE_ROLE, },
"write": {settings.RIGHTS_EVERYONE_ROLE, },
},
'style_type': style_type,
}
publications.update_publication(username, db_info)
pubs = publications.get_publication_infos(username, publication_type)
assert pubs[(username, publication_type, publication_name)].get('name') == publication_name
assert pubs[(username, publication_type, publication_name)].get('title') == publication_title2
assert pubs[(username, publication_type, publication_name)].get('uuid') == uuid_str

db_info = {"name": publication_name,
"title": publication_title,
"actor_name": username,
"publ_type_name": publication_type,
"access_rights": {"read": {settings.RIGHTS_EVERYONE_ROLE, },
"write": {settings.RIGHTS_EVERYONE_ROLE, },
},
'style_type': style_type,
}
publications.update_publication(username, db_info)
pubs = publications.get_publication_infos(username, publication_type)
assert pubs[(username, publication_type, publication_name)].get('name') == publication_name
assert pubs[(username, publication_type, publication_name)].get('title') == publication_title
assert pubs[(username, publication_type, publication_name)].get('uuid') == uuid_str

publications.delete_publication(username, publication_type, publication_name)
pubs = publications.get_publication_infos(username, publication_type)
assert pubs.get((username, publication_type, publication_name)) is None

workspaces.delete_workspace(username)

publications_by_type('test_publication_basic_layer',
LAYER_TYPE,
'sld',
)
publications_by_type('test_publication_basic_map',
MAP_TYPE,
None,
)


class TestSelectPublicationsBasic:
workspace1 = 'test_select_publications_basic_workspace1'
workspace2 = 'test_select_publications_basic_workspace2'
publications = [(workspace1, LAYER_TYPE, 'test_select_publications_publication1le', {}),
(workspace1, LAYER_TYPE, 'test_select_publications_publication1le_qml', {'style_type': 'qml'}),
(workspace1, MAP_TYPE, 'test_select_publications_publication1me', {'style_type': None}),
(workspace2, LAYER_TYPE, 'test_select_publications_publication2le', {}),
]

@pytest.fixture(scope="class")
def provide_data(self):
for workspace, type, name, kwargs in self.publications:
if type == LAYER_TYPE:
kwargs = {
**kwargs,
'geodata_type': settings.GEODATA_TYPE_VECTOR,
'wfs_wms_status': settings.EnumWfsWmsStatus.AVAILABLE.value,
}
prime_db_schema_client.post_workspace_publication(publication_type=type,
workspace=workspace,
name=name,
**kwargs, )
yield
prime_db_schema_client.clear_workspaces([self.workspace1, self.workspace2])

@staticmethod
@pytest.mark.parametrize('query_params, expected_publications', [
({'workspace_name': workspace1, 'pub_type': LAYER_TYPE},
[(workspace1, LAYER_TYPE, 'test_select_publications_publication1le'),
(workspace1, LAYER_TYPE, 'test_select_publications_publication1le_qml'),
]),
({'workspace_name': workspace1, 'pub_type': MAP_TYPE}, [(workspace1, MAP_TYPE, 'test_select_publications_publication1me'), ]),
({'workspace_name': workspace1, 'style_type': 'qml'},
[(workspace1, LAYER_TYPE, 'test_select_publications_publication1le_qml'), ]),
({'workspace_name': workspace1, 'style_type': 'sld'},
[(workspace1, LAYER_TYPE, 'test_select_publications_publication1le'), ]),
({'workspace_name': workspace1}, [(workspace1, LAYER_TYPE, 'test_select_publications_publication1le'),
(workspace1, LAYER_TYPE, 'test_select_publications_publication1le_qml'),
(workspace1, MAP_TYPE, 'test_select_publications_publication1me'),
]),
({}, [(workspace1, LAYER_TYPE, 'test_select_publications_publication1le'),
(workspace1, LAYER_TYPE, 'test_select_publications_publication1le_qml'),
(workspace1, MAP_TYPE, 'test_select_publications_publication1me'),
(workspace2, LAYER_TYPE, 'test_select_publications_publication2le'),
]),
])
@pytest.mark.usefixtures('ensure_layman', 'provide_data')
def test_get_publications(query_params, expected_publications):
with app.app_context():
infos = publications.get_publication_infos(**query_params)
info_publications = list(infos.keys())
assert expected_publications == info_publications


class TestWorldBboxFilter:
workspace = 'test_world_bbox_filter_workspace'
layer_prefix = 'test_world_bbox_filter_layer'

@pytest.fixture(scope="class")
def provide_data(self):
for crs, values in crs_def.CRSDefinitions.items():
layer = self.layer_prefix + '_' + crs.split(':')[1]
prime_db_schema_client.post_workspace_publication(LAYER_TYPE, self.workspace, layer,
geodata_type=settings.GEODATA_TYPE_VECTOR,
wfs_wms_status=settings.EnumWfsWmsStatus.AVAILABLE.value,
)
bbox = values.max_bbox or values.default_bbox
with app.app_context():
publications.set_bbox(self.workspace, LAYER_TYPE, layer, bbox, crs)
yield
prime_db_schema_client.clear_workspace(self.workspace)

@staticmethod
@pytest.mark.parametrize('crs', crs_def.CRSDefinitions.keys())
@pytest.mark.usefixtures('provide_data')
def test_world_bbox_filter(crs):
with app.app_context():
publications.get_publication_infos_with_metainfo(bbox_filter=(-100, -100, 100, 100),
bbox_filter_crs=crs)

@staticmethod
@pytest.mark.parametrize('crs', crs_def.CRSDefinitions.keys())
@pytest.mark.usefixtures('provide_data')
def test_world_bbox_ordering(crs):
with app.app_context():
publications.get_publication_infos_with_metainfo(ordering_bbox=(-100, -100, 100, 100),
ordering_bbox_crs=crs,
order_by_list=['bbox', ])


class TestExtremeCoordinatesFilter:
# pylint: disable=too-few-public-methods

workspace = 'test_extreme_coordinates_filter'
name_prefix = 'test_extreme_coordinates_filter_publication'
publ_type = LAYER_TYPE

@pytest.mark.parametrize('layer_suffix, x_coord_idx, y_coord_idx', [
('min_corner', 0, 1,),
('max_corner', 2, 3,),
])
@pytest.mark.parametrize('crs, crs_values', crs_def.CRSDefinitions.items())
def test_default_bbox_corner_filter(self, crs, crs_values, layer_suffix, x_coord_idx, y_coord_idx):
name = self.name_prefix + '_' + crs.split(':')[1] + '_' + layer_suffix
prime_db_schema_client.post_workspace_publication(self.publ_type, self.workspace, name,
geodata_type=settings.GEODATA_TYPE_VECTOR,
wfs_wms_status=settings.EnumWfsWmsStatus.AVAILABLE.value,
)
default_bbox = crs_values.default_bbox
point_bbox = (
default_bbox[x_coord_idx],
default_bbox[y_coord_idx],
default_bbox[x_coord_idx],
default_bbox[y_coord_idx]
)
with app.app_context():
publications.set_bbox(self.workspace, LAYER_TYPE, name, point_bbox, crs)

publication_infos = publications.get_publication_infos(workspace_name=self.workspace,
pub_type=self.publ_type,
)
info = publication_infos[(self.workspace, self.publ_type, name)]
native_bbox = info['native_bounding_box']
native_crs = info['native_crs']

bbox_3857 = info['bounding_box']
crs_3857 = crs_def.EPSG_3857

assert native_bbox == list(point_bbox)
assert native_crs == crs

with app.app_context():
publication_infos = publications.get_publication_infos_with_metainfo(workspace_name=self.workspace,
pub_type=self.publ_type,
bbox_filter=tuple(native_bbox),
bbox_filter_crs=native_crs,
)
assert (self.workspace, self.publ_type, name) in publication_infos['items']

with app.app_context():
publication_infos = publications.get_publication_infos_with_metainfo(workspace_name=self.workspace,
pub_type=self.publ_type,
bbox_filter=tuple(bbox_3857),
bbox_filter_crs=crs_3857,
)
assert (self.workspace, self.publ_type, name) in publication_infos['items']

with app.app_context():
publications.delete_publication(self.workspace, self.publ_type, name,)


def test_only_valid_names():
workspace_name = 'test_only_valid_names_workspace'
username = 'test_only_valid_names_user'
Expand Down
Loading