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_log_metric(self):
@task
def t_f_metric(a=5):
log_metric("t_f", a)
t = assert_run_task(t_f_metric.t())
assert (
t.ctrl.last_task_run.meta_files.get_metric_target("t_f").read().split()[1]
== "5"
)
def test_unknown_format(self):
task = TTextDataTask(text_data=scenario_path("data/some_unknown_ext.myext"))
assert_run_task(task)
def test_ret_dict(self):
@task(result=(output(name="o_a").csv[List[str]], "o_b"))
def t_f(a=5):
return {"o_a": [str(a)], "o_b": ["2"]}
t = assert_run_task(t_f.t(a=6))
assert "6\n" == t.o_a.read()
def test_feature_store_pipeline(self):
t = assert_run_task(CreateFeatureStoreViaClass())
assert_run_task(CalculateAdvancedFeatures(store=t.store))
assert_run_task(report_features.t(feature_store=t.store))
def test_prod_immutable_output_dict_dev(self):
t = assert_run_task(TProdImmutbaleOutputs())
print(t)
def test_custom_type_parameter(self):
t = assert_run_task(BuildMyData())
assert_run_task(MyDataReport(my_data=t.my_data))
def test_sanity(self):
assert_run_task(dbnd_sanity_check.task())
def test_simple_no_call(self):
@task
def t_f_nocall(a=5):
assert a == 6
t_f_nocall(a=6)
assert_run_task(t_f_nocall.t(a=6))
def test_input_lines_as_filename_args(self, target_1_2):
t_f_data_list_str_with_int_default(target_1_2.readlines(), 2)
target = t_f_data_list_str_with_int_default.t(target_1_2.path, 2)
assert_run_task(target)
def test_simple_func(self):
assert_run_task(
t_d_1.task(
a_str="1", b_datetime="2018-01-01T101010.1", c_timedelta="5d", d_int="1"
)