[AIRFLOW-2692] Add job_name as templated parameter in AWS Batch Operator

Closes #3557 from craigforster/master

airflow/contrib/operators/awsbatch_operator.py

@@ -36,7 +36,7 @@ class AWSBatchOperator(BaseOperator):
     .. warning: the queue parameter was renamed to job_queue to segreggate the
                 internal CeleryExecutor queue from the AWS Batch internal queue.
 
-    :param job_name: the name for the job that will run on AWS Batch
+    :param job_name: the name for the job that will run on AWS Batch (templated)
     :type job_name: str
     :param job_definition: the job definition name on AWS Batch
     :type job_definition: str
@@ -59,7 +59,7 @@ class AWSBatchOperator(BaseOperator):
     ui_color = '#c3dae0'
     client = None
     arn = None
-    template_fields = ('overrides',)
+    template_fields = ('job_name', 'overrides',)
 
     @apply_defaults
     def __init__(self, job_name, job_definition, job_queue, overrides, max_retries=4200,

tests/contrib/operators/test_awsbatch_operator.py

@@ -71,7 +71,7 @@ def test_init(self):
         self.aws_hook_mock.assert_called_once_with(aws_conn_id=None)
 
     def test_template_fields_overrides(self):
-        self.assertEqual(self.batch.template_fields, ('overrides',))
+        self.assertEqual(self.batch.template_fields, ('job_name', 'overrides',))
 
     @mock.patch.object(AWSBatchOperator, '_wait_for_task_ended')
     @mock.patch.object(AWSBatchOperator, '_check_success_task')
@@ -176,7 +176,7 @@ def test_check_success_tasks_raises_pending(self):
         # Ordering of str(dict) is not guaranteed.
         self.assertIn('This task is still pending ', str(e.exception))
 
-    def test_check_success_tasks_raises_mutliple(self):
+    def test_check_success_tasks_raises_multiple(self):
         client_mock = mock.Mock()
         self.batch.jobId = '8ba9d676-4108-4474-9dca-8bbac1da9b19'
         self.batch.client = client_mock