Fix slash escaping in entity keys and navigation paths

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Fixed
 
+- service: encode slashes within entity key literals while preserving navigation path separators (#282).
+
 - service: `update_entity` accepts `200 OK` as well as `204 No Content`, so updates against services such as SAP SuccessFactors no longer raise `HttpError` (#136) - Sena Köse
 
 ## [1.12.1]

pyodata/v2/service.py

@@ -26,6 +26,16 @@
 HTTP_CODE_NO_CONTENT = 204
 
 
+def _quote_path(path):
+    """Encode path literals without escaping navigation separators.
+
+    OData doubles embedded apostrophes, so splitting on apostrophes keeps
+    literal contents at odd indices even when a key contains an apostrophe.
+    """
+    return '%27'.join(quote(part, safe='' if index % 2 else '/')
+                      for index, part in enumerate(path.split("'")))
+
+
 def urljoin(*path):
     """Joins the passed string parts into a one string url"""
 
@@ -412,7 +422,7 @@ def expand(self, expand):
 
     def get_path(self):
         if self.get_encode_path():
-            return quote(self._entity_set_proxy.last_segment + self._entity_key.to_key_string())
+            return _quote_path(self._entity_set_proxy.last_segment + self._entity_key.to_key_string())
         return self._entity_set_proxy.last_segment + self._entity_key.to_key_string()
 
     def get_default_headers(self):
@@ -569,7 +579,7 @@ def __init__(self, url, connection, handler, entity_set, entity_key, encode_path
 
     def get_path(self):
         if self.get_encode_path():
-            return quote(self._entity_set.name + self._entity_key.to_key_string())
+            return _quote_path(self._entity_set.name + self._entity_key.to_key_string())
         return self._entity_set.name + self._entity_key.to_key_string()
 
     def get_encode_path(self):
@@ -613,7 +623,7 @@ def __init__(self, url, connection, handler, entity_set, entity_key, method="PAT
 
     def get_path(self):
         if self.get_encode_path():
-            return quote(self._entity_set.name + self._entity_key.to_key_string())
+            return _quote_path(self._entity_set.name + self._entity_key.to_key_string())
         return self._entity_set.name + self._entity_key.to_key_string()
 
     def get_method(self):
@@ -1349,7 +1359,7 @@ def filter(self, *args, **kwargs):
 
     def get_path(self):
         if self.get_encode_path():
-            path = quote(self._last_segment)
+            path = _quote_path(self._last_segment)
         else:
             path = self._last_segment
 

tests/test_service_v2.py

@@ -19,6 +19,69 @@
 URL_ROOT = 'http://odatapy.example.com'
 
 
+@pytest.mark.parametrize('operation', ['get_entity', 'update_entity', 'delete_entity'])
+@pytest.mark.parametrize('key', ['/BAZ/FOO', "O'/Neil", 'a%2Fb'])
+def test_entity_key_path_escaping(service, operation, key):
+    """Slashes in key literals do not become path separators."""
+    request = getattr(service.entity_sets.MasterEntities, operation)(key)
+    expected = quote("MasterEntities('%s')" % key.replace("'", "''"), safe='')
+    assert request.get_path() == expected
+
+
+@responses.activate
+@pytest.mark.parametrize('key', ['/BAZ/FOO', "O'/Neil", 'a%2Fb', "'wrapped'"])
+def test_entity_key_response_round_trip(service, key):
+    """JSON keys retain literal characters when reused in an encoded request."""
+    responses.add(responses.GET, f'{service.url}/MasterEntities',
+                  json={'d': {'results': [{'Key': key, 'Data': key}]}}, status=200)
+    entity = service.entity_sets.MasterEntities.get_entities().execute()[0]
+    assert entity.Key == key
+    assert entity.Data == key
+    request = service.entity_sets.MasterEntities.get_entity(entity.Key)
+    expected = quote("MasterEntities('%s')" % key.replace("'", "''"), safe='')
+    assert request.get_path() == expected
+    responses.add(responses.GET, f'{service.url}/{expected}',
+                  json={'d': {'Key': key, 'Data': key}}, status=200)
+    assert request.execute().Key == key
+
+
+def test_navigation_path_preserves_separator(service):
+    """Encode slashes in a parent key while retaining the navigation separator."""
+    request = service.entity_sets.Customers.get_entity("O'/Neil").nav('Orders').get_entities()
+    assert request.get_path() == quote("Customers('O''/Neil')", safe='') + '/Orders'
+
+
+def test_unencoded_entity_key_path(service):
+    """Explicitly disabling encoding retains the raw key path."""
+    request = service.entity_sets.MasterEntities.get_entity('/BAZ/FOO', encode_path=False)
+    assert request.get_path() == "MasterEntities('/BAZ/FOO')"
+
+
+def test_single_navigation_key_path_escaping(service):
+    """Single-entity navigation also preserves literal and separator slashes."""
+    request = service.entity_sets.Customers.get_entity("O'/Neil").nav('ReferredBy')
+    assert request.get_path() == quote("Customers('O''/Neil')", safe='') + '/ReferredBy'
+
+
+@pytest.mark.parametrize('encode_path', [True, False])
+def test_batch_key_path_escaping(service, encode_path):
+    """Batch serialization retains exactly one level of key encoding."""
+    request = service.entity_sets.MasterEntities.update_entity(
+        '/BAZ/FOO', encode_path=encode_path).set(Data='updated')
+    changeset = service.create_changeset('slash_changeset')
+    changeset.add_request(request)
+    batch = service.create_batch('slash_batch')
+    batch.add_request(changeset)
+    path = "MasterEntities('/BAZ/FOO')"
+    if encode_path:
+        path = quote(path, safe='')
+    expected = f'{request.get_method()} {path} HTTP/1.1'
+    body = batch.get_body()
+    assert expected in body
+    decoded = pyodata.v2.service.decode_multipart(body, batch.get_headers()['Content-Type'])
+    assert decoded[0][0][0].startswith(expected)
+
+
 @pytest.fixture
 def service(schema):
     """Service fixture"""
@@ -3204,4 +3267,4 @@ def hook(response):
 
 def test_service_without_response_hook_works(service):
     """response_hook defaults to None and does not affect normal operation"""
-    assert service.response_hook is None
+    assert service.response_hook is None