ubbd-tests/xfstests.py at master · DataTravelGuide/ubbd-tests · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
#!/usr/bin/env python


import os
import glob
import re
import shutil

import avocado
from avocado import Test
from avocado.utils import process, build, git, distro, partition
from avocado.utils import disk, data_structures, pmem
from avocado.utils import genio


class Xfstests(Test):
    proc = None

    def start_ubbdd_killer(self):
        cmd = str("bash %s/utils/start_ubbdd_killer.sh %s" % (self.ubbd_tests_dir, self.ubbdd_timeout))
        self.proc = process.get_sub_process_klass(cmd)(cmd)
        pid = self.proc.start()
        self.log.info("ubbdd killer started: pid: %s, %s", pid, self.proc)

    def stop_ubbdd_killer(self):
        if not self.proc:
            return

        self.proc.stop(1)
        self.log.info("ubbdd killer stopped")

    def do_config(self, dev):
        os.chdir(self.ubbd_dir)
        cmd = str("ubbdadm config --ubbdid %s --data-pages-reserve-percnt 0" % (dev))
        result = process.run(cmd, ignore_status=True)
        self.log.info("config result: %s" % (result))
        return (result.exit_status == 0)


    def setUp(self):
        self.xfstests_dir = self.params.get('xfstests_dir')
        self.ubbd_dir = self.params.get('ubbd_dir')
        self.ubbd_tests_dir = self.params.get('ubbd_tests_dir')
        self.scratch_mnt = self.params.get(
            'scratch_mnt', default='/mnt/scratch')
        self.test_mnt = self.params.get('test_mnt', default='/mnt/test')
        self.fs_to_test = self.params.get('fs', default='ext4')
        self.disk_type = self.params.get('disk_type')

        if process.system('which mkfs.%s' % self.fs_to_test,
                          ignore_status=True):
            self.cancel('Unknown filesystem %s' % self.fs_to_test)
        mount = True
        self.devices = []
        shutil.copyfile(self.get_data('local.config'),
                        os.path.join(self.xfstests_dir, 'local.config'))

        self.test_dev = self.params.get('disk_test', default=None)
        self.scratch_dev = self.params.get('disk_scratch', default=None)
        self.devices.extend([self.test_dev, self.scratch_dev])
        self.exclude = self.params.get('exclude', default=None)
        self.ubbdd_timeout = self.params.get('ubbdd_timeout', default=0)
        self.test_set = self.params.get('test_set', default="-g auto")

        process.run("dmesg -C ", sudo=True)
        # mkfs for devices
        if self.devices:
            cfg_file = os.path.join(self.xfstests_dir, 'local.config')
            self.mkfs_opt = self.params.get('mkfs_opt', default='')
            self.mount_opt = self.params.get('mount_opt', default='')
            with open(cfg_file, "r") as sources:
                lines = sources.readlines()
            with open(cfg_file, "w") as sources:
                for line in lines:
                    if line.startswith('export TEST_DEV'):
                        sources.write(
                            re.sub(r'export TEST_DEV=.*', 'export TEST_DEV=%s'
                                   % self.devices[0], line))
                    elif line.startswith('export TEST_DIR'):
                        sources.write(
                            re.sub(r'export TEST_DIR=.*', 'export TEST_DIR=%s'
                                   % self.test_mnt, line))
                    elif line.startswith('export SCRATCH_DEV'):
                        sources.write(re.sub(
                            r'export SCRATCH_DEV=.*', 'export SCRATCH_DEV=%s'
                                                      % self.devices[1], line))
                    elif line.startswith('export SCRATCH_MNT'):
                        sources.write(
                            re.sub(
                                r'export SCRATCH_MNT=.*',
                                'export SCRATCH_MNT=%s' %
                                self.scratch_mnt,
                                line))
                        break
            with open(cfg_file, "a") as sources:
                if self.mkfs_opt:
                    sources.write('MKFS_OPTIONS="%s"\n' % self.mkfs_opt)
                if self.mount_opt:
                    sources.write('MOUNT_OPTIONS="%s"\n' % self.mount_opt)

            for ite, dev in enumerate(self.devices):
                self.do_config(dev)
                dev_obj = partition.Partition(dev)
                dev_obj.mkfs(fstype=self.fs_to_test, args=self.mkfs_opt)

        if not os.path.exists(self.scratch_mnt):
            os.makedirs(self.scratch_mnt)
        if not os.path.exists(self.test_mnt):
            os.makedirs(self.test_mnt)

        if self.ubbdd_timeout:
            self.start_ubbdd_killer()

    def test(self):
        if (self.disk_type == "mem" and self.ubbdd_timeout):
            return
        failures = False
        os.chdir(self.xfstests_dir)

        args = ''
        if self.exclude:
            args = ' -e \"%s\"' % self.exclude
        cmd = './check %s %s' % (args, self.test_set)
        result = process.run(cmd, ignore_status=True, verbose=True)
        if result.exit_status == 0:
            self.log.info('OK: All Tests passed.')
        else:
            msg = self._parse_error_message(result.stdout)
            self.log.info('ERR: Test(s) failed. Message: %s', msg)
            failures = True

        # print out the error message.
        if failures:
            self.log.info('One or more tests failed. Please check the logs.')

    def tearDown(self):
        self.stop_ubbdd_killer()
        # In case if any test has been interrupted
        process.system('umount %s %s' % (self.scratch_mnt, self.test_mnt),
                       sudo=True, ignore_status=True)

        process.system('wipefs -a %s' % (self.test_dev),
                       sudo=True, ignore_status=True)
        self.whiteboard = process.system_output("dmesg").decode()

    @staticmethod
    def _parse_error_message(output):
        na_re = re.compile(r'Passed all 0 tests')
        na_detail_re = re.compile(r'(\d{3})\s*(\[not run\])\s*(.*)')
        failed_re = re.compile(r'Failed \d+ of \d+ tests')

        lines = output.decode("utf-8").split('\n')
        result_line = lines[-3]

        error_msg = None
        if na_re.match(result_line):
            detail_line = lines[-3]
            match = na_detail_re.match(detail_line)
            if match is not None:
                error_msg = match.groups()[2]
            else:
                error_msg = 'Test dependency failed, test will not run.'
        elif failed_re.match(result_line):
            error_msg = 'Test error. %s.' % result_line
        else:
            error_msg = 'Could not verify test result. Please check the logs.'

        return error_msg