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_stepping_tg_ramp_no_proportion(self):
self.configure({
'execution': {
'steps': 5,
'concurrency': 170,
'scenario': {
'script': RESOURCES_DIR + '/jmeter/jmx/stepping_ramp_up.jmx'},
'ramp-up': '1m',
'distributed': ['127.0.0.1'],
'hold-for': '2m'}})
self.obj.prepare()
modified_xml_tree = etree.fromstring(open(self.obj.modified_jmx, "rb").read())
mod_stepping_tgs = modified_xml_tree.findall(".//kg.apc.jmeter.threads.SteppingThreadGroup")
self.assertEqual(0, len(mod_stepping_tgs)) # generation of STG is obsolete
def test_logic_include_data_sources(self):
self.configure({
'scenarios': {
'login': {
'data-sources': [RESOURCES_DIR + "test1.csv"],
'requests': ['http://blazedemo.com/auth/${test1}']}},
'execution': {
'scenario': {
"data-sources": [RESOURCES_DIR + "test2.csv"],
"requests": [
{"include-scenario": "login"},
"http://example.com/${test2}"]}}})
self.obj.prepare()
xml_tree = etree.fromstring(open(self.obj.original_jmx, "rb").read())
thread_group = xml_tree.find('.//hashTree[@type="tg"]')
scenario_dataset = xml_tree.find('.//hashTree[@type="tg"]/CSVDataSet')
self.assertIsNotNone(scenario_dataset)
filename = scenario_dataset.find('stringProp[@name="filename"]')
self.assertEqual(filename.text, get_full_path(RESOURCES_DIR + "test2.csv"))
login_controler = thread_group.find('GenericController')
self.assertIsNotNone(login_controler)
login_ht = login_controler.getnext()
login_dataset = login_ht.find('CSVDataSet')
self.assertIsNotNone(login_dataset)
filename = scenario_dataset.find('stringProp[@name="filename"]')
self.assertEqual(filename.text, get_full_path(RESOURCES_DIR + "test2.csv"))
def test_plugin_extractor_config_reader(self):
xml_tree = etree.fromstring(open(RESOURCES_DIR + "jmeter/jmx/json_extractors.jmx", "rb").read())
target = {'pv1': {'default': 'pd1', 'from_variable': 'pJV', 'jsonpath': 'pe1'},
'pv2': {'default': None, 'from_variable': None, 'jsonpath': 'pe2'}}
self.assertEqual(target, self.get_plugin_json_extractor_config(xml_tree))
def test_xpath_assertion(self):
self.configure(json.loads(open(RESOURCES_DIR + "json/get-post.json").read()))
self.obj.prepare()
target_jmx = os.path.join(self.obj.engine.artifacts_dir, "requests.jmx")
modified_xml_tree = etree.fromstring(open(target_jmx, "rb").read())
assertions = modified_xml_tree.findall(".//XPathAssertion")
self.assertEqual(2, len(assertions))
simplified = assertions[0]
self.assertEqual(simplified.find(".//stringProp[@name='XPath.xpath']").text, "/note/to")
self.assertEqual(simplified.find(".//boolProp[@name='XPath.validate']").text, "false")
self.assertEqual(simplified.find(".//boolProp[@name='XPath.whitespace']").text, "true")
self.assertEqual(simplified.find(".//boolProp[@name='XPath.tolerant']").text, "false")
self.assertEqual(simplified.find(".//boolProp[@name='XPath.negate']").text, "false")
full_form = assertions[1]
self.assertEqual(full_form.find(".//stringProp[@name='XPath.xpath']").text, "/note/from")
self.assertEqual(full_form.find(".//boolProp[@name='XPath.validate']").text, "true")
self.assertEqual(full_form.find(".//boolProp[@name='XPath.whitespace']").text, "true")
self.assertEqual(full_form.find(".//boolProp[@name='XPath.tolerant']").text, "true")
self.assertEqual(full_form.find(".//boolProp[@name='XPath.negate']").text, "true")
self.configure({
"execution": {
"scenario": {
"requests": [{
"url": "http://blazedemo.com/",
"jsr223": {
"language": "javascript",
"script-file": script,
"parameters": "first second"
}
}]
}
}
})
self.obj.prepare()
xml_tree = etree.fromstring(open(self.obj.modified_jmx, "rb").read())
post_procs = xml_tree.findall(".//JSR223PostProcessor[@testclass='JSR223PostProcessor']")
self.assertEqual(1, len(post_procs))
jsr = post_procs[0]
self.assertEqual(script, jsr.find(".//stringProp[@name='filename']").text)
self.assertEqual("javascript", jsr.find(".//stringProp[@name='scriptLanguage']").text)
self.assertEqual("first second", jsr.find(".//stringProp[@name='parameters']").text)
self.assertEqual("true", jsr.find(".//stringProp[@name='cacheKey']").text)
def test_keep_cookies(self):
self.configure(scenario={"requests": ["http://localhost"],
"store-cookie": True
})
self.obj.save(self.jmx)
xml_tree = etree.fromstring(open(self.jmx, "rb").read())
include = xml_tree.find(".//boolProp[@name='CookieManager.clearEachIteration']")
self.assertIsNotNone(include)
self.assertEqual(include.text, 'false')
def test_jsonpath_assertion(self):
self.configure(json.loads(open(RESOURCES_DIR + "json/get-post.json").read()))
self.obj.prepare()
target_jmx = os.path.join(self.obj.engine.artifacts_dir, "requests.jmx")
modified_xml_tree = etree.fromstring(open(target_jmx, "rb").read())
path = ".//com.atlantbh.jmeter.plugins.jsonutils.jsonpathassertion.JSONPathAssertion"
assertions = modified_xml_tree.findall(path)
self.assertEqual(4, len(assertions))
vals = [
{'path': '$.', 'exp_val': None, 'valid': 'false',
'null': 'false', 'invert': 'false', 'regexp': 'true'},
{'path': '$.res[0].type', 'exp_val': 'some_value.1', 'valid': 'true',
'null': 'false', 'invert': 'false', 'regexp': 'true'},
{'path': '$.res[1].ip', 'exp_val': 'some_value.2', 'valid': 'true',
'null': 'false', 'invert': 'true', 'regexp': 'false'},
{'path': '$.res[2].default', 'exp_val': None, 'valid': 'false',
'null': 'true', 'invert': 'false', 'regexp': 'true'}]
for num in range(len(assertions)):
assertion = assertions[num]
val = vals[num]
def test_request_logic_set_vars(self):
self.configure({
'execution': {
'scenario': {
"requests": [{
"set-variables": {"foo": "bar"}}]}}})
self.obj.prepare()
xml_tree = etree.fromstring(open(self.obj.modified_jmx, "rb").read())
set_var_action = xml_tree.find(".//" + JMX.SET_VAR_ACTION)
self.assertIsNotNone(set_var_action)
self.assertEqual("foo", set_var_action.find(".//stringProp[@name='Argument.name']").text)
self.assertEqual("bar", set_var_action.find(".//stringProp[@name='Argument.value']").text)
def test_regexp_extractors(self):
self.configure({"execution":
{"scenario":
{"requests": [{
"url": "http://localhost",
"extract-regexp": {
"varname": {
"regexp": "???",
"scope": "variable",
"from-variable": "RESULT"}
}}]}}})
self.obj.prepare()
xml_tree = etree.fromstring(open(self.obj.modified_jmx, "rb").read())
self.assertEqual("body", xml_tree.findall(".//stringProp[@name='RegexExtractor.useHeaders']")[0].text)
self.assertEqual("???", xml_tree.findall(".//stringProp[@name='RegexExtractor.regex']")[0].text)
self.assertEqual("variable", xml_tree.findall(".//stringProp[@name='Sample.scope']")[0].text)
self.assertEqual("RESULT", xml_tree.findall(".//stringProp[@name='Scope.variable']")[0].text)
pass_fail.processors.append(pass_fail2)
obj.engine.reporters.append(pass_fail)
obj.engine.reporters.append(BlazeMeterUploader())
path_from_config = tempfile.mktemp(suffix='.xml', prefix='junit-xml_passfail', dir=obj.engine.artifacts_dir)
obj.parameters.merge({"filename": path_from_config, "data-source": "pass-fail"})
obj.prepare()
obj.last_second = DataPoint(0)
obj.post_process()
with open(obj.report_file_path, 'rb') as fds:
f_contents = fds.read()
ROOT_LOGGER.info("File: %s", f_contents)
xml_tree = etree.fromstring(f_contents)
self.assertEqual('testsuites', xml_tree.tag)
suite = xml_tree.getchildren()[0]
self.assertEqual('testsuite', suite.tag)
test_cases = suite.getchildren()
self.assertEqual(4, len(test_cases))
self.assertEqual('testcase', test_cases[0].tag)
self.assertEqual('error', test_cases[0].getchildren()[1].tag)
self.assertEqual('error', test_cases[2].getchildren()[1].tag)
sys_out = test_cases[0].getchildren()[0]
self.assertEqual('system-out', sys_out.tag)
self.assertIn('BlazeMeter report link: http://test/report/123', sys_out.text)