Chciałbym dodać kilka testów jednostkowych dla naszych DAG, ale nie mogłem znaleźć żadnych. Czy istnieje ramowy test jednostkowy dla DAG? Istnieje szkielet testowania End-to-End, który istnieje, ale myślę, że jest martwy: https://issues.apache.org/jira/browse/AIRFLOW-79. Proszę sugerować, dzięki!Test jednostki Pythona w powietrzu?
Obecnie nie byłem w stanie znaleźć coś lepszego niż po prostu za pomocą BashOperator:
with DAG('platform-test', start_date=datetime(2017, 8, 29)) as dag:
test_command = "python3 -m unittest --verbose {}".format(platform_test_fname)
op = BashOperator(
task_id="platform-test",
bash_command=test_command,
)
Sprawdź swoje operatorów tak:
class TestMyOperator(TestCase):
def test_execute(self):
dag = DAG(dag_id='foo', start_date=datetime.now())
task = MyOperator(dag=dag, task_id='foo')
ti = TaskInstance(task=task, execution_date=datetime.now())
result = task.execute(ti.get_template_context())
self.assertEqual(result, 'foo')
Ja pracowałem na wyodrębnienie wszystkich moich niestandardową logikę do katalogu './plugins' i tworzenie zestawu testów dla tamtej logiki. Nie od końca do końca, ale w międzyczasie może być dobrym rozwiązaniem: https://airflow.incubator.apache.org/plugins.html. Jeśli wkładasz do nietestowanej logiki do swojego katalogu './Dags', możesz rozważyć podzielenie tego na wtyczki. – Mike