Add support for whitelisted and custom XML namespaces

Recognizing that many xml namespaces ae same despite different hosting url
we added whitelisted urls for which we granite support. Also we added
option to pass custom dict with namespaces in case non of whitelisted
is used by your data source.

--- Jakub Filak <[email protected]> ----

This commit adds support for XML Namespaces used in #27

Author: mamiksik

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+# 1.2.3
+  * dca01a28 - model:  add support for whitelisted and custom namespaces - Martin Miksik
+
 # 1.2.2
   * ed1ae8a - model:  fix namespace parsing - Jakub Filak
 

docs/usage/initialization.rst

@@ -22,6 +22,27 @@ Basic initialization which is going to work for everybody:
     northwind = pyodata.Client(SERVICE_URL, requests.Session())
 
 
+Set custom namespaces
+---------------------
+
+Let's assume you need to work with a service  which uses namespaces not directly supported by this library e. g. ones
+hosted on private urls such as *customEdmxUrl.com* and *customEdmUrl.com*:
+
+.. code-block:: python
+
+    import pyodata
+    import requests
+
+    SERVICE_URL = 'http://services.odata.org/V2/Northwind/Northwind.svc/'
+
+    namespaces = {
+        'edmx': "customEdmxUrl.com"
+        'edm': 'customEdmUrl.com'
+    }
+
+    northwind = pyodata.Client(SERVICE_URL, requests.Session(), namespaces=namespaces)
+
+
 Get the service proxy client for an OData service requiring authentication
 --------------------------------------------------------------------------
 

pyodata/client.py

@@ -14,7 +14,7 @@ class Client:
 
     ODATA_VERSION_2 = 2
 
-    def __new__(cls, url, connection, odata_version=ODATA_VERSION_2):
+    def __new__(cls, url, connection, odata_version=ODATA_VERSION_2, namespaces=None):
         """Create instance of the OData Client for given URL"""
 
         logger = logging.getLogger('pyodata.client')
@@ -44,7 +44,7 @@ def __new__(cls, url, connection, odata_version=ODATA_VERSION_2):
 
             # create model instance from received metadata
             logger.info('Creating OData Schema (version: %d)', odata_version)
-            schema = pyodata.v2.model.schema_from_xml(resp.content)
+            schema = pyodata.v2.model.schema_from_xml(resp.content, namespaces=namespaces)
 
             # create service instance based on model we have
             logger.info('Creating OData Service (version: %d)', odata_version)

pyodata/exceptions.py

@@ -12,6 +12,10 @@ class PyODataModelError(PyODataException):
     """Raised when model error occurs"""
 
 
+class PyODataParserError(PyODataException):
+    """Raised when parser error occurs"""
+
+
 class ExpressionError(PyODataException):
     """Raise when runtime logical expression error occurs"""
 

pyodata/v2/model.py

@@ -16,7 +16,7 @@
 
 from lxml import etree
 
-from pyodata.exceptions import PyODataException, PyODataModelError
+from pyodata.exceptions import PyODataException, PyODataModelError, PyODataParserError
 
 LOGGER_NAME = 'pyodata.model'
 
@@ -2029,12 +2029,24 @@ def sap_attribute_get_bool(node, attr, default):
     raise TypeError('Not a bool attribute: {0} = {1}'.format(attr, value))
 
 
+EDMX_WHITELIST = [
+    'http://schemas.microsoft.com/ado/2007/06/edmx',
+    'http://docs.oasis-open.org/odata/ns/edmx',
+]
+
+
+EDM_WHITELIST = [
+    'http://schemas.microsoft.com/ado/2008/09/edm',
+    'http://docs.oasis-open.org/odata/ns/edm'
+]
+
 NAMESPACES = {
     'd': 'http://schemas.microsoft.com/ado/2007/08/dataservices',
     'm': 'http://schemas.microsoft.com/ado/2007/08/dataservices/metadata',
     'sap': 'http://www.sap.com/Protocols/SAPData',
-    'edmx': 'http://schemas.microsoft.com/ado/2007/06/edmx',
-    'edm': 'http://schemas.microsoft.com/ado/2008/09/edm'
+    'edmx': None,
+    'edm': None
+
 }
 
 
@@ -2064,21 +2076,56 @@ def __init__(self):
         super(Edmx, self).__init__()
 
     @staticmethod
-    def parse(metadata_xml):
+    def parse(metadata_xml, namespaces=None):
         """ Build model from the XML metadata"""
         if isinstance(metadata_xml, str):
             mdf = io.StringIO(metadata_xml)
         elif isinstance(metadata_xml, bytes):
             mdf = io.BytesIO(metadata_xml)
         else:
             raise TypeError('Expected bytes or str type on metadata_xml, got : {0}'.format(type(metadata_xml)))
-        # the first child element has name 'Edmx'
-        edmx = etree.parse(mdf)
+
+        NAMESPACES['edmx'] = None
+        NAMESPACES['edm'] = None
+        del NAMESPACES['edmx']
+        del NAMESPACES['edm']
+
+        if namespaces is not None:
+            NAMESPACES.update(namespaces)
+
+        xml = etree.parse(mdf)
+        edmx = xml.getroot()
+
+        try:
+            dataservices = next((child for child in edmx if etree.QName(child.tag).localname == 'DataServices'))
+        except StopIteration:
+            raise PyODataParserError('Metadata document is missing the element DataServices')
+
+        try:
+            schema = next((child for child in dataservices if etree.QName(child.tag).localname == 'Schema'))
+        except StopIteration:
+            raise PyODataParserError('Metadata document is missing the element Schema')
+
+        if 'edmx' not in NAMESPACES:
+            namespace = etree.QName(edmx.tag).namespace
+
+            if namespace not in EDMX_WHITELIST:
+                raise PyODataParserError(f'Unsupported Edmx namespace - {namespace}')
+
+            NAMESPACES['edmx'] = namespace
+
+        if 'edm' not in NAMESPACES:
+            namespace = etree.QName(schema.tag).namespace
+
+            if namespace not in EDM_WHITELIST:
+                raise PyODataParserError(f'Unsupported Schema namespace - {namespace}')
+
+            NAMESPACES['edm'] = namespace
 
         # aliases - http://docs.oasis-open.org/odata/odata/v4.0/odata-v4.0-part3-csdl.html
-        Edmx.update_global_variables_with_alias(Edmx.get_aliases(edmx))
+        Edmx.update_global_variables_with_alias(Edmx.get_aliases(xml))
 
-        edm_schemas = edmx.xpath('/edmx:Edmx/edmx:DataServices/edm:Schema', namespaces=NAMESPACES)
+        edm_schemas = xml.xpath('/edmx:Edmx/edmx:DataServices/edm:Schema', namespaces=NAMESPACES)
         schema = Schema.from_etree(edm_schemas)
         return schema
 
@@ -2114,7 +2161,7 @@ def update_global_variables_with_alias(aliases):
                 SAP_VALUE_HELPER_DIRECTIONS[alias + '.' + suffix] = SAP_VALUE_HELPER_DIRECTIONS[direction_key]
 
 
-def schema_from_xml(metadata_xml):
+def schema_from_xml(metadata_xml, namespaces=None):
     """Parses XML data and returns Schema representing OData Metadata"""
 
-    return Edmx.parse(metadata_xml)
+    return Edmx.parse(metadata_xml, namespaces=namespaces)

tests/conftest.py

@@ -313,38 +313,89 @@ def metadata():
 def metadata_builder_factory():
     """Skeleton OData metadata"""
 
-    # pylint: disable=line-too-long
-
-    class MetadaBuilder(object):
+    class MetadaBuilder:
         """Helper class for building XML metadata document"""
 
-        PROLOGUE = """
-        <edmx:Edmx xmlns:edmx="http://schemas.microsoft.com/ado/2007/06/edmx" xmlns:m="http://schemas.microsoft.com/ado/2007/08/dataservices/metadata" xmlns:sap="http://www.sap.com/Protocols/SAPData" Version="1.0">
-          <edmx:Reference xmlns:edmx="http://docs.oasis-open.org/odata/ns/edmx" Uri="https://example.sap.corp/sap/opu/odata/IWFND/CATALOGSERVICE;v=2/Vocabularies(TechnicalName='%2FIWBEP%2FVOC_COMMON',Version='0001',SAP__Origin='LOCAL')/$value">
-           <edmx:Include Namespace="com.sap.vocabularies.Common.v1" Alias="Common"/>
-          </edmx:Reference>
-         <edmx:DataServices m:DataServiceVersion="2.0">
-         """
+        # pylint: disable=too-many-instance-attributes,line-too-long
+        def __init__(self):
+            self.reference_is_enabled = True
+            self.data_services_is_enabled = True
+            self.schema_is_enabled = True
 
-        EPILOGUE = """
-         </edmx:DataServices>
-        </edmx:Edmx>
-        """
+            self.namespaces = {
+                'edmx': "http://schemas.microsoft.com/ado/2007/06/edmx",
+                'sap': 'http://www.sap.com/Protocols/SAPData',
+                'edm': 'http://schemas.microsoft.com/ado/2008/09/edm',
+                'm': 'http://schemas.microsoft.com/ado/2007/08/dataservices/metadata',
+                'd': 'http://schemas.microsoft.com/ado/2007/08/dataservices',
+            }
+
+            self.custom_edmx_prologue = None
+            self.custom_edmx_epilogue = None
+
+            self.custom_data_services_prologue = None
+            self.custom_data_services_epilogue = None
+
+            self._reference = '\n<edmx:Reference xmlns:edmx="http://docs.oasis-open.org/odata/ns/edmx" Uri="https://example.sap.corp/sap/opu/odata/IWFND/CATALOGSERVICE;v=2/Vocabularies(TechnicalName=\'%2FIWBEP%2FVOC_COMMON\',Version=\'0001\',SAP__Origin=\'LOCAL\')/$value">' + \
+                              '\n<edmx:Include Namespace="com.sap.vocabularies.Common.v1" Alias="Common"/>' + \
+                              '\n</edmx:Reference>'
 
-        def __init__(self):
             self._schemas = ''
 
         def add_schema(self, namespace, xml_definition):
             """Add schema element"""
-
-            self._schemas += '<Schema xmlns:d="http://schemas.microsoft.com/ado/2007/08/dataservices" xmlns:m="http://schemas.microsoft.com/ado/2007/08/dataservices/metadata" xmlns="http://schemas.microsoft.com/ado/2008/09/edm" Namespace="{0}" xml:lang="en" sap:schema-version="1">'.format(namespace)
-            self._schemas += xml_definition
-            self._schemas += '</Schema>'
+            self._schemas += f""""\n<Schema xmlns:d="{self.namespaces["d"]}" xmlns:m="{self.namespaces["m"]}" xmlns="{
+            self.namespaces["edm"]}" Namespace="{namespace}" xml:lang="en" sap:schema-version="1">"""
+            self._schemas += "\n" + xml_definition
+            self._schemas += '\n</Schema>'
 
         def serialize(self):
             """Returns full metadata XML document"""
-
-            return MetadaBuilder.PROLOGUE + self._schemas + MetadaBuilder.EPILOGUE
+            result = self._edmx_prologue()
+
+            if self.reference_is_enabled:
+                result += self._reference
+
+            if self.data_services_is_enabled:
+                result += self._data_services_prologue()
+
+            if self.schema_is_enabled:
+                result += self._schemas
+
+            if self.data_services_is_enabled:
+                result += self._data_services_epilogue()
+
+            result += self._edmx_epilogue()
+
+            return result
+
+        def _edmx_prologue(self):
+            if self.custom_edmx_prologue:
+                prologue = self.custom_edmx_prologue
+            else:
+                prologue = f"""<edmx:Edmx  xmlns:edmx="{self.namespaces["edmx"]}" xmlns:m="{self.namespaces["m"]}" xmlns:sap="{self.namespaces["sap"]}" Version="1.0">"""
+            return prologue
+
+        def _edmx_epilogue(self):
+            if self.custom_edmx_epilogue:
+                epilogue = self.custom_edmx_epilogue
+            else:
+                epilogue = '\n</edmx:Edmx>'
+            return epilogue
+
+        def _data_services_prologue(self):
+            if self.custom_data_services_prologue:
+                prologue = self.custom_data_services_prologue
+            else:
+                prologue = '\n<edmx:DataServices m:DataServiceVersion="2.0">'
+            return prologue
+
+        def _data_services_epilogue(self):
+            if self.custom_data_services_epilogue:
+                prologue = self.custom_data_services_epilogue
+            else:
+                prologue = '\n</edmx:DataServices>'
+            return prologue
 
     return MetadaBuilder
 

tests/test_model_v2.py

@@ -5,7 +5,7 @@
 import pytest
 from pyodata.v2.model import Edmx, Typ, StructTypeProperty, Types, EntityType, EdmStructTypeSerializer,\
     Association, AssociationSet, EndRole, AssociationSetEndRole
-from pyodata.exceptions import PyODataException, PyODataModelError
+from pyodata.exceptions import PyODataException, PyODataModelError, PyODataParserError
 
 
 def test_edmx(schema):
@@ -807,3 +807,74 @@ def test_edmx_association_set_end_by_entity_set():
 
     assert association_set.end_by_entity_set(end_from.entity_set_name) == end_from
     assert association_set.end_by_entity_set(end_to.entity_set_name) == end_to
+
+
+def test_missing_data_service(metadata_builder_factory):
+    """Test correct handling of missing DataService tag in xml"""
+
+    builder = metadata_builder_factory()
+    builder.data_services_is_enabled = False
+    xml = builder.serialize()
+
+    try:
+        Edmx.parse(xml)
+    except PyODataParserError as ex:
+        assert str(ex) == 'Metadata document is missing the element DataServices'
+
+
+def test_missing_schema(metadata_builder_factory):
+    """Test correct handling of missing Schema tag in xml"""
+
+    builder = metadata_builder_factory()
+    builder.schema_is_enabled = False
+    xml = builder.serialize()
+
+    try:
+        Edmx.parse(xml)
+    except PyODataParserError as ex:
+        assert str(ex) == 'Metadata document is missing the element Schema'
+
+
+def test_namespace_whitelist(metadata_builder_factory):
+    """Test correct handling of whitelisted namespaces"""
+
+    builder = metadata_builder_factory()
+    builder.namespaces['edmx'] = 'http://docs.oasis-open.org/odata/ns/edmx'
+    builder.namespaces['edm'] = 'http://docs.oasis-open.org/odata/ns/edm'
+    builder.add_schema('', '')
+    xml = builder.serialize()
+    Edmx.parse(xml)
+
+
+def test_unsupported_edmx_n(metadata_builder_factory):
+    """Test correct handling of non-whitelisted Edmx namespaces"""
+
+    builder = metadata_builder_factory()
+    edmx = 'wedonotsupportthisnamespace.com'
+    builder.namespaces['edmx'] = edmx
+    builder.add_schema('', '')
+    xml = builder.serialize()
+
+    Edmx.parse(xml, {'edmx': edmx})
+
+    try:
+        Edmx.parse(xml)
+    except PyODataParserError as ex:
+        assert str(ex) == f'Unsupported Edmx namespace - {edmx}'
+
+
+def test_unsupported_schema_n(metadata_builder_factory):
+    """Test correct handling of non-whitelisted Schema namespaces"""
+
+    builder = metadata_builder_factory()
+    edm = 'wedonotsupportthisnamespace.com'
+    builder.namespaces['edm'] = edm
+    builder.add_schema('', '')
+    xml = builder.serialize()
+
+    Edmx.parse(xml, {'edm': edm})
+
+    try:
+        Edmx.parse(xml)
+    except PyODataParserError as ex:
+        assert str(ex) == f'Unsupported Schema namespace - {edm}'