libcurl4-openssl-dev \
libssl-dev \
wget
+
+ENV LC_ALL C.UTF-8
+ENV LANG C.UTF-8
\ No newline at end of file
import pkg_resources
import logging
from datetime import datetime
+from typing import Any, Dict
def wrap_text(text, width):
help="do not return the control immediately, but keep it "
"until the operation is completed, or timeout",
)
+@click.option("--vca", default=None, help="VCA to be used in this VIM account")
@click.pass_context
def vim_create(
ctx,
sdn_controller,
sdn_port_mapping,
wait,
+ vca,
):
"""creates a new VIM account"""
logger.debug("")
vim["vim-type"] = account_type
vim["description"] = description
vim["config"] = config
+ if vca:
+ vim["vca"] = vca
if sdn_controller or sdn_port_mapping:
ctx.obj.vim.create(name, vim, sdn_controller, sdn_port_mapping, wait=wait)
else:
# exit(1)
+###########################
+# VCA operations
+###########################
+
+
+@cli_osm.command(name="vca-add", short_help="adds a VCA (Juju controller) to OSM")
+@click.argument("name")
+@click.option(
+ "--endpoints",
+ prompt=True,
+ help="Comma-separated list of IP or hostnames of the Juju controller",
+)
+@click.option("--user", prompt=True, help="Username with admin priviledges")
+@click.option("--secret", prompt=True, help="Password of the specified username")
+@click.option("--cacert", prompt=True, help="CA certificate")
+@click.option(
+ "--lxd-cloud",
+ prompt=True,
+ help="Name of the cloud that will be used for LXD containers (LXD proxy charms)",
+)
+@click.option(
+ "--lxd-credentials",
+ prompt=True,
+ help="Name of the cloud credentialsto be used for the LXD cloud",
+)
+@click.option(
+ "--k8s-cloud",
+ prompt=True,
+ help="Name of the cloud that will be used for K8s containers (K8s proxy charms)",
+)
+@click.option(
+ "--k8s-credentials",
+ prompt=True,
+ help="Name of the cloud credentialsto be used for the K8s cloud",
+)
+@click.option(
+ "--model-config",
+ default={},
+ help="Configuration options for the models",
+)
+@click.option("--description", default=None, help="human readable description")
+@click.pass_context
+def vca_add(
+ ctx,
+ name,
+ endpoints,
+ user,
+ secret,
+ cacert,
+ lxd_cloud,
+ lxd_credentials,
+ k8s_cloud,
+ k8s_credentials,
+ model_config,
+ description,
+):
+ """adds a VCA to OSM
+
+ NAME: name of the VCA
+ """
+ check_client_version(ctx.obj, ctx.command.name)
+ vca = {}
+ vca["name"] = name
+ vca["endpoints"] = endpoints.split(",")
+ vca["user"] = user
+ vca["secret"] = secret
+ vca["cacert"] = cacert
+ vca["lxd-cloud"] = lxd_cloud
+ vca["lxd-credentials"] = lxd_credentials
+ vca["k8s-cloud"] = k8s_cloud
+ vca["k8s-credentials"] = k8s_credentials
+ if description:
+ vca["description"] = description
+ if model_config:
+ model_config = load(model_config)
+ vca["model-config"] = model_config
+ ctx.obj.vca.create(name, vca)
+
+
+def load(data: Any):
+ if os.path.isfile(data):
+ return load_file(data)
+ else:
+ try:
+ return json.loads(data)
+ except ValueError as e:
+ raise ClientException(e)
+
+
+def load_file(file_path: str) -> Dict:
+ content = None
+ with open(file_path, "r") as f:
+ content = f.read()
+ try:
+ return yaml.safe_load(content)
+ except yaml.scanner.ScannerError:
+ pass
+ try:
+ return json.loads(content)
+ except ValueError:
+ pass
+ raise ClientException(f"{file_path} must be a valid yaml or json file")
+
+
+@cli_osm.command(name="vca-update", short_help="updates a K8s cluster")
+@click.argument("name")
+@click.option(
+ "--endpoints", help="Comma-separated list of IP or hostnames of the Juju controller"
+)
+@click.option("--user", help="Username with admin priviledges")
+@click.option("--secret", help="Password of the specified username")
+@click.option("--cacert", help="CA certificate")
+@click.option(
+ "--lxd-cloud",
+ help="Name of the cloud that will be used for LXD containers (LXD proxy charms)",
+)
+@click.option(
+ "--lxd-credentials",
+ help="Name of the cloud credentialsto be used for the LXD cloud",
+)
+@click.option(
+ "--k8s-cloud",
+ help="Name of the cloud that will be used for K8s containers (K8s proxy charms)",
+)
+@click.option(
+ "--k8s-credentials",
+ help="Name of the cloud credentialsto be used for the K8s cloud",
+)
+@click.option(
+ "--model-config",
+ help="Configuration options for the models",
+)
+@click.option("--description", default=None, help="human readable description")
+@click.pass_context
+def vca_update(
+ ctx,
+ name,
+ endpoints,
+ user,
+ secret,
+ cacert,
+ lxd_cloud,
+ lxd_credentials,
+ k8s_cloud,
+ k8s_credentials,
+ model_config,
+ description,
+):
+ """updates a K8s cluster
+
+ NAME: name or ID of the K8s cluster
+ """
+ check_client_version(ctx.obj, ctx.command.name)
+ vca = {}
+ vca["name"] = name
+ if endpoints:
+ vca["endpoints"] = endpoints.split(",")
+ if user:
+ vca["user"] = user
+ if secret:
+ vca["secret"] = secret
+ if cacert:
+ vca["cacert"] = cacert
+ if lxd_cloud:
+ vca["lxd-cloud"] = lxd_cloud
+ if lxd_credentials:
+ vca["lxd-credentials"] = lxd_credentials
+ if k8s_cloud:
+ vca["k8s-cloud"] = k8s_cloud
+ if k8s_credentials:
+ vca["k8s-credentials"] = k8s_credentials
+ if description:
+ vca["description"] = description
+ if model_config:
+ model_config = load(model_config)
+ vca["model-config"] = model_config
+ ctx.obj.vca.update(name, vca)
+
+
+@cli_osm.command(name="vca-delete", short_help="deletes a K8s cluster")
+@click.argument("name")
+@click.option(
+ "--force", is_flag=True, help="forces the deletion from the DB (not recommended)"
+)
+@click.pass_context
+def vca_delete(ctx, name, force):
+ """deletes a K8s cluster
+
+ NAME: name or ID of the K8s cluster to be deleted
+ """
+ check_client_version(ctx.obj, ctx.command.name)
+ ctx.obj.vca.delete(name, force=force)
+
+
+@cli_osm.command(name="vca-list")
+@click.option(
+ "--filter",
+ default=None,
+ multiple=True,
+ help="restricts the list to the VCAs matching the filter",
+)
+@click.option("--literal", is_flag=True, help="print literally, no pretty table")
+@click.option("--long", is_flag=True, help="get more details")
+@click.pass_context
+def vca_list(ctx, filter, literal, long):
+ """list VCAs"""
+ check_client_version(ctx.obj, ctx.command.name)
+ if filter:
+ filter = "&".join(filter)
+ resp = ctx.obj.vca.list(filter)
+ if literal:
+ print(yaml.safe_dump(resp, indent=4, default_flow_style=False))
+ return
+ if long:
+ table = PrettyTable(
+ ["Name", "Id", "Project", "Operational State", "Detailed Status"]
+ )
+ project_list = ctx.obj.project.list()
+ else:
+ table = PrettyTable(["Name", "Id", "Operational State"])
+ for vca in resp:
+ logger.debug("VCA details: {}".format(yaml.safe_dump(vca)))
+ if long:
+ project_id, project_name = get_project(project_list, vca)
+ detailed_status = vca.get("_admin", {}).get("detailed-status", "-")
+ table.add_row(
+ [
+ vca["name"],
+ vca["_id"],
+ project_name,
+ vca.get("_admin", {}).get("operationalState", "-"),
+ wrap_text(text=detailed_status, width=40),
+ ]
+ )
+ else:
+ table.add_row(
+ [
+ vca["name"],
+ vca["_id"],
+ vca.get("_admin", {}).get("operationalState", "-"),
+ ]
+ )
+ table.align = "l"
+ print(table)
+
+
+@cli_osm.command(name="vca-show", short_help="shows the details of a K8s cluster")
+@click.argument("name")
+@click.option("--literal", is_flag=True, help="print literally, no pretty table")
+@click.pass_context
+def vca_show(ctx, name, literal):
+ """shows the details of a K8s cluster
+
+ NAME: name or ID of the K8s cluster
+ """
+ # try:
+ resp = ctx.obj.vca.get(name)
+ if literal:
+ print(yaml.safe_dump(resp, indent=4, default_flow_style=False))
+ return
+ table = PrettyTable(["key", "attribute"])
+ for k, v in list(resp.items()):
+ table.add_row([k, wrap_text(text=json.dumps(v, indent=2), width=100)])
+ table.align = "l"
+ print(table)
+
+
###########################
# Repo operations
###########################
--- /dev/null
+# Copyright 2021 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import json
+import unittest
+from unittest.mock import Mock, patch
+import yaml
+
+
+from click.testing import CliRunner
+from osmclient.scripts import osm
+
+
+@patch("builtins.print")
+@patch("osmclient.scripts.osm.PrettyTable")
+@patch("osmclient.scripts.osm.client.Client")
+@patch("osmclient.scripts.osm.check_client_version")
+@patch("osmclient.scripts.osm.get_project")
+class TestVca(unittest.TestCase):
+ def setUp(self):
+ self.runner = CliRunner()
+ self.ctx_obj = Mock()
+ self.table = Mock()
+ self.vca_data = {
+ "name": "name",
+ "_id": "1234",
+ "_admin": {
+ "detailed-status": "status",
+ "operationalState": "state",
+ },
+ }
+
+ def test_vca_list(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ mock_pretty_table.return_value = self.table
+ mock_get_project.return_value = ("5678", "project")
+ self.ctx_obj.vca.list.return_value = [self.vca_data]
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-list", "--filter", "somefilter"],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.list.assert_called_with("somefilter")
+ mock_pretty_table.assert_called_with(["Name", "Id", "Operational State"])
+ self.table.add_row.assert_called_with(["name", "1234", "state"])
+ mock_print.assert_called_with(self.table)
+
+ def test_vca_list_long(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ mock_pretty_table.return_value = self.table
+ mock_get_project.return_value = ("5678", "project")
+ self.ctx_obj.vca.list.return_value = [self.vca_data]
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-list", "--filter", "somefilter", "--long"],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.list.assert_called_with("somefilter")
+ mock_pretty_table.assert_called_with(
+ ["Name", "Id", "Project", "Operational State", "Detailed Status"]
+ )
+ self.table.add_row.assert_called_with(
+ ["name", "1234", "project", "state", "status"]
+ )
+ mock_print.assert_called_with(self.table)
+
+ def test_vca_list_literal(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ self.ctx_obj.vca.list.return_value = [self.vca_data]
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-list", "--literal"],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.list.assert_called()
+ mock_pretty_table.assert_not_called()
+ self.table.add_row.assert_not_called()
+ mock_print.assert_called_with(
+ yaml.safe_dump([self.vca_data], indent=4, default_flow_style=False)
+ )
+
+ def test_vca_show(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ mock_pretty_table.return_value = self.table
+ self.ctx_obj.vca.get.return_value = self.vca_data
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-show", "name"],
+ )
+ self.ctx_obj.vca.get.assert_called_with("name")
+ mock_pretty_table.assert_called_with(["key", "attribute"])
+ self.assertEqual(self.table.add_row.call_count, len(self.vca_data))
+ mock_print.assert_called_with(self.table)
+
+ def test_vca_show_literal(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ self.ctx_obj.vca.get.return_value = self.vca_data
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-show", "name", "--literal"],
+ )
+ self.ctx_obj.vca.get.assert_called_with("name")
+ mock_pretty_table.assert_not_called()
+ self.table.add_row.assert_not_called()
+ mock_print.assert_called_with(
+ yaml.safe_dump(self.vca_data, indent=4, default_flow_style=False)
+ )
+
+ def test_vca_update(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-update", "name"],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.update.assert_called_with("name", {"name": "name"})
+ mock_pretty_table.assert_not_called()
+ self.table.add_row.assert_not_called()
+ mock_print.assert_not_called()
+
+ def test_vca_update_with_args(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ self.runner.invoke(
+ osm.cli_osm,
+ [
+ "vca-update",
+ "name",
+ "--endpoints",
+ "1.2.3.4:17070",
+ "--user",
+ "user",
+ "--secret",
+ "secret",
+ "--cacert",
+ "cacert",
+ "--lxd-cloud",
+ "lxd_cloud",
+ "--lxd-credentials",
+ "lxd_credentials",
+ "--k8s-cloud",
+ "k8s_cloud",
+ "--k8s-credentials",
+ "k8s_credentials",
+ "--description",
+ "description",
+ "--model-config",
+ json.dumps({"juju-https-proxy": "http://squid:3128"}),
+ ],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.update.assert_called_with(
+ "name",
+ {
+ "name": "name",
+ "endpoints": ["1.2.3.4:17070"],
+ "user": "user",
+ "secret": "secret",
+ "cacert": "cacert",
+ "lxd-cloud": "lxd_cloud",
+ "lxd-credentials": "lxd_credentials",
+ "k8s-cloud": "k8s_cloud",
+ "k8s-credentials": "k8s_credentials",
+ "description": "description",
+ "model-config": {"juju-https-proxy": "http://squid:3128"},
+ },
+ )
+ mock_pretty_table.assert_not_called()
+ self.table.add_row.assert_not_called()
+ mock_print.assert_not_called()
+
+ def test_vca_add(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ self.runner.invoke(
+ osm.cli_osm,
+ [
+ "vca-add",
+ "name",
+ "--endpoints",
+ "1.2.3.4:17070",
+ "--user",
+ "user",
+ "--secret",
+ "secret",
+ "--cacert",
+ "cacert",
+ "--lxd-cloud",
+ "lxd_cloud",
+ "--lxd-credentials",
+ "lxd_credentials",
+ "--k8s-cloud",
+ "k8s_cloud",
+ "--k8s-credentials",
+ "k8s_credentials",
+ "--description",
+ "description",
+ "--model-config",
+ json.dumps({"juju-https-proxy": "http://squid:3128"}),
+ ],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.create.assert_called_with(
+ "name",
+ {
+ "name": "name",
+ "endpoints": ["1.2.3.4:17070"],
+ "user": "user",
+ "secret": "secret",
+ "cacert": "cacert",
+ "lxd-cloud": "lxd_cloud",
+ "lxd-credentials": "lxd_credentials",
+ "k8s-cloud": "k8s_cloud",
+ "k8s-credentials": "k8s_credentials",
+ "description": "description",
+ "model-config": {"juju-https-proxy": "http://squid:3128"},
+ },
+ )
+ mock_pretty_table.assert_not_called()
+ self.table.add_row.assert_not_called()
+ mock_print.assert_not_called()
+
+ def test_vca_delete(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ mock_client.return_value = self.ctx_obj
+ self.runner.invoke(
+ osm.cli_osm,
+ ["vca-delete", "name"],
+ )
+ mock_check_client_version.assert_called()
+ self.ctx_obj.vca.delete.assert_called_with("name", force=False)
+ mock_pretty_table.assert_not_called()
+ self.table.add_row.assert_not_called()
+ mock_print.assert_not_called()
+
+ def test_load(
+ self,
+ mock_get_project,
+ mock_check_client_version,
+ mock_client,
+ mock_pretty_table,
+ mock_print,
+ ):
+ data = osm.load(json.dumps({"juju-https-proxy": "http://squid:3128"}))
+ self.assertEqual(data, {"juju-https-proxy": "http://squid:3128"})
from osmclient.sol005 import role
from osmclient.sol005 import pdud
from osmclient.sol005 import k8scluster
+from osmclient.sol005 import vca
from osmclient.sol005 import repo
from osmclient.sol005 import osmrepo
from osmclient.common import package_tool
self.role = role.Role(self._http_client, client=self)
self.pdu = pdud.Pdu(self._http_client, client=self)
self.k8scluster = k8scluster.K8scluster(self._http_client, client=self)
+ self.vca = vca.VCA(self._http_client, client=self)
self.repo = repo.Repo(self._http_client, client=self)
self.osmrepo = osmrepo.OSMRepo(self._http_client, client=self)
self.package_tool = package_tool.PackageTool(client=self)
self._apiName, self._apiVersion, self._apiResource
)
- def create(self, name, k8s_cluster):
- def get_vim_account_id(vim_account):
- vim = self._client.vim.get(vim_account)
- if vim is None:
- raise NotFound("cannot find vim account '{}'".format(vim_account))
- return vim["_id"]
+ def _get_vim_account(self, vim_account):
+ vim = self._client.vim.get(vim_account)
+ if vim is None:
+ raise NotFound("cannot find vim account '{}'".format(vim_account))
+ return vim
+ def create(self, name, k8s_cluster):
self._client.get_token()
- k8s_cluster["vim_account"] = get_vim_account_id(k8s_cluster["vim_account"])
+ vim_account = self._get_vim_account(k8s_cluster["vim_account"])
+ k8s_cluster["vim_account"] = vim_account["_id"]
+ if "vca" in vim_account:
+ k8s_cluster["vca_id"] = vim_account["vca"]
http_code, resp = self._http.post_cmd(
endpoint=self._apiBase, postfields_dict=k8s_cluster
)
def update(self, name, k8s_cluster):
self._client.get_token()
cluster = self.get(name)
+ if "vim_account" in k8s_cluster:
+ vim_account = self._get_vim_account(k8s_cluster["vim_account"])
+ k8s_cluster["vim_account"] = vim_account["_id"]
+ if "vca" in vim_account:
+ k8s_cluster["vca_id"] = vim_account["vca"]
http_code, resp = self._http.put_cmd(
endpoint="{}/{}".format(self._apiBase, cluster["_id"]),
postfields_dict=k8s_cluster,
vim_account_id[vim_account] = vim["_id"]
return vim["_id"]
+ def get_vca_id(vim_id):
+ vim = self._client.vim.get(vim_id)
+ return vim.get("vca")
+
def get_wim_account_id(wim_account):
self._logger.debug("")
# wim_account can be False (boolean) to indicate not use wim account
wim_account_id[wim_account] = wim["_id"]
return wim["_id"]
+ vim_id = get_vim_account_id(account)
+ vca_id = get_vca_id(vim_id)
ns = {}
ns["nsdId"] = nsd["_id"]
ns["nsName"] = nsr_name
ns["nsDescription"] = description
- ns["vimAccountId"] = get_vim_account_id(account)
+ ns["vimAccountId"] = vim_id
+ if vca_id:
+ ns["vcaId"] = vca_id
# ns['userdata'] = {}
# ns['userdata']['key1']='value1'
# ns['userdata']['key2']='value2'
--- /dev/null
+# Copyright 2021 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import unittest
+from unittest.mock import Mock, patch
+
+
+from osmclient.common.exceptions import ClientException, NotFound
+from osmclient.sol005.vca import VCA
+
+
+class TestVca(unittest.TestCase):
+ def setUp(self):
+ self.vca = VCA(Mock(), Mock())
+ self.vca_data = {
+ "name": "name",
+ "endpoints": ["127.0.0.1:17070"],
+ "user": "user",
+ "secret": "secret",
+ "cacert": "cacert",
+ "lxd-cloud": "lxd_cloud",
+ "lxd-credentials": "lxd_credentials",
+ "k8s-cloud": "k8s_cloud",
+ "k8s-credentials": "k8s_credentials",
+ "description": "description",
+ "model-config": {},
+ }
+
+ @patch("builtins.print")
+ def test_create_success(self, mock_print):
+ self.vca._http.post_cmd.return_value = (200, '{"id": "1234"}')
+ self.vca.create("vca_name", self.vca_data)
+ self.vca._client.get_token.assert_called()
+ self.vca._http.post_cmd.assert_called()
+ mock_print.assert_called_with("1234")
+
+ @patch("builtins.print")
+ def test_create_missing_id(self, mock_print):
+ self.vca._http.post_cmd.return_value = (404, None)
+ with self.assertRaises(ClientException):
+ self.vca.create("vca_name", self.vca_data)
+ self.vca._client.get_token.assert_called()
+ self.vca._http.post_cmd.assert_called()
+ mock_print.assert_not_called()
+
+ def test_update_sucess(self):
+ self.vca.get = Mock()
+ self.vca.get.return_value = {"_id": "1234"}
+ self.vca.update("vca_name", self.vca_data)
+ self.vca._http.patch_cmd.assert_called_with(
+ endpoint="/admin/v1/vca/1234", postfields_dict=self.vca_data
+ )
+
+ def test_get_id_sucess(self):
+ self.vca_data.update({"_id": "1234"})
+ self.vca.list = Mock()
+ self.vca.list.return_value = [self.vca_data]
+ vca_id = self.vca.get_id("name")
+ self.assertEqual(vca_id, "1234")
+
+ def test_get_id_not_found(self):
+ self.vca.list = Mock()
+ self.vca.list.return_value = []
+ with self.assertRaises(NotFound):
+ self.vca.get_id("name")
+
+ @patch("osmclient.sol005.vca.utils")
+ @patch("builtins.print")
+ def test_delete_success_202(self, mock_print, mock_utils):
+ mock_utils.validate_uuid4.return_value = False
+ self.vca.get_id = Mock()
+ self.vca.get_id.return_value = "1234"
+ self.vca._http.delete_cmd.return_value = (202, None)
+ self.vca.delete("vca_name")
+ self.vca._client.get_token.assert_called()
+ self.vca._http.delete_cmd.assert_called_with("/admin/v1/vca/1234")
+ mock_print.assert_called_with("Deletion in progress")
+
+ @patch("osmclient.sol005.vca.utils")
+ @patch("builtins.print")
+ def test_delete_success_204(self, mock_print, mock_utils):
+ mock_utils.validate_uuid4.return_value = False
+ self.vca.get_id = Mock()
+ self.vca.get_id.return_value = "1234"
+ self.vca._http.delete_cmd.return_value = (204, None)
+ self.vca.delete("vca_name", force=True)
+ self.vca._client.get_token.assert_called()
+ self.vca._http.delete_cmd.assert_called_with("/admin/v1/vca/1234?FORCE=True")
+ mock_print.assert_called_with("Deleted")
+
+ @patch("osmclient.sol005.vca.utils")
+ @patch("builtins.print")
+ def test_delete_success_404(self, mock_print, mock_utils):
+ mock_utils.validate_uuid4.return_value = False
+ self.vca.get_id = Mock()
+ self.vca.get_id.return_value = "1234"
+ self.vca._http.delete_cmd.return_value = (404, "Not found")
+ with self.assertRaises(ClientException):
+ self.vca.delete("vca_name")
+ self.vca._client.get_token.assert_called()
+ self.vca._http.delete_cmd.assert_called_with("/admin/v1/vca/1234")
+ mock_print.assert_not_called()
+
+ def test_list_success(self):
+ self.vca._http.get2_cmd.return_value = (None, '[{"_id": "1234"}]')
+ vca_list = self.vca.list("my_filter")
+ self.vca._client.get_token.assert_called()
+ self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca?my_filter")
+ self.assertEqual(vca_list, [{"_id": "1234"}])
+
+ def test_list_no_response(self):
+ self.vca._http.get2_cmd.return_value = (None, None)
+ vca_list = self.vca.list()
+ self.vca._client.get_token.assert_called()
+ self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca")
+ self.assertEqual(vca_list, [])
+
+ @patch("osmclient.sol005.vca.utils")
+ def test_get_success(self, mock_utils):
+ self.vca_data.update({"_id": "1234"})
+ mock_utils.validate_uuid4.return_value = False
+ self.vca.get_id = Mock()
+ self.vca.get_id.return_value = "1234"
+ self.vca._http.get2_cmd.return_value = (404, json.dumps(self.vca_data))
+ vca = self.vca.get("vca_name")
+ self.vca._client.get_token.assert_called()
+ self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca/1234")
+ self.assertEqual(vca, self.vca_data)
+
+ @patch("osmclient.sol005.vca.utils")
+ def test_get_client_exception(self, mock_utils):
+ mock_utils.validate_uuid4.return_value = False
+ self.vca.get_id = Mock()
+ self.vca.get_id.return_value = "1234"
+ self.vca._http.get2_cmd.return_value = (404, json.dumps({}))
+ with self.assertRaises(ClientException):
+ self.vca.get("vca_name")
+ self.vca._client.get_token.assert_called()
+ self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca/1234")
+
+ @patch("osmclient.sol005.vca.utils")
+ def test_get_not_exception(self, mock_utils):
+ mock_utils.validate_uuid4.return_value = False
+ self.vca.get_id = Mock()
+ self.vca.get_id.return_value = "1234"
+ self.vca._http.get2_cmd.side_effect = NotFound()
+ with self.assertRaises(NotFound):
+ self.vca.get("vca_name")
+ self.vca._client.get_token.assert_called()
+ self.vca._http.get2_cmd.assert_called_with("/admin/v1/vca/1234")
--- /dev/null
+# Copyright 2021 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+OSM K8s cluster API handling
+"""
+
+from osmclient.common import utils
+from osmclient.common.exceptions import NotFound
+from osmclient.common.exceptions import ClientException
+import json
+
+
+class VCA(object):
+ def __init__(self, http=None, client=None):
+ self._http = http
+ self._client = client
+ self._apiName = "/admin"
+ self._apiVersion = "/v1"
+ self._apiResource = "/vca"
+ self._apiBase = "{}{}{}".format(
+ self._apiName, self._apiVersion, self._apiResource
+ )
+
+ def create(self, name, vca):
+ self._client.get_token()
+ http_code, resp = self._http.post_cmd(
+ endpoint=self._apiBase, postfields_dict=vca
+ )
+ resp = json.loads(resp) if resp else {}
+ if "id" not in resp:
+ raise ClientException("unexpected response from server - {}".format(resp))
+ print(resp["id"])
+
+ def update(self, name, vca):
+ self._client.get_token()
+ vca_id = self.get(name)["_id"]
+ self._http.patch_cmd(
+ endpoint="{}/{}".format(self._apiBase, vca_id),
+ postfields_dict=vca,
+ )
+
+ def get_id(self, name):
+ """Returns a VCA id from a VCA name"""
+ for vca in self.list():
+ if name == vca["name"]:
+ return vca["_id"]
+ raise NotFound("VCA {} not found".format(name))
+
+ def delete(self, name, force=False):
+ self._client.get_token()
+ vca_id = name
+ if not utils.validate_uuid4(name):
+ vca_id = self.get_id(name)
+ querystring = "?FORCE=True" if force else ""
+ http_code, resp = self._http.delete_cmd(
+ "{}/{}{}".format(self._apiBase, vca_id, querystring)
+ )
+ if http_code == 202:
+ print("Deletion in progress")
+ elif http_code == 204:
+ print("Deleted")
+ else:
+ msg = resp or ""
+ raise ClientException("failed to delete VCA {} - {}".format(name, msg))
+
+ def list(self, cmd_filter=None):
+ """Returns a list of K8s clusters"""
+ self._client.get_token()
+ filter_string = ""
+ if cmd_filter:
+ filter_string = "?{}".format(cmd_filter)
+ _, resp = self._http.get2_cmd("{}{}".format(self._apiBase, filter_string))
+ if resp:
+ return json.loads(resp)
+ return list()
+
+ def get(self, name):
+ """Returns a VCA based on name or id"""
+ self._client.get_token()
+ vca_id = name
+ if not utils.validate_uuid4(name):
+ vca_id = self.get_id(name)
+ try:
+ _, resp = self._http.get2_cmd("{}/{}".format(self._apiBase, vca_id))
+ resp = json.loads(resp) if resp else {}
+ if "_id" not in resp:
+ raise ClientException("failed to get VCA info: {}".format(resp))
+ return resp
+ except NotFound:
+ raise NotFound("VCA {} not found".format(name))
def create(
self, name, vim_access, sdn_controller=None, sdn_port_mapping=None, wait=False
):
+ vca_id = None
+
+ def get_vca_id(vca):
+ vca = self._client.vca.get(vca)
+ if vca is None:
+ raise NotFound("cannot find vca '{}'".format(vca))
+ return vca["_id"]
+
self._logger.debug("")
self._client.get_token()
+ if "vca" in vim_access:
+ vca_id = get_vca_id(vim_access["vca"])
if "vim-type" not in vim_access:
# 'openstack' not in vim_access['vim-type']):
raise Exception("vim type not provided")
-
vim_account = {}
vim_account["name"] = name
vim_account = self.update_vim_account_dict(vim_account, vim_access)
-
+ if vca_id:
+ vim_account["vca"] = vca_id
vim_config = {}
if "config" in vim_access and vim_access["config"] is not None:
vim_config = yaml.safe_load(vim_access["config"])
from osmclient.v1 import nsd
from osmclient.v1 import vim
from osmclient.v1 import package
-from osmclient.v1 import vca
from osmclient.v1 import utils
from osmclient.common import http
from osmclient.common import package_tool
self.package = package.Package(
http=http_client, upload_http=upload_client, client=self, **kwargs
)
- self.vca = vca.Vca(http_client, client=self, **kwargs)
self.utils = utils.Utils(http_client, **kwargs)
self.package_tool = package_tool.PackageTool(client=self)
projects / osm / osmclient.git / commitdiff