codekingpro/portable-devtools

1# Copyright 2018 The Kubernetes Authors.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15import json16import os17import unittest18 19from unittest import mock20 21from .config_exception import ConfigException22from .exec_provider import ExecProvider23from .kube_config import ConfigNode24 25 26class ExecProviderTest(unittest.TestCase):27 28    def setUp(self):29        self.input_ok = ConfigNode('test', {30            'command': 'aws-iam-authenticator',31            'args': ['token', '-i', 'dummy'],32            'apiVersion': 'client.authentication.k8s.io/v1beta1',33            'env': None34        })35        self.input_with_cluster = ConfigNode('test', {36            'command': 'aws-iam-authenticator',37            'args': ['token', '-i', 'dummy'],38            'apiVersion': 'client.authentication.k8s.io/v1beta1',39            'provideClusterInfo': True,40            'env': None41        })42        self.output_ok = """43        {44            "apiVersion": "client.authentication.k8s.io/v1beta1",45            "kind": "ExecCredential",46            "status": {47                "token": "dummy"48            }49        }50        """51 52    def test_missing_input_keys(self):53        exec_configs = [ConfigNode('test1', {}),54                        ConfigNode('test2', {'command': ''}),55                        ConfigNode('test3', {'apiVersion': ''})]56        for exec_config in exec_configs:57            with self.assertRaises(ConfigException) as context:58                ExecProvider(exec_config, None)59            self.assertIn('exec: malformed request. missing key',60                          context.exception.args[0])61 62    @mock.patch('subprocess.Popen')63    def test_error_code_returned(self, mock):64        instance = mock.return_value65        instance.wait.return_value = 166        instance.communicate.return_value = ('', '')67        with self.assertRaises(ConfigException) as context:68            ep = ExecProvider(self.input_ok, None)69            ep.run()70        self.assertIn('exec: process returned %d' %71                      instance.wait.return_value, context.exception.args[0])72 73    @mock.patch('subprocess.Popen')74    def test_nonjson_output_returned(self, mock):75        instance = mock.return_value76        instance.wait.return_value = 077        instance.communicate.return_value = ('', '')78        with self.assertRaises(ConfigException) as context:79            ep = ExecProvider(self.input_ok, None)80            ep.run()81        self.assertIn('exec: failed to decode process output',82                      context.exception.args[0])83 84    @mock.patch('subprocess.Popen')85    def test_missing_output_keys(self, mock):86        instance = mock.return_value87        instance.wait.return_value = 088        outputs = [89            """90            {91                "kind": "ExecCredential",92                "status": {93                    "token": "dummy"94                }95            }96            """, """97            {98                "apiVersion": "client.authentication.k8s.io/v1beta1",99                "status": {100                    "token": "dummy"101                }102            }103            """, """104            {105                "apiVersion": "client.authentication.k8s.io/v1beta1",106                "kind": "ExecCredential"107            }108            """109        ]110        for output in outputs:111            instance.communicate.return_value = (output, '')112            with self.assertRaises(ConfigException) as context:113                ep = ExecProvider(self.input_ok, None)114                ep.run()115            self.assertIn('exec: malformed response. missing key',116                          context.exception.args[0])117 118    @mock.patch('subprocess.Popen')119    def test_mismatched_api_version(self, mock):120        instance = mock.return_value121        instance.wait.return_value = 0122        wrong_api_version = 'client.authentication.k8s.io/v1'123        output = """124        {125            "apiVersion": "%s",126            "kind": "ExecCredential",127            "status": {128                "token": "dummy"129            }130        }131        """ % wrong_api_version132        instance.communicate.return_value = (output, '')133        with self.assertRaises(ConfigException) as context:134            ep = ExecProvider(self.input_ok, None)135            ep.run()136        self.assertIn(137            'exec: plugin api version %s does not match' %138            wrong_api_version,139            context.exception.args[0])140 141    @mock.patch('subprocess.Popen')142    def test_ok_01(self, mock):143        instance = mock.return_value144        instance.wait.return_value = 0145        instance.communicate.return_value = (self.output_ok, '')146        ep = ExecProvider(self.input_ok, None)147        result = ep.run()148        self.assertTrue(isinstance(result, dict))149        self.assertTrue('token' in result)150 151    @mock.patch('subprocess.Popen')152    def test_run_in_dir(self, mock):153        instance = mock.return_value154        instance.wait.return_value = 0155        instance.communicate.return_value = (self.output_ok, '')156        ep = ExecProvider(self.input_ok, '/some/directory')157        ep.run()158        self.assertEqual(mock.call_args[1]['cwd'], '/some/directory')159 160    @mock.patch('subprocess.Popen')161    def test_ok_no_console_attached(self, mock):162        instance = mock.return_value163        instance.wait.return_value = 0164        instance.communicate.return_value = (self.output_ok, '')165        mock_stdout = unittest.mock.patch(166            'sys.stdout', new=None)  # Simulate detached console167        with mock_stdout:168            ep = ExecProvider(self.input_ok, None)169            result = ep.run()170            self.assertTrue(isinstance(result, dict))171            self.assertTrue('token' in result)172 173    @mock.patch('subprocess.Popen')174    def test_with_cluster_info(self, mock):175        instance = mock.return_value176        instance.wait.return_value = 0177        instance.communicate.return_value = (self.output_ok, '')178        ep = ExecProvider(self.input_with_cluster, None, ConfigNode("cluster", {'server': 'name.company.com'}))179        result = ep.run()180        self.assertTrue(isinstance(result, dict))181        self.assertTrue('token' in result)182 183        obj = json.loads(mock.call_args.kwargs['env']['KUBERNETES_EXEC_INFO'])184        self.assertEqual(obj['spec']['cluster']['server'], 'name.company.com')185 186    @mock.patch("subprocess.Popen")187    def test_with_cluster_info_from_exec_extension(self, mock):188        instance = mock.return_value189        instance.wait.return_value = 0190        instance.communicate.return_value = (self.output_ok, "")191        ep = ExecProvider(192            self.input_with_cluster,193            None,194            ConfigNode(195                "cluster",196                {197                    "server": "name.company.com",198                    "extensions": [199                        {200                            "name": "client.authentication.k8s.io/exec",201                            "extension": {202                                "namespace": "myproject",203                                "name": "mycluster",204                            },205                        },206                    ],207                },208            ),209        )210        result = ep.run()211        self.assertTrue(isinstance(result, dict))212        self.assertTrue("token" in result)213 214        obj = json.loads(mock.call_args.kwargs["env"]["KUBERNETES_EXEC_INFO"])215        self.assertEqual(obj["spec"]["cluster"]["server"], "name.company.com")216        self.assertEqual(obj["spec"]["cluster"]["config"]["namespace"], "myproject")217        self.assertEqual(obj["spec"]["cluster"]["config"]["name"], "mycluster")218 219if __name__ == '__main__':220    unittest.main()221