Refactor auth-provider code paths a little. Add Azure support.

Author: brendandburns

config/kube_config.py

@@ -178,23 +178,35 @@ def _load_authentication(self):
         """
         if not self._user:
             return
-        if self._load_gcp_token():
+        if self._load_auth_provider_token():
             return
         if self._load_user_token():
             return
-        if self._load_oid_token():
-            return
         self._load_user_pass_token()
 
-    def _load_gcp_token(self):
+    def _load_auth_provider_token(self):
         if 'auth-provider' not in self._user:
             return
         provider = self._user['auth-provider']
         if 'name' not in provider:
             return
-        if provider['name'] != 'gcp':
+        if provider['name'] == 'gcp':
+            return self._load_gcp_token(provider)
+        if provider['name'] == 'azure':
+            return self._load_azure_token(provider)
+        if provider['name'] == 'oidc':
+            return self._load_oid_token(provider)
+
+    def _load_azure_token(self, provider):
+        if 'config' not in provider:
+            return
+        if 'access-token' not in provider['config']:
             return
+        # TODO: Refresh token here...
+        self.token = 'Bearer %s' % provider['config']['access-token']
+        return self.token
 
+    def _load_gcp_token(self, provider):
         if (('config' not in provider) or
                 ('access-token' not in provider['config']) or
                 ('expiry' in provider['config'] and
@@ -215,15 +227,8 @@ def _refresh_gcp_token(self):
         if self._config_persister:
             self._config_persister(self._config.value)
 
-    def _load_oid_token(self):
-        if 'auth-provider' not in self._user:
-            return
-        provider = self._user['auth-provider']
-
-        if 'name' not in provider or 'config' not in provider:
-            return
-
-        if provider['name'] != 'oidc':
+    def _load_oid_token(self, provider):
+        if 'config' not in provider:
             return
 
         parts = provider['config']['id-token'].split('.')

config/kube_config_test.py

@@ -618,7 +618,7 @@ def test_load_gcp_token_no_refresh(self):
             active_context="gcp",
             get_google_credentials=lambda: _raise_exception(
                 "SHOULD NOT BE CALLED"))
-        self.assertTrue(loader._load_gcp_token())
+        self.assertTrue(loader._load_auth_provider_token())
         self.assertEqual(BEARER_TOKEN_FORMAT % TEST_DATA_BASE64,
                          loader.token)
 
@@ -632,7 +632,7 @@ def cred(): return None
             active_context="expired_gcp",
             get_google_credentials=lambda: cred)
         original_expiry = _get_expiry(loader)
-        self.assertTrue(loader._load_gcp_token())
+        self.assertTrue(loader._load_auth_provider_token())
         new_expiry = _get_expiry(loader)
         # assert that the configs expiry actually updates
         self.assertTrue(new_expiry > original_expiry)
@@ -644,7 +644,7 @@ def test_oidc_no_refresh(self):
             config_dict=self.TEST_KUBE_CONFIG,
             active_context="oidc",
         )
-        self.assertTrue(loader._load_oid_token())
+        self.assertTrue(loader._load_auth_provider_token())
         self.assertEqual(TEST_OIDC_TOKEN, loader.token)
 
     @mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')
@@ -669,7 +669,7 @@ def test_oidc_with_refresh(self, mock_ApiClient, mock_OAuth2Session):
             config_dict=self.TEST_KUBE_CONFIG,
             active_context="expired_oidc",
         )
-        self.assertTrue(loader._load_oid_token())
+        self.assertTrue(loader._load_auth_provider_token())
         self.assertEqual("Bearer abc123", loader.token)
 
     @mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')
@@ -695,7 +695,7 @@ def test_oidc_with_refresh_nocert(
             config_dict=self.TEST_KUBE_CONFIG,
             active_context="expired_oidc_nocert",
         )
-        self.assertTrue(loader._load_oid_token())
+        self.assertTrue(loader._load_auth_provider_token())
         self.assertEqual("Bearer abc123", loader.token)
 
     def test_user_pass(self):