[AIRFLOW-6585] Fixed Timestamp bug in RefreshKubeConfigLoader (apache…

…#11219)

Co-authored-by: Jan Brusch <[email protected]>
Co-authored-by: Marian Cepok <[email protected]>

airflow/kubernetes/refresh_config.py

@@ -24,15 +24,20 @@
 import logging
 import os
 import time
-from datetime import datetime
-from typing import Optional
+from typing import Optional, cast
 
+import pendulum
 import yaml
 from kubernetes.client import Configuration
 from kubernetes.config.exec_provider import ExecProvider
 from kubernetes.config.kube_config import KUBE_CONFIG_DEFAULT_LOCATION, KubeConfigLoader
 
 
+def _parse_timestamp(ts_str: str) -> int:
+    parsed_dt = cast(pendulum.DateTime, pendulum.parse(ts_str))
+    return calendar.timegm(parsed_dt.timetuple())
+
+
 class RefreshKubeConfigLoader(KubeConfigLoader):
     """
     Patched KubeConfigLoader, this subclass takes expirationTimestamp into
@@ -59,9 +64,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,37 @@
+# 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 pendulum.parsing import ParserError
+
+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)
+
+    def test_parse_timestamp_should_throw_exception(self):
+        with self.assertRaises(ParserError):
+            _parse_timestamp("foobar")