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_no_double_format_failure(self):
to_format = "{{b}}"
final_value = "{b}"
format_variables = {"b": final_value}
formatted = format_keys(to_format, format_variables)
assert formatted == final_value
formatted_2 = format_keys(formatted, {})
assert formatted_2 == final_value
def test_no_double_format_failure(self):
to_format = "{{b}}"
final_value = "{b}"
format_variables = {"b": final_value}
formatted = format_keys(to_format, format_variables)
assert formatted == final_value
formatted_2 = format_keys(formatted, {})
assert formatted_2 == final_value
def __init__(self, client, rspec, test_block_config):
expected = {"topic", "payload", "json", "qos", "retain"}
check_expected_keys(expected, rspec)
publish_args = get_publish_args(rspec, test_block_config)
self._prepared = functools.partial(client.publish, **publish_args)
# Need to do this here because get_publish_args will modify the original
# input, which we might want to use to format. No error handling because
# all the error handling is done in the previous call
self._original_publish_args = format_keys(rspec, test_block_config["variables"])
def test_bad_format_string_extra(self):
"""Extra things in format string"""
with pytest.raises(exceptions.InvalidFormattedJsonError):
format_keys(ForceIncludeToken("{fd}gg"), {"fd": "123"})
if test_spec.get("includes"):
# Need to do this separately here so there is no confusion between global and included stages
for included in test_spec["includes"]:
for stage in included.get("stages", {}):
if stage["id"] in stage_ids(available_stages):
msg = "Stage id '{}' defined in stage-included test which was already defined in global configuration - this will be an error in future!".format(
stage["id"]
)
logger.warning(msg)
warnings.warn(msg, FutureWarning)
included_stages = []
for included in test_spec["includes"]:
if "variables" in included:
formatted_include = format_keys(
included["variables"], {"tavern": tavern_box}
)
test_block_config["variables"].update(formatted_include)
for stage in included.get("stages", []):
if stage["id"] in stage_ids(included_stages):
raise exceptions.DuplicateStageDefinitionError(
"Stage with specified id already defined: {}".format(
stage["id"]
)
)
included_stages.append(stage)
else:
included_stages = []
return included_stages
def test_format_success(self):
to_format = {"a": "{b}"}
final_value = "formatted"
format_variables = {"b": final_value}
assert format_keys(to_format, format_variables)["a"] == final_value
# skipif: 'https' in '{hostname}'
# skipif: '{hostname}'.contains('ignoreme')
fmt_vars = {}
global_cfg = load_global_cfg(self.config)
fmt_vars.update(**global_cfg.get("variables", {}))
included = test_spec.get("includes", [])
for i in included:
fmt_vars.update(**i.get("variables", {}))
# Needed if something in a config file uses tavern.env_vars
tavern_box = Box({"tavern": {"env_vars": dict(os.environ)}})
try:
fmt_vars = format_keys(fmt_vars, tavern_box)
except exceptions.MissingFormatError as e:
# eg, if we have {tavern.env_vars.DOESNT_EXIST}
msg = "Tried to use tavern format variable that did not exist"
raise_from(exceptions.MissingFormatError(msg), e)
return fmt_vars
def test_format_missing_raises(self):
to_format = {"a": "{b}"}
with pytest.raises(exceptions.MissingFormatError):
format_keys(to_format, {})
Yields:
YamlItem: Tavern YAML test
"""
item = YamlItem(test_spec["test_name"], self, test_spec, self.fspath)
original_marks = test_spec.get("marks", [])
if original_marks:
fmt_vars = self._get_test_fmt_vars(test_spec)
pytest_marks = []
formatted_marks = []
for m in original_marks:
if isinstance(m, str):
# a normal mark
m = format_keys(m, fmt_vars)
pytest_marks.append(getattr(pytest.mark, m))
elif isinstance(m, dict):
# skipif or parametrize (for now)
for markname, extra_arg in m.items():
# NOTE
# cannot do 'skipif' and rely on a parametrized
# argument.
try:
extra_arg = format_keys(extra_arg, fmt_vars)
except exceptions.MissingFormatError as e:
msg = "Tried to use mark '{}' (with value '{}') in test '{}' but one or more format variables was not in any configuration file used by the test".format(
markname, extra_arg, test_spec["test_name"]
)
# NOTE
# we could continue and let it fail in the test, but
# this gives a better indication of what actually
def delay(stage, when, variables):
"""Look for delay_before/delay_after and sleep
Args:
stage (dict): test stage
when (str): 'before' or 'after'
"""
try:
length = format_keys(stage["delay_{}".format(when)], variables)
except KeyError:
pass
else:
logger.debug("Delaying %s request for %.2f seconds", when, length)
time.sleep(length)