Assert external table is in supported CRS

src/db/util.py

@@ -88,14 +88,14 @@ def get_srid(crs):
     return srid
 
 
-def get_crs(srid):
+def get_crs(srid, conn_cur=None, *, use_internal_srid=True):
     crs = next((
         crs_code for crs_code, crs_item_def in crs_def.CRSDefinitions.items()
         if crs_item_def.srid == srid
-    ), None)
+    ), None) if use_internal_srid else None
     if not crs:
         sql = 'select auth_name, auth_srid from spatial_ref_sys where srid = %s;'
-        auth_name, auth_srid = run_query(sql, (srid, ))[0]
+        auth_name, auth_srid = run_query(sql, (srid, ), conn_cur=conn_cur)[0]
         if auth_name or auth_srid:
             crs = f'{auth_name}:{auth_srid}'
     return crs

src/layman/layer/db/__init__.py

@@ -544,10 +544,10 @@ def get_bbox(schema, table_name, conn_cur=None, column=settings.OGR_DEFAULT_GEOM
     return result
 
 
-def get_crs(schema, table_name, conn_cur=None, column=settings.OGR_DEFAULT_GEOMETRY_COLUMN):
+def get_crs(schema, table_name, conn_cur=None, column=settings.OGR_DEFAULT_GEOMETRY_COLUMN, *, use_internal_srid=True):
     query = 'select Find_SRID(%s, %s, %s);'
     srid = db_util.run_query(query, (schema, table_name, column), conn_cur=conn_cur)[0][0]
-    crs = db_util.get_crs(srid)
+    crs = db_util.get_crs(srid, conn_cur, use_internal_srid=use_internal_srid)
     return crs
 
 

src/layman/layer/util.py

@@ -12,7 +12,7 @@
 from layman.common import redis as redis_util, tasks as tasks_util, metadata as metadata_common
 from layman.common.util import PUBLICATION_NAME_PATTERN, PUBLICATION_MAX_LENGTH, clear_publication_info as common_clear_publication_info
 from . import get_layer_sources, LAYER_TYPE, get_layer_type_def, get_layer_info_keys
-from .db import get_all_table_column_names
+from .db import get_all_table_column_names, get_crs
 
 LAYERNAME_PATTERN = PUBLICATION_NAME_PATTERN
 LAYERNAME_MAX_LENGTH = PUBLICATION_MAX_LENGTH
@@ -374,6 +374,15 @@ def parse_and_validate_external_table_uri_str(external_table_uri_str):
             }
         })
 
+    crs = get_crs(schema, table, conn_cur=conn_cur, column=geo_column, use_internal_srid=False)
+    if crs not in settings.INPUT_SRS_LIST:
+        raise LaymanError(2, {
+            'parameter': 'db_connection',
+            'message': 'Unsupported CRS of table data.',
+            'supported_values': settings.INPUT_SRS_LIST,
+            'found': crs,
+        })
+
     # https://stackoverflow.com/a/20537829
     query = '''
 SELECT

src/layman/layer/util_test.py

@@ -422,6 +422,18 @@ def test_parse_external_table_uri_str(external_table_uri_str, exp_result):
                 },
             },
         }, id='table_without_geo_column'),
+    pytest.param(
+        'postgresql://docker:docker@postgresql:5432/external_test_db?schema=schema_name&table=table_name_32635',
+        {
+            'http_code': 400,
+            'code': 2,
+            'detail': {
+                'parameter': 'db_connection',
+                'message': 'Unsupported CRS of table data.',
+                'found': 'EPSG:32635',
+                'supported_values': settings.INPUT_SRS_LIST,
+            },
+        }, id='table_in_crs_32635'),
 ])
 def test_validate_external_table_uri_str(external_table_uri_str, exp_error):
     with pytest.raises(LaymanError) as exc_info: