Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import os
import sys
import tempfile
from bzt.engine import Configuration
fp, filename = tempfile.mkstemp()
os.write(fp, sys.stdin.read())
conf = Configuration()
conf.load([filename])
conf.dump(filename, Configuration.YAML)
sys.stdout.write(open(filename).read())
def get_token():
a = Configuration()
a.load([os.path.expanduser("~/.bzt-rc")])
return a['modules']['blazemeter']['token']
def test_filtering(self):
obj = Configuration()
obj.merge({
"drop": "me",
"also-drop": {"this": "drop"},
"and-also-drop": ["thelist"],
"but-keep": "value",
"and-also-keep": {
"nested": "value",
"while-dropping": "some"
},
"filter-subitems": {
"keep": "value",
"drop": "some"
}
})
rules = {
def test_merge(self):
obj = Configuration()
configs = [
RESOURCES_DIR + "yaml/test.yml",
RESOURCES_DIR + "json/merge1.json",
RESOURCES_DIR + "json/merge2.json",
]
obj.load(configs)
fname = temp_file()
obj.dump(fname, Configuration.JSON)
with open(fname) as fh:
ROOT_LOGGER.debug("JSON:\n%s", fh.read())
jmeter = obj['modules']['jmeter']
classval = jmeter['class']
self.assertEquals("bzt.modules.jmeter.JMeterExecutor", classval)
self.assertEquals("value", obj['key'])
self.assertEquals(6, len(obj["list-append"]))
self.assertEquals(2, len(obj["list-replace"]))
def test_elementwise_merge(self):
obj = Configuration()
obj.merge({
"execution": [{
"executor": "jmeter",
"iterations": 10,
}],
})
obj.merge({
"$execution": [{"iterations": 20}],
})
self.assertEqual(obj["execution"][0]["iterations"], 20)
def test_save(self):
obj = Configuration()
obj.merge({
"str": "text",
"uc": six.u("ucstring")
})
fname = temp_file()
obj.dump(fname, Configuration.YAML)
with open(fname) as fh:
written = fh.read()
ROOT_LOGGER.debug("YAML:\n%s", written)
self.assertNotIn("unicode", written)
def test_elementwise_merge_right_is_bigger(self):
obj = Configuration()
obj.merge({
"execution": [{
"executor": "jmeter",
"iterations": 10,
}],
})
obj.merge({
"$execution": [{"iterations": 20}, {"iterations": 30}],
})
self.assertEqual(obj["execution"][0]["iterations"], 20)
self.assertEqual(obj["execution"][1]["iterations"], 30)
:type configs: list
:return: list
"""
jmxes = []
for filename in configs[:]:
if filename.lower().endswith(".jmx"):
jmxes.append(filename)
configs.remove(filename)
if jmxes:
self.log.debug("Adding JMX shorthand config for: %s", jmxes)
fds = NamedTemporaryFile(prefix="jmx_", suffix=".json")
fname = fds.name
fds.close()
config = Configuration()
for jmx_file in jmxes:
piece = BetterDict.from_dict({"executor": "jmeter", "scenario": {"script": jmx_file}})
config.get(EXEC, [], force_set=True).append(piece) # Does it brake single execution?
config.dump(fname, Configuration.JSON)
return [fname]
else:
return []
:type configs: list
:return: list
"""
jtls = []
for filename in configs[:]:
if filename.lower().endswith(".jtl"):
jtls.append(filename)
configs.remove(filename)
if jtls:
self.log.debug("Adding JTL shorthand config for: %s", jtls)
fds = NamedTemporaryFile(prefix="jtl_", suffix=".json")
fname = fds.name
fds.close()
config = Configuration()
for jtl in jtls:
piece = BetterDict.from_dict({"executor": "external-results-loader", "data-file": jtl})
config.get(EXEC, [], force_set=True).append(piece)
config.dump(fname, Configuration.JSON)
return [fname]
else:
return []