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_organization_schema_not_found(self):
lc = LocalCKAN('visitor')
assert_raises(NotFound,
lc.action.scheming_organization_schema_show,
type='elmo')
def test_group_schema_not_found(self):
lc = LocalCKAN('visitor')
assert_raises(NotFound,
lc.action.scheming_group_schema_show,
type='bert')
def test_no_choices_on_not_found(self, LocalCKAN):
lc = Mock()
lc.action.datastore_search.side_effect = NotFound()
LocalCKAN.return_value = lc
assert_equals(scheming_datastore_choices(
{'datastore_choices_resource': 'not-found'}), [])
lc.action.datastore_search.assert_called_once()
def test_no_choices_on_not_authorized(self, LocalCKAN):
lc = Mock()
lc.action.datastore_search.side_effect = NotFound()
LocalCKAN.return_value = lc
assert_equals(scheming_datastore_choices(
{'datastore_choices_resource': 'not-allowed'}), [])
lc.action.datastore_search.assert_called_once()
err("Failed to open metatab '{}': {}".format(m.mt_file, e))
c = RemoteCKAN(m.ckan_url, apikey=m.api_key)
ckanid = doc.find_first_value('Root.Ckanid')
identifier = doc.find_first_value('Root.Identitfier')
name = doc.as_version(None).find_first('Root.Name')
ckan_name = name.value.replace('.','-')
id_name = ckanid or ckan_name
try:
pkg = c.action.package_show(name_or_id=id_name)
prt("Updating CKAN dataset for '{}'".format(id_name))
except NotFound as e:
e.__traceback__ = None
traceback.clear_frames(e.__traceback__)
try:
pkg = c.action.package_create(name=ckan_name)
except Exception as e:
err("Failed to create package for name '{}': {} ".format(ckan_name, e))
prt("Adding CKAN dataset for '{}'".format(ckan_name))
pkg['title'] = doc.find_first_value('Root.Title')
if not pkg['title']:
pkg['title'] = doc.find_first_value('Root.Description')
try:
pkg['notes'] = doc.markdown #doc.find_first_value('Root.Description')
resource_id = field['datastore_choices_resource']
limit = field.get('datastore_choices_limit', 1000)
columns = field.get('datastore_choices_columns')
fields = None
if columns:
fields = [columns['value'], columns['label']]
# anon user must be able to read choices or this helper
# could be used to leak data from private datastore tables
lc = LocalCKAN(username='')
try:
result = lc.action.datastore_search(
resource_id=resource_id,
limit=limit,
fields=fields)
except (NotFound, NotAuthorized):
return []
if not fields:
fields = [f['id'] for f in result['fields'] if f['id'] != '_id']
return [{'value': r[fields[0]], 'label': r[fields[1]]}
for r in result['records']]