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_cli_raises1(self):
"""
Verify that we also read from the config when we build tasks from the
command line parsers.
"""
dbnd_run_cmd(
[
"TConfigTask",
"-s",
"TConfigTask.t_param=124",
"-s",
"TConfigTask.t_param=123",
]
)
run_locally__raises(
DatabandRunError,
[
"TConfigTask",
"-s",
"TConfigTask.t_param=123",
"-s",
"TConfigTask.t_param=124",
def test_use_config_class_with_configuration(self):
result = dbnd_run_cmd(
[
"MyConfigTester",
"--set",
json.dumps({"MyConfig": {"mc_p": "123", "mc_q": "345"}}),
]
)
actual = result.task.t_output.load(object)
assert actual == [123, 345]
def dbnd_run_task_with_output(
self, run_args, task=TTask, output_parameter=TTask.t_output, call_f=dbnd_run_cmd
):
local_file = str(self.tmpdir.join("output_file.txt"))
run_args = [
TTask.task_definition.full_task_family,
"--set",
"TTask.t_output=%s" % local_file,
] + run_args
logging.info("Running command:%s", subprocess.list2cmdline(run_args))
dbnd_run_cmd(run_args)
assert os.path.exists(local_file), (
"Output file %s wasn't created by task!" % local_file
)
my_target.write_df(pandas_data_frame)
@task
def t_f_defaults_cmdline(
a_str="",
b_datetime=datetime.datetime.utcnow(),
c_timedelta=datetime.timedelta(),
d_int=0,
):
assert a_str == "1"
assert b_datetime.isoformat() == "2018-01-01T10:10:10.100000+00:00"
assert c_timedelta == datetime.timedelta(days=5)
assert d_int == 1
return pandas_data_frame
dbnd_run_cmd(
[
"t_f_defaults_cmdline",
"-r",
"a_str=1",
"-r",
"b_datetime=2018-01-01T101010.1",
"-r",
"c_timedelta=5d",
"--set",
json.dumps({"t_f_defaults_cmdline": {"d_int": 1}}),
]
def dbnd_run_task_with_output(
self, run_args, task=TTask, output_parameter=TTask.t_output, call_f=dbnd_run_cmd
):
local_file = str(self.tmpdir.join("output_file.txt"))
run_args = [
TTask.task_definition.full_task_family,
"--set",
"TTask.t_output=%s" % local_file,
] + run_args
logging.info("Running command:%s", subprocess.list2cmdline(run_args))
dbnd_run_cmd(run_args)
assert os.path.exists(local_file), (
"Output file %s wasn't created by task!" % local_file
)
def test_bool_false_default(self):
result = dbnd_run_cmd(["Baz"])
assert result.task.bool is False
def test_local_params(self):
class MyTask(TTask):
param1 = parameter[int]
param2 = parameter.value(default=False)
def run(self):
super(MyTask, self).run()
assert self.param1 == 1 and self.param2
assert dbnd_run_cmd("MyTask -r param1=1 -r param2=True")
def test_bool_false_cmdline(self):
result = dbnd_run_cmd(["BazTrue", "-r", "bool=False"])
assert result.task.bool is False
def test_function_cmd_dynamic_name(self):
dbnd_test_name = "unique_test_name_" + uuid.uuid4().hex.upper()[0:6]
cmdline = ["dbnd_sanity_check", "--task-name", dbnd_test_name]
result = dbnd_run_cmd(cmdline)
assert result.task.task_name == dbnd_test_name
def test_not_a_class(self):
with pytest.raises(TaskClassNotFoundException):
dbnd_run_cmd(["NotAClass"])