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

How to use prepare_environment method in localstack

Best Python code snippet using localstack_python

platform.py

Source:platform.py

...33        self.__name = get_random_name()34        while self.__name in inst_names:35            self.__name = get_random_name()36        return self.__name37    def prepare_environment(self):38        """39        Prepare the running environment.40        Duplicate the host filesystem if required (sometimes it is modified41        by the guest system).42        Prepare the SD card image file if required (if sd=true in config.yaml),43        and provide the path to qemu.44        """45        #Â Create a temporary directory46        self.__tempDir = tempfile.mkdtemp(prefix='piotr-')47        debug('piotr.qemu:prepare_environment', 'Created temporary directory %s' % self.__tempDir)48        # Copy our host filesystem in it49        try:50            debug('piotr.qemu:prepare_environment', 'Copy selected host FS to temporary directory')51            tempHostFs = join(self.__tempDir, basename(self.device.getHostFsPath()))52            copyfile(53                self.device.getHostFsPath(),54                tempHostFs55            )56            self.__hostFsPath = tempHostFs57            debug('piot.qemu:prepare_environment', 'Host fs is now stored at: %s' % self.__hostFsPath)58            # Embed our target file system if required (if guestfs is set to 'embed')59            # This method is designed to work for old versions of kernel, in order to60            #Â avoid using 9P as it may not be available61            debug('piotr.qemu:prepare_environment', 'Checking if guest FS must be embedded into our duplicate host FS ...')62            if self.device.mustEmbedGuestFS():63                debug('piotr.qemu:prepare_environment', 'Guest FS must be embedded')64                debug('piotr.qemu:prepare_environment', 'Creating mounting point for host FS')65                # First, create a temporary mount point for our filesystem66                tempHostFsMP = join(self.__tempDir, 'host')67                mkdir(tempHostFsMP)68                #Â Then, we mount our host FS onto this mounting point69                debug('piotr.qemu:prepare_environment', 'Mounting Host FS onto %s' % tempHostFsMP)70                if subprocess.call('mount %s %s' % (71                    tempHostFs,72                    tempHostFsMP73                    ), shell=True) == 0:74                    debug('piotr.qemu:prepare_environment', 'Host FS successfully mounted !')75                    #Â Copy guest filesystem into our host fs76                    debug('piotr.qemu:prepare_environment', 'Copying guest FS into host FS ...')77                    try:78                        subprocess.call('cp -rf %s %s' % (79                            join(self.device.getPath(), 'rootfs/*'),80                            join(tempHostFsMP, 'target'),81                        ), shell=True)82                        debug('piotr.qemu:prepare_environment', 'Guest FS successfully copied to host FS !')83                    except Error as shutilErr:84                        debug('piotr.qemu:prepare_environment', 'Unable to copy guest FS into host FS')85                        warning('Piotr:QemuGuest', 'Cannot install guest FS into host FS (copy failed)')86                        # Failure87                        subprocess.call('umount %s' % tempHostFsMP)88                        return False89                    #Â Unmount host FS90                    debug('piotr.qemu:prepare_environment', 'Unmounting host FS ...')91                    if subprocess.call('umount %s' % tempHostFsMP, shell=True) == 0:92                        debug('piotr.qemu:prepare_environment', 'Host FS successfully unmounted.')93                    else:94                        debug('piotr.qemu:prepare_environment', 'Unable to unmount host FS')95                        warning('Piotr:QemuGuest', 'An error occurred while unmounting host FS')96                        # Failure97                        return False98                else:99                    debug('piotr.qemu:prepare_environment', 'Unable to mount host FS onto %s' % tempHostFsMP)100                    warning('Piotr:QemuGuest', 'Cannot mount host FS to %s (required to embed guest FS)' % tempHostFsMP)101            #Â Prepare our SD card if any102            if self.device.hasSd():103                #Â Compute sd card image size (fs size + 50M)104                sd_image = join(105                    self.__tempDir,106                    'sdcard.img'107                )108                sd_directory = Path(self.device.getSdImage())109                image_size = sum(f.stat().st_size for f in sd_directory.glob('**/*') if f.is_file()) + (50*(2**20))110                #Â Allocate sd card image111                image_alloc = subprocess.call(112                    'dd if=/dev/zero of=%s bs=%d count=1' % (113                        sd_image,114                        image_size115                    ),116                    shell=True,117                    stdout=subprocess.DEVNULL,118                    stderr=subprocess.DEVNULL119                )120                if image_alloc == 0:121                    #Â create an ext2 filesystem122                    image_fs = subprocess.call(123                        'mkfs.ext2 %s' % sd_image,124                        shell=True,125                        stdout=subprocess.DEVNULL,126                        stderr=subprocess.DEVNULL127                    )128                    if image_fs == 0:129                        # Mount filesystem and copy sdcard files130                        sd_mp = join(self.__tempDir, 'sdcard')131                        subprocess.call('mkdir %s' % sd_mp, shell=True)132                        subprocess.call('mount -t ext2 %s %s' % (sd_image, sd_mp), shell=True)133                        subprocess.call('cp -rf %s/* %s' % (sd_directory, sd_mp), shell=True)134                        #Â unmount filesystem135                        subprocess.call('umount %s' % sd_mp, shell=True)136                        # SD card image is ready to use !137                        self.__sdImagePath = sd_image138                    else:139                        debug('piotr.qemu:prepare_environment', 'cannot format sdcard image to ext2')140                        error('QemuGuest','Cannot format SD card, mkfs.ext2 error occurred.')141                        return False142                else:143                    debug('piotr.qemu:prepare_environment', 'cannot create sdcard image in temporary directory')144                    error('QemuGuest', 'Cannot create SD card image in %s' % self.__tempDir)145                    return False146        except Exception as exc:147            print(exc)148            debug('piotr.qemu:prepare_environment', 'An exception occurred: %s' % str(exc))149            error('QemuGuest', 'An error occurred, unable to prepare the guest environment.')150            return False151        return True152    def clean_environment(self):153        """154        Clean environment.155        Delete the temporary directory and remove all files.156        """157        debug('piotr.qemu:clean_environment', 'Remove temporary environment at %s' % self.__tempDir)158        rmtree(self.__tempDir)159        debug('piotr.qemu:clean_environment', 'Temporary environment removed')160    def append_arg(self, argument):161        """162        Append an argument to qemu arguments list.`argument` may be a string163        or a list of arguments.164        @param  argument    object  Argument to append.165        """166        if isinstance(argument, str):167            self.__qemu_args.append(argument)168        elif isinstance(argument, list):169            self.__qemu_args.extend(argument)170        else:171            raise ValueError()172    def set_platform(self, platform):173        """174        Add qemu platform arguments175        @param  str     platform    Qemu platform to use176        """177        #Â declare machine178        self.append_arg([179            '-M',180            platform181        ])182    def set_cpu(self, cpu):183        """184        Add qemu platform argument.185        @param  str     cpu     CPU to use.186        """187        self.append_arg([188            '-cpu',189            cpu190        ])191    def set_memory(self, memory):192        """193        Add qemu memory argument.194        @param  str     memory  Memory to use.195        """196        self.append_arg([197            '-m',198            str(memory)199        ])200    def set_kernel(self, kernel):201        """202        Add qemu kernel argument.203        @param  str     kernel  Kernel file to use.204        """205        self.append_arg([206            '-kernel',207            kernel208        ])209    def set_dtb(self, dtb):210        """211        Add qemu DTB file path argument.212        @param  str     dtb     DTB file path.213        """214        self.append_arg([215            '-dtb',216            dtb217        ])218    def set_sd(self, sd_image_path):219        """220        Add qemu SD card arguments.221        @param  str     sd_image_path   SD card image file path.222        """223        self.append_arg([224            '-drive',225            'file=%s,if=none,format=raw,id=hd1,index=1' % sd_image_path,226            '-device',227            'virtio-blk-device,drive=hd1'228        ])229    def set_host_drive(self, hostfs, drive_type):230        """231        Add qemu host drive with corresponding image.232        @param  str     hostfs  Host FS path.233        """234        #Â Handle 'virtio' special drive.235        if drive_type== 'virtio':236            self.append_arg([237                '-drive',238                'file=%s,if=none,format=raw,id=hd0' % hostfs,239                '-device',240                'virtio-blk-device,drive=hd0'241            ])242        else:243            # Otherwise use the provided drive type.244            self.append_arg([245                '-drive',246                'file=%s,%s' % (247                    hostfs,248                    drive_type249                )250            ])251    def set_bootargs(self, bootargs):252        """253        Add qemu bootargs argument.254        @param  str     bootargs    Bootargs to use.255        """256        self.append_arg([257            '-append',258            '"%s"' % bootargs259        ])260    def set_nic(self, nic, redirects=[]):261        """262        Add qemu nic arguments.263        @param str  nic     Nic parameters to use.264        """265        # If nic is in 'user' mode, then add a simple network device with current266        # nic index.267        if nic['type'] == 'user':268            # Compile port forward rules269            redir_rules = []270            for redir in redirects:271                redir_rules.append(272                    'hostfwd=%s::%d-:%d' % (273                        redir['proto'],274                        redir['local'],275                        redir['dest']276                    )277                )278            #Â Declare interface279            if len(redir_rules) == 0:280                self.append_arg([281                    '-netdev',282                    'user,id=net%d' % self.nic_index,283                    '-device',284                    'virtio-net-%s,netdev=net%d' % (285                        self.get_virtio_config(),286                        self.nic_index287                    )288                ])289            else:290                self.append_arg([291                    '-netdev',292                    'user,id=net%d,%s' % (293                        self.nic_index,294                        ','.join(redir_rules)295                    ),296                    '-device',297                    'virtio-net-%s,netdev=net%d' % (298                        self.get_virtio_config(),299                        self.nic_index300                    )301                ])302        elif nic['type'] == 'tap':303            self.append_arg([304                '-netdev',305                'type=tap,id=net%d,ifname=%s,script=no,downscript=no' % (306                    self.nic_index,307                    nic['interface']308                ),309                '-device',310                'virtio-net-%s,netdev=net%d' % (311                    self.get_virtio_config(),312                    self.nic_index,313                )314            ])315        #Â Increment nic index316        self.nic_index += 1317    def set_network_tap(self):318        """319        Add a network tap interface.320        """321        self.append_arg([322            '-netdev',323            'type=tap,id=net%d,ifname=tap0,script=no,downscript=no' % self.nic_index,324            '-device',325            'virtio-net-device,netdev=net%d' % self.nic_index,326        ])327        self.nic_index += 1328    def set_guest_fs(self, rootfs, fstype):329        """330        Add qemu guest FS arguments.331        @param  str rootfs  Guest root filesystem path.332        @param  str fstype  Bus name (must be 'pci' or 'device')333        """334        self.append_arg([335            '-fsdev',336            'local,path=%s,security_model=passthrough,id=host0' % (337                rootfs338                ),339            '-device',340            '%s,fsdev=host0,mount_tag=host0' % fstype,341        ])342    def set_no_screen(self):343        """344        Add qemu nographic option.345        """346        self.append_arg('-nographic')347    def set_agent_vport(self, vport, device):348        """349        Set agent virtual serial port.350        @param  str     vport   Virtual port UNIX socket path351        @param  str     device  Virtual serial port device type352        """353        self.append_arg([354            '-chardev',355            'socket,path=%s,server,nowait,id=piotr' % vport,356            '-device',357            '%s' % device,358            '-device',359            'virtserialport,chardev=piotr,name=piotr'360        ])361    def prepare_args(self):362        """363        Prepare arguments array for qemu-system-arm.364        """365        #Â Generate command line to start Qemu guest366        self.__qemu_args = []367        # Use mainline qemu-system-arm368        self.append_arg('qemu-system-arm')369        #Â Add platform arguments370        self.set_platform(self.device.getPlatform())371        # Add cpu arguments372        if self.device.getCpu() is not None:373            self.set_cpu(self.device.getCpu())374        # Add memory arguments375        self.set_memory(self.device.getMemory())376        #Â Add kernel377        self.set_kernel(self.device.getKernelPath())378        # Add DTB argument if provided379        if self.device.hasDtb():380            self.set_dtb(self.device.getDtbPath())381        # Add sd card arguments if required382        if self.device.hasSd():383            self.set_sd(self.__sdImagePath)384        # Add our host drive385        self.set_host_drive(self.__hostFsPath, self.device.getDriveType())386        #Â Add bootargs arguments387        self.set_bootargs(self.device.getBootArgs())388        #Â Check network redirections consistency389        applied_redirects = []390        if self.device.hasRedirects() and self.device.hasNetwork():391            rules = self.device.getRedirects()392            nics = self.device.getNics()393            nic_names = [nic['name'] for nic in nics]394            nics_type = {}395            for nic in nics:396                nics_type[nic['name']] = nic['type']397            for rule in rules:398                if rule['iface'] not in nic_names:399                    warning('piotr.qemu.platform', 'Port redirection `%s` cannot be applied (unknown network interface `%s`)' % (400                        rule['name'],401                        rule['iface']402                    ))403                elif nics_type[rule['iface']] == 'user':404                    applied_redirects.append(rule)405                else:406                    warning('piotr.qemu.platform', 'Port redirection `%s` cannot be applied (wrong NIC type for interface `%s`)' % (407                        rule['name'],408                        rule['iface']409                    ))410        elif not self.device.hasNetwork():411            warning('piotr.qemu.platform', 'Target device has network port redirections but no NIC !')412        # Prepare networking interfaces413        if self.device.hasNetwork():414            nics = self.device.getNics()415            for nic in nics:416                # Setup interface with associated redirections (if any)417                self.set_nic(nic, filter(lambda x: (x['iface'] == nic['name']), applied_redirects))418        # set our virtfs share if required419        if not self.device.mustEmbedGuestFS():420            self.set_guest_fs(421                self.device.getRootfsPath(),422                self.device.getVirtio9pConfig()423            )424        # No screen425        self.set_no_screen()426        #Â Add a virtual serial port to allow host/guest communication427        virtport = join(self.__tempDir, 'piotr')428        self.set_agent_vport(virtport, self.device.getVirtioSerialConfig())429    def start(self, background=False):430        """431        Start Qemu guest432        """433        # Prepare our environment434        if self.prepare_environment():435            self.prepare_args()436            # Start process with dedicated environment437            try:438                # Save device name and socket in environment variables439                environ['PIOTR_GUEST'] = self.device.getDeviceName()440                environ['PIOTR_SOCK'] = join(self.__tempDir, 'piotr')441                environ['PIOTR_INSTNAME'] = self.generate_name()442                debug('piotr.qemu:cmdline', ' '.join(self.__qemu_args))443                debug('piotr.qemu:start', ' '.join(self.__qemu_args))444                if background:445                    #Â Launch Qemu-system-arm in background, redirect stdout446                    #Â and stderr into a pipe447                    self.process = subprocess.Popen(448                        ' '.join(self.__qemu_args),...

test_application_environment.py

Source:test_application_environment.py

...26)27class ApplicationEnvironmentTests(TestCase):28    def test_raises_notimplementederror(self):29        self.assertThat(30            lambda: ApplicationEnvironment().prepare_environment(None, None),31            raises(32                NotImplementedError("Sub-classes must implement this method.")33            )34        )35class GtkApplicationEnvironmentTests(TestCase):36    def setUp(self):37        super(GtkApplicationEnvironmentTests, self).setUp()38        self.app_environment = GtkApplicationEnvironment()39    def test_does_not_alter_app(self):40        fake_app = self.getUniqueString()41        app, args = self.app_environment.prepare_environment(fake_app, [])42        self.assertEqual(fake_app, app)43    @patch("autopilot.application._environment.os")44    def test_modules_patched(self, patched_os):45        patched_os.getenv.return_value = ""46        fake_app = self.getUniqueString()47        app, args = self.app_environment.prepare_environment(fake_app, [])48        patched_os.putenv.assert_called_once_with('GTK_MODULES', ':autopilot')49    @patch("autopilot.application._environment.os")50    def test_modules_not_patched_twice(self, patched_os):51        patched_os.getenv.return_value = "autopilot"52        fake_app = self.getUniqueString()53        app, args = self.app_environment.prepare_environment(fake_app, [])54        self.assertFalse(patched_os.putenv.called)55class QtApplicationEnvironmentTests(TestCase):56    def setUp(self):57        super(QtApplicationEnvironmentTests, self).setUp()58        self.app_environment = QtApplicationEnvironment()59    def test_does_not_alter_app(self):60        fake_app = self.getUniqueString()61        app, args = self.app_environment.prepare_environment(fake_app, [])62        self.assertEqual(fake_app, app)63    def test_inserts_testability_with_no_args(self):64        app, args = self.app_environment.prepare_environment('some_app', [])65        self.assertEqual(['-testability'], args)66    def test_inserts_testability_before_normal_argument(self):67        app, args = self.app_environment.prepare_environment('app', ['-l'])68        self.assertEqual(['-testability', '-l'], args)69    def test_inserts_testability_after_qt_version_arg(self):70        app, args = self.app_environment.prepare_environment(71            'app',72            ['-qt=qt5']73        )74        self.assertEqual(['-qt=qt5', '-testability'], args)75    def test_does_not_insert_testability_if_already_present(self):76        app, args = self.app_environment.prepare_environment(77            'app', ['-testability']78        )...

actions.py

Source:actions.py

...7from pisi.actionsapi import crosstools8from pisi.actionsapi import pisitools9from pisi.actionsapi import shelltools10from pisi.actionsapi import get11def prepare_environment():12    crosstools.environment["CXXFLAGS"] = ""13    crosstools.environment["instDIR"] = get.installDIR()14    crosstools.environment["workDIR"] = get.workDIR()15    crosstools.environment["srcVER"] = get.srcVERSION()16    crosstools.environment["CFLAGS"] = "-Du32=__u32 -O2 \17                                        -I%(workDIR)s/iproute2-%(srcVER)s/include/linux \18                                        -I%(workDIR)s/iproute2-%(srcVER)s/include/ \19                                        -I%(RootDir)s/usr/include \20                                        -L%(RootDir)s/usr/lib -L%(RootDir)s/lib" % crosstools.environment21    crosstools.environment["LDFLAGS"] = "%(LDFLAGS)s -L%(RootDir)s/usr/lib -L%(RootDir)s/lib"  % crosstools.environment22def setup():23    pisitools.dosed("configure", "gcc", "%(CC)s" % crosstools.environment)24    crosstools.configure()25def build():26    shelltools.export("LC_ALL", "C")27    prepare_environment()28    crosstools.prepare()29    crosstools.make('CC="%(CC)s" \30                     KERNEL_INCLUDE="%(SysRoot)s/usr/include" \31                     RPM_OPT_FLAGS="%(CFLAGS)s" \32                     SUBDIRS="lib tc ip" \33                     CFLAGS="%(CFLAGS)s" \34                     LDFLAGS="%(LDFLAGS)s"' % crosstools.environment)35def install():36    prepare_environment()37    # FIXME: something's missing.38    crosstools.rawInstall('DESTDIR="%(instDIR)s" \39                           SBINDIR="/sbin" \40                           DOCDIR="/%(docDIR)s" \41                           MANDIR="/usr/share/man" \42                           ROOTDIR="%(SysRoot)s" \43                           SUBDIRS="lib tc ip"' % crosstools.environment)44    pisitools.removeDir("/var")45    #pisitools.dodir("/usr/sbin")...

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.