123456789101112131415161718192021222324252627282930313233343536373839404142434445464748 |
- def _azure_dladmin_backup_storage_data_actions_check(
- config: Dict[str, Any],
- auth_client: AuthorizationManagementClient,
- resource_client: ResourceManagementClient,
- azure_data_required_data_actions: List[str],
- ) -> None: # pragma: no cover
- # noqa: D401,E501
- sub_id: str = get_config_value(config=config, key="infra:azure:subscription_id")
- rg_name: str = get_config_value(config=config, key="infra:azure:metagroup:name")
- storage_name: str = get_config_value(config=config, key="env:azure:storage:name")
- backup_path: str = get_config_value(
- config=config, key="env:azure:storage:path:backup"
- )
- datalake_admin: str = get_config_value(
- config=config, key="env:azure:role:name:datalake_admin"
- )
- parsed_logger_path = parse_adls_path(backup_path)
- container_name = parsed_logger_path[1]
- role_assignments = get_role_assignments(
- auth_client=auth_client,
- resource_client=resource_client,
- identity_name=datalake_admin,
- subscription_id=sub_id,
- resource_group=rg_name,
- )
- proper_scope = get_storage_container_scope(
- sub_id, rg_name, storage_name, container_name
- )
- _, missing_data_actions = check_for_actions(
- auth_client=auth_client,
- role_assigments=role_assignments,
- proper_scope=proper_scope,
- required_actions=[],
- required_data_actions=azure_data_required_data_actions,
- )
- if missing_data_actions:
- fail(
- AZURE_IDENTITY_MISSING_DATA_ACTIONS_FOR_LOCATION,
- subjects=[
- datalake_admin,
- f"storageAccounts/{storage_name}/blobServices/default/containers/{container_name}", # noqa: E501
- ],
- resources=missing_data_actions,
- )
|