1# 2# SPDX-License-Identifier: MIT 3# 4 5import re 6import time 7 8from oeqa.runtime.case import OERuntimeTestCase 9from oeqa.core.decorator.depends import OETestDepends 10from oeqa.core.decorator.data import skipIfQemu 11 12class StorageBase(OERuntimeTestCase): 13 def storage_mount(cls, tmo=1): 14 15 (status, output) = cls.target.run('mkdir -p %s' % cls.mount_point) 16 (status, output) = cls.target.run('mount %s %s' % (cls.device, cls.mount_point)) 17 msg = ('Mount failed: %s.' % status) 18 cls.assertFalse(output, msg = msg) 19 time.sleep(tmo) 20 (status, output) = cls.target.run('cat /proc/mounts') 21 match = re.search('%s' % cls.device, output) 22 if match: 23 msg = ('Device %s not mounted.' % cls.device) 24 cls.assertTrue(match, msg = msg) 25 26 (status, output) = cls.target.run('mkdir -p %s' % cls.test_dir) 27 28 (status, output) = cls.target.run('rm -f %s/*' % cls.test_dir) 29 msg = ('Failed to cleanup files @ %s/*' % cls.test_dir) 30 cls.assertFalse(output, msg = msg) 31 32 33 def storage_basic(cls): 34 # create file on device 35 (status, output) = cls.target.run('touch %s/%s' % (cls.test_dir, cls.test_file)) 36 msg = ('File %s not created on %s' % (cls.test_file, cls.device)) 37 cls.assertFalse(status, msg = msg) 38 # move file 39 (status, output) = cls.target.run('mv %s/%s %s/%s1' % 40 (cls.test_dir, cls.test_file, cls.test_dir, cls.test_file)) 41 msg = ('File %s not moved to %s' % (cls.test_file, cls.device)) 42 cls.assertFalse(status, msg = msg) 43 # remove file 44 (status, output) = cls.target.run('rm %s/%s1' % (cls.test_dir, cls.test_file)) 45 msg = ('File %s not removed on %s' % (cls.test_file, cls.device)) 46 cls.assertFalse(status, msg = msg) 47 48 def storage_read(cls): 49 # check if message is in file 50 (status, output) = cls.target.run('cat %s/%s' % 51 (cls.test_dir, cls.test_file)) 52 53 match = re.search('%s' % cls.test_msg, output) 54 msg = ('Test message %s not in file %s.' % (cls.test_msg, cls.test_file)) 55 cls.assertEqual(status, 0, msg = msg) 56 57 def storage_write(cls): 58 # create test message in file on device 59 (status, output) = cls.target.run('echo "%s" > %s/%s' % 60 (cls.test_msg, cls.test_dir, cls.test_file)) 61 msg = ('File %s not create test message on %s' % (cls.test_file, cls.device)) 62 cls.assertEqual(status, 0, msg = msg) 63 64 def storage_umount(cls, tmo=1): 65 time.sleep(tmo) 66 (status, output) = cls.target.run('umount %s' % cls.mount_point) 67 68 if status == 32: 69 # already unmounted, should it fail? 70 return 71 else: 72 msg = ('Device not unmount %s' % cls.mount_point) 73 cls.assertEqual(status, 0, msg = msg) 74 75 (status, output) = cls.target.run('cat /proc/mounts') 76 match = re.search('%s' % cls.device, output) 77 if match: 78 msg = ('Device %s still mounted.' % cls.device) 79 cls.assertTrue(match, msg = msg) 80 81 82class UsbTest(StorageBase): 83 ''' 84 This is to mimic the usb test previously done in manual bsp-hw.json 85 ''' 86 @classmethod 87 def setUpClass(self): 88 self.test_msg = "Hello World - USB" 89 self.mount_point = "/media/usb" 90 self.device = "/dev/sda1" 91 self.test_file = "usb.tst" 92 self.test_dir = os.path.join(self.mount_point, "oeqa") 93 94 @skipIfQemu() 95 @OETestDepends(['ssh.SSHTest.test_ssh']) 96 def test_usb_mount(self): 97 self.storage_umount(2) 98 self.storage_mount(5) 99 100 @skipIfQemu() 101 @OETestDepends(['storage.UsbTest.test_usb_mount']) 102 def test_usb_basic_operations(self): 103 self.storage_basic() 104 105 @skipIfQemu() 106 @OETestDepends(['storage.UsbTest.test_usb_basic_operations']) 107 def test_usb_basic_rw(self): 108 self.storage_write() 109 self.storage_read() 110 111 @skipIfQemu() 112 @OETestDepends(['storage.UsbTest.test_usb_mount']) 113 def test_usb_umount(self): 114 self.storage_umount(2) 115 116 117class MMCTest(StorageBase): 118 ''' 119 This is to mimic the usb test previously done in manual bsp-hw.json 120 ''' 121 @classmethod 122 def setUpClass(self): 123 self.test_msg = "Hello World - MMC" 124 self.mount_point = "/media/mmc" 125 self.device = "/dev/mmcblk1p1" 126 self.test_file = "mmc.tst" 127 self.test_dir = os.path.join(self.mount_point, "oeqa") 128 129 @skipIfQemu() 130 @OETestDepends(['ssh.SSHTest.test_ssh']) 131 def test_mmc_mount(self): 132 self.storage_umount(2) 133 self.storage_mount() 134 135 @skipIfQemu() 136 @OETestDepends(['storage.MMCTest.test_mmc_mount']) 137 def test_mmc_basic_operations(self): 138 self.storage_basic() 139 140 @skipIfQemu() 141 @OETestDepends(['storage.MMCTest.test_mmc_basic_operations']) 142 def test_mmc_basic_rw(self): 143 self.storage_write() 144 self.storage_read() 145 146 @skipIfQemu() 147 @OETestDepends(['storage.MMCTest.test_mmc_mount']) 148 def test_mmc_umount(self): 149 self.storage_umount(2) 150