[AIRFLOW-3251] KubernetesPodOperator now uses 'image_pull_secrets' ar…

…gument when creating Pods (apache#4188)

airflow/contrib/kubernetes/pod.py

@@ -54,6 +54,10 @@ class Pod:
     :type result: any
     :param image_pull_policy: Specify a policy to cache or always pull an image
     :type image_pull_policy: str
+    :param image_pull_secrets: Any image pull secrets to be given to the pod.
+                               If more than one secret is required, provide a
+                               comma separated list: secret_a,secret_b
+    :type image_pull_secrets: str
     :param affinity: A dict containing a group of affinity scheduling rules
     :type affinity: dict
     """

airflow/contrib/operators/kubernetes_pod_operator.py

@@ -45,6 +45,12 @@ class KubernetesPodOperator(BaseOperator):
     :param arguments: arguments of to the entrypoint. (templated)
         The docker image's CMD is used if this is not provided.
     :type arguments: list of str
+    :param image_pull_policy: Specify a policy to cache or always pull an image
+    :type image_pull_policy: str
+    :param image_pull_secrets: Any image pull secrets to be given to the pod.
+                               If more than one secret is required, provide a
+                               comma separated list: secret_a,secret_b
+    :type image_pull_secrets: str
     :param volume_mounts: volumeMounts for launched pod
     :type volume_mounts: list of VolumeMount
     :param volumes: volumes for launched pod. Includes ConfigMaps and PersistentVolumes
@@ -108,6 +114,7 @@ def execute(self, context):
             pod.secrets = self.secrets
             pod.envs = self.env_vars
             pod.image_pull_policy = self.image_pull_policy
+            pod.image_pull_secrets = self.image_pull_secrets
             pod.annotations = self.annotations
             pod.resources = self.resources
             pod.affinity = self.affinity

tests/contrib/minikube/test_kubernetes_pod_operator.py

@@ -84,6 +84,29 @@ def test_config_path(self, client_mock, launcher_mock):
                                        cluster_context='default',
                                        config_file=file_path)
 
+    @mock.patch("airflow.contrib.kubernetes.pod_launcher.PodLauncher.run_pod")
+    @mock.patch("airflow.contrib.kubernetes.kube_client.get_kube_client")
+    def test_image_pull_secrets_correctly_set(self, client_mock, launcher_mock):
+        from airflow.utils.state import State
+
+        fake_pull_secrets = "fakeSecret"
+        k = KubernetesPodOperator(
+            namespace='default',
+            image="ubuntu:16.04",
+            cmds=["bash", "-cx"],
+            arguments=["echo 10"],
+            labels={"foo": "bar"},
+            name="test",
+            task_id="task",
+            image_pull_secrets=fake_pull_secrets,
+            in_cluster=False,
+            cluster_context='default'
+        )
+        launcher_mock.return_value = (State.SUCCESS, None)
+        k.execute(None)
+        self.assertEqual(launcher_mock.call_args[0][0].image_pull_secrets,
+                         fake_pull_secrets)
+
     @staticmethod
     def test_working_pod():
         k = KubernetesPodOperator(