Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import logging
from datetime import date, timedelta
from typing import Dict
import pandas as pd
from databand import parameters
from dbnd import PipelineTask, output, parameter, task
from dbnd.testing.helpers_pytest import assert_run_task
from dbnd_test_scenarios.test_common.task.factories import TTask
class DummyTask(TTask):
float_param = parameters.FloatParameter()
expected_param = parameters.FloatParameter()
timedelta_param = parameters.TimeDeltaParameter()
expected_timedelta_param = parameters.TimeDeltaParameter()
def run(self):
assert isinstance(self.float_param, float)
assert isinstance(self.timedelta_param, timedelta)
assert self.float_param == self.expected_param
assert self.timedelta_param == self.expected_timedelta_param
super(DummyTask, self).run()
class DummyWrapper(PipelineTask):
defaults = {DummyTask.float_param: "0.1", DummyTask.timedelta_param: "4d"}
output = output
def test_params_parse(self):
class TParseTask(TTask):
bool_param = parameter[bool]
int_param = parameter[int]
float_param = parameters.FloatParameter()
date_param = parameter[date]
timedelta_param = parameters.TimeDeltaParameter()
def run(self):
assert isinstance(self.float_param, float)
super(TParseTask, self).run()
target = TParseTask(
bool_param="yes",
int_param="666",
float_param="123.456",
date_param="2018-11-10",
timedelta_param="44d",
)
assert_run_task(target)
tasks.DataSourceTask,
tasks.PipelineTask,
targets.Target,
targets.DataTarget,
dbnd._core.task_build.task_namespace.namespace,
parameters.Parameter,
parameters.DateHourParameter,
parameters.DateMinuteParameter,
parameters.DateSecondParameter,
parameters.DateParameter,
parameters.MonthParameter,
parameters.YearParameter,
parameters.DateIntervalParameter,
parameters.TimeDeltaParameter,
parameters.IntParameter,
parameters.FloatParameter,
parameters.BoolParameter,
]
assert len(expected) > 0
import logging
from datetime import date, timedelta
from typing import Dict
import pandas as pd
from databand import parameters
from dbnd import PipelineTask, output, parameter, task
from dbnd.testing.helpers_pytest import assert_run_task
from dbnd_test_scenarios.test_common.task.factories import TTask
class DummyTask(TTask):
float_param = parameters.FloatParameter()
expected_param = parameters.FloatParameter()
timedelta_param = parameters.TimeDeltaParameter()
expected_timedelta_param = parameters.TimeDeltaParameter()
def run(self):
assert isinstance(self.float_param, float)
assert isinstance(self.timedelta_param, timedelta)
assert self.float_param == self.expected_param
assert self.timedelta_param == self.expected_timedelta_param
super(DummyTask, self).run()
class DummyWrapper(PipelineTask):
defaults = {DummyTask.float_param: "0.1", DummyTask.timedelta_param: "4d"}
output = output