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):
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 in suite:
test.setUpEnv()
break
else:
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 "Python script error" in line:
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.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 ("Python script error" 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()
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 """
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
morse_initialized = False
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
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("localhost", 4000))
s.send(b"id1 simulation quit\n")
with open(self.logfile_name) as log:
lines = follow(log)
for line in lines:
if "EXITING SIMULATION" in line:
return
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