Best Python code snippet using avocado_python
test_ec2_volume_benchmark.py
Source:test_ec2_volume_benchmark.py  
...86            content = test_content.text_content(87                "Min time: " + str(reference_time[0]) + "s, " +88                "Max time: " + str(reference_time[1]) + "s")89            self.addDetail("AWS", content)90    def _check_test(self):91        reference_time = self._get_benchmark_result()92        if reference_time is not None:93            self.assertLessEqual(self.test_time, float(reference_time[1]),94                str(self.test_time) + "s (current) > " +95                reference_time[1] + "s (AWS)")96    @decorators.attr(type='benchmark')97    def test_001_attach_volume(self):98        """Attach volume"""99        if self.ctx.ssh is None:100            raise self.skipException("Booting failed")101        self._start_test()102        # NOTE(apavlov): ec2-create-volume -z ZONE -s SIZE_GB103        zone = self.ctx.instance.placement104        volume = self.ec2_client.create_volume(self.volume_size, zone)105        self.addResourceCleanUp(self.destroy_volume_wait, volume)106        self.ctx.volume = volume107        # NOTE(apavlov): wait it (ec2-describe-volumes VOLUME)108        self.assertVolumeStatusWait(volume, "available")109        # NOTE(apavlov): ec2-attach-volume -d /dev/XXX -i INSTANCE VOLUME110        # and wait until it will be available111        self.ctx.part_lines = self.ctx.ssh.get_partitions().split('\n')112        volume.attach(self.ctx.instance.id, "/dev/" + self.volume_attach_name)113        # NOTE(apavlov): "attaching" invalid EC2 status #1074901114        self.assertVolumeStatusWait(self._volume_state, "in-use")115        boto_wait.re_search_wait(self._volume_state, "in-use")116        boto_wait.state_wait(self._part_state, 1)117        part_lines_new = self.ctx.ssh.get_partitions().split('\n')118        volume_name = utils.detect_new_volume(self.ctx.part_lines,119                                              part_lines_new)120        self.ctx.part_lines = part_lines_new121        self._end_test("Create and attach volume")122        self.ctx.ssh.exec_command("PATH=$PATH:/usr/sbin:/usr/bin "123            "&& sudo mkfs.ext3 /dev/" + volume_name)124        self.ctx.ssh.exec_command("sudo mkdir -m 777 /vol "125            "&& sudo mount /dev/" + volume_name + " /vol")126        self.ctx.volume_ready = True127        self._check_test()128    @decorators.attr(type='benchmark')129    def test_002_fill_volume(self):130        """Fill volume with data"""131        if self.ctx.ssh is None:132            raise self.skipException("Booting failed")133        if not self.ctx.volume_ready:134            raise self.skipException("Volume preparation failed")135        self._start_test()136        self.ctx.ssh.exec_command("sudo mkdir -m 777 /vol/data")137        file_lines = 102 * int(self.volume_size)138        for i in xrange(int(self.volume_fill)):139            self.ctx.ssh.exec_command("cat /dev/urandom "140                                      "| tr -d -c 'a-zA-Z0-9' "141                                      "| fold -w 1020 "142                                      "| head -n " + str(file_lines) +143                                      " > /vol/data/file" + str(i))144        self._end_test("Volume filling")145        self.ctx.volume_filled = True146        self._check_test()147    @decorators.attr(type='benchmark')148    def test_003_snapshot_volume(self):149        """Snapshot volume"""150        if self.ctx.ssh is None:151            raise self.skipException("Booting failed")152        if not self.ctx.volume_filled:153            raise self.skipException("Volume filling failed")154        self._start_test()155        snapshot = self.ec2_client.create_snapshot(self.ctx.volume.id)156        self.addResourceCleanUp(self.destroy_snapshot_wait, snapshot)157        self.assertSnapshotStatusWait(snapshot, "completed")158        self._end_test("Snapshot creation")159        self.ctx.snapshot = snapshot160        self._check_test()161    @decorators.attr(type='benchmark')162    def test_004_clone_volume_snapshot(self):163        """Clone volume"""164        if self.ctx.ssh is None:165            raise self.skipException("Booting failed")166        if self.ctx.snapshot is None:167            raise self.skipException("Snapshot of volume failed")168        self._start_test()169        zone = self.ctx.instance.placement170        volume2 = self.ec2_client.create_volume(171            self.volume_size, zone, snapshot=self.ctx.snapshot)172        self.addResourceCleanUp(self.destroy_volume_wait, volume2)173        # NOTE(apavlov): wait it (ec2-describe-volumes VOLUME)174        self.assertVolumeStatusWait(volume2, "available")175        self._end_test("Volume creation by snapshot")176        self._check_test()177    @decorators.attr(type='benchmark')178    def test_005_detach_volume(self):179        """Detach volume"""180        if self.ctx.ssh is None:181            raise self.skipException("Booting failed")182        if not self.ctx.volume_ready:183            raise self.skipException("Volume preparation failed")184        self._start_test()185        self.ctx.ssh.exec_command("sudo umount /vol")186        self.ctx.volume.detach()187        # NOTE(apavlov): "detaching" invalid EC2 status #1074901188        self.assertVolumeStatusWait(self._volume_state, "available")189        boto_wait.re_search_wait(self._volume_state, "available")190        self._end_test("Detach volume")191        boto_wait.state_wait(self._part_state, -1)...test_file.py
Source:test_file.py  
...10ASSETS_PATH = Path(__file__).parent.resolve() / "assets"11class TestFile:12    """Tests the CVE Bin Tool file binary checker."""13    @pytest.mark.asyncio14    async def _check_test(self, file_type):15        """Helper function to parse a binary file and check whether16        the given string is in the parsed result"""17        async with NamedTemporaryFile("w+b", suffix=file_type, delete=False) as f:18            if file_type == "out":19                # write magic signature20                await f.write(b"\x7f\x45\x4c\x46\x02\x01\x01\x03\n")21                await f.seek(0)22                assert await aio_is_binary(f.name)23            else:24                await f.write(b"some other data\n")25                await f.seek(0)26                assert not await aio_is_binary(f.name)27        await aio_rmfile(f.name)28    @pytest.mark.asyncio29    async def test_binary_out_file(self):30        """file *.out"""31        await self._check_test("out")32    @pytest.mark.asyncio33    async def test_source_file(self):34        """file *.c"""35        await self._check_test("c")36    @pytest.mark.asyncio37    async def test_single_byte_file(self):38        """file single-byte"""39        assert not await aio_is_binary(str(ASSETS_PATH / "single-byte.txt"))40    @pytest.mark.asyncio41    async def test_windows(self):42        """file single-byte"""...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
