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