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_application_todict_obj(self):
MyObject = namedtuple('MyObject', 'a b')
the_input = MyObject('x', 'y')
actual = todict(the_input)
expected = {'a': 'x', 'b': 'y'}
self.assertEqual(actual, expected)
def test_application_todict_dict_with_datetime(self):
the_input = datetime(2017, 10, 13, 1, 23, 45)
actual = todict(the_input)
expected = the_input.isoformat()
self.assertEqual(actual, expected)
del path_comps[-1]
if implicit_preview_info:
preview_kwargs = implicit_preview_info.__dict__.copy()
preview_kwargs['object_type'] = 'command'
previews.append(ImplicitPreviewItem(**preview_kwargs))
colorama.init()
for d in deprecations:
print(d.message, file=sys.stderr)
for p in previews:
print(p.message, file=sys.stderr)
colorama.deinit()
cmd_result = parsed_args.func(params)
cmd_result = todict(cmd_result)
event_data = {'result': cmd_result}
self.cli_ctx.raise_event(EVENT_INVOKER_TRANSFORM_RESULT, event_data=event_data)
self.cli_ctx.raise_event(EVENT_INVOKER_FILTER_RESULT, event_data=event_data)
return CommandResultItem(event_data['result'],
exit_code=0,
table_transformer=cmd_tbl[parsed_args.command].table_transformer,
is_query_active=self.data['query_active'])
assignments_client = factory.role_assignments
definitions_client = factory.role_definitions
if show_all:
if resource_group_name or scope:
raise CLIError('group or scope are not required when --all is used')
scope = None
else:
scope = _build_role_scope(resource_group_name, scope,
definitions_client.config.subscription_id)
assignments = _search_role_assignments(cmd.cli_ctx, assignments_client, definitions_client,
scope, assignee, role,
include_inherited, include_groups)
results = todict(assignments) if assignments else []
if include_classic_administrators:
results += _backfill_assignments_for_co_admins(cmd.cli_ctx, factory, assignee)
if not results:
return []
# 1. fill in logic names to get things understandable.
# (it's possible that associated roles and principals were deleted, and we just do nothing.)
# 2. fill in role names
role_defs = list(definitions_client.list(
scope=scope or ('/subscriptions/' + definitions_client.config.subscription_id)))
worker = MultiAPIAdaptor(cmd.cli_ctx)
role_dics = {i.id: worker.get_role_property(i, 'role_name') for i in role_defs}
for i in results:
if not i.get('roleDefinitionName'):
if role_dics.get(worker.get_role_property(i, 'roleDefinitionId')):
result = cmd_copy(params)
if cmd_copy.supports_no_wait and getattr(expanded_arg, 'no_wait', False):
result = None
elif cmd_copy.no_wait_param and getattr(expanded_arg, cmd_copy.no_wait_param, False):
result = None
transform_op = cmd_copy.command_kwargs.get('transform', None)
if transform_op:
result = transform_op(result)
if _is_poller(result):
result = LongRunningOperation(cmd_copy.cli_ctx, 'Starting {}'.format(cmd_copy.name))(result)
elif _is_paged(result):
result = list(result)
result = todict(result, AzCliCommandInvoker.remove_additional_prop_layer)
event_data = {'result': result}
cmd_copy.cli_ctx.raise_event(EVENT_INVOKER_TRANSFORM_RESULT, event_data=event_data)
return event_data['result']
except Exception as ex: # pylint: disable=broad-except
if cmd_copy.exception_handler:
cmd_copy.exception_handler(ex)
return CommandResultItem(None, exit_code=1, error=ex)
six.reraise(*sys.exc_info())
def list_role_assignments(cmd, assignee_object_id, scope=None):
'''
:param include_groups: include extra assignments to the groups of which the user is a
member(transitively).
'''
graph_client = _graph_client_factory(cmd.cli_ctx)
factory = _auth_client_factory(cmd.cli_ctx)
assignments_client = factory.role_assignments
definitions_client = factory.role_definitions
assignments = _search_role_assignments(assignments_client, assignee_object_id)
results = todict(assignments) if assignments else []
if not results:
return []
# 1. fill in logic names to get things understandable.
# (it's possible that associated roles and principals were deleted, and we just do nothing.)
# 2. fill in role names
role_defs = list(definitions_client.list(
scope=(scope if scope else '/subscriptions/' + definitions_client.config.subscription_id)))
role_dics = {i.id: i.role_name for i in role_defs}
for i in results:
if role_dics.get(i['roleDefinitionId']):
i['roleDefinitionName'] = role_dics[i['roleDefinitionId']]
# fill in principal names
principal_ids = set(i['principalId'] for i in results if i['principalId'])
def _to_dict(obj):
if isinstance(obj, ErrorResponse):
return _to_dict(todict(obj))
if isinstance(obj, dict):
result = OrderedDict()
# Complex objects should be displayed last
sorted_keys = sorted(obj.keys(), key=lambda k: (isinstance(obj[k], dict), isinstance(obj[k], list), k))
for key in sorted_keys:
if obj[key] is None or obj[key] == [] or obj[key] == {}:
continue
result[key] = _to_dict(obj[key])
return result
if isinstance(obj, list):
return [_to_dict(v) for v in obj]