Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _init_target(self):
# todo: allow empty address in apiritif (HTTPTarget.__init__)
default_address = self.scenario.get("default-address", "")
target_call = ast_call(
func=ast_attr("apiritif.http.target"),
args=[self._gen_expr(default_address)])
target = ast.Assign(
targets=[ast_attr("self.target")],
value=target_call)
return target
method = "self.assertNotEqual"
assertion_elements.append(
ast.Assign(
targets=[ast.Name(id="re_pattern")],
value=ast_call(
func=ast_attr("re.compile"),
args=[ast.Str(val)])))
assertion_elements.append(ast.Expr(
ast_call(
func=ast_attr(method),
args=[
ast.Num(0),
ast_call(
func=ast.Name(id="len"),
args=[ast_call(
func=ast_attr("re.findall"),
args=[ast.Name(id="re_pattern"), ast.Name(id="body")])]),
ast.Str("Assertion: %s" % assert_message)])))
else:
if reverse:
method = "self.assertNotIn"
else:
method = "self.assertIn"
assertion_elements.append(
ast.Expr(
ast_call(
func=ast_attr(method),
args=[
ast.Str(val),
ast.Name(id="body"),
raise TaurusConfigError("Only 'body' subject supported ")
assert_message = "'%s' " % val
if not reverse:
assert_message += 'not '
assert_message += 'found in BODY'
if regexp:
if reverse:
method = "self.assertEqual"
else:
method = "self.assertNotEqual"
assertion_elements.append(
ast.Assign(
targets=[ast.Name(id="re_pattern")],
value=ast_call(
func=ast_attr("re.compile"),
args=[ast.Str(val)])))
assertion_elements.append(ast.Expr(
ast_call(
func=ast_attr(method),
args=[
ast.Num(0),
ast_call(
func=ast.Name(id="len"),
args=[ast_call(
func=ast_attr("re.findall"),
args=[ast.Name(id="re_pattern"), ast.Name(id="body")])]),
ast.Str("Assertion: %s" % assert_message)])))
else:
lines.append(ast.Assign(
targets=[ast.Name(id="body")],
value=ast_attr("self.driver.page_source")))
for assert_config in req.config.get("assert"):
lines.extend(self._gen_sel_assertion(assert_config))
else:
lines.extend(self._gen_assertions(req))
lines.extend(self._gen_jsonpath_assertions(req))
lines.extend(self._gen_xpath_assertions(req))
lines.extend(self._gen_extractors(req))
if think_time:
lines.append(ast.Expr(
ast_call(
func=ast_attr("sleep"),
args=[self._gen_expr(think_time)])))
return lines
if "quoted" in source:
quoted = ast.keyword()
quoted.arg = "quoted"
quoted.value = ast.Name(id=source.get("quoted"))
keywords.append(quoted)
if "delimiter" in source:
delimiter = ast.keyword()
delimiter.arg = "delimiter"
delimiter.value = ast.Str(s=source.get("delimiter"))
keywords.append(delimiter)
csv_file = self.scenario.engine.find_file(source["path"])
reader = ast.Assign(
targets=[ast.Name(id="reader_%s" % idx)],
value=ast_call(
func=ast_attr("apiritif.CSVReaderPerThread"),
args=[ast.Str(s=csv_file)],
keywords=keywords))
readers.append(reader)
if readers:
readers.append(gen_empty_line_stmt())
return readers
def _gen_locator(self, tag, selector):
return ast_call(
func=ast_attr("self.driver.find_element"),
args=[
ast_attr("By.%s" % self.BYS[tag]),
self._gen_expr(selector)])
targets=[ast_attr("self.driver")],
value=ast_call(
func=ast_attr("webdriver.Chrome"),
keywords=[
ast.keyword(
arg="service_log_path",
value=ast.Str(self.wdlog)),
ast.keyword(
arg="chrome_options",
value=ast.Name(id="options"))])))
elif browser == 'remote':
keys = sorted(self.capabilities.keys())
values = [str(self.capabilities[key]) for key in keys]
body.append(ast.Assign(
targets=[ast_attr("self.driver")],
value=ast_call(
func=ast_attr("webdriver.Remote"),
keywords=[
ast.keyword(
arg="command_executor",
value=ast.Str(self.remote_address)),
ast.keyword(
arg="desired_capabilities",
value=ast.Dict(
keys=[ast.Str(key) for key in keys],
values=[ast.Str(value) for value in values]))])))
else:
if headless:
self.log.warning("Browser %r doesn't support headless mode" % browser)
body.append(ast.Assign(
targets=[ast_attr("self.driver")],
def _gen_master_test_method(self, slave_method_names):
if not slave_method_names:
raise TaurusConfigError("Supported trasactions not found, test is empty")
body = []
for slave_name in slave_method_names:
body.append(ast.Expr(ast_call(func=ast_attr("self." + slave_name))))
name = 'test_' + create_method_name(self.label)
return self._gen_test_method(name=name, body=body)
def _gen_class_setup(self):
if self.test_mode == "apiritif":
target_init = self._gen_api_target()
else:
target_init = self._gen_webdriver()
data_sources = [self._gen_default_vars()]
for idx in range(len(self.data_sources)):
data_sources.append(ast.Expr(ast_call(func=ast_attr("reader_%s.read_vars" % (idx + 1)))))
for idx in range(len(self.data_sources)):
extend_vars = ast_call(
func=ast_attr("self.vars.update"),
args=[ast_call(
func=ast_attr("reader_%s.get_vars" % (idx + 1)))])
data_sources.append(ast.Expr(extend_vars))
handlers = []
if self.generate_markers:
func_name = "add_flow_markers"
self.selenium_extras.add(func_name)
handlers.append(ast.Expr(ast_call(func=func_name)))
stored_vars = {"func_mode": str(False)} # todo: make func_mode optional
if target_init:
if self.test_mode == "selenium":
stored_vars["driver"] = "self.driver"
value=ast_call(
func=ast_attr("webdriver.ChromeOptions"))))
body.append(ast.Expr(
ast_call(
func=ast_attr("options.add_argument"),
args=[ast.Str("%s" % "--no-sandbox")]
)))
body.append(ast.Expr(
ast_call(
func=ast_attr("options.add_argument"),
args=[ast.Str("%s" % "--disable-dev-shm-usage")]
)))
body.extend(headless_setup)
body.append(ast.Assign(
targets=[ast_attr("self.driver")],
value=ast_call(
func=ast_attr("webdriver.Chrome"),
keywords=[
ast.keyword(
arg="service_log_path",
value=ast.Str(self.wdlog)),
ast.keyword(
arg="chrome_options",
value=ast.Name(id="options"))])))
elif browser == 'remote':
keys = sorted(self.capabilities.keys())
values = [str(self.capabilities[key]) for key in keys]
body.append(ast.Assign(
targets=[ast_attr("self.driver")],
value=ast_call(
func=ast_attr("webdriver.Remote"),
keywords=[