How to use copy_to_container method in avocado

Best Python code snippet using avocado_python

container_file_actions.py

Source:container_file_actions.py

...9        container = infrastructure().get_wordpress_container()10        if not container:11            raise ActionException("This WP project must first be started")12        self.run_with_container()13    def copy_to_container(self, local_path, container_path, owner = None):14        self.docker_exec(['cp', local_path, '{}:{}'.format(infrastructure().get_wordpress_container().id, container_path)])15        if owner:16            self.exec_on_container(['chown', '-R', '{}:{}'.format(owner, owner), container_path])17    def copy_from_container(self, container_path, local_path):18        self.docker_exec(['cp', '{}:{}'.format(infrastructure().get_wordpress_container().id, container_path), local_path])19    def exec_on_container(self, command_line):20        full_command_line = ['exec', '{}'.format(infrastructure().get_wordpress_container().id)]21        full_command_line.extend(command_line)22        self.docker_exec(full_command_line)23    def docker_exec(self, args):24        command_line = ['docker']25        command_line.extend(args)26        subprocess.call(command_line)27    def run_on_container(self, local_script, remote_script):28        self.copy_to_container(local_script, remote_script)29        self.exec_on_container(['chmod', 'a+rx', remote_script])30        self.exec_on_container(['/bin/sh', '-c', remote_script])31class Install(ContainerFileAction):32    def run_with_container(self):33        self.copy_components('themes')34        self.copy_components('plugins')35    def copy_components(self, component_type):36        if os.path.isdir(component_type):37            self.copy_to_container('{}/.'.format(component_type), '/var/www/html/wp-content/{}/'.format(component_type), 'www-data')38        else:39            logger.debug("%s is not a component dir" % os.path.abspath(component_type))40class Uninstall(ContainerFileAction):41    def run_with_container(self):42        self.remove_components('themes')43        self.remove_components('plugins')44    def remove_components(self, component_type):45        if os.path.isdir(component_type):46            for dir_item in os.listdir('{}'.format(component_type)):47                self.docker_exec(['exec', '{}'.format(infrastructure().get_wordpress_container().id), 'rm', '-rf', '/var/www/html/wp-content/{}/{}'.format(component_type, dir_item)])48ENABLE_MULTISITE_SCRIPT = \49"""50# Below command inspired by https://github.com/docker-library/wordpress/issues/195#issuecomment-27138240351sed -r -e 's/\\r$//' /var/www/html/wp-config.php | awk '/^\/\*.*stop editing.*\*\/$/ { print("define( \\"WP_ALLOW_MULTISITE\\", true );") } { print }' > temp.php52chown --reference /var/www/html/wp-config.php temp.php53mv temp.php /var/www/html/wp-config.php54"""55class EnableMultisite(ContainerFileAction):56    def run_with_container(self):57        script_file = '/tmp/enable_multisite.sh'58        with open(script_file, 'w') as f:59            f.write(ENABLE_MULTISITE_SCRIPT)60        self.run_on_container(script_file, script_file)61class CompleteMultisite(ContainerFileAction):62    def run_with_container(self):63        self.edit_config()64        self.edit_htaccess()65    def edit_config(self):66        config_file = '/var/www/html/wp-config.php'67        temp_file = '/tmp/wp-config.php'68        self.edit_file(config_file, temp_file)69    def edit_file(self, remote_file, temp_file):70        self.copy_from_container(remote_file, temp_file)71        subprocess.call(['vim', temp_file])72        self.copy_to_container(temp_file, remote_file)73    def edit_htaccess(self):74        htaccess_file = '/var/www/html/.htaccess'75        temp_file = '/tmp/wp-config.php'...

run_docker.py

Source:run_docker.py

...7from io import BytesIO, StringIO8import docker9DOCKER_IMAGE = "gs_test"10client = docker.from_env()11def copy_to_container(container, artifact_file, path='/autograder'):12    with create_archive(artifact_file) as archive:13        container.put_archive(path=path, data=archive)14def create_archive(artifact_file):15    pw_tarstream = BytesIO()16    pw_tar = tarfile.TarFile(fileobj=pw_tarstream, mode='w')17    file_data = open(artifact_file, 'rb').read()18    tarinfo = tarfile.TarInfo(name=artifact_file)19    tarinfo.size = len(file_data)20    tarinfo.mtime = time.time()21    # tarinfo.mode = 060022    pw_tar.addfile(tarinfo, BytesIO(file_data))23    pw_tar.close()24    pw_tarstream.seek(0)25    return pw_tarstream26def run_submission(sd, docker_image=DOCKER_IMAGE):27    image = client.images.get(docker_image)28    container = client.containers.create(image, command="/bin/bash", tty=True,29                                         stdin_open=True, auto_remove=False)30    container.start()31    try:32        log = container.exec_run('mkdir /autograder/submission',33                                 stdout=True,34                                 stderr=True,35                                 stream=True)36        for line in log[1]:37            print(line.decode("utf-8"), end='')38        copy_to_container(container, sd, '/autograder/submission')39        #copy_to_container(container, 'run_autograder')40        #container.exec_run('cp /autograder/source/run_autograder /autograder')41        #container.exec_run('chmod ugo+x /autograder/run_autograder')42        container.exec_run(43            'rm -rf /autograder/submission/template/build',44            workdir='/autograder',45            privileged=True)46        log = container.exec_run('timeout 3m /autograder/run_autograder',47                                 stdout=True,48                                 stderr=True,49                                 stream=True,50                                 workdir='/autograder',51                                 privileged=True)52        for line in log[1]:53            print(line.decode("utf-8"), end='')...

blob.py

Source:blob.py

1# from azure.storage.blob import BlockBlobService2from azure.storage.blob.aio import BlobServiceClient3from azure.identity.aio import DefaultAzureCredential4import io5import shutil6class BlobService(object):7    """8    """9    def __init__(self, credential, copy_from_container, copy_to_container, blob_name, blob_url) -> None:10        self.copy_from_container = copy_from_container11        self.copy_to_container = copy_to_container12        self.blob_name = blob_name13        self.credential = credential14        self.blob_url = blob_url15        self.blob_service_client = BlobServiceClient(account_url=blob_url, credential=self.credential)16        17        pass18    async def get_blob(self):19        """20        """21        print("Downloading file from container using managed identity")22        async with self.blob_service_client as blob_service_client, self.credential:23            24            blob_client = blob_service_client.get_blob_client(25                container=self.copy_from_container, blob=self.blob_name26            )27        28            stream = await blob_client.download_blob()29            print("Download complete using managed identity")30            return_data = io.StringIO((await stream.readall()).decode("utf-8"))31            with open("./data/{}_raw.{}".format(self.blob_name.split('.')[0],self.blob_name.split('.')[1]),'w',encoding = 'utf-8') as f:32                return_data.seek(0)33                shutil.copyfileobj(return_data, f)34            await blob_client.close()35            return None36    async def write_blob(self):37        """38        """39        print("Uploading file to container using managed identity")40        async with self.blob_service_client as blob_service_client, self.credential:41            42            blob_client = blob_service_client.get_blob_client(43                container=self.copy_to_container, blob=self.blob_name44            )45            with open("./data/{}_raw.{}".format(self.blob_name.split('.')[0],self.blob_name.split('.')[1]), "rb") as data:46                print("Uploaded complete using managed identity")47                await blob_client.upload_blob(data, overwrite=True)48            await blob_client.close()...

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.