How to use read_bytes method in Airtest

Best Python code snippet using Airtest

fixtures.py

Source:fixtures.py

2from ward import fixture3TEST_PATH = Path(__file__).parent4@fixture(scope='global')5def flac_0_duration():6	return Path(TEST_PATH, 'data', 'flac', 'flac-0-duration.bin').read_bytes()7@fixture(scope='global')8def flac_0_size_block():9	return Path(TEST_PATH, 'data', 'flac', 'flac-block-0-size.bin').read_bytes()10@fixture(scope='global')11def flac_application_block():12	return Path(TEST_PATH, 'data', 'flac', 'flac-application-block.bin').read_bytes()13@fixture(scope='global')14def flac_application_data(flac_application_block=flac_application_block):15	return flac_application_block[4:]16@fixture(scope='global')17def flac_cuesheet_block():18	return Path(TEST_PATH, 'data', 'flac', 'flac-cuesheet-block.bin').read_bytes()19@fixture(scope='global')20def flac_cuesheet_data(flac_cuesheet_block=flac_cuesheet_block):21	return flac_cuesheet_block[4:]22@fixture(scope='global')23def flac_cuesheet_index_1():24	return Path(TEST_PATH, 'data', 'flac', 'flac-cuesheet-index-1.bin').read_bytes()25@fixture(scope='global')26def flac_cuesheet_index_2():27	return Path(TEST_PATH, 'data', 'flac', 'flac-cuesheet-index-2.bin').read_bytes()28@fixture(scope='global')29def flac_cuesheet_track_1():30	return Path(TEST_PATH, 'data', 'flac', 'flac-cuesheet-track-1.bin').read_bytes()31@fixture(scope='global')32def flac_cuesheet_track_2():33	return Path(TEST_PATH, 'data', 'flac', 'flac-cuesheet-track-2.bin').read_bytes()34@fixture(scope='global')35def flac_id3v2():36	return Path(TEST_PATH, 'data', 'flac', 'flac-id3v2.bin').read_bytes()37@fixture(scope='global')38def flac_invalid_block():39	return Path(TEST_PATH, 'data', 'flac', 'flac-invalid-block.bin').read_bytes()40@fixture(scope='global')41def flac_padding_block():42	return Path(TEST_PATH, 'data', 'flac', 'flac-padding-block.bin').read_bytes()43@fixture(scope='global')44def flac_padding_data(flac_padding_block=flac_padding_block):45	return flac_padding_block[4:]46@fixture(scope='global')47def flac_picture_block():48	return Path(TEST_PATH, 'data', 'flac', 'flac-picture-block.bin').read_bytes()49@fixture(scope='global')50def flac_picture_data(flac_picture_block=flac_picture_block):51	return flac_picture_block[4:]52@fixture(scope='global')53def flac_reserved_block():54	return Path(TEST_PATH, 'data', 'flac', 'flac-reserved-block.bin').read_bytes()55@fixture(scope='global')56def flac_seektable_block():57	return Path(TEST_PATH, 'data', 'flac', 'flac-seektable-block.bin').read_bytes()58@fixture(scope='global')59def flac_seektable_data(flac_seektable_block=flac_seektable_block):60	return flac_seektable_block[4:]61@fixture(scope='global')62def flac_streaminfo_block():63	return Path(TEST_PATH, 'data', 'flac', 'flac-streaminfo-block.bin').read_bytes()64@fixture(scope='global')65def flac_streaminfo_data(flac_streaminfo_block=flac_streaminfo_block):66	return flac_streaminfo_block[4:]67@fixture(scope='global')68def flac_vorbis():69	return Path(TEST_PATH, 'data', 'flac', 'flac-vorbis.bin').read_bytes()70@fixture(scope='global')71def flac_vorbis_comment_block():72	return Path(TEST_PATH, 'data', 'flac', 'flac-vorbis-comment-block.bin').read_bytes()73@fixture(scope='global')74def flac_vorbis_comment_data(flac_vorbis_comment_block=flac_vorbis_comment_block):75	return flac_vorbis_comment_block[4:]76@fixture(scope='global')77def id3v1():78	return Path(TEST_PATH, 'data', 'id3', 'id3v1.bin').read_bytes()79@fixture(scope='global')80def id3v22():81	return Path(TEST_PATH, 'data', 'id3', 'id3v22.bin').read_bytes()82@fixture(scope='global')83def id3v23():84	return Path(TEST_PATH, 'data', 'id3', 'id3v23.bin').read_bytes()85@fixture(scope='global')86def id3v24():87	return Path(TEST_PATH, 'data', 'id3', 'id3v24.bin').read_bytes()88@fixture(scope='global')89def id3v24_unsync():90	return Path(TEST_PATH, 'data', 'id3', 'id3v24-unsync.bin').read_bytes()91@fixture(scope='global')92def id3v2_header():93	return Path(TEST_PATH, 'data', 'id3', 'id3v2-header.bin').read_bytes()94@fixture(scope='global')95def id3v22_picture_frame():96	return Path(TEST_PATH, 'data', 'id3', 'id3v22-picture-frame.bin').read_bytes()97@fixture(scope='global')98def id3v22_picture(id3v22_picture_frame=id3v22_picture_frame):99	return id3v22_picture_frame[6:]100@fixture(scope='global')101def id3v22_picture_data(id3v22_picture=id3v22_picture):102	return id3v22_picture[-92:]103@fixture(scope='global')104def id3v24_picture_frame():105	return Path(TEST_PATH, 'data', 'id3', 'id3v24-picture-frame.bin').read_bytes()106@fixture(scope='global')107def id3v24_picture(id3v24_picture_frame=id3v24_picture_frame):108	return id3v24_picture_frame[10:]109@fixture(scope='global')110def id3v24_picture_data(id3v24_picture=id3v24_picture):111	return id3v24_picture[-92:]112@fixture(scope='global')113def lame_header():114	return Path(TEST_PATH, 'data', 'mp3', 'lame-header.bin').read_bytes()115@fixture(scope='global')116def lame_replay_gain():117	return Path(TEST_PATH, 'data', 'mp3', 'lame-replay-gain.bin').read_bytes()118@fixture(scope='global')119def lame_replay_gain_negative():120	return Path(TEST_PATH, 'data', 'mp3', 'lame-replay-gain-negative.bin').read_bytes()121@fixture(scope='global')122def lame_replay_gain_null():123	return Path(TEST_PATH, 'data', 'mp3', 'lame-replay-gain-null.bin').read_bytes()124@fixture(scope='global')125def mp3_cbr_2_frames():126	return Path(TEST_PATH, 'data', 'mp3', 'mp3-cbr-2-frames.bin').read_bytes()127@fixture(scope='global')128def mp3_lame_vbr():129	return Path(TEST_PATH, 'data', 'mp3', 'mp3-lame-vbr.bin').read_bytes()130@fixture(scope='global')131def mp3_sync_branch():132	return Path(TEST_PATH, 'data', 'mp3', 'mp3-sync-branch.bin').read_bytes()133@fixture(scope='global')134def mpeg_frame():135	return Path(TEST_PATH, 'data', 'mp3', 'mpeg-frame.bin').read_bytes()136@fixture(scope='global')137def null():138	return Path(TEST_PATH, 'data', 'null.bin').read_bytes()139@fixture(scope='global')140def vbri_header():141	return Path(TEST_PATH, 'data', 'mp3', 'vbri-header.bin').read_bytes()142@fixture(scope='global')143def vorbis_comments():144	return Path(TEST_PATH, 'data', 'vorbis_comments', 'vorbis-comments.bin').read_bytes()145@fixture(scope='global')146def xing_header_no_lame():147	return Path(TEST_PATH, 'data', 'mp3', 'xing-header-no-lame.bin').read_bytes()148@fixture(scope='global')149def xing_toc():150	return Path(TEST_PATH, 'data', 'mp3', 'xing-toc.bin').read_bytes()151@fixture(scope='global')152def wave_riff_tags_subchunk():153	return Path(TEST_PATH, 'data', 'wave', 'wave-riff-tags-subchunk.bin').read_bytes()154@fixture(scope='global')155def wave_riff_tags_data():156	return Path(TEST_PATH, 'data', 'wave', 'wave-riff-tags-subchunk.bin').read_bytes()[8:]157@fixture(scope='global')158def wave_streaminfo_subchunk():159	return Path(TEST_PATH, 'data', 'wave', 'wave-streaminfo-subchunk.bin').read_bytes()160@fixture(scope='global')161def wave_streaminfo_data():...

file_reader.py

Source:file_reader.py

1import os2import numpy34def getFileSize(path):5    try:6        size = os.path.getsize(path)7        return size8    except Exception as err:9        print(err)1011class FileReader(object):12    def __init__(self):13        self._file_name=''14        self._file_pointer=None15        self._current_bytes=016        self._max_bytes=017            18    def initialize(self, file_name, max_bytes):19        print('Opening File : %s' % file_name)20        self._file_name=file_name            21        self._file_pointer=open(file_name, 'rb')22        self._current_bytes=023        24        file_bytes=getFileSize(file_name)25        if max_bytes<file_bytes:26            self._max_bytes=max_bytes27        else:28            self._max_bytes=file_bytes        29        30    @property31    def get_file_name(self):32        return self._file_name33    34    @property35    def get_file_pointer(self):36        return self._file_pointer37    38    @property39    def get_max_bytes(self):40        return self._max_bytes41    42    @property43    def get_current_bytes(self):44        return self._current_bytes45    46    def skip_bytes(self, read_bytes):47        self._file_pointer.seek(read_bytes,1)48        print('Skipped %dGB'%(read_bytes//1024**3))49    50    def read_data(self, read_bytes, isloop, dtype):51        if read_bytes > self._max_bytes:52            read_bytes=self._max_bytes53        if self._current_bytes + read_bytes <= self._max_bytes:54            buf = self._file_pointer.read(read_bytes)55            data = numpy.frombuffer(buf, dtype=dtype)56            self._current_bytes += read_bytes57        else:58            if isloop==False:59                self._file_pointer=open(self._file_name, 'rb')60                buf = self._file_pointer.read(read_bytes)61                data = numpy.frombuffer(buf, dtype=dtype)62                self._current_bytes = read_bytes63            else:64                buf = self._file_pointer.read(self._max_bytes-self._current_bytes)65                data1 = numpy.frombuffer(buf, dtype=dtype)66                self._file_pointer=open(self._file_name, 'rb')67                buf = self._file_pointer.read(read_bytes-(self._max_bytes-self._current_bytes))68                data2 = numpy.frombuffer(buf, dtype=dtype)               69                data=numpy.concatenate((data1,data2))70                self._current_bytes = read_bytes-(self._max_bytes-self._current_bytes)71        return data72    73    def read_data_skip(self, read_bytes, isloop, dtype):74        if read_bytes > self._max_bytes:75            read_bytes=self._max_bytes76        if self._current_bytes + read_bytes <= self._max_bytes:77            self.skip_bytes(read_bytes)78            self._current_bytes += read_bytes79        else:80            if isloop==False:81                self._file_pointer=open(self._file_name, 'rb')82                self._file_pointer.read(read_bytes)83                self._current_bytes = read_bytes84            else:85                self._file_pointer=open(self._file_name, 'rb')86                self._file_pointer.read(read_bytes-(self._max_bytes-self._current_bytes))87                self._current_bytes = read_bytes-(self._max_bytes-self._current_bytes)
...

test_http.py

Source:test_http.py

1import unittest2import responses3import smart_open.http4import smart_open.s35BYTES = b'i tried so hard and got so far but in the end it doesn\'t even matter'6URL = 'http://localhost'7HEADERS = {8    'Content-Length': str(len(BYTES)),9    'Accept-Ranges': 'bytes',10}11def request_callback(request):12    try:13        range_string = request.headers['range']14    except KeyError:15        return (200, HEADERS, BYTES)16    start, end = range_string.replace('bytes=', '').split('-', 1)17    start = int(start)18    if end:19        end = int(end)20    else:21        end = len(BYTES)22    return (200, HEADERS, BYTES[start:end])23class HttpTest(unittest.TestCase):24    @responses.activate25    def test_read_all(self):26        responses.add(responses.GET, URL, body=BYTES, stream=True)27        reader = smart_open.http.SeekableBufferedInputBase(URL)28        read_bytes = reader.read()29        self.assertEqual(BYTES, read_bytes)30    @responses.activate31    def test_seek_from_start(self):32        responses.add_callback(responses.GET, URL, callback=request_callback)33        reader = smart_open.http.SeekableBufferedInputBase(URL)34        reader.seek(10)35        self.assertEqual(reader.tell(), 10)36        read_bytes = reader.read(size=10)37        self.assertEqual(reader.tell(), 20)38        self.assertEqual(BYTES[10:20], read_bytes)39        reader.seek(20)40        read_bytes = reader.read(size=10)41        self.assertEqual(BYTES[20:30], read_bytes)42        reader.seek(0)43        read_bytes = reader.read(size=10)44        self.assertEqual(BYTES[:10], read_bytes)45    @responses.activate46    def test_seek_from_current(self):47        responses.add_callback(responses.GET, URL, callback=request_callback)48        reader = smart_open.http.SeekableBufferedInputBase(URL)49        reader.seek(10)50        read_bytes = reader.read(size=10)51        self.assertEqual(BYTES[10:20], read_bytes)52        self.assertEqual(reader.tell(), 20)53        reader.seek(10, whence=smart_open.s3.CURRENT)54        self.assertEqual(reader.tell(), 30)55        read_bytes = reader.read(size=10)56        self.assertEqual(reader.tell(), 40)57        self.assertEqual(BYTES[30:40], read_bytes)58    @responses.activate59    def test_seek_from_end(self):60        responses.add_callback(responses.GET, URL, callback=request_callback)61        reader = smart_open.http.SeekableBufferedInputBase(URL)62        reader.seek(-10, whence=smart_open.s3.END)63        self.assertEqual(reader.tell(), len(BYTES) - 10)64        read_bytes = reader.read(size=10)65        self.assertEqual(reader.tell(), len(BYTES))...

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.