Source code for morse.testing.testing

import logging
#testrunnerlogger = logging.getLogger("test.runner")
testlogger = logging.getLogger("morsetesting.general")

import sys, os
from abc import ABCMeta, abstractmethod
import unittest
import inspect
import tempfile
from time import sleep
import threading # Used to be able to timeout when waiting for Blender initialization
import subprocess
import signal

from morse.testing.exceptions import MorseTestingError

BLENDER_INITIALIZATION_TIMEOUT = 15 # seconds

[docs]class MorseTestRunner(unittest.TextTestRunner):
[docs] def setup_logging(self): logger = logging.getLogger('morsetesting') logger.setLevel(logging.DEBUG) ch = logging.StreamHandler() ch.setLevel(logging.DEBUG) formatter = logging.Formatter('[%(asctime)s (%(levelname)s)] %(message)s') ch.setFormatter(formatter) logger.addHandler(ch)
[docs] def run(self, suite): self.setup_logging() return unittest.TextTestRunner.run(self, suite)
[docs]def follow(file): """ Really emulate tail -f See http://stackoverflow.com/questions/1475950/tail-f-in-python-with-no-time-sleep for a detailled discussion on the subject """ while True: line = file.readline() if not line: sleep(0.1) # Sleep briefly continue yield line
[docs]class MorseTestCase(unittest.TestCase): # Make this an abstract class __metaclass__ = ABCMeta
[docs] def setUpMw(self): """ This method can be overloaded by subclasses to define environment setup, before the launching of the Morse environment pass """ pass
def _checkMorseException(self): """ Check in the Morse output if some python error happens""" with open(self.logfile_name) as log: lines = follow(log) for line in lines: # Python Error Case if "[ERROR][MORSE]" in line: testlogger.error(line.strip()) testlogger.error("Exception detected in Morse execution : " "see %s for details." " Exiting the current test !" % self.logfile_name) os.kill(os.getpid(), signal.SIGINT) return # End of simulation, exit the thread if "EXITING SIMULATION" in line: return
[docs] def setUp(self): testlogger.info("Starting test " + self.id()) self.logfile_name = self.__class__.__name__ + ".log" # Wait for a second # to wait for ports open in previous tests to be closed sleep(1) self.morse_initialized = False self.setUpMw() self.startmorse(self) self.t = threading.Thread(target=self._checkMorseException) self.t.start()
[docs] def tearDownMw(self): """ This method can be overloaded by subclasses to clean up environment setup """ pass
[docs] def tearDown(self): self.stopmorse() self.tearDownMw() self.logfile.close() # force to flush self.t.join()
@abstractmethod
[docs] def setUpEnv(self): """ This method must be overloaded by subclasses to define a simulation environment. The code must follow the :doc:`Builder API <morse/dev/builder>` convention (without the import of the `morsebuilder` module which is automatically added). """ pass
[docs] def wait_initialization(self): """ Wait until Morse is initialized """ testlogger.info("Waiting for MORSE to initialize... (timeout: %s sec)" % \ BLENDER_INITIALIZATION_TIMEOUT) with open(self.logfile_name) as log: lines = follow(log) for line in lines: if ("[ERROR][MORSE]" in line) or ("INITIALIZATION ERROR" in line): testlogger.error("Error during MORSE initialization! Check " "the log file.") return if "SCENE INITIALIZED" in line: self.morse_initialized = True return
[docs] def run(self, result=None): """ Overwrite unittest.TestCase::run Detect KeyBoardInterrupt exception , due to user or a SIGINIT In particular, it can happen if we detect an exception in the Morse execution. In this case, clean up correctly the environnement. """ try: return unittest.TestCase.run(self, result) except KeyboardInterrupt as e: self.tearDownMw() if self.pid: os.kill(self.pid, signal.SIGKILL) if result: result.addError(self, sys.exc_info())
def _extract_pid(self): """ Extract the pid from the log file. We can not simply rely on Popen.subprocess.pid because we need the PID of the Blender process itself, not the (Python) MORSE process. """ with open(self.logfile_name) as log: for line in log: if "PID" in line: words = line.split() return int(words[-1])
[docs] def startmorse(self, test_case): """ This starts MORSE in a new process, passing the script itself as parameter (to build the scene via the Builder API). """ temp_builder_script = self.generate_builder_script(test_case) try: original_script_name = os.path.abspath(inspect.stack()[-1][1]) try: prefix = os.environ['MORSE_ROOT'] except KeyError: prefix="" if prefix == "": cmd = 'morse' else: cmd = prefix + "/bin/morse" self.logfile = open(self.logfile_name, 'w') self.morse_process = subprocess.Popen([cmd, 'run', temp_builder_script], stdout=self.logfile, stderr=subprocess.STDOUT) except OSError as ose: testlogger.error("Error while launching MORSE! Check you can run it from command-line\n" + \ " and if you use the $MORSE_ROOT env variable, check it points to a correct " + \ " place!") raise ose t = threading.Thread(target=self.wait_initialization) t.start() t.join(BLENDER_INITIALIZATION_TIMEOUT) if self.morse_initialized: self.pid = self._extract_pid() testlogger.info("MORSE successfully initialized with PID %s" % self.pid) else: self.morse_process.terminate() raise MorseTestingError("MORSE did not start successfully! Check %s " "for details." % self.logfile_name)
[docs] def stopmorse(self): """ Cleanly stop MORSE """ import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: sock.connect(("localhost", 4000)) sock.send(b"id1 simulation quit\n") except (socket.error, KeyboardInterrupt): sock.close() sock = None testlogger.info("MORSE crashed") if sock: sock.close() with open(self.logfile_name) as log: lines = follow(log) for line in lines: if "EXITING SIMULATION" in line: return if self.pid: os.kill(self.pid, signal.SIGKILL) testlogger.info("MORSE stopped")
[docs] def generate_builder_script(self, test_case): tmp_name = "" # We need to generate a temp builder file in case of running # several test cases with different environment: # Blender must be restarted and called again with the right # environment. with tempfile.NamedTemporaryFile(delete = False) as tmp: tmp.write(b"from morse.builder import *\n") tmp.write(b"from morse.builder.actuators import *\n") tmp.write(b"from morse.builder.sensors import *\n") tmp.write(b"from morse.builder.blenderobjects import *\n") tmp.write(b"class MyEnv():\n") tmp.write(inspect.getsource(test_case.setUpEnv).encode()) tmp.write(b"MyEnv().setUpEnv()\n") tmp.flush() tmp_name = tmp.name testlogger.info("Created a temporary builder file for test-case " +\ test_case.__class__.__name__) return tmp_name
[docs]class MorseBuilderFailureTestCase(MorseTestCase): """ This subclass of MorseTestCase can be used to test MORSE handles properly ill-constructed Builder scripts. It will *fail* if the Blender Game Engine get started. """ # Make this an abstract class __metaclass__ = ABCMeta
[docs] def wait_initialization(self): """ Wait until Morse is initialized """ testlogger.info("Waiting for MORSE to parse the scene... (timeout: %s sec)" % \ BLENDER_INITIALIZATION_TIMEOUT) # we assume we will correctly detect Builder script issue, so wait_initialization # 'succeed'. self.morse_initialized = True with open(self.logfile_name) as log: lines = follow(log) for line in lines: if "Blender Game Engine Started" in line: testlogger.error("Blender Game Engine started!" " This is not expected." " See %s for details." % self.logfile_name) os.kill(os.getpid(), signal.SIGINT) return elif "[ERROR][MORSE]" in line: # use 'info' since we suppose to get this error testlogger.info("MORSE initialization error: %s"%line.strip()) return
def _checkMorseException(self): return
[docs]def main(*test_cases): import sys if sys.argv[0].endswith('blender'): # If we arrive here from within MORSE, we have probably run # morse [exec|run] my_test.py # If this case, simply build the environment based on the # setUpEnv of the first test. for test_class in test_cases: test_class().setUpEnv() return import unittest suite = unittest.TestSuite() loader = unittest.TestLoader() for test_class in test_cases: tests = loader.loadTestsFromTestCase(test_class) suite.addTests(tests) sys.exit(not MorseTestRunner().run(suite).wasSuccessful())