[Jenkins]Jenkins

Code Coverage

Cobertura Coverage Report > osmclient.sol005 >

vca.py

Trend

File Coverage summary

NameClassesLinesConditionals
vca.py
100%
1/1
100%
73/73
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
vca.py
100%
73/73
N/A

Source

osmclient/sol005/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
1
"""




16

OSM K8s cluster API handling




17

"""




18






19
1
from osmclient.common import utils




20
1
from osmclient.common.exceptions import NotFound




21
1
from osmclient.common.exceptions import ClientException




22
1
import json




23
1
import logging




24






25






26
1
class VCA(object):




27
1
    def __init__(self, http=None, client=None):




28
1
        self._http = http




29
1
        self._client = client




30
1
        self._logger = logging.getLogger("osmclient")




31
1
        self._apiName = "/admin"




32
1
        self._apiVersion = "/v1"




33
1
        self._apiResource = "/vca"




34
1
        self._apiBase = "{}{}{}".format(




35

            self._apiName, self._apiVersion, self._apiResource




36

        )




37






38
1
    def create(self, name, vca):




39
1
        self._logger.debug("")




40
1
        self._client.get_token()




41
1
        http_code, resp = self._http.post_cmd(




42

            endpoint=self._apiBase, postfields_dict=vca




43

        )




44
1
        resp = json.loads(resp) if resp else {}




45
1
        if "id" not in resp:




46
1
            raise ClientException("unexpected response from server - {}".format(resp))




47
1
        print(resp["id"])




48






49
1
    def update(self, name, vca):




50
1
        self._logger.debug("")




51
1
        self._client.get_token()




52
1
        vca_id = self.get(name)["_id"]




53
1
        self._http.patch_cmd(




54

            endpoint="{}/{}".format(self._apiBase, vca_id),




55

            postfields_dict=vca,




56

        )




57






58
1
    def get_id(self, name):




59
1
        self._logger.debug("")




60
1
        """Returns a VCA id from a VCA name"""




61
1
        for vca in self.list():




62
1
            if name == vca["name"]:




63
1
                return vca["_id"]




64
1
        raise NotFound("VCA {} not found".format(name))




65






66
1
    def delete(self, name, force=False):




67
1
        self._logger.debug("")




68
1
        self._client.get_token()




69
1
        vca_id = name




70
1
        if not utils.validate_uuid4(name):




71
1
            vca_id = self.get_id(name)




72
1
        querystring = "?FORCE=True" if force else ""




73
1
        http_code, resp = self._http.delete_cmd(




74

            "{}/{}{}".format(self._apiBase, vca_id, querystring)




75

        )




76
1
        if http_code == 202:




77
1
            print("Deletion in progress")




78
1
        elif http_code == 204:




79
1
            print("Deleted")




80

        else:




81
1
            msg = resp or ""




82
1
            raise ClientException("failed to delete VCA {} - {}".format(name, msg))




83






84
1
    def list(self, cmd_filter=None):




85

        """Returns a list of K8s clusters"""




86
1
        self._logger.debug("")




87
1
        self._client.get_token()




88
1
        filter_string = ""




89
1
        if cmd_filter:




90
1
            filter_string = "?{}".format(cmd_filter)




91
1
        _, resp = self._http.get2_cmd("{}{}".format(self._apiBase, filter_string))




92
1
        if resp:




93
1
            return json.loads(resp)




94
1
        return list()




95






96
1
    def get(self, name):




97

        """Returns a VCA based on name or id"""




98
1
        self._logger.debug("")




99
1
        self._client.get_token()




100
1
        vca_id = name




101
1
        if not utils.validate_uuid4(name):




102
1
            vca_id = self.get_id(name)




103
1
        try:




104
1
            _, resp = self._http.get2_cmd("{}/{}".format(self._apiBase, vca_id))




105
1
            resp = json.loads(resp) if resp else {}




106
1
            if "_id" not in resp:




107
1
                raise ClientException("failed to get VCA info: {}".format(resp))




108
1
            return resp




109
1
        except NotFound:




110
1
            raise NotFound("VCA {} not found".format(name))