1 # Copyright 2020 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 from unittest
import TestCase
17 from unittest
.mock
import Mock
, patch
20 from n2vc
.tests
.unit
.utils
import AsyncMock
21 from n2vc
.vca
import connection
24 class TestConnection(TestCase
):
26 self
.loop
= asyncio
.get_event_loop()
27 self
.store
= AsyncMock()
29 def test_load_from_store(self
):
30 self
.loop
.run_until_complete(connection
.get_connection(self
.store
, "vim_id"))
32 self
.store
.get_vca_connection_data
.assert_called_once()
34 def test_cloud_properties(self
):
35 conn
= self
.loop
.run_until_complete(
36 connection
.get_connection(self
.store
, "vim_id")
39 conn
._data
.lxd_cloud
= "name"
40 conn
._data
.k8s_cloud
= "name"
41 conn
._data
.lxd_credentials
= "credential"
42 conn
._data
.k8s_credentials
= "credential"
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")
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",
60 "credentials": "credentials",
61 "k8s_cloud": "k8s_cloud",
62 "k8s_credentials": "k8s_credentials",
64 "api-proxy": "api_proxy",
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()