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_get_materializeviews_tasks(self):
with open(GET_XML_MATERIALIZEVIEWS_TASK, "rb") as f:
response_xml = f.read().decode("utf-8")
with requests_mock.mock() as m:
m.get('{}/{}'.format(
self.server.tasks.baseurl, TaskItem.Type.MaterializeViews), text=response_xml)
all_tasks, pagination_item = self.server.tasks.get(task_type=TaskItem.Type.MaterializeViews)
task = all_tasks[0]
self.assertEqual('a462c148-fc40-4670-a8e4-39b7f0c58c7f', task.target.id)
self.assertEqual('workbook', task.target.type)
self.assertEqual('b22190b4-6ac2-4eed-9563-4afc03444413', task.schedule_id)
self.assertEqual(parse_datetime('2019-12-09T22:30:00Z'), task.schedule_item.next_run_at)
self.assertEqual(parse_datetime('2019-12-09T20:45:04Z'), task.last_run_at)
def _parse_element(schedule_xml, ns):
id = schedule_xml.get('id', None)
name = schedule_xml.get('name', None)
state = schedule_xml.get('state', None)
created_at = parse_datetime(schedule_xml.get('createdAt', None))
updated_at = parse_datetime(schedule_xml.get('updatedAt', None))
schedule_type = schedule_xml.get('type', None)
frequency = schedule_xml.get('frequency', None)
next_run_at = parse_datetime(schedule_xml.get('nextRunAt', None))
end_schedule_at = parse_datetime(schedule_xml.get('endScheduleAt', None))
execution_order = schedule_xml.get('executionOrder', None)
priority = schedule_xml.get('priority', None)
if priority:
priority = int(priority)
interval_item = None
frequency_detail_elem = schedule_xml.find('.//t:frequencyDetails', namespaces=ns)
if frequency_detail_elem is not None:
interval_item = ScheduleItem._parse_interval_item(frequency_detail_elem, frequency, ns)
return id, name, state, created_at, updated_at, schedule_type, \
next_run_at, end_schedule_at, execution_order, priority, interval_item
@staticmethod
def _parse_element(datasource_xml):
id = datasource_xml.get('id', None)
name = datasource_xml.get('name', None)
datasource_type = datasource_xml.get('type', None)
content_url = datasource_xml.get('contentUrl', None)
created_at = parse_datetime(datasource_xml.get('createdAt', None))
updated_at = parse_datetime(datasource_xml.get('updatedAt', None))
tags = None
tags_elem = datasource_xml.find('.//t:tags', namespaces=NAMESPACE)
if tags_elem is not None:
tags = TagItem.from_xml_element(tags_elem)
project_id = None
project_name = None
project_elem = datasource_xml.find('.//t:project', namespaces=NAMESPACE)
if project_elem is not None:
project_id = project_elem.get('id', None)
project_name = project_elem.get('name', None)
owner_id = None
owner_elem = datasource_xml.find('.//t:owner', namespaces=NAMESPACE)
if owner_elem is not None:
@staticmethod
def _parse_element(datasource_xml):
id = datasource_xml.get('id', None)
name = datasource_xml.get('name', None)
datasource_type = datasource_xml.get('type', None)
content_url = datasource_xml.get('contentUrl', None)
created_at = parse_datetime(datasource_xml.get('createdAt', None))
updated_at = parse_datetime(datasource_xml.get('updatedAt', None))
tags = None
tags_elem = datasource_xml.find('.//t:tags', namespaces=NAMESPACE)
if tags_elem is not None:
tags = TagItem.from_xml_element(tags_elem)
project_id = None
project_name = None
project_elem = datasource_xml.find('.//t:project', namespaces=NAMESPACE)
if project_elem is not None:
project_id = project_elem.get('id', None)
project_name = project_elem.get('name', None)
owner_id = None
owner_elem = datasource_xml.find('.//t:owner', namespaces=NAMESPACE)
def _parse_element(schedule_xml, ns):
id = schedule_xml.get('id', None)
name = schedule_xml.get('name', None)
state = schedule_xml.get('state', None)
created_at = parse_datetime(schedule_xml.get('createdAt', None))
updated_at = parse_datetime(schedule_xml.get('updatedAt', None))
schedule_type = schedule_xml.get('type', None)
frequency = schedule_xml.get('frequency', None)
next_run_at = parse_datetime(schedule_xml.get('nextRunAt', None))
end_schedule_at = parse_datetime(schedule_xml.get('endScheduleAt', None))
execution_order = schedule_xml.get('executionOrder', None)
priority = schedule_xml.get('priority', None)
if priority:
priority = int(priority)
interval_item = None
frequency_detail_elem = schedule_xml.find('.//t:frequencyDetails', namespaces=ns)
if frequency_detail_elem is not None:
interval_item = ScheduleItem._parse_interval_item(frequency_detail_elem, frequency, ns)
return id, name, state, created_at, updated_at, schedule_type, \
def _parse_element(schedule_xml, ns):
id = schedule_xml.get('id', None)
name = schedule_xml.get('name', None)
state = schedule_xml.get('state', None)
created_at = parse_datetime(schedule_xml.get('createdAt', None))
updated_at = parse_datetime(schedule_xml.get('updatedAt', None))
schedule_type = schedule_xml.get('type', None)
frequency = schedule_xml.get('frequency', None)
next_run_at = parse_datetime(schedule_xml.get('nextRunAt', None))
end_schedule_at = parse_datetime(schedule_xml.get('endScheduleAt', None))
execution_order = schedule_xml.get('executionOrder', None)
priority = schedule_xml.get('priority', None)
if priority:
priority = int(priority)
interval_item = None
frequency_detail_elem = schedule_xml.find('.//t:frequencyDetails', namespaces=ns)
if frequency_detail_elem is not None:
interval_item = ScheduleItem._parse_interval_item(frequency_detail_elem, frequency, ns)
def _parse_element(user_xml, ns):
id = user_xml.get('id', None)
name = user_xml.get('name', None)
site_role = user_xml.get('siteRole', None)
last_login = parse_datetime(user_xml.get('lastLogin', None))
external_auth_user_id = user_xml.get('externalAuthUserId', None)
fullname = user_xml.get('fullName', None)
email = user_xml.get('email', None)
auth_setting = user_xml.get('authSetting', None)
domain_name = None
domain_elem = user_xml.find('.//t:domain', namespaces=ns)
if domain_elem is not None:
domain_name = domain_elem.get('name', None)
return id, name, site_role, last_login, external_auth_user_id, fullname, email, auth_setting, domain_name
def _parse_element(datasource_xml, ns):
id_ = datasource_xml.get('id', None)
name = datasource_xml.get('name', None)
datasource_type = datasource_xml.get('type', None)
content_url = datasource_xml.get('contentUrl', None)
created_at = parse_datetime(datasource_xml.get('createdAt', None))
updated_at = parse_datetime(datasource_xml.get('updatedAt', None))
certification_note = datasource_xml.get('certificationNote', None)
certified = str(datasource_xml.get('isCertified', None)).lower() == 'true'
tags = None
tags_elem = datasource_xml.find('.//t:tags', namespaces=ns)
if tags_elem is not None:
tags = TagItem.from_xml_element(tags_elem, ns)
project_id = None
project_name = None
project_elem = datasource_xml.find('.//t:project', namespaces=ns)
if project_elem is not None:
project_id = project_elem.get('id', None)
project_name = project_elem.get('name', None)
owner_id = None
def _parse_element(workbook_xml, ns):
id = workbook_xml.get('id', None)
name = workbook_xml.get('name', None)
content_url = workbook_xml.get('contentUrl', None)
created_at = parse_datetime(workbook_xml.get('createdAt', None))
updated_at = parse_datetime(workbook_xml.get('updatedAt', None))
size = workbook_xml.get('size', None)
if size:
size = int(size)
show_tabs = string_to_bool(workbook_xml.get('showTabs', ''))
project_id = None
project_name = None
project_tag = workbook_xml.find('.//t:project', namespaces=ns)
if project_tag is not None:
project_id = project_tag.get('id', None)
project_name = project_tag.get('name', None)
owner_id = None