Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_long_buf(self):
""" subprocess (tast) became blocked and blocks parent (shellexec)
if exchange buffer (PIPE) is full because of wait() """
file_name = temp_file()
if is_windows():
task = "type "
buf_len = 2 ** 10 * 4 # 4K
else:
task = "tail "
buf_len = 2 ** 10 * 64 # 64K
task += file_name
buf = '*' * (buf_len + 1)
with open(file_name, "w+") as _file:
_file.write(buf)
self.obj.parameters.merge({"prepare": [task]})
self.obj.prepare()
self.obj.startup()
self.obj.shutdown()
out = self.log_recorder.debug_buff.getvalue()
self.assertIn(buf, out)
@skipUnless(is_windows(), "Windows-only")
def test_jmeter_4_windows_jmeter_home_var(self):
self.configure({
"execution": {
"iterations": 1,
"scenario": {
"script": RESOURCES_DIR + "/jmeter/jmx/dummy.jmx"
}
}
})
self.obj.settings.merge({"version": "4.0"})
self.obj.prepare()
self.obj._execute = lambda *args, **kwargs: None
self.obj.startup()
jmeter_home = self.obj.env.get("JMETER_HOME")
self.assertEqual(jmeter_home, get_full_path(self.obj.settings.get("path"), step_up=2))
self.assertEqual(jmeter_home, get_full_path(RESOURCES_DIR))
def full_run(self, config):
self.configure(config)
dummy = RESOURCES_DIR + 'selenium/ruby/ruby' + ('.bat' if is_windows() else '.sh')
self.obj.prepare()
self.obj.runner.settings.merge({"interpreter": dummy})
self.obj.startup()
while not self.obj.check():
time.sleep(self.obj.engine.check_interval)
self.obj.shutdown()
def test_properties_migration(self):
self.obj.execution.merge({
"scenario": {
"keepalive": True,
"requests": ["http://blazedemo.com/"]}
})
self.obj._execute = lambda *args, **kwargs: None
self.obj.prepare()
self.obj.startup()
if is_windows():
form = '%s'
else:
form = '%r'
self.assertIn("gatling.http.ahc.allowPoolingConnections=" + form % 'true', self.obj.env.get("JAVA_OPTS"))
self.assertIn("gatling.http.ahc.keepAlive=" + form % 'true', self.obj.env.get("JAVA_OPTS"))
def test_background_task_output(self):
temp = temp_file()
try:
with open(temp, "at") as temp_f:
temp_f.write("*" * (2 ** 16 + 1))
if is_windows():
cmd = "type"
else:
cmd = "cat"
command1 = "%s %s" % (cmd, temp)
command2 = "sleep 1"
task = {"command": command1, "background": True}
blocking_task = {"command": command2, "background": False}
self.obj.parameters.merge({"prepare": [task, blocking_task]})
self.obj.prepare()
self.obj.check()
self.obj.shutdown()
out = self.log_recorder.debug_buff.getvalue()
self.assertIn("code 0: %s" % command1, out)
self.assertIn("code 0: %s" % command2, out)
finally:
os.remove(temp)
"additional-classpath": [jars[1]],
"simulation": "mytest.BasicSimulation"}})
self.obj.prepare()
self.obj.startup()
self.obj.shutdown()
modified_launcher = self.obj.tool.tool_path
with open(modified_launcher) as modified:
modified_lines = modified.readlines()
for var in ("JAVA_CLASSPATH", "COMPILATION_CLASSPATH"):
self.assertNotIn(jars[0], self.obj.env.get(var))
self.assertIn(jars[1], self.obj.env.get(var))
for line in modified_lines:
if not is_windows() and '"$JAVA"' in line and not line.startswith("bash"):
self.assertTrue(line.startswith('eval'))
self.assertFalse(line.startswith('set COMPILATION_CLASSPATH=""')) # win
if line.startswith('COMPILATION_CLASSPATH='): # linux
self.assertTrue(line.endswith(':"${COMPILATION_CLASSPATH}"\n'))
def __redirect_streams(self):
if self.__streams_redirected:
return
if isinstance(self.screen, DummyScreen):
return
if sys.stdout.isatty():
if not is_windows():
self.__detect_console_logger()
if self.orig_streams:
raise TaurusInternalException("Console: original streams already set")
elif self.logger_handlers and not self.orig_streams:
self.log.debug("Overriding logging streams")
for handler in self.logger_handlers:
self.orig_streams[handler] = handler.stream
handler.stream = self.temp_stream
self.log.debug("Redirected logging streams, %s/%s", self.logger_handlers, self.orig_streams)
self.__streams_redirected = True
else:
self.log.info("Did not mute console logging")
def install(self):
dest = self.get_driver_dir()
if not os.path.exists(dest):
os.makedirs(dest)
self.log.info("Will install %s into %s", self.tool_name, dest)
dist = self._download(use_link=True)
try:
self.log.info("Unzipping %s to %s", dist, dest)
unzip(dist, dest)
finally:
os.remove(dist)
if not is_windows():
os.chmod(self.tool_path, 0o755)
if not self.check_if_installed():
raise ToolError("Unable to find %s after installation!" % self.tool_name)
def _get_screen_type(self):
screen_type = self.settings.get("screen", "console")
if screen_type not in ("console", "gui", "dummy"):
self.log.info("Invalid screen type %r, trying 'console'", screen_type)
screen_type = "console"
if not sys.stdout.isatty():
self.log.debug("Not in terminal, using dummy screen")
screen_type = "dummy"
if screen_type == "console":
if ConsoleScreen is DummyScreen or is_windows():
self.log.debug("Can't use 'console' screen, trying 'gui'")
screen_type = "gui"
if screen_type == "gui" and GUIScreen is DummyScreen:
self.log.debug("Can't use 'gui' screen, trying 'dummy'")
screen_type = "dummy"
return screen_type