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