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_activity_creation_from_dict():
"""test that activities are created correctly from a dictionary"""
d = {'name': 'Project Fuzz',
'description': 'hipster beard dataset',
'used': [{'reference': {'targetId': 'syn12345', 'versionNumber': 42}, 'wasExecuted': True}]}
a = Activity(data=d)
assert_equals(a['name'], 'Project Fuzz')
assert_equals(a['description'], 'hipster beard dataset')
usedEntities = a['used']
assert_equals(len(usedEntities), 1)
u = usedEntities[0]
assert_true(u['wasExecuted'])
assert_equals(u['reference']['targetId'], 'syn12345')
assert_equals(u['reference']['versionNumber'], 42)
def test_private_getStringList():
act = Activity()
url_string = \
'https://github.com/Sage-Bionetworks/ampAdScripts/blob/master/Broad-Rush/migrateROSMAPGenotypesFeb2015.R'
act.used([{'wasExecuted': True,
'concreteType': 'org.sagebionetworks.repo.model.provenance.UsedURL',
'url': url_string}
])
assert_equals([url_string], act._getStringList())
:returns: An updated :py:class:`synapseclient.activity.Activity` object
"""
# Assert that the entity was generated by a given Activity.
if 'id' in activity:
# We're updating provenance
uri = '/activity/%s' % activity['id']
activity = Activity(data=self.restPUT(uri, json.dumps(activity)))
else:
activity = self.restPOST('/activity', body=json.dumps(activity))
# assert that an entity is generated by an activity
#TODO IS THIS REALLY CORRECT? WHAT HAPPENS IF WE WANT TO REUSE AN ACTIVITY?
uri = '/entity/%s/generatedBy?generatedBy=%s' % (id_of(entity), activity['id'])
activity = Activity(data=self.restPUT(uri))
return activity
:param version: The version of the Entity to retrieve.
Gets the most recent version if omitted
:returns: An Activity object or
raises exception if no provenance record exists
"""
# Get versionNumber from Entity
if version is None and 'versionNumber' in entity:
version = entity['versionNumber']
if version:
uri = '/entity/%s/version/%d/generatedBy' % (id_of(entity), version)
else:
uri = '/entity/%s/generatedBy' % id_of(entity)
return Activity(data=self.restGET(uri))
def updateActivity(self, activity):
"""
Modifies an existing Activity.
:returns: An updated Activity object
"""
uri = '/activity/%s' % activity['id']
return Activity(data=self.restPUT(uri, json.dumps(activity)))
def setProvenance(self, entity, activity):
"""
Stores a record of the code and data used to derive a Synapse entity.
:param entity: An Entity or Synapse ID to modify
:param activity: a :py:class:`synapseclient.activity.Activity`
:returns: An updated :py:class:`synapseclient.activity.Activity` object
"""
# Assert that the entity was generated by a given Activity.
if 'id' in activity:
# We're updating provenance
uri = '/activity/%s' % activity['id']
activity = Activity(data=self.restPUT(uri, json.dumps(activity)))
else:
activity = self.restPOST('/activity', body=json.dumps(activity))
# assert that an entity is generated by an activity
#TODO IS THIS REALLY CORRECT? WHAT HAPPENS IF WE WANT TO REUSE AN ACTIVITY?
uri = '/entity/%s/generatedBy?generatedBy=%s' % (id_of(entity), activity['id'])
activity = Activity(data=self.restPUT(uri))
return activity
annotations['etag'] = properties['etag']
annotations = self.setAnnotations(properties, annotations)
properties['etag'] = annotations['etag']
# If the parameters 'used' or 'executed' are given, create an Activity object
activity = kwargs.get('activity', None)
used = kwargs.get('used', None)
executed = kwargs.get('executed', None)
if used or executed:
if activity is not None:
## TODO: move this argument check closer to the front of the method
raise SynapseProvenanceError('Provenance can be specified as an Activity object or as used/executed item(s), but not both.')
activityName = kwargs.get('activityName', None)
activityDescription = kwargs.get('activityDescription', None)
activity = Activity(name=activityName, description=activityDescription, used=used, executed=executed)
# If we have an Activity, set it as the Entity's provenance record
if activity:
activity = self.setProvenance(properties, activity)
# 'etag' has changed, so get the new Entity
properties = self._getEntity(properties)
# Return the updated Entity object
return Entity.create(properties, annotations, local_state)