add support for templates schema validation after transform

src/airflow_declarative/schema.py

@@ -234,10 +234,13 @@ def represent_callable(self, obj):
 
 WITH_ITEMS = List(ANY) | Dict(using=CALLBACK) | Dict(from_stdout=STRING)
 
+TEMPLATE_OPERATORS = Mapping(STRING, ANY)
+TEMPLATE_SENSORS = Mapping(STRING, ANY)
+
 DO_TEMPLATE = Dict(
     {
-        OptionalKey("operators"): OPERATORS,
-        OptionalKey("sensors"): SENSORS,
+        OptionalKey("operators"): TEMPLATE_OPERATORS,
+        OptionalKey("sensors"): TEMPLATE_SENSORS,
         OptionalKey("flow"): FLOW,
         Key("with_items"): WITH_ITEMS,
     }

src/airflow_declarative/transformer.py

@@ -108,7 +108,7 @@ def transform(schema):
     schema0 = ensure_schema(schema)
     schema1 = transform_templates(schema0)
     schema2 = transform_defaults(schema1)
-    return schema2
+    return ensure_schema(schema2)
 
 
 def transform_templates(schema):

tests/dags/good/template-with-invalid-schema-before-do.yaml

@@ -0,0 +1,59 @@
+#
+# Copyright 2019, Rambler Digital Solutions
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+dags:
+  callback_dag:
+    args:
+      start_date: 2017-07-27
+      schedule_interval: 1d
+    do:
+    - operators:
+        operator_{{ item.name }}:
+          callback: 'tests.utils:{{ item.callback }}'
+          callback_args:
+            '{{ item.args }}'
+      with_items:
+      - name: multi
+        callback: MultiParamOperator
+        args:
+          param1: egg
+          param2: bacon
+          param3: spam
+      - name: simple
+        callback: Operator
+        args:
+          param: egg
+    - operators:
+        operator_{{ item.name }}:
+          class: '{{ item.class }}'
+          args:
+            '{{ item.args }}'
+      with_items:
+      - name: bash
+        class: airflow.operators.bash_operator:BashOperator
+        args:
+          bash_command: 'echo "test"'
+          end_date: 2019-10-25
+    - sensors:
+        sensor_{{ item.name }}:
+          callback: 'tests.utils:{{ item.callback }}'
+      flow:
+        sensor_{{item.name}}:
+        - operator_bash
+        - operator_simple
+      with_items:
+      - name: simple
+        callback: sensor

tests/test_template_with_invalid_schema_before_do.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2019, Rambler Digital Solutions
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+from datetime import date
+
+import pytest
+from airflow.operators.bash_operator import BashOperator
+from tests.utils import MultiParamOperator, Operator, sensor
+
+import airflow_declarative
+
+
+@pytest.fixture()
+def dag(good_dag_path):
+    path = good_dag_path("template-with-invalid-schema-before-do")
+    dags = airflow_declarative.from_path(path)
+
+    assert len(dags) == 1
+
+    dag = dags[0]
+
+    return dag
+
+
+def test_callback_params(dag):
+    operator_multi = dag.task_dict["operator_multi"]
+    assert operator_multi._callback == MultiParamOperator
+    assert operator_multi._callback_args == {
+        "param1": "egg",
+        "param2": "bacon",
+        "param3": "spam",
+    }
+
+    operator_simple = dag.task_dict["operator_simple"]
+    assert operator_simple._callback == Operator
+    assert operator_simple._callback_args == {"param": "egg"}
+
+    operator_bash = dag.task_dict["operator_bash"]
+    assert isinstance(operator_bash, BashOperator)
+    assert operator_bash.bash_command == 'echo "test"'
+    assert operator_bash.end_date == date(2019, 10, 25)
+
+    sensor_simple = dag.task_dict["sensor_simple"]
+    assert sensor_simple._callback == sensor