osmclient/sol005/tests/test_vca.py

   1 # Copyright 2021 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
  15 import json
  16 import unittest
  17 from unittest.mock import Mock, patch
  18
  19
  20 from osmclient.common.exceptions import ClientException, NotFound
  21 from osmclient.sol005.vca import VCA
  22
  23
  24 class TestVca(unittest.TestCase):
  25     def setUp(self):
  26         self.vca = VCA(Mock(), Mock())
  27         self.vca_data = {
  28             "name": "name",
  29             "endpoints": ["127.0.0.1:17070"],
  30             "user": "user",
  31             "secret": "secret",
  32             "cacert": "cacert",
  33             "lxd-cloud": "lxd_cloud",
  34             "lxd-credentials": "lxd_credentials",
  35             "k8s-cloud": "k8s_cloud",
  36             "k8s-credentials": "k8s_credentials",
  37             "description": "description",
  38             "model-config": {},
  39         }
  40
  41     @patch("builtins.print")
  42     def test_create_success(self, mock_print):
  43         self.vca._http.post_cmd.return_value = (200, '{"id": "1234"}')
  44         self.vca.create("vca_name", self.vca_data)
  45         self.vca._client.get_token.assert_called()
  46         self.vca._http.post_cmd.assert_called()
  47         mock_print.assert_called_with("1234")
  48
  49     @patch("builtins.print")
  50     def test_create_missing_id(self, mock_print):
  51         self.vca._http.post_cmd.return_value = (404, None)
  52         with self.assertRaises(ClientException):
  53             self.vca.create("vca_name", self.vca_data)
  54         self.vca._client.get_token.assert_called()
  55         self.vca._http.post_cmd.assert_called()
  56         mock_print.assert_not_called()
  57
  58     def test_update_sucess(self):
  59         self.vca.get = Mock()
  60         self.vca.get.return_value = {"_id": "1234"}
  61         self.vca.update("vca_name", self.vca_data)
  62         self.vca._http.patch_cmd.assert_called_with(
  63             endpoint="/admin/v1/vca/1234", postfields_dict=self.vca_data
  64         )
  65
  66     def test_get_id_sucess(self):
  67         self.vca_data.update({"_id": "1234"})
  68         self.vca.list = Mock()
  69         self.vca.list.return_value = [self.vca_data]
  70         vca_id = self.vca.get_id("name")
  71         self.assertEqual(vca_id, "1234")
  72
  73     def test_get_id_not_found(self):
  74         self.vca.list = Mock()
  75         self.vca.list.return_value = []
  76         with self.assertRaises(NotFound):
  77             self.vca.get_id("name")
  78
  79     @patch("osmclient.sol005.vca.utils")
  80     @patch("builtins.print")
  81     def test_delete_success_202(self, mock_print, mock_utils):
  82         mock_utils.validate_uuid4.return_value = False
  83         self.vca.get_id = Mock()
  84         self.vca.get_id.return_value = "1234"
  85         self.vca._http.delete_cmd.return_value = (202, None)
  86         self.vca.delete("vca_name")
  87         self.vca._client.get_token.assert_called()
  88         self.vca._http.delete_cmd.assert_called_with("/admin/v1/vca/1234")
  89         mock_print.assert_called_with("Deletion in progress")
  90
  91     @patch("osmclient.sol005.vca.utils")
  92     @patch("builtins.print")
  93     def test_delete_success_204(self, mock_print, mock_utils):
  94         mock_utils.validate_uuid4.return_value = False
  95         self.vca.get_id = Mock()
  96         self.vca.get_id.return_value = "1234"
  97         self.vca._http.delete_cmd.return_value = (204, None)
  98         self.vca.delete("vca_name", force=True)
  99         self.vca._client.get_token.assert_called()
 100         self.vca._http.delete_cmd.assert_called_with("/admin/v1/vca/1234?FORCE=True")
 101         mock_print.assert_called_with("Deleted")
 102
 103     @patch("osmclient.sol005.vca.utils")
 104     @patch("builtins.print")
 105     def test_delete_success_404(self, mock_print, mock_utils):
 106         mock_utils.validate_uuid4.return_value = False
 107         self.vca.get_id = Mock()
 108         self.vca.get_id.return_value = "1234"
 109         self.vca._http.delete_cmd.return_value = (404, "Not found")
 110         with self.assertRaises(ClientException):
 111             self.vca.delete("vca_name")
 112         self.vca._client.get_token.assert_called()
 113         self.vca._http.delete_cmd.assert_called_with("/admin/v1/vca/1234")
 114         mock_print.assert_not_called()
 115
 116     def test_list_success(self):
 117         self.vca._http.get2_cmd.return_value = (None, '[{"_id": "1234"}]')
 118         vca_list = self.vca.list("my_filter")
 119         self.vca._client.get_token.assert_called()
 120         self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca?my_filter")
 121         self.assertEqual(vca_list, [{"_id": "1234"}])
 122
 123     def test_list_no_response(self):
 124         self.vca._http.get2_cmd.return_value = (None, None)
 125         vca_list = self.vca.list()
 126         self.vca._client.get_token.assert_called()
 127         self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca")
 128         self.assertEqual(vca_list, [])
 129
 130     @patch("osmclient.sol005.vca.utils")
 131     def test_get_success(self, mock_utils):
 132         self.vca_data.update({"_id": "1234"})
 133         mock_utils.validate_uuid4.return_value = False
 134         self.vca.get_id = Mock()
 135         self.vca.get_id.return_value = "1234"
 136         self.vca._http.get2_cmd.return_value = (404, json.dumps(self.vca_data))
 137         vca = self.vca.get("vca_name")
 138         self.vca._client.get_token.assert_called()
 139         self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca/1234")
 140         self.assertEqual(vca, self.vca_data)
 141
 142     @patch("osmclient.sol005.vca.utils")
 143     def test_get_client_exception(self, mock_utils):
 144         mock_utils.validate_uuid4.return_value = False
 145         self.vca.get_id = Mock()
 146         self.vca.get_id.return_value = "1234"
 147         self.vca._http.get2_cmd.return_value = (404, json.dumps({}))
 148         with self.assertRaises(ClientException):
 149             self.vca.get("vca_name")
 150         self.vca._client.get_token.assert_called()
 151         self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca/1234")
 152
 153     @patch("osmclient.sol005.vca.utils")
 154     def test_get_not_exception(self, mock_utils):
 155         mock_utils.validate_uuid4.return_value = False
 156         self.vca.get_id = Mock()
 157         self.vca.get_id.return_value = "1234"
 158         self.vca._http.get2_cmd.side_effect = NotFound()
 159         with self.assertRaises(NotFound):
 160             self.vca.get("vca_name")
 161         self.vca._client.get_token.assert_called()
 162         self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca/1234")