Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# raise exception
with pytest.raises(UnknownParameterError) as e:
with config(
{
"TTask": {
"t_parammm": 2,
"validate_no_extra_params": ParamValidation.error,
}
}
):
TTask()
assert "Did you mean: t_param" in e.value.help_msg
# log warning to log
with config(
{
"TTask": {
"t_parammm": 2,
"validate_no_extra_params": ParamValidation.warn,
}
}
):
TTask()
# tried to add a capsys assert here but couldn't get it to work
# do nothing
with config(
{
"TTask": {
"t_parammm": 2,
"validate_no_extra_params": ParamValidation.disabled,
def test_error_on_same_from(self):
with pytest.raises(Exception):
with config(
{"unknown_task_with_from": {"_from": "unknown_task_with_from"}}
):
get_task_registry().build_dbnd_task("unknown_task_with_from")
def test_read_environ_config(self):
os.environ["DBND__TEST_SECTION__TEST_KEY"] = "TEST_VALUE"
actual = read_environ_config()
with config(actual):
assert config.get("test_section", "test_key") == "TEST_VALUE"
def test_override_config_values_from_context(self):
with config(
{"task_from_config": {"parameter_from_config": "from_context_override"}}
):
assert (
config.get("task_from_config", "parameter_from_config")
== "from_context_override"
)
task_from_config.dbnd_run(expected="from_context_override")
def test_case_insensitive_parameter_building(self):
# First run with correct case
with config(
{
"CaseSensitiveParameterTask": {
"TParam": 2,
"validate_no_extra_params": ParamValidation.error,
}
}
):
dbnd_run = CaseSensitiveParameterTask().dbnd_run()
assert dbnd_run.task.TParam == 2
# Second run with incorrect lower case
with config(
{
"CaseSensitiveParameterTask": {
"tparam": 3,
"validate_no_extra_params": ParamValidation.error,
}
def test_prod_immutable_output_dict_prod(self):
env = get_databand_context().env
prod_env = env.clone(production=True)
task = TProdImmutbaleOutputs(task_env=prod_env)
assert task.task_enabled_in_prod
assert task.task_signature[:5] not in str(task.splits)
config.log_current_config()
actual = assert_run_task(task)
print(actual)
def test_list(self):
with config({"ListFoo": {"my_list": "1,2"}}) as c:
ListFoo().dbnd_run()
assert ListFoo._val == [1, 2]
def databand_config():
config.load_system_configs()
return config
def dbnd_kube_check(check_time=datetime.datetime.now(), sleep_time_sec=120):
# type: ( datetime.datetime, int)-> str
config.log_current_config(as_table=True)
logger.info("Running Kube Sanity Check!")
if sleep_time_sec:
logger.info("sleeping for %s", sleep_time_sec)
sleep(sleep_time_sec)
return "Databand checked at %s" % check_time
def wrapper(*args, **kwargs):
from dbnd import new_dbnd_context, config
from dbnd._core.settings import CoreConfig
with config({CoreConfig.tracker: ""}, source="fast_dbnd_context"):
with new_dbnd_context(name="fast_dbnd_context", autoload_modules=False):
f(*args, **kwargs)