PrivacyScanData/target/Azure/validate_azure_dladmin_identity_8.py

def _azure_dladmin_data_storage_actions_check(
    config: Dict[str, Any],
    auth_client: AuthorizationManagementClient,
    resource_client: ResourceManagementClient,
    azure_data_required_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")
    data_path: str = get_config_value(config=config, key="env:azure:storage:path:data")
    datalake_admin: str = get_config_value(
        config=config, key="env:azure:role:name:datalake_admin"
    )

    parsed_data_path = parse_adls_path(data_path)
    container_name = parsed_data_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_actions, _ = check_for_actions(
        auth_client=auth_client,
        role_assigments=role_assignments,
        proper_scope=proper_scope,
        required_actions=azure_data_required_actions,
        required_data_actions=[],
    )

    if missing_actions:
        fail(
            AZURE_IDENTITY_MISSING_ACTIONS_FOR_LOCATION,
            subjects=[
                datalake_admin,
                f"storageAccounts/{storage_name}/blobServices/default/containers/{container_name}",  # noqa: E501
            ],
            resources=missing_actions,
        )