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

How to use readchunk method in autotest

Best Python code snippet using autotest_python

runtime.py

Source:runtime.py

1from ctypes import *2from .instructions.bytecode import BinaryR3from .instructions.bytecode import UnaryR4from .instructions.bytecode import BinaryC5from .instructions.bytecode import UnaryC6from .instructions.bytecode import SingleC7from .instructions.bytecode import SingleR8from .instructions.bytecode import JumpR9from timeit import default_timer as timer1011command_registers = [0,0]12timeout_flag = 013timeout_set = False14timeout_value = 015timestamp = 016registers = [0,0,0,0,0,0,0,0,0,0,0,0,0,0]1718class Runtime:19    def __init__ (self, filename):20        with open (filename) as f:21            self.binary = f.read ()22        self.length = int (len (self.binary))23        self.pc = 024        self.offset = 02526        # print (BinaryR.sizeInBits)27        # print (UnaryR.sizeInBits)28        # print (BinaryC.sizeInBits)29        # print (UnaryC.sizeInBits)30        # print (SingleC.sizeInBits)31        # print (SingleR.sizeInBits)32        # print (JumpR.sizeInBits)3334    def readChunk (self, size):35        ret = int (self.binary [self.offset:self.offset + size], 2)36        self.offset += size37        return ret3839    def dumpReg (self):40        global registers41        print (registers)4243    def decode (self):44        if self.pc >= self.length:45            raise SystemExit (f"Program finished.")46        instrType = self.readChunk (3)47        opcode = self.readChunk (6)48        match instrType:49            case 0:50                lRegType = self.readChunk (2)51                lReg = self.readChunk (4)52                rRegType = self.readChunk (2)53                rReg = self.readChunk (4)54                dRegType = self.readChunk (2)55                dReg = self.readChunk (4)56                self.readChunk (5)57                dispatcher [(instrType, opcode)] (lRegType, lReg, rRegType, rReg, dRegType, dReg)58            case 1:59                srcRegType = self.readChunk (2)60                srcReg = self.readChunk (4)61                destRegType = self.readChunk (2)62                destReg = self.readChunk (4)63                self.readChunk (3)64                dispatcher [(instrType, opcode)] (srcRegType, srcReg, destRegType, destReg)65            case 2:66                const = self.readChunk (32)67                srcRegType = self.readChunk (2)68                srcReg = self.readChunk (4)69                destRegType = self.readChunk (2)70                destReg = self.readChunk (4)71                self.readChunk (3)72                dispatcher [(instrType, opcode)] (const, srcRegType, srcReg, destRegType, destReg)73            case 3:74                const = self.readChunk (32)75                regType = self.readChunk (2)76                reg = self.readChunk (4)77                self.readChunk (1)78                dispatcher [(instrType, opcode)] (const, regType, reg)79            case 4:80                const = self.readChunk (32)81                self.readChunk (7)82                dispatcher [(instrType, opcode)] (const)83            case 5:84                regType = self.readChunk (2)85                reg = self.readChunk (4)86                self.readChunk (1)87                dispatcher [(instrType, opcode)] ()88            case 6:89                regType = self.readChunk (2)90                reg = self.readChunk (4)91                jDest = self.readChunk (16)92                self.readChunk (1)93                self.pc = dispatcher [(instrType, opcode)] (regType, reg, jDest, self.offset)94                self.offset = self.pc95                return96            case _:97                raise SystemExit (f"unknown type: {instrType}, {opcode}")98        self.pc = self.offset99        100101def readReg (regType, reg):102    global registers, command_registers, timeout_flag103    if regType == 0:104        return command_registers [reg]105    elif regType == 1:106        return timeout_flag107    elif regType == 2:108        return registers [reg]109110def writeReg (regType, reg, value):111    global registers, command_registers, timeout_flag112    if regType == 0:113        command_registers [reg] = value114    elif regType == 1:115        timeout_flag = value116    elif regType == 2:117        registers [reg] = value118119def execAdd (lRegType, lReg, rRegType, rReg, dRegType, dReg):120    to_write = readReg (lRegType, lReg) + readReg (rRegType, rReg)121    writeReg (dRegType, dReg, to_write)122123def execMovr (srcRegType, srcReg, destRegType, destReg):124    writeReg (destRegType, destReg, readReg (srcRegType, srcReg))125126def execUNot (srcRegType, srcReg, destRegType, destReg):127    writeReg (destRegType, destReg, int (not readReg (srcRegType, srcReg)))128129def execMovc (const, regType, reg):130    writeReg (regType, reg, const)131132def execUSubc (const, regType, reg):133    writeReg (regType, reg, -const)134135def execInvc (const, regType, reg):136    writeReg (regType, reg, ~const)137138def execNotc (const, regType, reg):139    writeReg (regType, reg, not const)140141def execTimeoutcr (const):142    global timeout_value, timeout_set, timeout_flag143    timeout_flag = 0144    timeout_set = True145    timeout_value = const146    print (f"Relative timeout set for {const}")147148def execFetch (const):149    global timeout_value, timeout_set, timeout_flag150    if timeout_set:151        start = timer ()152        writeReg (2, 1, int (input (f"Requesting telemetry {const}: ")))153        timeout_value -= (timer () - start) # Subtract the elapsed time154        if timeout_value <= 0:155            print ("Timed out!")156            timeout_flag = 1157            timeout_set = False158    else:159        timeout_set = False160        timeout_flag = 0161        writeReg (2, 1, int (input (f"Requesting telemetry {const}: ")))162163def execJmpzr (regType, reg, jDest, nextInstr):164    return jDest * 8 if readReg (regType, reg) == 0 else nextInstr165166def execJmpnzr (regType, reg, jDest, nextInstr):167    return jDest * 8 if readReg (regType, reg) != 0 else nextInstr168169def execGtc (const, srcRegType, srcReg, destRegType, destReg):170    to_write = readReg (srcRegType, srcReg) > const171    writeReg (destRegType, destReg, to_write)172173def execLtc (const, srcRegType, srcReg, destRegType, destReg):174    to_write = readReg (srcRegType, srcReg) < const175    writeReg (destRegType, destReg, to_write)176177def execShiftcl (const, srcRegType, srcReg, destRegType, destReg):178    to_write = readReg (srcRegType, srcReg) << const179    writeReg (destRegType, destReg, to_write)180181def rshiftHelp (val, n): return (val % 0x100000000) >> n182183def execShiftcrl (const, srcRegType, srcReg, destRegType, destReg):184    to_write = rshiftHelp (readReg (srcRegType, srcReg), const)185    writeReg (destRegType, destReg, to_write)186187def execOrrb (lRegType, lReg, rRegType, rReg, dRegType, dReg):188    to_write = readReg (lRegType, lReg) | readReg (rRegType, rReg)189    writeReg (dRegType, dReg, to_write)190191def execSend (const):192    cmd = c_uint32 (readReg (0, 0))193    cmd = f"{cmd.value:0{32}b}"194    print (f"Sending command {const}: {cmd}")195196dispatcher = {197    (0, 0) : execAdd,198    (0, 1) : execAdd, 199    (0, 2) : execAdd,200    (0, 41) : execOrrb,201    (1, 0) : execMovr,202    (1, 4) : execUNot,203    (2, 34) : execGtc,204    (2, 37) : execLtc,205    (2, 42) : execShiftcl,206    (2, 43) : execShiftcrl,207    (3, 0) : execMovc,208    (3, 1) : execUSubc,209    (3, 2) : execUSubc,210    (3, 3) : execInvc,211    (3, 4) : execNotc,212    (4, 2) : execTimeoutcr,213    (4, 4) : execFetch,214    (4, 6) : execSend,215    (6, 0) : execJmpzr,216    (6, 1) : execJmpnzr
...

_entityreader.py

Source:_entityreader.py

...16        elif isinstance(entity_size, int):17            self.__class__ = SimpleEntityReader18        else:19            raise ValueError('invalid entity_size %r' % entity_size)20    def readchunk(self):21        """Read a chunk of data from the reader's file object, of whatever22        size it makes sense for this subtype of EntityReader.23        24        Returns a tuple of (transfer_decoded_data, raw_data)"""25        raise NotImplementedError26    def __iter__(self):27        while True:28            data, raw_data = chunk = self.readchunk()29            if not raw_data:30                raise StopIteration31            yield chunk32    33#======================================================================34class SimpleEntityReader(EntityReader):35    """:class:`EntityReader` subclass for HTTP entities whose size is36    accurately reflected by the :mailheader:`Content-Length` header. """37    _pos = 038    read_size = 409639    def readchunk(self):40        EntityReader.readchunk.__doc__41        size_left = self._size - self._pos42        this_read_size = min(self.read_size, size_left)43        44        assert this_read_size >= 0, "over-read the data stream!"45        if this_read_size == 0:46            return '', ''47        48        raw_data = self._fileobj.read(this_read_size)49        if not raw_data:50            msg = "data stream ended at %r with %r bytes remaining" % (51                    self._pos, size_left)52            raise exc.EntityReadError(msg, raw_data)53        self._pos += len(raw_data)54        return raw_data, raw_data55#======================================================================56class TilCloseEntityReader(EntityReader):57    """:class:`EntityReader` subclass for HTTP entities who should be read58    until the end of file, or the socket closes."""59    read_size = 409660    def readchunk(self):61        EntityReader.readchunk.__doc__62        raw_data = self._fileobj.read(self.read_size)63        if not raw_data:64            return '', ''65        66        return raw_data, raw_data67#======================================================================68class ChunkedEntityReader(EntityReader):69    """:class:`EntityReader` subclass for HTTP entities transfered via the 70    *chunked* :mailheader:`Transfer-Encoding`."""71    finished = False72    73    def _verify_read(self, check_data, msg, raw_data):74        if not check_data:75            msg = 'no data read during %r' % msg76            raise exc.EntityReadError(msg, raw_data)77    def readchunk(self):78        EntityReader.readchunk.__doc__79        # we're expecting a chunk to consist of the following:80        # 1. the size of the chunk in hex, followed by '\r\n'81        # 2. the chunk data, of exactly the size specified in (1)82        # 3. and end-chunk separator, consisting exactly of '\r\n'83        #84        # The last chunk there is to read will be the zero chunk (ie, the85        # size will be zero, with no chunk data).86        # did we previously read the zero chunk?87        if self.finished:88            return '', ''89        # read the chunk size from the fileobj90        raw_data = self._fileobj.readline()91        self._verify_read(raw_data, 'read chunk size', raw_data)92        try:93            size = int(raw_data.strip(),16)94        except ValueError:95            msg = 'invalid chunk size %r' % raw_data.strip()96            raise exc.EntityReadError(msg, raw_data)97        # read the chunk data98        if size == 0:99            chunk_data = ''100            self.finished = True101        else:102            chunk_data = self._fileobj.read(size)103            self._verify_read(chunk_data, 'read chunk data', raw_data)104            if len(chunk_data) != size:105                msg = 'expected chunk of size %r, but size was %r' % (106                        size, len(chunk_data))107                raise exc.EntityReadError(msg, chunk_data)108            raw_data += chunk_data109        # read the end separator110        sep = self._fileobj.readline()111        self._verify_read(sep, 'read chunk separator', raw_data)112        if sep != '\r\n':113            msg = r"end separator wrong; expected '\r\n', found %r" % sep114            raise exc.EntityReadError(msg, raw_data+sep)115        raw_data += sep116        117        return chunk_data, raw_data118#======================================================================119class MultipartEntityReader(EntityReader):120    """:class:`EntityReader` subclass...  Not implemented yet."""121    122    123    def readchunk(self):124        EntityReader.readchunk.__doc__125        raise NotImplementedError126#======================================================================127class NullEntityReader(EntityReader):128    """:class:`EntityReader` subclass for HTTP entities of zero length."""129    def readchunk(self):130        EntityReader.readchunk.__doc__...

integer_to_english_words.py

Source:integer_to_english_words.py

1'''2Convert a non-negative integer to its english words representation. Given input is guaranteed to be less than 2^31 - 1.3For example,4123 -> "One Hundred Twenty Three"512345 -> "Twelve Thousand Three Hundred Forty Five"61234567 -> "One Million Two Hundred Thirty Four Thousand Five Hundred Sixty Seven"7Hint:8Did you see a pattern in dividing the number into chunk of words? For example, 123 and 123000.9Group the number by thousands (3 digits). You can write a helper function that takes a number less than 1000 and convert just that chunk to words.10There are many edge cases. What are some good test cases? Does your code work with input such as 0? Or 1000010? (middle chunk is zero and should not be printed out)11'''12class Solution(object):13    base = ["", "Thousand", "Million", "Billion"]14    digit = ["", "One", "Two", "Three", "Four", "Five", "Six", "Seven", "Eight", "Nine", "Ten", "Eleven", "Twelve", "Thirteen", "Fourteen", "Fifteen", "Sixteen", "Seventeen", "Eighteen", "Nineteen"]15    tens = ["", "", "Twenty", "Thirty", "Forty", "Fifty", "Sixty", "Seventy", "Eighty", "Ninety"]16    def readChunk(self, num):17        output = list()18        if num >= 100:19            output.append(Solution.digit[num / 100])20            output.append("Hundred")21        num = num % 10022        if num < 20:23            output.append(Solution.digit[num])24        else:25            output.append(Solution.tens[num / 10])26            output.append(Solution.digit[num % 10])27        return " ".join(output)28    def numberToWords(self, num):29        """30        :type num: int31        :rtype: str32        """33        if num == 0:34            return "Zero"35        output = list()36        count = 037        while num > 0:38            chunk = num % 100039            if chunk > 0:40                output.append(self.readChunk(chunk).strip() + " " + Solution.base[count])41            num /= 100042            count += 143        return " ".join(output[::-1]).strip()44solution = Solution()45print solution.readChunk(21)46print solution.readChunk(11)47print solution.readChunk(8)48print solution.readChunk(111)49print solution.readChunk(191)50print solution.readChunk(90)51print solution.numberToWords(121)52print solution.numberToWords(1000)53print solution.numberToWords(1119)...

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.