KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use usb_device method in Airtest
Best Python code snippet using Airtest
usbdeviceforensics.py
Source:usbdeviceforensics.py 
...414 if len(parts_serial_no) == 2:415 serial_no = parts_serial_no[0]416 else:417 serial_no = device_sk.name()418 usb_device = get_usb_device(serial_number=serial_no,419 vendor=vendor,420 product=product,421 version=version)422 if usb_device is None:423 continue424 for sub_key_device in device_sk.subkeys():425 if sub_key_device.name().lower() != 'properties':426 continue427 key64 = get_key(sub_key_device, r'{83da6326-97a6-4088-9453-a1923f573b29}\00000064\00000000')428 if key64 is not None:429 value64 = get_reg_value(key64, 'Data')430 if value64 is not None:431 usb_device.usbstor_datetime64 = value64.value()432 write_debug(name='USBSTOR date/time (64)', value=usb_device.usbstor_datetime64.strftime('%Y-%m-%dT%H:%M:%S'))433 else:434 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\00000064\\00000000 is None')435 key65 = get_key(sub_key_device, r'{83da6326-97a6-4088-9453-a1923f573b29}\00000065\00000000')436 if key65 is not None:437 value65 = get_reg_value(key65, 'Data')438 if value65 is not None:439 usb_device.usbstor_datetime65 = value65.value()440 write_debug(name='USBSTOR date/time (65)', value=usb_device.usbstor_datetime65.strftime('%Y-%m-%dT%H:%M:%S'))441 else:442 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\00000065\\00000000 is None')443 key64win8 = get_key(sub_key_device, r'{83da6326-97a6-4088-9453-a1923f573b29}\0064')444 if key64win8 is not None:445 value64win8 = get_reg_value(key64win8, '(default)')446 if value64win8 is not None:447 usb_device.usbstor_datetime64 = value64win8.value()448 write_debug(name='USBSTOR date/time (64)', value=usb_device.usbstor_datetime64.strftime('%Y-%m-%dT%H:%M:%S'))449 else:450 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\0064 is None')451 key65win8 = get_key(sub_key_device, r'{83da6326-97a6-4088-9453-a1923f573b29}\0065')452 if key65win8 is not None:453 value65win8 = get_reg_value(key65win8, '(default)')454 if not value65win8 is None:455 usb_device.usbstor_datetime65 = value65win8.value()456 write_debug(name='USBSTOR date/time (65)', value=usb_device.usbstor_datetime65.strftime('%Y-%m-%dT%H:%M:%S'))457 else:458 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\0065 is None')459 key66 = get_key(sub_key_device, r'{83da6326-97a6-4088-9453-a1923f573b29}\0066')460 if key66 is not None:461 value66 = get_reg_value(key66, '(default)')462 if value66 is not None:463 usb_device.usbstor_datetime66 = value66.value()464 write_debug(name='USBSTOR date/time (66)', value=usb_device.usbstor_datetime66.strftime('%Y-%m-%dT%H:%M:%S'))465 else:466 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\0066\\(default) is None')467 else:468 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\0066 is None')469 key67 = get_key(sub_key_device, r'{83da6326-97a6-4088-9453-a1923f573b29}\0067')470 if key67 is not None:471 value67 = get_reg_value(key67, '(default)')472 if value67 is not None:473 usb_device.usbstor_datetime67 = value67.value()474 write_debug(name='USBSTOR date/time (67)', value=usb_device.usbstor_datetime67.strftime('%Y-%m-%dT%H:%M:%S'))475 else:476 write_debug(data='{83da6326-97a6-4088-9453-a1923f573b29}\\0067 is None')477def process_usb(registry):478 """Processes the CCS \Enum\USB keys"""479 write_debug(data='Method: process_usb')480 ccs = []481 root_key = registry.root()482 for k in root_key.subkeys():483 if 'ControlSet' in k.name():484 ccs.append(k.name())485 for c in ccs:486 key = registry.open(c + '\\Enum\\USB')487 for sub_key in key.subkeys():488 if (not 'vid' in sub_key.name().lower() and489 not 'pid' in sub_key.name().lower()):490 continue491 # Get the serial number which is the next key492 for serial_key in sub_key.subkeys():493 usb_device = get_usb_device(serial_number=serial_key.name())494 if usb_device is None:495 continue496 vid_pid = sub_key.name().split('&')497 usb_device.vid = vid_pid[0]498 write_debug(name='VID', value=usb_device.vid)499 usb_device.pid = vid_pid[1]500 write_debug(name='PID', value=usb_device.pid)501 usb_device.vid_pid_datetime = sub_key.timestamp()502 write_debug(name='VID/PID datetime', value=usb_device.vid_pid_datetime.strftime('%Y-%m-%dT%H:%M:%S'))503def process_mounted_devices(registry):504 """Processes the MountedDevices keys"""505 write_debug(data='Method: process_mounted_devices')506 global usb_devices507 reg_key = registry.root().find_key('MountedDevices')508 if reg_key is None:509 write_debug(data='MountedDevices key does not exist')510 return511 for usb_device in usb_devices:512 for reg_value in reg_key.values():513 # Example Data: USBSTOR#Disk&Ven_SanDisk&Prod_Cruzer&Rev_7.01#2444120C4E80D827&514 data = reg_value.value()515 data = remove_non_ascii_characters(data)516 if '\\DosDevices\\' in reg_value.name():517 if len(data) == 12: # Drive Sig (DWORD) Partition Offset (DWORD DWORD)518 dos_device = bytearray(data[0:4])519 usb_device.disk_signature = ''.join('{:02x}'.format(byte) for byte in dos_device)520 write_debug(name='Disk sig', value=usb_device.disk_signature)521 if len(usb_device.parent_prefix_id) > 0:522 if '\\DosDevices\\' in reg_value.name():523 if usb_device.parent_prefix_id in str(data):524 usb_device.drive_letter = reg_value.name().replace('\\DosDevices\\', '')525 write_debug(name='Drive letter', value=usb_device.drive_letter)526 if '\\Volume{' in reg_value.name():527 if usb_device.parent_prefix_id in data:528 guid = reg_value.name()[11:]529 guid = guid[:len(guid)-1]530 usb_device.guid = guid531 write_debug(name='GUID', value=usb_device.guid)532 usb_device.mountpoint = data[4:]533 write_debug(name='Mountpoint', value=usb_device.mountpoint)534 # If the drive letter is missing from being identified by the535 # ParentPrefixId then try matching the full device string536 if len(usb_device.drive_letter) == 0:537 if '\\DosDevices\\' in reg_value.name():538 if ('USBSTOR#Disk&' +539 usb_device.vendor + '&' +540 usb_device.product + '&' +541 usb_device.version + '#' +542 usb_device.serial_number) in data:543 usb_device.drive_letter = reg_value.name().replace('\\DosDevices\\', '')544 write_debug(name='Drive letter', value=usb_device.drive_letter)545 # If the GUID is missing from being identified by the546 # ParentPrefixId then try matching the full device string547 if len(usb_device.guid) == 0:548 if '\\Volume{' in reg_value.name():549 temp = str(('USBSTOR#Disk&' +550 usb_device.vendor + '&' +551 usb_device.product + '&' +552 usb_device.version + '#' +553 usb_device.serial_number))554 if temp.lower() in data.lower():555 guid = reg_value.name()[11:]556 guid = guid[:len(guid)-1]557 usb_device.guid = guid558 write_debug(name='GUID', value=usb_device.guid)559 usb_device.mountpoint = data[4:]560 write_debug(name='Mountpoint', value=usb_device.mountpoint)561def process_device_classes(registry):562 """Processes the CCS \Control\DeviceClasses\{53f56307-b6bf-11d0-94f2-00a0c91efb8b keys"""563 write_debug(data='Method: process_device_classes')564 ccs = []565 root_key = registry.root()566 for k in root_key.subkeys():567 if 'ControlSet' in k.name():568 ccs.append(k.name())569 global usb_devices570 # Get the DeviceClasses related information e.g. VID & PID + USB Date/Time571 for usb_device in usb_devices:572 if len(usb_device.mountpoint.strip()) == 0:573 continue574 for c in ccs:575 key = registry.open(c + '\\Control\\DeviceClasses\\{53f56307-b6bf-11d0-94f2-00a0c91efb8b}')576 for sub_key in key.subkeys():577 if usb_device.mountpoint in sub_key.name():578 usb_device.device_classes_datetime_53f56307b6bf11d094f200a0c91efb8b = sub_key.timestamp()579 write_debug(name='Dev Classes date/time (53f56)',580 value=usb_device.device_classes_datetime_53f56307b6bf11d094f200a0c91efb8b.strftime('%Y-%m-%dT%H:%M:%S'))581 continue582 if usb_device.serial_number in sub_key.name():583 usb_device.device_classes_datetime_53f56307b6bf11d094f200a0c91efb8b = sub_key.timestamp()584 write_debug(name='Dev Classes date/time (53f56)',585 value=usb_device.device_classes_datetime_53f56307b6bf11d094f200a0c91efb8b.strftime('%Y-%m-%dT%H:%M:%S'))586 continue587 key = registry.open(c + '\\Control\\DeviceClasses\\{10497b1b-ba51-44e5-8318-a65c837b6661}')588 for sub_key in key.subkeys():589 if usb_device.mountpoint in sub_key.name():590 usb_device.device_classes_datetime_10497b1bba5144e58318a65c837b6661 = sub_key.timestamp()591 write_debug(name='Dev Classes date/time (10497)',592 value=usb_device.device_classes_datetime_10497b1bba5144e58318a65c837b6661.strftime('%Y-%m-%dT%H:%M:%S'))593 continue594 if usb_device.serial_number in sub_key.name():595 usb_device.device_classes_datetime_10497b1bba5144e58318a65c837b6661 = sub_key.timestamp()596 write_debug(name='Dev Classes date/time (10497)',597 value=usb_device.device_classes_datetime_10497b1bba5144e58318a65c837b6661.strftime('%Y-%m-%dT%H:%M:%S'))598 continue599# Software Hive Methods #######################################################600def process_windows_portable_devices(registry):601 """Processes the Microsoft\Windows Portable Devices\Devices key"""602 write_debug(data='Method: process_windows_portable_devices')603 try:604 key = registry.open('Microsoft\\Windows Portable Devices\\Devices')605 global usb_devices606 # Get the DeviceClasses related information e.g. VID & PID + USB Date/Time607 for sub_key in key.subkeys():608 if not '_##_USBSTOR' in sub_key.name() and not '_??_USBSTOR' in sub_key.name():609 continue610 for usb_device in usb_devices:611 if not usb_device.serial_number in sub_key.name():612 continue613 friendly_name = get_reg_value(sub_key, 'FriendlyName')614 if len(friendly_name.value()) == 0:615 write_debug(data='FriendlyName value not defined')616 continue617 temp = friendly_name.value()618 if '(' in temp:619 usb_device.drive_letter = temp[temp.index('('):]620 usb_device.drive_letter = usb_device.drive_letter.replace('(', '')621 usb_device.drive_letter = usb_device.drive_letter.replace(')', '')622 usb_device.volume_name = temp[0:temp.index('(')]623 write_debug(name='Drive letter', value=usb_device.drive_letter)624 write_debug(name='Volume name', value=usb_device.volume_name)625 elif ':\\' in temp:626 usb_device.drive_letter = temp627 write_debug(name='Drive letter', value=usb_device.drive_letter)628 else:629 usb_device.volume_name = temp630 write_debug(name='Volume name', value=usb_device.volume_name)631 except Registry.RegistryKeyNotFoundException:632 return633def process_emd_mgmt(registry):634 """Processes SOFTWARE\Microsoft\Windows NT\CurrentVersion\EMDMgmt key"""635 write_debug(data='Method: process_emd_mgmt')636 try:637 key = registry.open('Microsoft\\Windows NT\\CurrentVersion\\EMDMgmt')638 global usb_devices639 for sub_key in key.subkeys():640 if not '_##_USBSTOR#Disk&' in sub_key.name() and not '_??_USBSTOR#Disk&' in sub_key.name():641 continue642 volume_serial_no = ''643 volume_name = ''644 write_debug(data=sub_key.name())645 if not '{53f56307-b6bf-11d0-94f2-00a0c91efb8b}' in sub_key.name().lower():646 write_debug(data='Key name does not contain {53f56307-b6bf-11d0-94f2-00a0c91efb8b}')647 continue648 index = sub_key.name().index('#{53f56307-b6bf-11d0-94f2-00a0c91efb8b}')649 mountpoint = sub_key.name()[0:index + 39]650 mountpoint = mountpoint.replace('_??_', '')651 mountpoint = mountpoint.replace('_##_', '')652 write_debug(name='Mountpoint', value=mountpoint)653 if len(mountpoint) == 0:654 continue655 # Now get the string data that is after the GUID656 data = sub_key.name()[index + 39:] # 38 is the length of the '{53f56....' string/GUID657 if '_' in data:658 volume_serial_no = data[data.rfind('_') + 1:]659 volume_name = data[0:data.rfind('_')]660 for usb_device in usb_devices:661 if usb_device.mountpoint.lower() != mountpoint.lower():662 continue663 emdMgmt = EmdMgmt()664 emdMgmt.volume_serial_num = volume_serial_no665 write_debug(name='EMDMgmt serial no.', value=emdMgmt.volume_serial_num)666 emdMgmt.volume_name = volume_name667 write_debug(name='EMDMgmt volume name', value=emdMgmt.volume_name)668 emdMgmt.timestamp = sub_key.timestamp()669 write_debug(name='EMDMgmt date/time', value=emdMgmt.timestamp.strftime('%Y-%m-%dT%H:%M:%S'))670 if len(volume_serial_no) > 0:671 temp_vsn = int(volume_serial_no)672 emdMgmt.volume_serial_num_hex = "%x" % temp_vsn673 write_debug(name='EMDMgmt serial no. (hex)', value=emdMgmt.volume_serial_num_hex)674 usb_device.emdmgmt.append(emdMgmt)675 except Registry.RegistryKeyNotFoundException:676 return677##### NTUSER Hive Methods ############################################################################################678def process_mountpoints2(registry, reg_file_path):679 """Processes the Software\Microsoft\Windows\CurrentVersion\Explorer\MountPoints2 key"""680 try:681 key = registry.open('Software\\Microsoft\\Windows\\CurrentVersion\\Explorer\\MountPoints2')682 global usb_devices683 for usb_device in usb_devices:684 # If there is no GUID then we cannot match so just continue685 if len(usb_device.guid) == 0:686 continue687 for sub_key in key.subkeys():688 if '{' + usb_device.guid + '}' != sub_key.name():689 continue690 mp2 = MountPoint2()691 mp2.file = reg_file_path692 mp2.timestamp = sub_key.timestamp()693 usb_device.mountpoint2.append(mp2)694 write_debug(name='Mountpoint2 file', value=mp2.file)695 write_debug(name='Mountpoint2 date/time', value=mp2.timestamp.strftime('%Y-%m-%dT%H:%M:%S'))696 except Registry.RegistryKeyNotFoundException:697 return698# Log File Methods ############################################################699def process_log_file(file):700 """"""701 regexXp1 = '^\[([0-9]+/[0-9]+/[0-9]+\s[0-9]+:[0-9]+:[0-9]+)\s[0-9]+.[0-9]+\sDriver\sInstall\]'702 regexXp2 = '#-019 .*? ID\(s\): usb\\(.+)'703 regexXp3 = '#I121.*? "(.*)"'704 regexVista1 = '>>> *\[Device Install \(Hardware initiated\) - USBSTOR\\(.+)\]'705 regexVista2 = '>>>\s\sSection\sstart\s([0-9]+/[0-9]+/[0-9]+\s[0-9]+:[0-9]+:[0-9]+\.[0-9]+)'706 regexWin78 = '>>>\s\s\[Device\sInstall\s\(Hardware\sinitiated\) - SWD\\WPDBUSENUM\\_\?\?_USBSTOR#(.*)\]'707 with open(file) as f:708 lines = f.readlines()709 f.close()710 install_times = {}711 index = 0712 while index < len(lines) - 1:713 index += 1714 line = lines[index].strip()715 timestamp = ''716 key = ''717 if os == WindowsVersions.WindowsXP or os == WindowsVersions.WindowsXPx64:718 match = re.search(regexXp1, line, re.I)719 if match:720 timestamp = match.group(1)721 if len(timestamp) == 0:722 continue723 match = re.search(regexXp3, line, re.I)724 if match:725 key = match.group(1)726 if len(key) == 0:727 continue728 if 'USB\\' not in key and 'USBSTOR\\' not in key:729 continue730 if key in install_times is False:731 install_times[key] = timestamp732 elif os == WindowsVersions.WindowsVista:733 if not '>>> [Device Install (Hardware initiated) - USBSTOR' in line:734 continue735 match = re.search(regexVista1, line, re.I)736 if match:737 key = 'USBSTOR\\' + match.group(1)738 if len(key) == 0:739 continue740 # Get the line below741 index += 1742 line = lines[index].strip()743 match = re.search(regexVista2, line, re.I)744 if match:745 timestamp = match.group(1)746 if len(timestamp) == 0:747 continue748 if key not in install_times:749 install_times[key] = timestamp750 else:751 #'>>>\s\s\[Device\sInstall\s\(Hardware\sinitiated\) - SWD\\WPDBUSENUM\\_\?\?_USBSTOR#(.*)\]'752 if not '>>> [Device Install (Hardware initiated) - SWD\\WPDBUSENUM\\_??_USBSTOR#' in line:753 continue754 #'>>>\s\s\[Device\sInstall\s\(Hardware\sinitiated\) - SWD\\WPDBUSENUM\\_\?\?_USBSTOR#(.*)\]'755 match = re.match('>>>\s\s\[Device\sInstall\s\(Hardware\sinitiated\) - SWD\\\\WPDBUSENUM\\\\_\?\?_USBSTOR#(.*)\]', line, re.I)756 if match:757 key = 'USBSTOR\\' + match.group(1)758 if len(key) == 0:759 continue760 # Now turn the top value into the bottom value761 # usbstor\disk&ven_frespons&prod_tactical_subject&rev_0.00#00000022928277&0#{53f56307-b6bf-11d0-94f2-00a0c91efb8b}762 # usbstor\disk&ven_frespons&prod_tactical_subject&rev_0.00\00000022928277&0763 key = key.replace("\\", "#")764 # Get the line below765 index += 1766 line = lines[index].strip()767 match = re.search(regexVista2, line, re.I)768 if match:769 timestamp = match.group(1)770 if len(timestamp) == 0:771 continue772 if key not in install_times:773 install_times[key] = timestamp774 # Now update the install date/time for the devices775 for key, timestamp in install_times.iteritems():776 for device in usb_devices:777 if os == WindowsVersions.WindowsXP or os == WindowsVersions.WindowsXPx64:778 if (device.vid.lower() + '&' + device.pid.lower() + "\\" + device.serial_number.lower()) in key.lower():779 device.install_datetime = datetime.strptime(timestamp, "%Y/%m/%d %H:%M:%S.%f")780 elif (device.vendor.lower() + "&" + device.product.lower() + "&" + device.version.lower() + "\\" + device.serial_number.lower()) in key.lower():781 #I121 "USBSTOR\DISK&VEN_USB_2.0&PROD_&REV_1100\6&12202299&0782 device.install_datetime = datetime.strptime(timestamp, "%Y/%m/%d %H:%M:%S.%f")783 else:784 if len(device.mountpoint) == 0:785 continue786 parts = device.mountpoint.split('#')787 if len(parts) == 4:788 temp = parts[0] + "#" + parts[1] + "#" + parts[2] + '#{53f56307-b6bf-11d0-94f2-00a0c91efb8b}'789 if temp.lower() in key.lower():790 write_debug(data='Matched install log timestamp using mountpoint: ' + key.lower())791 device.install_datetime = datetime.strptime(timestamp, "%Y/%m/%d %H:%M:%S.%f")792 else:793 write_debug(data='Unable to match install log timestamp using mountpoint: ' + key.lower())794 else:795 print('The mountpoint does not contain 4 delimited (#) parts: ' + device.mountpoint)796# Helper Methods ##############################################################797def load_file(file):798 """Loads a file as a registry hive"""799 try:800 print('Loading file: ' + file)801 registry = Registry.Registry(file)802 return registry803 except Exception:804 return None805def parse_windows_timestamp(qword):806 """see http://integriography.wordpress.com/2010/01/16/using-phython-to-parse-and-present-windows-64-bit-timestamps"""807 return datetime.utcfromtimestamp(float(qword) * 1e-7 - 11644473600)808def does_usb_device_exist(usb_device):809 """Iterates through the module level list of USB device objects and determines if the object already exists"""810 global usb_devices811 for device in usb_devices:812 if (device.serial_number == usb_device.serial_number and813 device.vendor == usb_device.vendor and814 device.product == usb_device.product and815 device.version == usb_device.version and816 device.parent_prefix_id == usb_device.parent_prefix_id):817 return True818 return False819def get_usb_device(serial_no):820 """Iterates through the module level list of USB device objects and returns the object if it exists"""821 global usb_devices822 for device in usb_devices:823 if device.serial_number == serial_no:824 return device825 return None826def get_usb_device(**kwargs):827 """Iterates through the module level list of USB device objects and returns the object if it exists"""828 global usb_devices829 for device in usb_devices:830 if len(kwargs) == 1:831 if device.serial_number == kwargs['serial_number']:832 write_debug(data='Located USB device by serial number')833 return device834 elif len(kwargs) == 4:835 write_debug(data=kwargs['serial_number'] + '#' + kwargs['vendor'] + '#' + kwargs['product'] + '#' + kwargs['version'])836 write_debug(data=device.serial_number + '#' + device.vendor + '#' + device.product + '#' + device.version)837 if (device.serial_number == kwargs['serial_number'] and838 device.vendor == kwargs['vendor'] and839 device.product == kwargs['product'] and840 device.version == kwargs['version']):...
test_ntfs.py
Source:test_ntfs.py
1"""2This test checks the target for the presence of a USB device. It checks3whether the device is large enough for the test to execute. It then formats4the USB as NTFS, creates partitions and a device label. The device is then5mounted and a large image is copied over to the USB (virtual size 13G). That6image is then imported into glance and a VM is booted from the image using a7newly created flavour.8"""9from pytest import skip10from utils.tis_log import LOG11from utils.clients.ssh import ControllerClient12from consts.stx import FlavorSpec13from keywords import vm_helper, nova_helper, glance_helper, host_helper, system_helper14from testfixtures.fixture_resources import ResourceCleanup15def locate_usb(host_type="controller", min_size=13):16 """17 Try to locate a USB device on a host of the type specified.18 Arguments:19 - host_type (string) - e.g. controller, compute, storage20 - min_size (int) - minimum size of USB required (GiB)21 Returns:22 - hostname, e.g. controller-023 - usb_device, e.g. /dev/sdb24 """25 LOG.tc_step("Check all hosts of type {} for USB devices".format(host_type))26 hosts = system_helper.get_hosts(personality=host_type)27 for host in hosts:28 with host_helper.ssh_to_host(host) as host_ssh:29 cmd = "ls --color=none -ltrd /dev/disk/by-id/usb*"30 rc, out = host_ssh.exec_cmd(cmd)31 if rc == 0:32 usb_device = "/dev/" + (out.splitlines()[0])[-3:]33 LOG.info("Found USB device {} on host {}".format(usb_device, host))34 cmd = "blockdev --getsize64 {}".format(usb_device)35 usb_bytes = host_ssh.exec_sudo_cmd(cmd)[1]36 gib = int(usb_bytes) / (1024 * 1024 * 1024)37 if gib > min_size:38 LOG.info("Size of USB device is sufficient for test")39 return host, usb_device40 else:41 skip("Size of USB device is insufficient for test")42 return (None, None)43def umount_usb(host_ssh, mount_point="/media/ntfs"):44 """45 Unmount a USB device.46 Arguments:47 - host_ssh - ssh session to host with USB48 - mount_point (string) - e.g. /media/ntfs49 Returns50 - Nothing51 """52 LOG.tc_step("Unmounting {}".format(mount_point))53 cmd = "umount {}".format(mount_point)54 rc = host_ssh.exec_sudo_cmd(cmd)[0]55 assert rc == 0 or rc == 3256 if rc == 0:57 LOG.info("Umount was successful")58 if rc == 32:59 LOG.info("Umount was unsuccessful. Maybe device was already unmounted?")60def wipe_usb(host_ssh, usb_device):61 """62 Wipe a USB device, including all existing partitions.63 Arguments:64 - host_ssh - ssh session to host with USB65 - usb_device (string) - name of usb device66 Returns:67 - nothing68 """69 LOG.tc_step("Wipe the USB completely")70 cmd = "dd if=/dev/zero of={} bs=1k count=2048".format(usb_device)71 rc = host_ssh.exec_sudo_cmd(cmd, fail_ok=False)[0]72 assert rc == 0, "USB wipe failed"73def create_usb_label(host_ssh, usb_device=None, label="msdos"):74 """75 Create a label on a USB device.76 Arguments:77 - host_ssh - ssh session to host with USB78 - usb_device (string) - e.g. /dev/sdb79 - label (string) - e.g. "msdos"80 Returns:81 - Nothing82 """83 LOG.tc_step("Create label and partition table on the USB")84 cmd = "parted {} mklabel {} -s".format(usb_device, label)85 rc = host_ssh.exec_sudo_cmd(cmd, fail_ok=False)[0]86 assert rc == 0, "Label creation failed"87def create_usb_partition(host_ssh, usb_device, startpt="0", endpt="0"):88 """89 Create a partition on a USB device.90 Arguments:91 - host_ssh - ssh session to host with USB92 - usb_device (string) - e.g. /dev/sdb93 - startpt (string) - partition start point, e.g. "0"94 - endpt (string) - partition end point, e.g. "2048"95 Returns:96 - Nothing97 """98 cmd = "parted -a none {} mkpart primary ntfs {} {}".format(usb_device, startpt, endpt)99 rc = host_ssh.exec_sudo_cmd(cmd)[0]100 assert rc == 0, "Primary partition creation failed"101def format_usb(host_ssh, usb_device, partition):102 """103 This formats a particular partition on a usb device.104 Arguments:105 - host_ssh - ssh session to host with USB106 - usb_device (string) - e.g. /dev/sdb107 - partition (string) - e.g. "2" for /dev/sdb2108 Returns:109 - Nothing110 """111 LOG.tc_step("Format device {} as NTFS".format(usb_device + partition))112 cmd = "mkfs.ntfs -f {}{}".format(usb_device, partition)113 rc = host_ssh.exec_sudo_cmd(cmd)[0]114 assert rc == 0, "Failed to format device"115def mount_usb(host_ssh, usb_device=None, partition="2", mount_type="ntfs", mount_point="/media/ntfs"):116 """117 This creates a mount point and then mounts the desired device.118 Arguments:119 - host_ssh - ssh session to host with USB120 - usb_device (string) - e.g. /dev/sdb121 - mount_point (string) - where the usb should be mounted122 Returns:123 - Nothing124 """125 LOG.tc_step("Check if mount point exists")126 cmd = "test -d {}".format(mount_point)127 rc = host_ssh.exec_sudo_cmd(cmd)[0]128 if rc == 1:129 LOG.tc_step("Create mount point")130 cmd = "mkdir -p {}".format(mount_point)131 rc = host_ssh.exec_sudo_cmd(cmd)[0]132 assert rc == 0, "Mount point creation failed"133 LOG.tc_step("Mount ntfs device")134 cmd = "mount -t {} {} {}".format(mount_type, usb_device + partition, mount_point)135 rc = host_ssh.exec_sudo_cmd(cmd)[0]136 assert rc == 0, "Unable to mount device"137# Wendy says just testing one node is enough.138# @mark.parametrize("host_type", ['controller', 'compute', 'storage'])139def test_ntfs(stx_openstack_required, host_type="controller"):140 """141 This test will test NTFS mount and NTFS formatted device creation on a TiS142 system.143 Arguments:144 - host_type (string) - host type to be tested, e.g. controller, compute,145 storage146 Returns:147 - Nothing148 Test Steps:149 1. Check if desired host has USB inserted. If not, skip150 2. Wipe USB151 3. Change label of device152 4. Create partitions on NTFS device153 5. Format partitions154 4. Copy large image to NTFS mount point155 5. Test mount and big file creation on NTFS mounted device156 """157 # Could pass these in through parametrize instead158 mount_type = "ntfs"159 mount_point = "/media/ntfs/"160 guest_os = 'win_2012'161 boot_source = "image"162 host, usb_device = locate_usb(host_type, min_size=13)163 if not host:164 skip("No USB hardware found on {} host type".format(host_type))165 hosts_with_image_backing = host_helper.get_hosts_in_storage_backing(storage_backing='image')166 if len(hosts_with_image_backing) == 0:167 skip("No hosts with image backing present")168 # if the host with the USB is not the active controler, swact controllers169 con_ssh = ControllerClient.get_active_controller()170 active_controller = system_helper.get_active_controller_name(con_ssh)171 if host != active_controller:172 host_helper.swact_host()173 with host_helper.ssh_to_host(host) as host_ssh:174 wipe_usb(host_ssh, usb_device)175 umount_usb(host_ssh, mount_point=mount_point)176 create_usb_label(host_ssh, usb_device, label="msdos")177 create_usb_partition(host_ssh, usb_device, startpt="0", endpt="2048")178 format_usb(host_ssh, usb_device, partition="1")179 create_usb_partition(host_ssh, usb_device, startpt="2049", endpt="100%")180 format_usb(host_ssh, usb_device, partition="2")181 mount_usb(host_ssh, usb_device, partition="2", mount_type=mount_type, mount_point=mount_point)182 LOG.tc_step("Copy the windows guest image to the mount point")183 src_img = glance_helper.scp_guest_image(img_os=guest_os, dest_dir=mount_point, con_ssh=con_ssh)184 LOG.tc_step("Create flavor for windows guest image")185 flv_id = nova_helper.create_flavor(name=guest_os, vcpus=4, ram=8192, storage_backing="local_image",186 guest_os=guest_os)[1]187 nova_helper.set_flavor(flv_id, **{FlavorSpec.CPU_POLICY: "dedicated"})188 ResourceCleanup.add("flavor", flv_id)189 LOG.tc_step("Import image into glance")190 glance_helper.create_image(name=guest_os, source_image_file=src_img, disk_format="qcow2",191 container_format="bare", con_ssh=con_ssh, cleanup="function")192 LOG.tc_step("Boot VM")193 vm_id = vm_helper.boot_vm(name=guest_os, flavor=flv_id, guest_os=guest_os, source=boot_source, cleanup="function")[1]194 LOG.tc_step("Ping vm and ssh to it")195 vm_helper.wait_for_vm_pingable_from_natbox(vm_id)196 with vm_helper.ssh_to_vm_from_natbox(vm_id) as vm_ssh:197 output = vm_ssh.exec_cmd('pwd', fail_ok=False)[1]...
origin-usb-monitor
Source:origin-usb-monitor
1#!/usr/bin/env python2import sys3import os4fullBinPath=os.path.abspath(os.getcwd() + "/" + sys.argv[0])5fullLibPath=os.path.abspath(os.path.dirname(os.path.dirname(fullBinPath))+"/lib")6sys.path.append(fullLibPath)7import time8import random9import calendar10import time11import sys12import origin13if len(sys.argv) > 1:14 configBundle = sys.argv[1]15 origin.configure(configBundle)16else:17 origin.configure("site")18from origin.client import monitoring_record19from origin.client import server20from origin.client import server_connection21from origin import current_time, timestamp22#Create usb interface23from devices import originIO24io = originIO.originIO()25# just a proxy for actually doing something useful26def makeTempMeasurement():27 return random.random()28# something that represents the connection to the server29# might need arguments.... idk30serv = server()31# perhaps print some useful message. Perhaps try to reconnect....32# print "problem establishing connection to server..."33# sys.exit(1)34# This might need to be more complicated, but you get the gist. Keep sending records forever35while True:36 print "sending...."37 usb_device_info = io.get_all_devices_info()38 print("Number of detected devices: ")39 print(len(usb_device_info))40 for usb_device in usb_device_info:41 #print(usb_device)42 brand = usb_device['brand']43 type = usb_device['type']44 serial = usb_device['serial']45 stream_name = usb_device['name']46 if stream_name == '':47 stream_name = brand+'_'+type+'_'+serial48 49 print(stream_name)50 # this should be changed to not reregister everytime51 connection = serv.registerStream(52 stream=stream_name.encode("ascii"),53 records={ 'set_point':'float',54 'sensor':'float',55 'V':'float',56 'I':'float'})57 record = {58 'set_point':float(usb_device['set_point']),59 'sensor' :float(usb_device['sensor']),60 'V' :float(usb_device['V']),61 'I' :float(usb_device['I'])62 }63 #time.sleep(1)64 record['timestamp'] = current_time(origin.config)65 connection.send(**record)...
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 Airtest 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?
