import shutil
import contextlib
import pprint
-import traceback
from tempfile import TemporaryDirectory
self._runtime_log_aggregation = os.path.join(self.__runtime_log, "runtime.log")
self._run_command_count = 0
+ self._is_test_modules_loaded = False
self.special_env_variables = {"LTTNG_UST_DEBUG": "1",
"LTTNG_APP_SOCKET_TIMEOUT": "-1",
def subprocess_signal(self, subprocess_uuid, signal):
self.__subproces[subprocess_uuid].send_signal(signal)
- def subprocess_terminate(self, subprocess_uuid, timeout=60):
+ def subprocess_terminate(self, subprocess_uuid, timeout=60, check_return=True):
process = self.__subprocess[subprocess_uuid]
process.terminate()
- process.wait(timeout)
+ try:
+ process.wait(timeout)
+ except subprocess.TimeoutExpired:
+ # Force kill
+ return self.subprocess_kill(subprocess_uuid)
stdout, stderr = self.__stdout_stderr[subprocess_uuid]
stdout.close()
stderr.close()
+ if check_return:
+ if process.returncode != 0:
+ raise subprocess.CalledProcessError(process.returncode, process.args)
return process
def subprocess_kill(self, subprocess_uuid):
stderr.close()
return process
+ def subprocess_wait(self, subprocess_uuid, check_return=True):
+ process = self.__subprocess[subprocess_uuid]
+ process.wait()
+ stdout, stderr = self.__stdout_stderr[subprocess_uuid]
+ stdout.close()
+ stderr.close()
+ if check_return:
+ if process.returncode != 0:
+ raise subprocess.CalledProcessError(process.returncode, process.args)
+ return process
+
def get_subprocess_stdout_path(self, subprocess_uuid):
stdout, stderr = self.__stdout_stderr[subprocess_uuid]
return stdout.name
def load_test_module(self):
# Base directory is provided by env
self.run("modprobe lttng-test")
+ self._is_test_modules_loaded = True
def unload_test_module(self, check_return=True):
# Base directory is provided by env
- self.run("modprobe -r lttng-test", check_return=check_return)
+ if self._is_test_modules_loaded:
+ self.run("modprobe -r lttng-test", check_return=check_return)
def close(self):
for key, subp in self.__subprocess.items():
- subp.terminate()
- for key, subp in self.__subprocess.items():
- # TODO move timeout to settings
- subp.wait(timeout=60)
- for key, (stdout, stderr) in self.__stdout_stderr.items():
- stdout.close()
- stderr.close()
+ self.subprocess_terminate(key, check_return=False)
# Always try to remove test module but do not perform check on return
# value.