_logger.debug("Spawned sub pid: {} args: {} stdout: {} stderr{}".format(p.pid, p.args, out_path, err_path))
return tmp_id
- def run(self, command_line, cwd=None, check_return=True):
+ def run(self, command_line, cwd=None, check_return=True, ld_preload=""):
"""
Run the command and return a tuple of a (CompletedProcess, stdout_path,
stderr_path). The subprocess is already executed and returned. The
args = shlex.split(command_line)
env = self.get_env()
+ if ld_preload:
+ env['LD_PRELOAD'] = ld_preload
+
+
tmp_id = self._run_command_count
self._run_command_count += 1
env_path = os.path.join(self.__runtime_log, str(tmp_id) + ".env")
with open(env_path, 'w') as env_out:
- pprint.pprint(env, stream=env_out)
+ for key, value in env.items():
+ env_out.write('{}={}\n'.format(key, value))
cp = subprocess.run(args, stdout=stdout, stderr=stderr, env=env, cwd=cwd)
_logger.debug("Command #{} args: {} stdout: {} stderr{}".format(tmp_id, cp.args, out_path, err_path))
env[var] = value
return env
+ def load_test_module(self):
+ # Base directory is provided by env
+ self.run("modprobe lttng-test")
+
+ def unload_test_module(self, check_return=True):
+ # Base directory is provided by env
+ self.run("modprobe -r lttng-test", check_return=check_return)
+
def close(self):
for key, subp in self.__subprocess.items():
subp.terminate()
stdout.close()
stderr.close()
+ # Always try to remove test module but do not perform check on return
+ # value.
+ self.unload_test_module(False)
+
# Copy the lttng_home used at runtime using hardlink to prevent useless
# data duplication
shutil.copytree(self.lttng_home, self.__post_runtime_lttng_home_path, copy_function=os.link)