KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
