[AIRFLOW-6585] Fixed Timestamp bug in RefreshKubeConfigLoader

airflow/kubernetes/refresh_config.py

@@ -32,6 +32,15 @@
 from kubernetes.config.kube_config import KUBE_CONFIG_DEFAULT_LOCATION, KubeConfigLoader
 
 
+def _parse_timestamp(ts_str: str) -> int:
+    if ts_str[-1] == 'Z':
+        ts_str = ts_str[:-1] + '+0000'
+    expire_ts = calendar.timegm(
+        datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S%z").timetuple()
+    )
+    return expire_ts
+
+
 class RefreshKubeConfigLoader(KubeConfigLoader):
     """
     Patched KubeConfigLoader, this subclass takes expirationTimestamp into
@@ -57,9 +66,7 @@ def _load_from_exec_plugin(self):
             self.token = "Bearer %s" % status['token']  # pylint: disable=W0201
             ts_str = status.get('expirationTimestamp')
             if ts_str:
-                self.api_key_expire_ts = calendar.timegm(
-                    datetime.strptime(ts_str, "%Y-%m-%dT%H:%M:%S%z").timetuple(),
-                )
+                self.api_key_expire_ts = _parse_timestamp(ts_str)
             return True
         except Exception as e:  # pylint: disable=W0703
             logging.error(str(e))

tests/kubernetes/test_refresh_config.py

@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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 unittest import TestCase
+
+from airflow.kubernetes.refresh_config import _parse_timestamp
+
+
+class TestRefreshKubeConfigLoader(TestCase):
+
+    def test_parse_timestamp_should_convert_z_timezone_to_unix_timestamp(self):
+        ts = _parse_timestamp("2020-01-13T13:42:20Z")
+        self.assertEqual(1578922940, ts)
+
+    def test_parse_timestamp_should_convert_regular_timezone_to_unix_timestamp(self):
+        ts = _parse_timestamp("2020-01-13T13:42:20+0600")
+        self.assertEqual(1578922940, ts)