Test geometry types in external DB

test_tools/external_db.py

@@ -1,3 +1,6 @@
+import os
+import subprocess
+from urllib import parse
 import pytest
 from db import util as db_util
 from layman import settings, app
@@ -6,6 +9,11 @@
 URI = f'''postgresql://{settings.LAYMAN_PG_USER}:{settings.LAYMAN_PG_PASSWORD}@{settings.LAYMAN_PG_HOST}:{settings.LAYMAN_PG_PORT}/{EXTERNAL_DB_NAME}'''
 
 
+def uri_to_ogr2ogr(uri_str):
+    uri = parse.urlparse(uri_str)
+    return f"PG:host={uri.hostname} port={uri.port} dbname={uri.path[1:]} user={uri.username} password={uri.password}"
+
+
 @pytest.fixture(scope="session")
 def ensure_db():
     statement = f"""CREATE DATABASE {EXTERNAL_DB_NAME} TEMPLATE {settings.LAYMAN_PG_TEMPLATE_DBNAME}"""
@@ -23,6 +31,27 @@ def ensure_table(schema, name, geo_column):
     db_util.run_statement(statement, conn_cur=conn_cur)
 
 
+def import_table(input_file_path, *, table=None, schema='public'):
+    table = table or os.path.splitext(os.path.basename(input_file_path))[0]
+    target_db = uri_to_ogr2ogr(URI)
+
+    bash_args = [
+        'ogr2ogr',
+        '-nln', table,
+        # '-nlt', 'GEOMETRY',
+        '-lco', f'SCHEMA={schema}',
+        '-f', 'PostgreSQL',
+        target_db,
+        input_file_path,
+    ]
+
+    process = subprocess.Popen(bash_args, stdout=subprocess.PIPE,
+                               stderr=subprocess.STDOUT)
+    stdout, stderr = process.communicate()
+    return_code = process.poll()
+    assert return_code == 0 and not stdout and not stderr
+
+
 def drop_table(schema, name):
     statement = f'''drop table if exists {schema}.{name}'''
     conn_cur = db_util.create_connection_cursor(URI)

tests/dynamic_data/publications/layer_external_db/geometry_types_test.py

@@ -0,0 +1,71 @@
+import os
+import pytest
+
+from db import util as db_util
+from test_tools import process_client, external_db
+from tests import EnumTestTypes, Publication
+from tests.dynamic_data import base_test
+
+DIRECTORY = os.path.dirname(os.path.abspath(__file__))
+
+pytest_generate_tests = base_test.pytest_generate_tests
+
+TEST_CASES = {
+    'all': {
+        'exp_geometry_type': 'GEOMETRY',
+    },
+    'geometrycollection': {
+        'exp_geometry_type': 'GEOMETRYCOLLECTION',
+    },
+    'linestring': {
+        'exp_geometry_type': 'LINESTRING',
+    },
+    'multilinestring': {
+        'exp_geometry_type': 'MULTILINESTRING',
+    },
+    'multipoint': {
+        'exp_geometry_type': 'MULTIPOINT',
+    },
+    'multipolygon': {
+        'exp_geometry_type': 'MULTIPOLYGON',
+    },
+    'point': {
+        'exp_geometry_type': 'POINT',
+    },
+    'polygon': {
+        'exp_geometry_type': 'POLYGON',
+    },
+}
+
+
+@pytest.mark.usefixtures('ensure_external_db')
+class TestLayer(base_test.TestSingleRestPublication):
+
+    workspace = 'dynamic_test_workspace_layer_external_db_geometry_type'
+
+    publication_type = process_client.LAYER_TYPE
+
+    rest_parametrization = []
+
+    test_cases = [base_test.TestCaseType(key=key,
+                                         type=EnumTestTypes.MANDATORY,
+                                         rest_args={
+                                             'db_connection': f"{external_db.URI}?table=public.{key}&geo_column=wkb_geometry",
+                                         },
+                                         params=value,
+                                         ) for key, value in TEST_CASES.items()]
+
+    @staticmethod
+    def test_style_xml(layer: Publication, key, rest_method, rest_args, params):
+        """Parametrized using pytest_generate_tests"""
+        file_path = f"sample/data/geometry-types/{key}.geojson"
+
+        external_db.import_table(file_path, table=key)
+        conn_cur = db_util.create_connection_cursor(external_db.URI)
+        query = f'''select type from geometry_columns where f_table_schema = %s and f_table_name = %s and f_geometry_column = %s'''
+        result = db_util.run_query(query, ('public', key, 'wkb_geometry'), conn_cur=conn_cur)
+        assert result[0][0] == params['exp_geometry_type']
+
+        rest_method(layer, args=rest_args)
+
+        # external_db.drop_table('public', key)