1# Copyright (C) 2016 Intel Corporation 2# Released under the MIT license (see COPYING.MIT) 3 4from . import OETestDecorator, registerDecorator 5 6import signal 7from threading import Timer 8 9from oeqa.core.threaded import OETestRunnerThreaded 10from oeqa.core.exception import OEQATimeoutError 11 12@registerDecorator 13class OETimeout(OETestDecorator): 14 attrs = ('oetimeout',) 15 16 def setUpDecorator(self): 17 self.logger.debug("Setting up a %d second(s) timeout" % self.oetimeout) 18 19 if isinstance(self.case.tc.runner, OETestRunnerThreaded): 20 self.timeouted = False 21 def _timeoutHandler(): 22 self.timeouted = True 23 24 self.timer = Timer(self.oetimeout, _timeoutHandler) 25 self.timer.start() 26 else: 27 timeout = self.oetimeout 28 def _timeoutHandler(signum, frame): 29 raise OEQATimeoutError("Timed out after %s " 30 "seconds of execution" % timeout) 31 32 self.alarmSignal = signal.signal(signal.SIGALRM, _timeoutHandler) 33 signal.alarm(self.oetimeout) 34 35 def tearDownDecorator(self): 36 if isinstance(self.case.tc.runner, OETestRunnerThreaded): 37 self.timer.cancel() 38 self.logger.debug("Removed Timer handler") 39 if self.timeouted: 40 raise OEQATimeoutError("Timed out after %s " 41 "seconds of execution" % self.oetimeout) 42 else: 43 signal.alarm(0) 44 signal.signal(signal.SIGALRM, self.alarmSignal) 45 self.logger.debug("Removed SIGALRM handler") 46