1# Copyright 2016 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import os, re, glob, logging, shutil
6from autotest_lib.client.common_lib import error
7from autotest_lib.client.bin import test, utils
8
9class xfstests(test.test):
10    """
11    Runs a single test of the xfstests suite.
12    """
13
14    XFS_TESTS_PATH='/usr/local/xfstests'
15    XFS_EXCLUDE_FILENAME = '/tmp/.xfstests.exclude'
16    version = 2
17
18    PASSED_RE = re.compile(r'Passed all \d+ tests')
19    FAILED_RE = re.compile(r'Failed \d+ of \d+ tests')
20    TEST_RE = re.compile(r'(?P<name>\d+)\.out')
21    NA_RE = re.compile(r'Passed all 0 tests')
22    NA_DETAIL_RE = re.compile(r'(\d{3})\s*(\[not run\])\s*(.*)')
23
24
25    def _get_available_tests(self, fs):
26        os.chdir(os.path.join(self.XFS_TESTS_PATH, 'tests', fs))
27        tests = glob.glob('*.out*')
28        tests_list = []
29        for t in tests:
30            t_m = self.TEST_RE.match(t)
31            if t_m:
32                t_name = t_m.group('name')
33                if t_name not in tests_list and os.path.exists(t_name):
34                    tests_list.append(t_name)
35        tests_list.sort()
36        return tests_list
37
38
39    def _run_sub_test(self, test):
40        os.chdir(self.XFS_TESTS_PATH)
41        logging.debug("Environment variables: %s", os.environ)
42        output = utils.system_output(
43                'bash ./check %s' % os.path.join('tests', test),
44                ignore_status=True,
45                retain_output=True)
46        lines = output.split('\n')
47        result_line = lines[-2]
48        result_full = os.path.join('results', '.'.join([test, 'full']))
49        result_full_loc = os.path.join(self.XFS_TESTS_PATH, result_full)
50        if os.path.isfile(result_full_loc):
51            shutil.copyfile(result_full_loc,
52                            os.path.join(self.resultsdir, 'full'))
53
54        if self.NA_RE.match(result_line):
55            detail_line = lines[-3]
56            match = self.NA_DETAIL_RE.match(detail_line)
57            if match is not None:
58                error_msg = match.groups()[2]
59            else:
60                error_msg = 'Test dependency failed, test not run'
61            raise error.TestNAError(error_msg)
62
63        elif self.FAILED_RE.match(result_line):
64            raise error.TestError('Test error, check debug logs for complete '
65                                  'test output')
66
67        elif self.PASSED_RE.match(result_line):
68            return
69
70        else:
71            raise error.TestError('Could not assert test success or failure, '
72                                  'assuming failure. Please check debug logs')
73
74
75    def _run_standalone(self, group):
76        os.chdir(self.XFS_TESTS_PATH)
77        logging.debug("Environment variables: %s", os.environ)
78        output = utils.system_output(
79                'bash ./check -E %s -g %s' % (self.XFS_EXCLUDE_FILENAME, group),
80                ignore_status=True,
81                retain_output=True)
82        lines = output.split('\n')
83        result_line = lines[-2]
84
85        if self.NA_RE.match(result_line):
86            raise error.TestNAError('Test dependency failed, no tests run')
87
88        elif self.FAILED_RE.match(result_line):
89            failures_line = re.match(r'Failures: (?P<tests>.*)', lines[-3])
90            if failures_line:
91                test_failures = failures_line.group('tests')
92                tests = test_failures.split(' ')
93                for test in tests:
94                    result_full = os.path.join('results',
95                                               '.'.join([test, 'full']))
96                    result_full_loc = os.path.join(self.XFS_TESTS_PATH,
97                                                   result_full)
98                    if os.path.isfile(result_full_loc):
99                        test_name = test.replace('/','_')
100                        shutil.copyfile(result_full_loc,
101                                        os.path.join(self.resultsdir,
102                                                     '%s.full' % test_name))
103            raise error.TestError('%s. Check debug logs for complete '
104                                  'test output' % result_line)
105
106        elif self.PASSED_RE.match(result_line):
107            return
108        else:
109            raise error.TestError('Could not assert success or failure, '
110                                  'assuming failure. Please check debug logs')
111
112
113    def run_once(self, test_dir='generic', test_number='000', group=None,
114                 exclude=[]):
115        if group:
116            excludeFile = open(self.XFS_EXCLUDE_FILENAME, 'w')
117            for test in exclude:
118                excludeFile.write('%s\n' % test)
119            excludeFile.close()
120            logging.debug("Running tests: group %s", group )
121            self._run_standalone(group)
122            if os.path.exists(self.XFS_EXCLUDE_FILENAME):
123                os.remove(self.XFS_EXCLUDE_FILENAME)
124        else:
125            if test_number == '000':
126                logging.debug('Dummy test to setup xfstests')
127                return
128
129            if test_number not in self._get_available_tests(test_dir):
130                raise error.TestNAError(
131                    'test file %s/%s not found' % (test_dir, test_number))
132
133            test_name = os.path.join(test_dir, test_number)
134            logging.debug("Running test: %s", test_name)
135            self._run_sub_test(test_name)
136