1# LTP Stress runtime
2#
3# Copyright (c) 2019 MontaVista Software, LLC
4#
5# SPDX-License-Identifier: MIT
6#
7
8import time
9import datetime
10import pprint
11
12from oeqa.runtime.case import OERuntimeTestCase
13from oeqa.core.decorator.depends import OETestDepends
14from oeqa.runtime.decorator.package import OEHasPackage
15from oeqa.core.decorator.data import skipIfQemu
16from oeqa.utils.logparser import LtpParser
17
18class LtpStressBase(OERuntimeTestCase):
19
20    @classmethod
21    def setUpClass(cls):
22        cls.ltp_startup()
23
24    @classmethod
25    def tearDownClass(cls):
26        cls.ltp_finishup()
27
28    @classmethod
29    def ltp_startup(cls):
30        cls.sections = {}
31        cls.failmsg = ""
32        test_log_dir = os.path.join(cls.td.get('WORKDIR', ''), 'testimage')
33        timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
34
35        cls.ltptest_log_dir_link = os.path.join(test_log_dir, 'ltpstress_log')
36        cls.ltptest_log_dir = '%s.%s' % (cls.ltptest_log_dir_link, timestamp)
37        os.makedirs(cls.ltptest_log_dir)
38
39        cls.tc.target.run("mkdir -p /opt/ltp/results")
40
41        if not hasattr(cls.tc, "extraresults"):
42            cls.tc.extraresults = {}
43        cls.extras = cls.tc.extraresults
44        cls.extras['ltpstressresult.rawlogs'] = {'log': ""}
45
46
47    @classmethod
48    def ltp_finishup(cls):
49        cls.extras['ltpstressresult.sections'] =  cls.sections
50
51        # update symlink to ltp_log
52        if os.path.exists(cls.ltptest_log_dir_link):
53            os.remove(cls.ltptest_log_dir_link)
54
55        os.symlink(os.path.basename(cls.ltptest_log_dir), cls.ltptest_log_dir_link)
56
57        if cls.failmsg:
58            cls.fail(cls.failmsg)
59
60class LtpStressTest(LtpStressBase):
61
62    def runltp(self, stress_group):
63            cmd = '/opt/ltp/runltp -f %s -p -q 2>@1 | tee /opt/ltp/results/%s' % (stress_group, stress_group)
64            starttime = time.time()
65            (status, output) = self.target.run(cmd)
66            endtime = time.time()
67            with open(os.path.join(self.ltptest_log_dir, "%s" % stress_group), 'w') as f:
68                f.write(output)
69
70            self.extras['ltpstressresult.rawlogs']['log'] = self.extras['ltpstressresult.rawlogs']['log'] + output
71
72            parser = LtpParser()
73            results, sections  = parser.parse(os.path.join(self.ltptest_log_dir, "%s" % stress_group))
74
75            runtime = int(endtime-starttime)
76            sections['duration'] = runtime
77            self.sections[stress_group] =  sections
78
79            failed_tests = {}
80            for test in results:
81                result = results[test]
82                testname = ("ltpstressresult." + stress_group + "." + test)
83                self.extras[testname] = {'status': result}
84                if result == 'FAILED':
85                    failed_tests[stress_group] = test
86
87            if failed_tests:
88                self.failmsg = self.failmsg + "Failed ptests:\n%s" % pprint.pformat(failed_tests)
89
90    # LTP stress runtime tests
91    #
92    @skipIfQemu()
93    @OETestDepends(['ssh.SSHTest.test_ssh'])
94    @OEHasPackage(["ltp"])
95    def test_ltp_stress(self):
96        self.tc.target.run("sed -i -r 's/^fork12.*//' /opt/ltp/runtest/crashme")
97        self.runltp('crashme')
98