add testcases
@@ -17,6 +17,7 @@ import datetime
import io
import json
import os
+from pprint import pprint
import shutil
import tempfile
import unittest
@@ -485,6 +486,13 @@ class TestKubeConfigLoader(BaseTestCase):
"user": "expired_oidc"
}
},
+ {
+ "name": "expired_oidc_with_idp_ca_file",
+ "context": {
+ "cluster": "default",
+ "user": "expired_oidc_with_idp_ca_file"
+ }
+ },
{
"name": "expired_oidc_nocert",
"context": {
@@ -799,6 +807,23 @@ class TestKubeConfigLoader(BaseTestCase):
}
}
},
+ {
+ "name": "expired_oidc_with_idp_ca_file",
+ "user": {
+ "auth-provider": {
+ "name": "oidc",
+ "config": {
+ "client-id": "tectonic-kubectl",
+ "client-secret": "FAKE_SECRET",
+ "id-token": TEST_OIDC_EXPIRED_LOGIN,
+ "idp-certificate-authority": TEST_CERTIFICATE_AUTH,
+ "idp-issuer-url": "https://example.org/identity",
+ "refresh-token":
+ "lucWJjEhlxZW01cXI3YmVlcYnpxNGhzk"
+ }
+ }
+ }
+ },
{
"name": "expired_oidc_nocert",
"user": {
@@ -1059,6 +1084,33 @@ class TestKubeConfigLoader(BaseTestCase):
self.assertTrue(loader._load_auth_provider_token())
self.assertEqual("Bearer abc123", loader.token)
+ @mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')
+ @mock.patch('kubernetes.config.kube_config.ApiClient.request')
+ def test_oidc_with_idp_ca_file_refresh(self, mock_ApiClient, mock_OAuth2Session):
+ mock_response = mock.MagicMock()
+ type(mock_response).status = mock.PropertyMock(
+ return_value=200
+ )
+ type(mock_response).data = mock.PropertyMock(
+ return_value=json.dumps({
+ "token_endpoint": "https://example.org/identity/token"
+ })
+ )
+
+ mock_ApiClient.return_value = mock_response
+
+ mock_OAuth2Session.return_value = {"id_token": "abc123",
+ "refresh_token": "newtoken123"}
+
+ loader = KubeConfigLoader(
+ config_dict=self.TEST_KUBE_CONFIG,
+ active_context="expired_oidc_with_idp_ca_file",
+ )
+
+
+ self.assertTrue(loader._load_auth_provider_token())
+ self.assertEqual("Bearer abc123", loader.token)
+
@mock.patch('kubernetes.config.kube_config.OAuth2Session.refresh_token')
@mock.patch('kubernetes.config.kube_config.ApiClient.request')
def test_oidc_with_refresh_nocert(
