fix template_fields in the decorator task.docker (#29586)

* fix template_fields in the decorator `task.docker`

* add tests

* check the `container_name` matched

* remove the docker run

* Update tests/providers/docker/decorators/test_docker.py

Co-authored-by: Andrey Anshin <[email protected]>

* fix creating TaskInstance

* manually fix code styles and other stuff, like use ' instead of "

* fix static check!!

* fix static check: No overload variant of "__add__" of "tuple" matches argument type "Sequence[str]"

* Update airflow/providers/docker/decorators/docker.py

Co-authored-by: Josh Fell <[email protected]>

* another static_check fix

---------

Co-authored-by: Andrey Anshin <[email protected]>
Co-authored-by: Josh Fell <[email protected]>

Author: 3 people

airflow/providers/docker/decorators/docker.py

@@ -77,7 +77,7 @@ class _DockerDecoratedOperator(DecoratedOperator, DockerOperator):
 
     custom_operator_name = "@task.docker"
 
-    template_fields: Sequence[str] = ("op_args", "op_kwargs")
+    template_fields: Sequence[str] = (*DockerOperator.template_fields, "op_args", "op_kwargs")
 
     # since we won't mutate the arguments, we should just do the shallow copy
     # there are some cases we can't deepcopy the objects (e.g protobuf).

tests/providers/docker/decorators/test_docker.py

@@ -20,6 +20,7 @@
 
 from airflow.decorators import task
 from airflow.exceptions import AirflowException
+from airflow.models import TaskInstance
 from airflow.models.dag import DAG
 from airflow.utils import timezone
 from airflow.utils.state import TaskInstanceState
@@ -60,6 +61,19 @@ def f(num_results):
         assert isinstance(result, list)
         assert len(result) == 50
 
+    def test_basic_docker_operator_with_template_fields(self, dag_maker):
+        @task.docker(image="python:3.9-slim", container_name="python_{{dag_run.dag_id}}", auto_remove="force")
+        def f():
+            raise RuntimeError("Should not executed")
+
+        with dag_maker():
+            ret = f()
+
+        dr = dag_maker.create_dagrun()
+        ti = TaskInstance(task=ret.operator, run_id=dr.run_id)
+        rendered = ti.render_templates()
+        assert rendered.container_name == f"python_{dr.dag_id}"
+
     def test_basic_docker_operator_multiple_output(self, dag_maker):
         @task.docker(image="python:3.9-slim", multiple_outputs=True, auto_remove="force")
         def return_dict(number: int):