Coverage for n2vc/tests/unit/test_connection.py: 100%

1# Copyright 2020 Canonical Ltd. 

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 at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# 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 and 

13# limitations under the License. 

14 

15import asyncio 

16from unittest import TestCase 

17from unittest.mock import Mock, patch 

18 

19 

20from n2vc.tests.unit.utils import AsyncMock 

21from n2vc.vca import connection 

22 

23 

24class TestConnection(TestCase): 

25 def setUp(self): 

26 self.loop = asyncio.get_event_loop() 

27 self.store = AsyncMock() 

28 

29 def test_load_from_store(self): 

30 self.loop.run_until_complete(connection.get_connection(self.store, "vim_id")) 

31 

32 self.store.get_vca_connection_data.assert_called_once() 

33 

34 def test_cloud_properties(self): 

35 conn = self.loop.run_until_complete( 

36 connection.get_connection(self.store, "vim_id") 

37 ) 

38 conn._data = Mock() 

39 conn._data.lxd_cloud = "name" 

40 conn._data.k8s_cloud = "name" 

41 conn._data.lxd_credentials = "credential" 

42 conn._data.k8s_credentials = "credential" 

43 

44 self.assertEqual(conn.lxd_cloud.name, "name") 

45 self.assertEqual(conn.lxd_cloud.credential_name, "credential") 

46 self.assertEqual(conn.k8s_cloud.name, "name") 

47 self.assertEqual(conn.k8s_cloud.credential_name, "credential") 

48 

49 @patch("n2vc.vca.connection.EnvironConfig") 

50 @patch("n2vc.vca.connection_data.base64_to_cacert") 

51 def test_load_from_env(self, mock_base64_to_cacert, mock_env): 

52 mock_base64_to_cacert.return_value = "cacert" 

53 mock_env.return_value = { 

54 "endpoints": "1.2.3.4:17070", 

55 "user": "user", 

56 "secret": "secret", 

57 "cacert": "cacert", 

58 "pubkey": "pubkey", 

59 "cloud": "cloud", 

60 "credentials": "credentials", 

61 "k8s_cloud": "k8s_cloud", 

62 "k8s_credentials": "k8s_credentials", 

63 "model_config": {}, 

64 "api-proxy": "api_proxy", 

65 } 

66 self.store.get_vca_endpoints.return_value = ["1.2.3.5:17070"] 

67 self.loop.run_until_complete(connection.get_connection(self.store)) 

68 self.store.get_vca_connection_data.assert_not_called()