Handle WFS version mismatch

owslib/feature/common.py

@@ -4,6 +4,10 @@
 from urllib.parse import urlencode, parse_qsl
 
 
+class CapabilitiesError(Exception):
+    pass
+
+
 class WFSCapabilitiesReader(object):
     """Read and parse capabilities document into a lxml.etree infoset
     """

owslib/feature/wfs100.py

@@ -28,6 +28,7 @@
 from owslib.namespaces import Namespaces
 from owslib.feature.schema import get_schema
 from owslib.feature.common import (
+    CapabilitiesError,
     WFSCapabilitiesReader,
     AbstractContentMetadata,
 )
@@ -112,6 +113,12 @@ def __init__(
         self._buildMetadata(parse_remote_metadata)
 
     def _buildMetadata(self, parse_remote_metadata=False):
+        # check if expected version matches actual version in XML
+        actual_wfs_version = self._capabilities.attrib.get("version")
+        if not actual_wfs_version.startswith(self.version):
+            error_message = f"WFS version mismatch. Actual: {actual_wfs_version}, Requested: {self.version}"
+            raise CapabilitiesError(error_message)
+
         """set up capabilities metadata objects: """
 
         self.updateSequence = self._capabilities.attrib.get("updateSequence")

owslib/feature/wfs110.py

@@ -36,6 +36,7 @@
 from owslib.feature.common import (
     WFSCapabilitiesReader,
     AbstractContentMetadata,
+    CapabilitiesError,
 )
 from owslib.namespaces import Namespaces
 from owslib.util import openURL
@@ -98,6 +99,12 @@ def __init__(
         self._buildMetadata(parse_remote_metadata)
 
     def _buildMetadata(self, parse_remote_metadata=False):
+        # check if expected version matches actual version in XML
+        actual_wfs_version = self._capabilities.attrib.get("version")
+        if not actual_wfs_version.startswith(self.version):
+            error_message = f"WFS version mismatch. Actual: {actual_wfs_version}, Requested: {self.version}"
+            raise CapabilitiesError(error_message)
+
         """set up capabilities metadata objects: """
 
         self.updateSequence = self._capabilities.attrib.get("updateSequence")

owslib/feature/wfs200.py

@@ -22,7 +22,8 @@
 from owslib.feature import WebFeatureService_
 from owslib.feature.common import (
     WFSCapabilitiesReader,
-    AbstractContentMetadata
+    AbstractContentMetadata,
+    CapabilitiesError,
 )
 from owslib.namespaces import Namespaces
 
@@ -87,6 +88,12 @@ def __init__(
         self._buildMetadata(parse_remote_metadata)
 
     def _buildMetadata(self, parse_remote_metadata=False):
+        # check if expected version matches actual version in XML
+        actual_wfs_version = self._capabilities.attrib.get("version")
+        if not actual_wfs_version.startswith(self.version):
+            error_message = f"WFS version mismatch. Actual: {actual_wfs_version}, Requested: {self.version}"
+            raise CapabilitiesError(error_message)
+
         """set up capabilities metadata objects: """
 
         self.updateSequence = self._capabilities.attrib.get("updateSequence")

tests/test_wfs_generic.py

@@ -1,3 +1,4 @@
+from owslib.feature.common import CapabilitiesError
 from owslib.wfs import WebFeatureService
 from owslib.util import ServiceException
 from owslib.fes import PropertyIsLike, etree
@@ -57,6 +58,40 @@ def test_verbOptions_wfs_100():
     verbOptions = [cm.verbOptions for cm in wfs.contents.values()]
     assert len(verbOptions[0]) == 2
 
+def create_mismatch_test_params():
+    """Build combinations of mismatching WFS versions"""
+    params = []
+    version_info = {
+        '1.0.0': "wfs_dov_getcapabilities_100_verbOptions.xml",
+        '1.1.0': "wfs_HSRS_GetCapabilities_1_1_0.xml",
+        '2.0.0': "wfs_CUZK_GetCapabilities_2_0_0.xml",
+    }
+
+    for expected_version in version_info.keys():
+        # get versions other than current one
+        other_versions = list(version_info.keys() - {expected_version})
+
+        for other_version in other_versions:
+            file_name = version_info[other_version]
+
+            def make_param(version):
+                test_id = f"expects '{version}', but gets '{other_version}'"
+                return pytest.param(version, file_name, id=test_id)
+
+            # create test case for both variants of the version number (e.g. '1.0' and '1.0.0')
+            version_variants = [expected_version[0:3], expected_version]
+            params.extend([make_param(variant) for variant in version_variants])
+
+    return params
+
+
+@pytest.mark.parametrize("expected_version, xml_path_with_other_version", create_mismatch_test_params() )
+def test_raises_error_on_version_mismatch(expected_version, xml_path_with_other_version):
+    DUMMY_URL = 'http://gis.bnhelp.cz/ows/crwfs'
+    xml = open(resource_file(xml_path_with_other_version), "rb").read()
+
+    with pytest.raises(CapabilitiesError):
+        WebFeatureService(DUMMY_URL, xml=xml, version=expected_version)
 
 @pytest.mark.online
 @pytest.mark.skipif(not service_ok(SERVICE_URL),

tests/test_wfs_schema.py

@@ -159,7 +159,7 @@ def test_get_schema_100(self, mp_wfs_100):
         assert list(wfs100.contents) == ['continents', 'cities']
 
 
-    def test_get_schema(self, mp_wfs_100, mp_remote_describefeaturetype):
+    def test_get_schema(self, mp_wfs_110, mp_remote_describefeaturetype):
         """Test the get_schema method for a standard schema.
 
         Parameters