Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use wait_copy_blob method in lisa

Best Python code snippet using lisa_python

platform_.py

Source:platform_.py

...1741            blob_client = container_client.get_blob_client(vhd_path)1742            blob_client.start_copy_from_url(1743                original_vhd_path, metadata=None, incremental_copy=False1744            )1745            wait_copy_blob(blob_client, vhd_path, log)1746        return full_vhd_path1747    def _get_vhd_details(self, vhd_path: str) -> Any:1748        matched = STORAGE_CONTAINER_BLOB_PATTERN.match(vhd_path)1749        assert matched, f"fail to get matched info from {vhd_path}"1750        sc_name = matched.group("sc")1751        container_name = matched.group("container")1752        blob_name = matched.group("blob")1753        storage_client = get_storage_client(self.credential, self.subscription_id)1754        sc = [x for x in storage_client.storage_accounts.list() if x.name == sc_name]1755        assert sc[1756            01757        ], f"fail to get storage account {sc_name} from {self.subscription_id}"1758        rg = get_matched_str(sc[0].id, RESOURCE_GROUP_PATTERN)1759        return {...

common.py

Source:common.py

...845            with global_credential_access_lock:846                rm_client.resource_groups.create_or_update(847                    resource_group_name, {"location": location}848                )849def wait_copy_blob(850    blob_client: Any,851    vhd_path: str,852    log: Logger,853    timeout: int = 60 * 60,854) -> None:855    log.info(f"copying vhd: {vhd_path}")856    timeout_timer = create_timer()857    while timeout_timer.elapsed(False) < timeout:858        props = blob_client.get_blob_properties()859        if props.copy.status == "success":860            break861        # the copy is very slow, it may need several minutes. check it every862        # 2 seconds.863        sleep(2)...

transformers.py

Source:transformers.py

...158        path = _generate_vhd_path(container_client, runbook.file_name_part)159        vhd_path = f"{container_client.url}/{path}"160        blob_client = container_client.get_blob_client(path)161        blob_client.start_copy_from_url(sas_url, metadata=None, incremental_copy=False)162        wait_copy_blob(blob_client, vhd_path, self._log)163        return vhd_path164    def _restore_vm(165        self, platform: AzurePlatform, virtual_machine: Any, node: Node166    ) -> None:167        runbook: VhdTransformerSchema = self.runbook168        self._log.debug("restoring vm...")169        # release the vhd export lock, so it can be started back170        compute_client = get_compute_client(platform)171        os_disk_name = virtual_machine.storage_profile.os_disk.name172        operation = compute_client.disks.begin_revoke_access(173            resource_group_name=runbook.resource_group_name,174            disk_name=os_disk_name,175        )176        wait_operation(operation)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.