osmclient/sol005/tests/test_sol005_ns.py

   1 # Copyright 2022 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 yaml
  16 import unittest
  17 from unittest.mock import Mock, MagicMock
  18
  19 from osmclient.common.exceptions import ClientException, NotFound
  20 from osmclient.sol005.ns import Ns
  21
  22
  23 class TestNs(unittest.TestCase):
  24     def setUp(self):
  25         self.ns = Ns(Mock(), Mock())
  26         self.set_http_header()
  27         self.endpoint = "/nslcm/v1/ns_instances_content"
  28         self.nsd_id = "nsd_id"
  29         self.nsd_name = {"name": "nsd_name"}
  30         self.nsr_name = {"name": "network service name"}
  31         self.ns._client.nsd.get.return_value = {"_id": self.nsd_id}
  32
  33     def set_http_header(self):
  34         headers = {"key_1": "value"}
  35         self.ns._client._headers = MagicMock()
  36         self.ns._client._headers.__getitem__.side_effect = headers.__getitem__
  37         self.ns._client._headers.items = headers.items
  38
  39     def test_create_ns_vim_account(self):
  40         vim_account = "my_vim_id"
  41         timeout = 15
  42         description = "description_for_my_NS"
  43
  44         self.ns._client.vim.get.return_value = {"_id": vim_account}
  45         self.ns._http.post_cmd.return_value = (200, '{"id": "1234"}')
  46
  47         ns = {
  48             "nsdId": self.nsd_id,
  49             "nsName": self.nsr_name,
  50             "nsDescription": description,
  51             "vimAccountId": vim_account,
  52             "timeout_ns_deploy": timeout,
  53         }
  54
  55         resp_id = self.ns.create(
  56             nsd_name=self.nsd_name,
  57             nsr_name=self.nsr_name,
  58             vim_account=vim_account,
  59             paas_account=None,
  60             description=description,
  61             timeout=timeout,
  62         )
  63
  64         self.ns._client.get_token.assert_called()
  65         self.ns._client.nsd.get.assert_called_with(self.nsd_name)
  66         self.ns._http.post_cmd.assert_called_with(
  67             endpoint=self.endpoint, postfields_dict=ns
  68         )
  69
  70         self.assertEqual(resp_id, "1234")
  71
  72     def test_create_ns_paas_account(self):
  73         paas_account = "my_paas_id"
  74         timeout = 10
  75         description = "description_for_my_NS"
  76         config = {
  77             "additionalParamsForNs": {},
  78             "additionalParamsForVnf": [
  79                 {"additional_param0": "val", "member-vnf-index": "index"}
  80             ],
  81         }
  82
  83         ns = {
  84             "nsdId": self.nsd_id,
  85             "nsName": self.nsr_name,
  86             "nsDescription": description,
  87             "paasAccountId": paas_account,
  88             "timeout_ns_deploy": timeout,
  89             "additionalParamsForNs": config["additionalParamsForNs"],
  90             "additionalParamsForVnf": config["additionalParamsForVnf"],
  91         }
  92
  93         self.ns._client.paas.get.return_value = {"_id": paas_account}
  94         self.ns._http.post_cmd.return_value = (200, '{"id": "1234"}')
  95
  96         resp_id = self.ns.create(
  97             nsd_name=self.nsd_name,
  98             nsr_name=self.nsr_name,
  99             vim_account=None,
 100             paas_account=paas_account,
 101             description=description,
 102             timeout=timeout,
 103             config=yaml.dump(config),
 104         )
 105
 106         self.ns._client.get_token.assert_called()
 107         self.ns._client.nsd.get.assert_called_with(self.nsd_name)
 108         self.ns._http.post_cmd.assert_called_with(
 109             endpoint=self.endpoint, postfields_dict=ns
 110         )
 111
 112         self.assertEqual(resp_id, "1234")
 113
 114     def test_create_ns_paas_account_does_not_exist(self):
 115         paas_account = "my_paas_id"
 116
 117         self.ns._client.paas.get.side_effect = NotFound()
 118
 119         with self.assertRaises(NotFound):
 120             self.ns.create(
 121                 nsd_name=self.nsd_name,
 122                 nsr_name=self.nsr_name,
 123                 vim_account=None,
 124                 paas_account=paas_account,
 125             )
 126
 127         self.ns._client.get_token.assert_called()
 128         self.ns._client.nsd.get.assert_called_with(self.nsd_name)
 129         self.ns._http.set_http_header.assert_not_called()
 130         self.ns._http.post_cmd.assert_not_called()
 131
 132     def test_create_ns_paas_account_post_raises_exception(self):
 133         paas_account = "my_paas_id"
 134
 135         ns = {
 136             "nsdId": self.nsd_id,
 137             "nsName": self.nsr_name,
 138             "nsDescription": "default description",
 139             "paasAccountId": paas_account,
 140         }
 141
 142         self.ns._client.paas.get.return_value = {"_id": paas_account}
 143         self.ns._http.post_cmd.side_effect = ClientException()
 144
 145         with self.assertRaises(ClientException):
 146             self.ns.create(
 147                 nsd_name=self.nsd_name,
 148                 nsr_name=self.nsr_name,
 149                 vim_account=None,
 150                 paas_account=paas_account,
 151             )
 152
 153         self.ns._client.get_token.assert_called()
 154         self.ns._client.nsd.get.assert_called_with(self.nsd_name)
 155         self.ns._http.post_cmd.assert_called_with(
 156             endpoint=self.endpoint, postfields_dict=ns
 157         )
 158
 159     def test_create_ns_paas_account_id_is_not_in_response(self):
 160         paas_account = "my_paas_id"
 161
 162         ns = {
 163             "nsdId": self.nsd_id,
 164             "nsName": self.nsr_name,
 165             "nsDescription": "default description",
 166             "paasAccountId": paas_account,
 167         }
 168
 169         self.ns._client.paas.get.return_value = {"_id": paas_account}
 170         self.ns._http.post_cmd.return_value = (200, "{}")
 171
 172         with self.assertRaises(ClientException):
 173             self.ns.create(
 174                 nsd_name=self.nsd_name,
 175                 nsr_name=self.nsr_name,
 176                 vim_account=None,
 177                 paas_account=paas_account,
 178             )
 179
 180         self.ns._client.get_token.assert_called()
 181         self.ns._client.nsd.get.assert_called_with(self.nsd_name)
 182         self.ns._http.post_cmd.assert_called_with(
 183             endpoint=self.endpoint, postfields_dict=ns
 184         )
 185
 186     def invalid_config_test(self, config, exception):
 187         paas_account = "my_paas_id"
 188
 189         self.ns._client.paas.get.return_value = {"_id": paas_account}
 190
 191         with self.assertRaises(ClientException) as e:
 192             self.ns.create(
 193                 nsd_name=self.nsd_name,
 194                 nsr_name=self.nsr_name,
 195                 vim_account=None,
 196                 paas_account=paas_account,
 197                 config=yaml.dump(config),
 198             )
 199             assert str(e.value) == exception
 200
 201         self.ns._client.get_token.assert_called()
 202         self.ns._client.nsd.get.assert_called_with(self.nsd_name)
 203         self.ns._http.set_http_header.assert_not_called()
 204         self.ns._http.post_cmd.assert_not_called()
 205
 206     def test_create_ns_paas_invalid_additional_params_ns(self):
 207         config = {"additionalParamsForNs": [], "additionalParamsForVnf": {}}
 208         exception = "Error at --config 'additionalParamsForNs' must be a dictionary"
 209         self.invalid_config_test(config, exception)
 210
 211     def test_create_ns_paas_invalid_additional_params_vnf(self):
 212         config = {"additionalParamsForNs": {}, "additionalParamsForVnf": {}}
 213         exception = "Error at --config 'additionalParamsForVnf' must be a list"
 214         self.invalid_config_test(config, exception)
 215
 216     def test_create_ns_paas_invalid_additional_param_vnf(self):
 217         config = {"additionalParamsForNs": {}, "additionalParamsForVnf": [[]]}
 218         exception = (
 219             "Error at --config 'additionalParamsForVnf' items must be dictionaries"
 220         )
 221         self.invalid_config_test(config, exception)
 222
 223     def test_create_ns_paas_invalid_config_member_vnf_index_missing(self):
 224         config = {
 225             "additionalParamsForNs": {},
 226             "additionalParamsForVnf": [{"additional_param0": "val"}],
 227         }
 228         exception = "Error at --config 'additionalParamsForVnf' items must contain 'member-vnf-index'"
 229         self.invalid_config_test(config, exception)
 230
 231     def test_create_ns_without_paas_or_vim_account_raises_exception(self):
 232         with self.assertRaises(ClientException) as e:
 233             self.ns.create(
 234                 nsd_name=self.nsd_name,
 235                 nsr_name=self.nsr_name,
 236                 vim_account=None,
 237                 paas_account=None,
 238             )
 239             error_msg = "Both of vim_account and paas_account options are empty."
 240             assert str(e.value) == error_msg
 241
 242         self.ns._client.get_token.assert_not_called()
 243         self.ns._client.nsd.get.assert_not_called()
 244         self.ns._http.set_http_header.assert_not_called()
 245         self.ns._http.post_cmd.assert_not_called()
 246
 247     def test_create_ns_with_paas_and_vim_account_raises_exception(self):
 248         with self.assertRaises(ClientException) as e:
 249             self.ns.create(
 250                 nsd_name=self.nsd_name,
 251                 nsr_name=self.nsr_name,
 252                 vim_account="vim_account",
 253                 paas_account="paas_account",
 254             )
 255             error_msg = "Both of vim_account and paas_account options are set."
 256             assert str(e.value) == error_msg
 257
 258         self.ns._client.get_token.assert_not_called()
 259         self.ns._client.nsd.get.assert_not_called()
 260         self.ns._http.set_http_header.assert_not_called()
 261         self.ns._http.post_cmd.assert_not_called()