fixing flipped sign in expiry time padding

Author: Zac Pustejovsky

config/kube_config.py

@@ -63,7 +63,7 @@ def _create_temp_file_with_content(content):
 
 
 def _is_expired(expiry):
-    return ((parse_rfc3339(expiry) + EXPIRY_SKEW_PREVENTION_DELAY) <=
+    return ((parse_rfc3339(expiry) - EXPIRY_SKEW_PREVENTION_DELAY) <=
             datetime.datetime.utcnow().replace(tzinfo=UTC))
 
 

config/kube_config_test.py

@@ -22,24 +22,38 @@
 
 import mock
 import yaml
-from six import PY3
+from six import PY3, next
 
 from .config_exception import ConfigException
-from .dateutil import parse_rfc3339
 from .kube_config import (ConfigNode, FileOrData, KubeConfigLoader,
                           _cleanup_temp_files, _create_temp_file_with_content,
                           list_kube_config_contexts, load_kube_config,
                           new_client_from_config)
 
 BEARER_TOKEN_FORMAT = "Bearer %s"
 
+EXPIRY_DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+# should be less than kube_config.EXPIRY_SKEW_PREVENTION_DELAY
+EXPIRY_TIMEDELTA = 2
+
 NON_EXISTING_FILE = "zz_non_existing_file_472398324"
 
 
 def _base64(string):
     return base64.encodestring(string.encode()).decode()
 
 
+def _format_expiry_datetime(dt):
+    return dt.strftime(EXPIRY_DATETIME_FORMAT)
+
+
+def _get_expiry(loader):
+    expired_gcp_conf = (item for item in loader._config.value.get("users")
+                        if item.get("name") == "expired_gcp")
+    return next(expired_gcp_conf).get("user").get("auth-provider") \
+        .get("config").get("expiry")
+
+
 def _raise_exception(st):
     raise Exception(st)
 
@@ -59,6 +73,8 @@ def _raise_exception(st):
 TEST_PASSWORD = "pass"
 # token for me:pass
 TEST_BASIC_TOKEN = "Basic bWU6cGFzcw=="
+TEST_TOKEN_EXPIRY = _format_expiry_datetime(
+    datetime.datetime.utcnow() - datetime.timedelta(minutes=EXPIRY_TIMEDELTA))
 
 TEST_SSL_HOST = "https://test-host"
 TEST_CERTIFICATE_AUTH = "cert-auth"
@@ -194,10 +210,12 @@ class TestConfigNode(BaseTestCase):
                                {"name": "test_name2",
                                 "value": {"key1", "test"}},
                                {"name": "test_name3", "value": [1, 2, 3]}],
-                "with_names_dup": [{"name": "test_name", "value": "test_value"},
-                                   {"name": "test_name",
-                                    "value": {"key1", "test"}},
-                                   {"name": "test_name3", "value": [1, 2, 3]}]}
+                "with_names_dup": [
+                    {"name": "test_name", "value": "test_value"},
+                    {"name": "test_name",
+                     "value": {"key1", "test"}},
+                    {"name": "test_name3", "value": [1, 2, 3]}
+    ]}
 
     def setUp(self):
         super(TestConfigNode, self).setUp()
@@ -213,7 +231,8 @@ def test_normal_map_array_operations(self):
         self.assertEqual(3, len(self.node['key2']))
 
         self.assertEqual("test_obj/key3", self.node['key3'].name)
-        self.assertEqual({"inner_key": "inner_value"}, self.node['key3'].value)
+        self.assertEqual({"inner_key": "inner_value"},
+                         self.node['key3'].value)
         self.assertEqual("inner_value", self.node['key3']["inner_key"])
         self.assertEqual(1, len(self.node['key3']))
 
@@ -255,7 +274,8 @@ def test_get_with_name_on_name_does_not_exists(self):
     def test_get_with_name_on_duplicate_name(self):
         self.expect_exception(
             lambda: self.node['with_names_dup'].get_with_name('test_name'),
-            "Expected only one object with name test_name in test_obj/with_names_dup list")
+            "Expected only one object with name test_name in "
+            "test_obj/with_names_dup list")
 
 
 class FakeConfig:
@@ -421,7 +441,8 @@ class TestKubeConfigLoader(BaseTestCase):
                 "name": "ssl",
                 "cluster": {
                     "server": TEST_SSL_HOST,
-                    "certificate-authority-data": TEST_CERTIFICATE_AUTH_BASE64,
+                    "certificate-authority-data":
+                        TEST_CERTIFICATE_AUTH_BASE64,
                 }
             },
             {
@@ -462,7 +483,7 @@ class TestKubeConfigLoader(BaseTestCase):
                         "name": "gcp",
                         "config": {
                             "access-token": TEST_DATA_BASE64,
-                            "expiry": "2000-01-01T12:00:00Z",  # always in past
+                            "expiry": TEST_TOKEN_EXPIRY,  # always in past
                         }
                     },
                     "token": TEST_DATA_BASE64,  # should be ignored
@@ -492,7 +513,8 @@ class TestKubeConfigLoader(BaseTestCase):
                             "id-token": TEST_OIDC_EXPIRED_LOGIN,
                             "idp-certificate-authority-data": TEST_OIDC_CA,
                             "idp-issuer-url": "https://example.org/identity",
-                            "refresh-token": "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"
+                            "refresh-token":
+                                "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"
                         }
                     }
                 }
@@ -578,7 +600,6 @@ def test_load_gcp_token_no_refresh(self):
                          loader.token)
 
     def test_load_gcp_token_with_refresh(self):
-
         def cred(): return None
         cred.token = TEST_ANOTHER_DATA_BASE64
         cred.expiry = datetime.datetime.now()
@@ -587,7 +608,11 @@ def cred(): return None
             config_dict=self.TEST_KUBE_CONFIG,
             active_context="expired_gcp",
             get_google_credentials=lambda: cred)
+        original_expiry = _get_expiry(loader)
         self.assertTrue(loader._load_gcp_token())
+        new_expiry = _get_expiry(loader)
+        # assert that the configs expiry actually updates
+        self.assertTrue(new_expiry > original_expiry)
         self.assertEqual(BEARER_TOKEN_FORMAT % TEST_ANOTHER_DATA_BASE64,
                          loader.token)