Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_entity_proxy_equals(service):
"""Two entity proxies are equal if they hold the same data"""
properties = {'Key': 'a', 'DataType': 'b', 'Data': 'c', 'DataName': 'd'}
fst_entity = EntityProxy(service, service.entity_sets.MasterEntities,
service.schema.entity_type('MasterEntity'), properties)
scn_entity = EntityProxy(service, service.entity_sets.MasterEntities,
service.schema.entity_type('MasterEntity'), properties)
properties['DataType'] = 'g'
thr_entity = EntityProxy(service, service.entity_sets.MasterEntities,
service.schema.entity_type('MasterEntity'), properties)
assert fst_entity.equals(fst_entity)
assert fst_entity.equals(scn_entity)
assert scn_entity.equals(fst_entity)
assert not fst_entity.equals(thr_entity)
assert not scn_entity.equals(thr_entity)
def test_entity_proxy_equals(service):
"""Two entity proxies are equal if they hold the same data"""
properties = {'Key': 'a', 'DataType': 'b', 'Data': 'c', 'DataName': 'd'}
fst_entity = EntityProxy(service, service.entity_sets.MasterEntities,
service.schema.entity_type('MasterEntity'), properties)
scn_entity = EntityProxy(service, service.entity_sets.MasterEntities,
service.schema.entity_type('MasterEntity'), properties)
properties['DataType'] = 'g'
thr_entity = EntityProxy(service, service.entity_sets.MasterEntities,
service.schema.entity_type('MasterEntity'), properties)
assert fst_entity.equals(fst_entity)
assert fst_entity.equals(scn_entity)
assert scn_entity.equals(fst_entity)
assert not fst_entity.equals(thr_entity)
assert not scn_entity.equals(thr_entity)
def get_entity_handler(response):
"""Gets entity from HTTP response"""
if response.status_code != requests.codes.ok:
raise HttpError('HTTP GET for Entity {0} failed with status code {1}'
.format(self._name, response.status_code), response)
entity = response.json()['d']
return EntityProxy(self._service, self._entity_set, self._entity_set.entity_type, entity)
if response.status_code != requests.codes.ok:
raise HttpError('HTTP GET for Entity {0} failed with status code {1}'
.format(self._name, response.status_code), response)
entity = response.json()['d']
return NavEntityProxy(parent, nav_property, navigation_entity_set.entity_type, entity)
self._logger.info(
'Getting the nav property %s of the entity %s for the key %s',
nav_property,
self._entity_set.entity_type.name,
master_key)
parent = EntityProxy(self._service, self, self._entity_set.entity_type, entity_key=master_key)
return NavEntityGetRequest(
partial(get_entity_handler, parent, nav_property, navigation_entity_set),
master_key,
self,
nav_property)
def _get_body(self):
"""Recursively builds a dictionary of values where some of the values
might be another entities.
"""
body = {}
for key, val in self._values.items():
# The value is either an entity or a scalar
if isinstance(val, EntityProxy):
body[key] = val._get_body() # pylint: disable=protected-access
else:
body[key] = val
return body
@property
def url(self):
"""URL of the real entity"""
service_url = self._service.url.rstrip('/')
entity_path = self.get_path()
return urljoin(service_url, entity_path)
def equals(self, other):
"""Returns true if the self and the other contains the same data"""
# pylint: disable=W0212
return self._cache == other._cache
class NavEntityProxy(EntityProxy):
"""Special case of an Entity access via 1 to 1 Navigation property"""
def __init__(self, parent_entity, prop_name, entity_type, entity):
# pylint: disable=protected-access
super(NavEntityProxy, self).__init__(parent_entity._service, parent_entity._entity_set, entity_type, entity)
self._parent_entity = parent_entity
self._prop_name = prop_name
def get_path(self):
"""Returns URL of the entity"""
return urljoin(self._parent_entity.get_path(), self._prop_name)
class GetEntitySetFilter:
"""Gets entity set from HTTP Response"""
if response.status_code != requests.codes.ok:
raise HttpError('HTTP GET for Entity Set {0} failed with status code {1}'
.format(self._name, response.status_code), response)
content = response.json()
if isinstance(content, int):
return content
entities = content['d']['results']
result = []
for props in entities:
entity = EntityProxy(self._service, self._entity_set, self._entity_set.entity_type, props)
result.append(entity)
return result
def create_entity_handler(response):
"""Gets newly created entity encoded in HTTP Response"""
if response.status_code != return_code:
raise HttpError('HTTP POST for Entity Set {0} failed with status code {1}'
.format(self._name, response.status_code), response)
entity_props = response.json()['d']
return EntityProxy(self._service, self._entity_set, self._entity_set.entity_type, entity_props)
if prop.name in proprties:
# entity type of navigation property
prop_etype = prop.to_role.entity_type
# cache value according to multiplicity
if prop.to_role.multiplicity in \
[model.EndRole.MULTIPLICITY_ONE,
model.EndRole.MULTIPLICITY_ZERO_OR_ONE]:
# cache None in case we receive nothing (null) instead of entity data
if proprties[prop.name] is None:
self._cache[prop.name] = None
else:
self._cache[prop.name] = EntityProxy(service, None, prop_etype, proprties[prop.name])
elif prop.to_role.multiplicity == model.EndRole.MULTIPLICITY_ZERO_OR_MORE:
# default value is empty array
self._cache[prop.name] = []
# if there are no entities available, received data consists of
# metadata properties only.
if 'results' in proprties[prop.name]:
# available entities are serialized in results array
for entity in proprties[prop.name]['results']:
self._cache[prop.name].append(EntityProxy(service, None, prop_etype, entity))
else:
raise PyODataException('Unknown multiplicity {0} of association role {1}'
.format(prop.to_role.multiplicity, prop.to_role.name))