title: Templates not rendered in Airflow unit test
body: <p>I have a Python module which is structured as follows:</p>
  
  <pre><code>my_module/
    ...
    tests/
      __init__.py
      my_test.py
      ...
  </code></pre>
  
  <p>where <code>my_test.py</code> is defined as follows:</p>
  
  <pre><code>from __future__ import print_function, unicode_literals
  
  import os
  import unittest
  from datetime import timedelta, date
  
  from airflow import configuration
  from airflow.models import TaskInstance as TI, DAG, DagRun
  from airflow.operators.python_operator import PythonOperator
  from airflow.settings import Session
  from airflow.utils import timezone
  from airflow.utils.state import State
  
  DEFAULT_DATE = timezone.datetime(2016, 1, 1)
  END_DATE = timezone.datetime(2016, 1, 2)
  INTERVAL = timedelta(hours=12)
  FROZEN_NOW = timezone.datetime(2016, 1, 2, 12, 1, 1)
  
  TI_CONTEXT_ENV_VARS = ['AIRFLOW_CTX_DAG_ID',
                         'AIRFLOW_CTX_TASK_ID',
                         'AIRFLOW_CTX_EXECUTION_DATE',
                         'AIRFLOW_CTX_DAG_RUN_ID']
  
  
  class Call:
      def __init__(self, *args, **kwargs):
          self.args = args
          self.kwargs = kwargs
  
  
  def build_recording_function(calls_collection):
      """
      We can not use a Mock instance as a PythonOperator callable function or some tests fail with a
      TypeError: Object of type Mock is not JSON serializable
      Then using this custom function recording custom Call objects for further testing
      (replacing Mock.assert_called_with assertion method)
      """
      def recording_function(*args, **kwargs):
          calls_collection.append(Call(*args, **kwargs))
      return recording_function
  
  
  class PythonOperatorTest(unittest.TestCase):
      @classmethod
      def setUpClass(cls):
          super(PythonOperatorTest, cls).setUpClass()
  
          session = Session()
  
          session.query(DagRun).delete()
          session.query(TI).delete()
          session.commit()
          session.close()
  
      def setUp(self):
          super(PythonOperatorTest, self).setUp()
          configuration.load_test_config()
          self.dag = DAG(
              'test_dag',
              default_args={
                  'owner': 'airflow',
                  'start_date': DEFAULT_DATE},
              schedule_interval=INTERVAL)
          self.addCleanup(self.dag.clear)
          self.clear_run()
          self.addCleanup(self.clear_run)
  
      def tearDown(self):
          super(PythonOperatorTest, self).tearDown()
  
          session = Session()
  
          session.query(DagRun).delete()
          session.query(TI).delete()
          print(len(session.query(DagRun).all()))
          session.commit()
          session.close()
  
          for var in TI_CONTEXT_ENV_VARS:
              if var in os.environ:
                  del os.environ[var]
  
      def do_run(self):
          self.run = True
  
      def clear_run(self):
          self.run = False
  
      def is_run(self):
          return self.run
  
      def _assertCallsEqual(self, first, second):
          self.assertIsInstance(first, Call)
          self.assertIsInstance(second, Call)
          self.assertTupleEqual(first.args, second.args)
          self.assertDictEqual(first.kwargs, second.kwargs)
  
      def test_python_callable_arguments_are_templatized(self):
          """Test PythonOperator op_args are templatized"""
          recorded_calls = []
  
          task = PythonOperator(
              task_id='python_operator',
              # a Mock instance cannot be used as a callable function or test fails with a
              # TypeError: Object of type Mock is not JSON serializable
              python_callable=(build_recording_function(recorded_calls)),
              op_args=[
                  4,
                  date(2019, 1, 1),
                  "dag {{dag.dag_id}} ran on {{ds}}."
              ],
              dag=self.dag)
  
          self.dag.create_dagrun(
              run_id='manual__' + DEFAULT_DATE.isoformat(),
              execution_date=DEFAULT_DATE,
              start_date=DEFAULT_DATE,
              state=State.RUNNING
          )
          task.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
  
          self.assertEqual(1, len(recorded_calls))
          self._assertCallsEqual(
              recorded_calls[0],
              Call(4,
                   date(2019, 1, 1),
                   "dag {} ran on {}.".format(self.dag.dag_id, DEFAULT_DATE.date().isoformat()))
          )
  </code></pre>
  
  <p>In a terminal, when I run <code>nosetests test/my_test.py</code>, the test fails because the Jinja templates are not correctly rendered. The full log is given below.</p>
  
  <pre><code>======================================================================
  FAIL: Test PythonOperator op_args are templatized
  ----------------------------------------------------------------------
  Traceback (most recent call last):
    File "/home/user/my_module/tests/my_test.py", line 120, in test_python_callable_arguments_are_templatized
      "dag {} ran on {}.".format(self.dag.dag_id, DEFAULT_DATE.date().isoformat()))
    File "/home/user/my_module/tests/my_test.py", line 88, in _assertCallsEqual
      self.assertTupleEqual(first.args, second.args)
  AssertionError: Tuples differ: (4, datetime.date(2019, 1, 1), 'dag {{dag.dag_id}} ran on {{ds}}.') != (4, datetime.date(2019, 1, 1), 'dag test_dag ran on 2016-01-01.')
  
  First differing element 2:
  'dag {{dag.dag_id}} ran on {{ds}}.'
  'dag test_dag ran on 2016-01-01.'
  
  - (4, datetime.date(2019, 1, 1), 'dag {{dag.dag_id}} ran on {{ds}}.')
  ?                                     ^^   ---------        ^^^^^^
  
  + (4, datetime.date(2019, 1, 1), 'dag test_dag ran on 2016-01-01.')
  ?                                     ^^^^^           ^^^^^^^^^^
  
  -------------------- &gt;&gt; begin captured logging &lt;&lt; --------------------
  airflow.utils.log.logging_mixin.LoggingMixin: INFO: Reading the config from /home/user/airflow/airflow.cfg
  airflow.settings: INFO: Configured default timezone &lt;Timezone [UTC]&gt;
  airflow.logging_config: DEBUG: Unable to load custom logging, using default config instead
  --------------------- &gt;&gt; end captured logging &lt;&lt; ---------------------
  </code></pre>
  
  <p>However, there is nothing wrong with the code in <code>my_test.py</code> since it is only a subset of the file <a href="https://github.com/apache/airflow/blob/master/tests/operators/test_python_operator.py" rel="nofollow noreferrer">test_python_operator.py</a> file from <a href="https://github.com/apache/airflow" rel="nofollow noreferrer">Airflow github repository</a> (<code>v1-10-stable</code> branch). Naively, I would expect this test to run just fine but it does not.</p>
  
  <blockquote>
    <p>What am I missing ?</p>
  </blockquote>
  
  <p><strong>EDIT</strong>: I'm using apache-airflow 1.10.2, Python 3.6.8 and nose 1.3.7.</p>