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_exception(self):
local = Local()
local.engine = EngineEmul()
local.engine.config.merge({EXEC: [{}]})
local.engine.config.get("settings")["default-executor"] = "mock"
local.engine.unify_config()
local.prepare()
local.startup()
local.check()
local.shutdown()
try:
local.post_process()
except ToolError as exc:
self.assertNotIn('DIAGNOSTICS', str(exc))
self.assertIsNotNone(exc.diagnostics)
self.assertEqual(exc.diagnostics, ['DIAGNOSTICS'])
except BaseException as exc:
self.fail("Was supposed to fail with ToolError, but crashed with %s" % exc)
def test_no_apache_benchmark(self):
"Checks that prepare() fails if ApacheBenchmark is not installed."
self.obj.settings.merge({"path": "*"})
self.obj.execution.merge({
"scenario": {
"requests": ["http://blazedemo.com"]}})
self.assertRaises(ToolError, self.obj.prepare)
def test_install_pbench(self):
self.obj.settings.merge({"path": "/notexistent"})
self.assertRaises(ToolError, self.obj.prepare)
with open(self.stdout.name) as out:
file_header = out.read(1024)
if wrong_line in file_header: # gatling can't select test scenario
scenarios = file_header[file_header.find(wrong_line) + len(wrong_line):].rstrip()
msg = 'Several gatling simulations are found, you must '
msg += 'specify one of them to use in "simulation" option: %s' % scenarios
raise TaurusConfigError(msg)
if 'started...' in file_header:
self.simulation_started = True
if self.retcode is None:
return False
elif self.retcode == 0:
return True
else:
raise ToolError("Gatling tool exited with non-zero code: %s" % self.retcode, self.get_error_diagnostics())
elif line.startswith('set GATLING_CLASSPATH='):
mod_success = True
line = line.rstrip() + ';%JAVA_CLASSPATH%\n' # add from env
else:
if line.startswith('COMPILER_CLASSPATH='):
mod_success = True
line = line.rstrip()[:-1] + '${COMPILATION_CLASSPATH}"\n' # add from env
elif line.startswith('GATLING_CLASSPATH='):
mod_success = True
line = line.rstrip()[:-1] + '${JAVA_CLASSPATH}"\n' # add from env
elif line.startswith('"$JAVA"'):
line = 'eval ' + line
modified_lines.append(line)
if not mod_success:
raise ToolError("Can't modify gatling launcher for jar usage, ability isn't supported")
return modified_lines
def check_if_installed(self):
self.log.debug("Trying %s: %s", self.tool_name, self.tool_path)
try:
out, err = self.call([self.tool_path, "--version"])
except CALL_PROBLEMS as exc:
self.log.warning("%s check failed: %s", self.tool_name, exc)
return False
version = out.strip()
if LooseVersion(version) < LooseVersion("1.4"):
raise ToolError("You must install molotov>=1.4 to use this executor (version %s detected)" % version)
if err:
out += err
self.log.debug("%s output: %s", self.tool_name, out)
return True
def install(self):
dest = get_full_path(self.tool_path, step_up=2)
self.log.info("Will install %s into %s", self.tool_name, dest)
grinder_dist = self._download(use_link=bool(self.download_link))
self.log.info("Unzipping %s", grinder_dist)
unzip(grinder_dist, dest, 'grinder-' + self.version)
os.remove(grinder_dist)
self.log.info("Installed grinder successfully")
if not self.check_if_installed():
raise ToolError("Unable to run %s after installation!" % self.tool_name)
def check(self):
retcode = self.process.poll()
if retcode is not None:
if retcode != 0:
raise ToolError("Phantom-benchmark exit code: %s" % retcode, self.get_error_diagnostics())
return True
return False
def __handle_taurus_exception(self, exc, log_level):
if isinstance(exc, TaurusConfigError):
self.log.log(log_level, "Config Error: %s", exc)
elif isinstance(exc, TaurusInternalException):
self.log.log(log_level, "Internal Error: %s", exc)
elif isinstance(exc, ToolError):
self.log.log(log_level, "Child Process Error: %s", exc)
if exc.diagnostics is not None:
for line in exc.diagnostics:
self.log.log(log_level, line)
elif isinstance(exc, TaurusNetworkError):
self.log.log(log_level, "Network Error: %s", exc)
else:
self.log.log(log_level, "Generic Taurus Error: %s", exc)
def check(self):
ret_code = self.process.poll()
if ret_code is None:
return False
if ret_code != 0:
raise ToolError("molotov exited with non-zero code: %s" % ret_code, self.get_error_diagnostics())
return True