Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use get_full_text_result method in autotest

Best Python code snippet using autotest_python

virtio_console.py

Source:virtio_console.py

...119                status = "PASS"120            else:121                status = "FAIL"122            return ("Subtest (%s): --> %s") % (result[1], status)123        def get_full_text_result(self):124            """125            @return string with text form of result126            """127            result = ""128            for res in self.result:129                result += self.result_to_string_debug(res) + "\n"130            return result131        def get_text_result(self):132            """133            @return string with text form of result134            """135            result = ""136            for res in self.result:137                result += self.result_to_string(res) + "\n"138            return result139    class Port(object):140        """141        Define structure to keep information about used port.142        """143        def __init__(self, sock, name, port_type, path):144            """145            @param vm: virtual machine object that port owned146            @param sock: Socket of port if port is open.147            @param name: Name of port for guest side.148            @param port_type: Type of port yes = console, no= serialport.149            @param path: Path to port on host side.150            """151            self.sock = sock152            self.name = name153            self.port_type = port_type154            self.path = path155            self.is_open = False156        def for_guest(self):157            """158            Format data for communication with guest side.159            """160            return [self.name, self.port_type]161        def open(self):162            """163            Open port on host side.164            """165            self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)166            self.sock.connect(self.path)167            self.is_open = True168        def clean_port(self):169            """170            Clean all data from opened port on host side.171            """172            if self.is_open:173                self.close()174            self.open()175            ret = select.select([self.sock], [], [], 1.0)176            if ret[0]:177                buf = self.sock.recv(1024)178                logging.debug("Rest in socket: " + buf)179        def close(self):180            """181            Close port.182            """183            self.sock.shutdown(socket.SHUT_RDWR)184            self.sock.close()185            self.is_open = False186        def __str__(self):187            """188            Convert to text.189            """190            return ("%s,%s,%s,%s,%d" % ("Socket", self.name, self.port_type,191                                        self.path, self.is_open))192    class ThSend(Thread):193        """194        Random data sender thread.195        """196        def __init__(self, port, data, event):197            """198            @param port: Destination port.199            @param data: The data intend to be send in a loop.200            @param event: Exit event.201            """202            Thread.__init__(self)203            self.port = port204            # FIXME: socket.send(data>>127998) without read blocks thread205            if len(data) > 102400:206                data = data[0:102400]207                logging.error("Data is too long, using only first %d bytes",208                              len(data))209            self.data = data210            self.exitevent = event211            self.idx = 0212        def run(self):213            logging.debug("ThSend %s: run", self.getName())214            while not self.exitevent.isSet():215                self.idx += self.port.send(self.data)216            logging.debug("ThSend %s: exit(%d)", self.getName(),217                          self.idx)218    class ThSendCheck(Thread):219        """220        Random data sender thread.221        """222        def __init__(self, port, event, queues, blocklen=1024):223            """224            @param port: Destination port225            @param event: Exit event226            @param queues: Queues for the control data (FIFOs)227            @param blocklen: Block length228            """229            Thread.__init__(self)230            self.port = port231            self.queues = queues232            # FIXME: socket.send(data>>127998) without read blocks thread233            if blocklen > 102400:234                blocklen = 102400235                logging.error("Data is too long, using blocklen = %d",236                              blocklen)237            self.blocklen = blocklen238            self.exitevent = event239            self.idx = 0240        def run(self):241            logging.debug("ThSendCheck %s: run", self.getName())242            too_much_data = False243            while not self.exitevent.isSet():244                # FIXME: workaround the problem with qemu-kvm stall when too245                # much data is sent without receiving246                for queue in self.queues:247                    while not self.exitevent.isSet() and len(queue) > 1048576:248                        too_much_data = True249                        time.sleep(0.1)250                ret = select.select([], [self.port], [], 1.0)251                if ret[1]:252                    # Generate blocklen of random data add them to the FIFO253                    # and send them over virtio_console254                    buf = ""255                    for i in range(self.blocklen):256                        ch = "%c" % random.randrange(255)257                        buf += ch258                        for queue in self.queues:259                            queue.append(ch)260                    target = self.idx + self.blocklen261                    while not self.exitevent.isSet() and self.idx < target:262                        idx = self.port.send(buf)263                        buf = buf[idx:]264                        self.idx += idx265            logging.debug("ThSendCheck %s: exit(%d)", self.getName(),266                          self.idx)267            if too_much_data:268                logging.error("ThSendCheck: workaround the 'too_much_data'"269                              "bug")270    class ThRecv(Thread):271        """272        Recieves data and throws it away.273        """274        def __init__(self, port, event, blocklen=1024):275            """276            @param port: Data source port.277            @param event: Exit event.278            @param blocklen: Block length.279            """280            Thread.__init__(self)281            self.port = port282            self._port_timeout = self.port.gettimeout()283            self.port.settimeout(0.1)284            self.exitevent = event285            self.blocklen = blocklen286            self.idx = 0287        def run(self):288            logging.debug("ThRecv %s: run", self.getName())289            while not self.exitevent.isSet():290                # TODO: Workaround, it didn't work with select :-/291                try:292                    self.idx += len(self.port.recv(self.blocklen))293                except socket.timeout:294                    pass295            self.port.settimeout(self._port_timeout)296            logging.debug("ThRecv %s: exit(%d)", self.getName(), self.idx)297    class ThRecvCheck(Thread):298        """299        Random data receiver/checker thread.300        """301        def __init__(self, port, buffer, event, blocklen=1024):302            """303            @param port: Source port.304            @param buffer: Control data buffer (FIFO).305            @param length: Amount of data we want to receive.306            @param blocklen: Block length.307            """308            Thread.__init__(self)309            self.port = port310            self.buffer = buffer311            self.exitevent = event312            self.blocklen = blocklen313            self.idx = 0314        def run(self):315            logging.debug("ThRecvCheck %s: run", self.getName())316            while not self.exitevent.isSet():317                ret = select.select([self.port], [], [], 1.0)318                if ret[0] and (not self.exitevent.isSet()):319                    buf = self.port.recv(self.blocklen)320                    if buf:321                        # Compare the recvd data with the control data322                        for ch in buf:323                            ch_ = self.buffer.popleft()324                            if not ch == ch_:325                                self.exitevent.set()326                                logging.error("Failed to recv %dth character",327                                              self.idx)328                                logging.error("%s != %s", repr(ch), repr(ch_))329                                logging.error("Recv = %s", repr(buf))330                                # sender might change the buffer :-(331                                time.sleep(1)332                                ch_ = ""333                                for buf in self.buffer:334                                    ch_ += buf335                                logging.error("Queue = %s", repr(ch_))336                                raise error.TestFail("ThRecvCheck: incorrect "337                                                     "data")338                        self.idx += len(buf)339            logging.debug("ThRecvCheck %s: exit(%d)", self.getName(),340                          self.idx)341    def process_stats(stats, scale=1.0):342        """343        Process and print the statistic.344        @param stats: List of measured data.345        """346        if not stats:347            return None348        for i in range((len(stats) - 1), 0, -1):349            stats[i] = stats[i] - stats[i - 1]350            stats[i] /= scale351        stats[0] /= scale352        stats = sorted(stats)353        return stats354    def _init_guest(vm, timeout=2):355        """356        Execute virtio_guest.py on guest, wait until it is initialized.357        @param vm: Informations about the guest.358        @param timeout: Timeout that will be used to verify if the script359                started properly.360        """361        logging.debug("compile virtio_guest.py on guest %s", vm[0].name)362        vm[1].sendline("python -OO /tmp/virtio_guest.py -c &&"363                       "echo -n 'PASS: Compile virtio_guest finished' ||"364                       "echo -n 'FAIL: Compile virtio_guest failed'")365        (match, data) = vm[1].read_until_last_line_matches(["PASS:", "FAIL:"],366                                                           timeout)367        if match != 0:368            raise error.TestFail("Command console_switch.py on guest %s failed."369                                 "\nreturn code: %s\n output:\n%s" %370                                 (vm[0].name, match, data))371        logging.debug("Starting virtio_guest.py on guest %s", vm[0].name)372        vm[1].sendline("python /tmp/virtio_guest.pyo &&"373                       "echo -n 'PASS: virtio_guest finished' ||"374                       "echo -n 'FAIL: virtio_guest failed'")375        (match, data) = vm[1].read_until_last_line_matches(["PASS:", "FAIL:"],376                                                           timeout)377        if match != 0:378            raise error.TestFail("Command console_switch.py on guest %s failed."379                                 "\nreturn code: %s\n output:\n%s" %380                                 (vm[0].name, match, data))381        # Let the system rest382        time.sleep(2)383    def init_guest(vm, consoles):384        """385        Prepares guest, executes virtio_guest.py and initialize for testing386        @param vm: Informations about the guest.387        @param consoles: Informations about consoles388        """389        conss = []390        for mode in consoles:391            for cons in mode:392                conss.append(cons.for_guest())393        _init_guest(vm, 10)394        on_guest("virt.init(%s)" % (conss), vm, 10)395    def _on_guest(command, vm, timeout=2):396        """397        Execute given command inside the script's main loop, indicating the vm398        the command was executed on.399        @param command: Command that will be executed.400        @param vm: Informations about the guest.401        @param timeout: Timeout used to verify expected output.402        @return: Tuple (match index, data)403        """404        logging.debug("Executing '%s' on virtio_guest.py loop, vm: %s," +405                      "timeout: %s", command, vm[0].name, timeout)406        vm[1].sendline(command)407        try:408            (match, data) = vm[1].read_until_last_line_matches(["PASS:",409                                                                "FAIL:"],410                                                               timeout)411        except (kvm_subprocess.ExpectTimeoutError):412            match = None413            data = None414        return (match, data)415    def on_guest(command, vm, timeout=2):416        """417        Wrapper around the _on_guest command which executes the command on418        guest. Unlike _on_guest command when the command fails it raises the419        test error.420        @param command: Command that will be executed.421        @param vm: Informations about the guest.422        @param timeout: Timeout used to verify expected output.423        @return: Tuple (match index, data)424        """425        match, data = _on_guest(command, vm, timeout)426        if match == 1 or match is None:427            raise error.TestFail("Failed to execute '%s' on virtio_guest.py, "428                                 "vm: %s, output:\n%s" %429                                 (command, vm[0].name, data))430        return (match, data)431    def _guest_exit_threads(vm, send_pts, recv_pts):432        """433        Safely executes on_guest("virt.exit_threads()") using workaround of434        the stuck thread in loopback in mode=virt.LOOP_NONE .435        @param vm: Informations about the guest.436        @param send_pts: list of possible send sockets we need to work around.437        @param recv_pts: list of possible recv sockets we need to read-out.438        """439        # in LOOP_NONE mode it might stuck in read/write440        match, tmp = _on_guest("virt.exit_threads()", vm, 10)441        if match == None:442            logging.debug("Workaround the stuck thread on guest")443            # Thread is stucked in read/write444            for send_pt in send_pts:445                send_pt.sock.sendall(".")446        elif match != 0:447            # Something else448            raise error.TestFail("Unexpected fail\nMatch: %s\nData:\n%s"449                                 % (match, tmp))450        # Read-out all remaining data451        for recv_pt in recv_pts:452            while select.select([recv_pt.sock], [], [], 0.1)[0]:453                recv_pt.sock.recv(1024)454        # This will cause fail in case anything went wrong.455        on_guest("print 'PASS: nothing'", vm, 10)456    def _vm_create(no_console=3, no_serialport=3):457        """458        Creates the VM and connects the specified number of consoles and serial459        ports.460        Ports are allocated by 2 per 1 virtio-serial-pci device starting with461        console. (3+2 => CC|CS|S; 0+2 => SS; 3+4 => CC|CS|SS|S, ...) This way462        it's easy to test communication on the same or different463        virtio-serial-pci device.464        Further in tests the consoles are being picked always from the first465        available one (3+2: 2xC => CC|cs|s <communication on the same PCI>;466        2xC,1xS => CC|cS|s <communication between 2 PCI devs)467        @param no_console: Number of desired virtconsoles.468        @param no_serialport: Number of desired virtserialports.469        @return: Tuple with (guest information, consoles information)470                guest informations = [vm, session, tmp_dir]471                consoles informations = [consoles[], serialports[]]472        """473        consoles = []474        serialports = []475        tmp_dir = tempfile.mkdtemp(prefix="virtio-console-", dir="/tmp/")476        if not params.get('extra_params'):477            params['extra_params'] = ''478        for i in range(0, no_console):479            # Spread consoles between multiple PCI devices (2 per a dev)480            if not i % 2:481                pci = "virtio-serial-pci%d" % (i / 2)482                params['extra_params'] += (" -device virtio-serial-pci,id="483                                           + pci)484                pci += ".0"485            params['extra_params'] += (" -chardev socket,path=%s/%d,id=vc%d,"486                                       "server,nowait" % (tmp_dir, i, i))487            params['extra_params'] += (" -device virtconsole,chardev=vc%d,"488                                      "name=console-%d,id=c%d,bus=%s"489                                      % (i, i, i, pci))490        for i in  range(no_console, no_console + no_serialport):491            # Spread seroal ports between multiple PCI devices (2 per a dev)492            if not i % 2:493                pci = "virtio-serial-pci%d" % (i / 2)494                params['extra_params'] += (" -device virtio-serial-pci,id="495                                           + pci)496                pci += ".0"497            params['extra_params'] += (" -chardev socket,path=%s/%d,id=vs%d,"498                                       "server,nowait" % (tmp_dir, i, i))499            params['extra_params'] += (" -device virtserialport,chardev=vs%d,"500                                       "name=serialport-%d,id=p%d,bus=%s"501                                       % (i, i, i, pci))502        logging.debug("Booting first guest %s", params.get("main_vm"))503        kvm_preprocessing.preprocess_vm(test, params, env,504                                        params.get("main_vm"))505        vm = kvm_utils.env_get_vm(env, params.get("main_vm"))506        session = kvm_test_utils.wait_for_login(vm, 0,507                                         float(params.get("boot_timeout", 240)),508                                         0, 2)509        # connect the sockets510        for i in range(0, no_console):511            consoles.append(Port(None ,"console-%d" % i,512                                 "yes", "%s/%d" % (tmp_dir, i)))513        for i in range(no_console, no_console + no_serialport):514            serialports.append(Port(None ,"serialport-%d" % i,515                                    "no", "%s/%d" % (tmp_dir, i)))516        return [vm, session, tmp_dir], [consoles, serialports]517    def topen(vm, port):518        """519        Open virtioconsole port.520        @param vm: Target virtual machine [vm, session, tmp_dir].521        @param port: Port identifier.522        """523        on_guest("virt.open('%s')" % (port.name), vm, 2)524        port.open()525    def tmulti_open(vm, port):526        """527        Multiopen virtioconsole port.528        @param vm: Target virtual machine [vm, session, tmp_dir].529        @param port: Port identifier.530        """531        on_guest("virt.close('%s')" % (port.name), vm, 2)532        on_guest("virt.open('%s')" % (port.name), vm, 2)533        match = _on_guest("virt.open('%s')" % (port.name), vm, 2)[0]534        # Console is permitted to open the device multiple times535        if match != 0 and port.port_type == "yes":536            raise error.TestFail("Unexpected fail of openning the console"537                                 " device for the 2nd time.")538        # Serial port is forbidden to open the device multiple times539        elif match != 1 and port.port_type == "no":540            raise error.TestFail("Unexpetded pass of oppening the serialport"541                                 " device for the 2nd time.")542        port.open()543    def tclose(vm, port):544        """545        Close socket.546        @param vm: Target virtual machine [vm, session, tmp_dir].547        @param port: Port to open.548        """549        on_guest("virt.close('%s')" % (port.name), vm, 2)550        port.close()551    def tpooling(vm, port):552        """553        Test try pooling function.554        @param vm: Target virtual machine [vm, session, tmp_dir].555        @param port: Port used in test.556        """557        # Poll (OUT)558        on_guest("virt.poll('%s', %s)" % (port.name, select.POLLOUT), vm,559                 2)560        # Poll (IN, OUT)561        port.sock.sendall("test")562        for test in [select.POLLIN, select.POLLOUT]:563            on_guest("virt.poll('%s', %s)" % (port.name, test), vm, 2)564        # Poll (IN HUP)565        # I store the socket informations and close the socket566        port.close()567        for test in [select.POLLIN, select.POLLHUP]:568            on_guest("virt.poll('%s', %s)" % (port.name, test), vm, 2)569        # Poll (HUP)570        on_guest("virt.recv('%s', 4, 1024, False)" % (port.name), vm, 2)571        on_guest("virt.poll('%s', %s)" % (port.name, select.POLLHUP), vm,572                 2)573        # Reconnect the socket574        port.open()575        # Redefine socket in consoles576        on_guest("virt.poll('%s', %s)" % (port.name, select.POLLOUT), vm,577                 2)578    def tsigio(vm, port):579        """580        Test try sigio function.581        @param vm: Target virtual machine [vm, session, tmp_dir].582        @param port: Port used in test.583        """584        if port.is_open:585            port.close()586        # Enable sigio on specific port587        on_guest("virt.async('%s', True, 0)" %588                 (port.name) , vm, 2)589        on_guest("virt.get_sigio_poll_return('%s')" % (port.name) , vm, 2)590        #Test sigio when port open591        on_guest("virt.set_pool_want_return('%s', select.POLLOUT)" %592                 (port.name), vm, 2)593        port.open()594        match = _on_guest("virt.get_sigio_poll_return('%s')" %595                          (port.name) , vm, 2)[0]596        if match == 1:597            raise error.TestFail("Problem with HUP on console port.")598        #Test sigio when port receive data599        on_guest("virt.set_pool_want_return('%s', select.POLLOUT |"600                 " select.POLLIN)" % (port.name), vm, 2)601        port.sock.sendall("0123456789")602        on_guest("virt.get_sigio_poll_return('%s')" % (port.name) , vm, 2)603        #Test sigio port close event604        on_guest("virt.set_pool_want_return('%s', select.POLLHUP |"605                 " select.POLLIN)" % (port.name), vm, 2)606        port.close()607        on_guest("virt.get_sigio_poll_return('%s')" % (port.name) , vm, 2)608        #Test sigio port open event and persistence of written data on port.609        on_guest("virt.set_pool_want_return('%s', select.POLLOUT |"610                 " select.POLLIN)" % (port.name), vm, 2)611        port.open()612        on_guest("virt.get_sigio_poll_return('%s')" % (port.name) , vm, 2)613        #Test event when erase data.614        on_guest("virt.clean_port('%s')" % (port.name), vm, 2)615        port.close()616        on_guest("virt.set_pool_want_return('%s', select.POLLOUT)"617                 % (port.name), vm, 2)618        port.open()619        on_guest("virt.get_sigio_poll_return('%s')" % (port.name) , vm, 2)620        # Disable sigio on specific port621        on_guest("virt.async('%s', False, 0)" %622                 (port.name) , vm, 2)623    def tlseek(vm, port):624        """625        Tests the correct handling of lseek (expected fail)626        @param vm: Target virtual machine [vm, session, tmp_dir].627        @param port: Port used in test.628        """629        # The virt.lseek returns PASS when the seek fails630        on_guest("virt.lseek('%s', 0, 0)" % (port.name), vm, 2)631    def trw_host_offline(vm, port):632        """633        Guest read/write from host when host is disconnected.634        @param vm: Target virtual machine [vm, session, tmp_dir].635        @param port: Port used in test.636        """637        if port.is_open:638            port.close()639        # Read should pass640        # FIXME: Console doesn't support polling (POLLHUP without host shoots641        # the guest OS)642        if port.port_type != "yes":643            on_guest("virt.recv('%s', 0, 1024, False)" % port.name, vm, 2)644        else: # Skip this test on virtio-console port645            raise error.TestFail("This test is unable to run with virtio"646                                 " console. Please see the in_code FIXME"647                                 " note for more info.\n"648                                 "Skipping this round.")649        # Write should timed-out650        match, tmp = _on_guest("virt.send('%s', 10, False)"651                                % port.name, vm, 2)652        if match != None:653            raise error.TestFail("Write on guest while host disconnected "654                                 "didn't timed out.\nOutput:\n%s"655                                 % tmp)656        port.open()657        if (port.sock.recv(1024) < 10):658            raise error.TestFail("Didn't received data from guest")659        # Now the _on_guest("virt.send('%s'... command should be finished660        on_guest("print 'PASS: nothing'", vm, 2)661    def trw_blocking_mode(vm, port):662        """663        Guest read\write data in blocking mode.664        @param vm: Target virtual machine [vm, session, tmp_dir].665        @param port: Port used in test.666        """667        # Blocking mode668        if not port.is_open:669            port.open()670        on_guest("virt.blocking('%s', True)" % port.name, vm, 2)671        # Recv should timed out672        match, tmp = _on_guest("virt.recv('%s', 10, 1024, False)" %673                               port.name, vm, 2)674        if match == 0:675            raise error.TestFail("Received data even when non were sent\n"676                                 "Data:\n%s" % tmp)677        elif match != None:678            raise error.TestFail("Unexpected fail\nMatch: %s\nData:\n%s" %679                                 (match, tmp))680        port.sock.sendall("1234567890")681        # Now guest received the data end escaped from the recv()682        on_guest("print 'PASS: nothing'", vm, 2)683    def trw_nonblocking_mode(vm, port):684        """685        Guest read\write data in nonblocking mode.686        @param vm: Target virtual machine [vm, session, tmp_dir].687        @param port: Port used in test.688        """689        # Non-blocking mode690        if not port.is_open:691            port.open()692        on_guest("virt.blocking('%s', False)" % port.name, vm, 2)693        # Recv should return FAIL with 0 received data694        match, tmp = _on_guest("virt.recv('%s', 10, 1024, False)" %695                              port.name, vm, 2)696        if match == 0:697            raise error.TestFail("Received data even when non were sent\n"698                                 "Data:\n%s" % tmp)699        elif match == None:700            raise error.TestFail("Timed out, probably in blocking mode\n"701                                 "Data:\n%s" % tmp)702        elif match != 1:703            raise error.TestFail("Unexpected fail\nMatch: %s\nData:\n%s" %704                                 (match, tmp))705        port.sock.sendall("1234567890")706        on_guest("virt.recv('%s', 10, 1024, False)" % port.name, vm, 2)707    def tbasic_loopback(vm, send_port, recv_port, data="Smoke test data"):708        """709        Easy loop back test with loop over only two port.710        @param vm: Target virtual machine [vm, session, tmp_dir].711        @param port: Port used in test.712        """713        if not send_port.is_open:714            send_port.open()715        if not recv_port.is_open:716            recv_port.open()717        on_guest("virt.loopback(['%s'], ['%s'], 1024, virt.LOOP_NONE)" %718                     (send_port.name, recv_port.name), vm, 2)719        send_port.sock.sendall(data)720        tmp = ""721        i = 0722        while i <= 10:723            i += 1724            ret = select.select([recv_port.sock], [], [], 1.0)725            if ret:726                tmp += recv_port.sock.recv(1024)727            if len(tmp) >= len(data):728                break729        if tmp != data:730            raise error.TestFail("Incorrect data: '%s' != '%s'",731                                 data, tmp)732        _guest_exit_threads(vm, [send_port], [recv_port])733    def test_smoke(test, vm, consoles, params):734        """735        Virtio console smoke test.736        Tests the basic functionalities (poll, read/write with and without737        connected host, etc.738        @param vm: Target virtual machine [vm, session, tmp_dir].739        @param consoles: Field of virtio ports with the minimum of 2 items.740        @param params: Test parameters '$console_type:$data;...'741        """742        # PREPARE743        for param in params.split(';'):744            if not param:745                continue746            logging.info("test_smoke: params: %s", param)747            param = param.split(':')748            if len(param) > 1:749                data = param[1]750            else:751                data = "Smoke test data"752            param = (param[0] == 'serialport')753            send_pt = consoles[param][0]754            recv_pt = consoles[param][1]755            test.do_test(topen, [vm, send_pt] , True)756            test.do_test(tclose, [vm, send_pt], True)757            test.do_test(tmulti_open, [vm, send_pt])758            test.do_test(tpooling, [vm, send_pt])759            test.do_test(tsigio, [vm, send_pt])760            test.do_test(tlseek, [vm, send_pt])761            test.do_test(trw_host_offline, [vm, send_pt])762            test.do_test(trw_nonblocking_mode, [vm, send_pt])763            test.do_test(trw_blocking_mode, [vm, send_pt])764            test.do_test(tbasic_loopback, [vm, send_pt, recv_pt, data], True)765    def tloopback(vm, consoles, params):766        """767        Virtio console loopback test.768        Creates loopback on the vm machine between send_pt and recv_pts769        ports and sends length amount of data through this connection.770        It validates the correctness of the data sent.771        @param vm: Target virtual machine [vm, session, tmp_dir].772        @param consoles: Field of virtio ports with the minimum of 2 items.773        @param params: test parameters, multiple recievers allowed.774            '$source_console_type@buffer_length:775             $destination_console_type1@$buffer_length:...:776             $loopback_buffer_length;...'777        """778        # PREPARE779        for param in params.split(';'):780            if not param:781                continue782            logging.info("test_loopback: params: %s", param)783            param = param.split(':')784            idx_serialport = 0785            idx_console = 0786            buf_len = []787            if (param[0].startswith('console')):788                send_pt = consoles[0][idx_console]789                idx_console += 1790            else:791                send_pt = consoles[1][idx_serialport]792                idx_serialport += 1793            if (len(param[0].split('@')) == 2):794                buf_len.append(int(param[0].split('@')[1]))795            else:796                buf_len.append(1024)797            recv_pts = []798            for parm in param[1:]:799                if (parm.isdigit()):800                    buf_len.append(int(parm))801                    break   # buf_len is the last portion of param802                if (parm.startswith('console')):803                    recv_pts.append(consoles[0][idx_console])804                    idx_console += 1805                else:806                    recv_pts.append(consoles[1][idx_serialport])807                    idx_serialport += 1808                if (len(parm[0].split('@')) == 2):809                    buf_len.append(int(parm[0].split('@')[1]))810                else:811                    buf_len.append(1024)812            # There must be sum(idx_*) consoles + last item as loopback buf_len813            if len(buf_len) == (idx_console + idx_serialport):814                buf_len.append(1024)815            for p in recv_pts:816                if not p.is_open:817                    p.open()818            if not send_pt.is_open:819                send_pt.open()820            if len(recv_pts) == 0:821                raise error.TestFail("test_loopback: incorrect recv consoles"822                                     "definition")823            threads = []824            queues = []825            for i in range(0, len(recv_pts)):826                queues.append(deque())827            tmp = "'%s'" % recv_pts[0].name828            for recv_pt in recv_pts[1:]:829                tmp += ", '%s'" % (recv_pt.name)830            on_guest("virt.loopback(['%s'], [%s], %d, virt.LOOP_POLL)"831                     % (send_pt.name, tmp, buf_len[-1]), vm, 2)832            exit_event = threading.Event()833            # TEST834            thread = ThSendCheck(send_pt.sock, exit_event, queues,835                                   buf_len[0])836            thread.start()837            threads.append(thread)838            for i in range(len(recv_pts)):839                thread = ThRecvCheck(recv_pts[i].sock, queues[i], exit_event,840                                       buf_len[i + 1])841                thread.start()842                threads.append(thread)843            time.sleep(60)844            exit_event.set()845            threads[0].join()846            tmp = "%d data sent; " % threads[0].idx847            for thread in threads[1:]:848                thread.join()849                tmp += "%d, " % thread.idx850            logging.info("test_loopback: %s data received and verified",851                         tmp[:-2])852            # Read-out all remaining data853            for recv_pt in recv_pts:854                while select.select([recv_pt.sock], [], [], 0.1)[0]:855                    recv_pt.sock.recv(1024)856            _guest_exit_threads(vm, [send_pt], recv_pts)857            del exit_event858            del threads[:]859    def tperf(vm, consoles, params):860        """861        Tests performance of the virtio_console tunel. First it sends the data862        from host to guest and than back. It provides informations about863        computer utilisation and statistic informations about the troughput.864        @param vm: Target virtual machine [vm, session, tmp_dir].865        @param consoles: Field of virtio ports with the minimum of 2 items.866        @param params: test parameters:867                '$console_type@$buffer_length:$test_duration;...'868        """869        for param in params.split(';'):870            if not param:871                continue872            logging.info("test_perf: params: %s", param)873            param = param.split(':')874            duration = 60.0875            if len(param) > 1:876                try:877                    duration = float(param[1])878                except:879                    pass880            param = param[0].split('@')881            if len(param) > 1 and param[1].isdigit():882                buf_len = int(param[1])883            else:884                buf_len = 1024885            param = (param[0] == 'serialport')886            port = consoles[param][0]887            if not port.is_open:888                port.open()889            data = ""890            for i in range(buf_len):891                data += "%c" % random.randrange(255)892            exit_event = threading.Event()893            slice = float(duration) / 100894            # HOST -> GUEST895            on_guest('virt.loopback(["%s"], [], %d, virt.LOOP_NONE)' %896                     (port.name, buf_len), vm, 2)897            thread = ThSend(port.sock, data, exit_event)898            stats = array.array('f', [])899            loads = utils.SystemLoad([(os.getpid(), 'autotest'),900                                      (vm[0].get_pid(), 'VM'), 0])901            loads.start()902            _time = time.time()903            thread.start()904            for i in range(100):905                stats.append(thread.idx)906                time.sleep(slice)907            _time = time.time() - _time - duration908            logging.info("\n" + loads.get_cpu_status_string()[:-1])909            logging.info("\n" + loads.get_mem_status_string()[:-1])910            exit_event.set()911            thread.join()912            # Let the guest read-out all the remaining data913            while not _on_guest("virt.poll('%s', %s)" %914                                (port.name, select.POLLIN), vm, 2)[0]:915                time.sleep(1)916            _guest_exit_threads(vm, [port], [])917            if (_time > slice):918                logging.error(919                "Test ran %fs longer which is more than one slice", _time)920            else:921                logging.debug("Test ran %fs longer", _time)922            stats = process_stats(stats[1:], slice * 1048576)923            logging.debug("Stats = %s", stats)924            logging.info("Host -> Guest [MB/s] (min/med/max) = %.3f/%.3f/%.3f",925                        stats[0], stats[len(stats) / 2], stats[-1])926            del thread927            # GUEST -> HOST928            exit_event.clear()929            stats = array.array('f', [])930            on_guest("virt.send_loop_init('%s', %d)" % (port.name, buf_len),931                     vm, 30)932            thread = ThRecv(port.sock, exit_event, buf_len)933            thread.start()934            loads.start()935            on_guest("virt.send_loop()", vm, 2)936            _time = time.time()937            for i in range(100):938                stats.append(thread.idx)939                time.sleep(slice)940            _time = time.time() - _time - duration941            logging.info("\n" + loads.get_cpu_status_string()[:-1])942            logging.info("\n" + loads.get_mem_status_string()[:-1])943            on_guest("virt.exit_threads()", vm, 2)944            exit_event.set()945            thread.join()946            if (_time > slice): # Deviation is higher than 1 slice947                logging.error(948                "Test ran %fs longer which is more than one slice", _time)949            else:950                logging.debug("Test ran %fs longer" % _time)951            stats = process_stats(stats[1:], slice * 1048576)952            logging.debug("Stats = %s", stats)953            logging.info("Guest -> Host [MB/s] (min/med/max) = %.3f/%.3f/%.3f",954                         stats[0], stats[len(stats) / 2], stats[-1])955            del thread956            del exit_event957    def clean_ports(vm, consoles):958        """959        Clean state of all ports and set port to default state.960        Default state:961           No data on port or in port buffer.962           Read mode = blocking.963        @param consoles: Consoles which should be clean.964        """965        # Check if python is still alive966        print "CLEANED"967        match, tmp = _on_guest("is_alive()", vm, 10)968        if (match == None) or (match != 0):969            logging.error("Python died/is stucked/have remaining threads")970            logging.debug(tmp)971            vm[1].close()972            vm[1] = kvm_test_utils.wait_for_login(vm[0], 0,973                                         float(params.get("boot_timeout", 240)),974                                         0, 2)975            (match, data) = _on_guest("killall -9 python "976                                      "&& echo -n PASS: python killed"977                                      "|| echo -n PASS: python was death",978                                      vm, 30)979            if (match == None):980                # killall -9 command didn't finished - python stucked981                raise error.TestFail("Python is really stucked - "982                                     "can't kill -9 it")983            #on_guest("rmmod -f virtio_console && echo -n PASS: rmmod "984            #         "|| echo -n FAIL: rmmod", vm, 10)985            #on_guest("modprobe virtio_console "986            #         "&& echo -n PASS: modprobe || echo -n FAIL: modprobe",987            #         vm, 10)988            init_guest(vm, consoles)989            (match, data) = _on_guest("virt.clean_port('%s'),1024" %990                                      consoles[0][0].name, vm, 2)991            if (match == None) or (match != 0):992                logging.error(data)993                logging.error("Virtio-console driver is irreparably"994                              " blocked. Every comd end with sig KILL."995                              "Try reboot vm for continue in testing.")996                vm[1] = kvm_test_utils.reboot(vm[0], vm[1], "system_reset")997                init_guest(vm, consoles)998                (match, data) = _on_guest("virt.clean_port('%s'),1024" %999                                      consoles[0][0].name, vm, 2)1000                if (match == None) or (match != 0):1001                    raise error.TestFail("Virtio-console driver is irreparably"1002                                         " blocked. Every comd end with sig"1003                                         " KILL. Neither the restart did not"1004                                         " help.")1005            else:1006                on_guest("virt.close('%s'),1024" % consoles[0][0].name, vm, 2)1007        for ctype in consoles:1008            for port in ctype:1009                openned = port.is_open1010                port.clean_port()1011                #on_guest("virt.blocking('%s', True)" % port.name, vm, 2)1012                on_guest("virt.clean_port('%s'),1024" % port.name, vm, 2)1013                if not openned:1014                    port.close()1015                    on_guest("virt.close('%s'),1024" % port.name, vm, 2)1016    # INITIALIZE1017    tsmoke_params = params.get('virtio_console_smoke', '')1018    tloopback_params = params.get('virtio_console_loopback', '')1019    tperf_params = params.get('virtio_console_perf', '')1020    no_serialports = 01021    no_consoles = 01022    # consoles required for Smoke test1023    if (tsmoke_params.count('serialport')):1024        no_serialports = max(2, no_serialports)1025    if (tsmoke_params.count('console')):1026        no_consoles = max(2, no_consoles)1027    # consoles required for Loopback test1028    for param in tloopback_params.split(';'):1029        no_serialports = max(no_serialports, param.count('serialport'))1030        no_consoles = max(no_consoles, param.count('console'))1031    # consoles required for Performance test1032    if (tperf_params.count('serialport')):1033        no_serialports = max(1, no_serialports)1034    if (tperf_params.count('console')):1035        no_consoles = max(1, no_consoles)1036    if (no_serialports + no_consoles) == 0:1037        raise error.TestFail("No tests defined, probably incorrect "1038                             "configuration in tests_base.cfg")1039    vm, consoles = _vm_create(no_consoles, no_serialports)1040    # Copy allocator.py into guests1041    pwd = os.path.join(os.environ['AUTODIR'], 'tests/kvm')1042    vksmd_src = os.path.join(pwd, "scripts/virtio_guest.py")1043    dst_dir = "/tmp"1044    if not vm[0].copy_files_to(vksmd_src, dst_dir):1045        raise error.TestFail("copy_files_to failed %s" % vm[0].name)1046    # ACTUAL TESTING1047    # Defines all available consoles; tests udev and sysfs1048    init_guest(vm, consoles)1049    test = SubTest()1050    test.set_cleanup_func(clean_ports, [vm, consoles])1051    #Test Smoke1052    test_smoke(test, vm, consoles, tsmoke_params)1053    #Test Loopback1054    test.do_test(tloopback, [vm, consoles, tloopback_params])1055    #Test Performance1056    test.do_test(tperf, [vm, consoles, tperf_params])1057    logging.info(("Summary: %d tests passed  %d test failed :\n" %1058                  (test.passed, test.failed)) + test.get_text_result())1059    logging.debug(("Summary: %d tests passed  %d test failed :\n" %1060                   (test.passed, test.failed)) + test.get_full_text_result())1061    if test.is_failed():1062        raise error.TestFail("Virtio_console test FAILED.")1063    # CLEANUP1064    vm[1].close()1065    vm[0].destroy(gracefully=False)...

run.py

Source:run.py

...54    opt.image_folder = temp_detection_dir55    # print(os.listdir(temp_detection_dir)) # for debugging56    # [[img_name, pred, confidence_score], ...]57    result = recognize(opt, model, converter)58    full_text = utils.get_full_text_result(result)59    print("[full text]")60    print(full_text)61    62    shutil.rmtree(temp_detection_dir)63    '''[ 4. visualize result ] '''64    # draw65    for box in fixed_row_list:66        tl = box[0]67        br = box[1]68        im = cv2.rectangle(im, tl, br, (0, 255, 0), 1)69    # save70    img_name = os.path.basename(args.image_path).split('.')[0]71    save_path = img_name + "_merged_detection.jpg"72    cv2.imshow('result', im)...

utils.py

Source:utils.py

...40        if bottom>height: bottom = height41        merged_row_list.append([(left, top),(right, bottom)])42    43    return merged_row_list44def get_full_text_result(origin_result):45    # [[img_name, pred, confidence_score], ...]46    result = [x[1] for x in origin_result]47    full_text = ''48    for string in result:49        full_text += (str(string) + " ")50    51    return full_text...

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.