Coverage for osmclient/sol005/pdud.py: 16%

1# Copyright 2018 Telefonica 

2# 

3# All Rights Reserved. 

4# 

5# Licensed under the Apache License, Version 2.0 (the "License"); you may 

6# not use this file except in compliance with the License. You may obtain 

7# a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, software 

12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

14# License for the specific language governing permissions and limitations 

15# under the License. 

16 

17""" 

18OSM pdud API handling 

19""" 

20 

21from osmclient.common.exceptions import NotFound 

22from osmclient.common.exceptions import ClientException 

23from osmclient.common import utils 

24import json 

25import logging 

26 

27 

28class Pdu(object): 

29 def __init__(self, http=None, client=None): 

30 self._http = http 

31 self._client = client 

32 self._logger = logging.getLogger("osmclient") 

33 self._apiName = "/pdu" 

34 self._apiVersion = "/v1" 

35 self._apiResource = "/pdu_descriptors" 

36 self._apiBase = "{}{}{}".format( 

37 self._apiName, self._apiVersion, self._apiResource 

38 ) 

39 

40 def _get_vim_account(self, vim_account): 

41 vim = self._client.vim.get(vim_account) 

42 if vim is None: 

43 raise NotFound("cannot find vim account '{}'".format(vim_account)) 

44 return vim 

45 

46 def list(self, filter=None): 

47 self._logger.debug("") 

48 self._client.get_token() 

49 filter_string = "" 

50 if filter: 

51 filter_string = "?{}".format(filter) 

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

53 if resp: 

54 return json.loads(resp) 

55 return list() 

56 

57 def get(self, name): 

58 self._logger.debug("") 

59 self._client.get_token() 

60 if utils.validate_uuid4(name): 

61 for pdud in self.list(): 

62 if name == pdud["_id"]: 

63 return pdud 

64 else: 

65 for pdud in self.list(): 

66 if "name" in pdud and name == pdud["name"]: 

67 return pdud 

68 raise NotFound("pdud {} not found".format(name)) 

69 

70 def get_individual(self, name): 

71 self._logger.debug("") 

72 pdud = self.get(name) 

73 # It is redundant, since the previous one already gets the whole pdudInfo 

74 # The only difference is that a different primitive is exercised 

75 try: 

76 _, resp = self._http.get2_cmd("{}/{}".format(self._apiBase, pdud["_id"])) 

77 except NotFound: 

78 raise NotFound("pdu '{}' not found".format(name)) 

79 # print(yaml.safe_dump(resp)) 

80 if resp: 

81 return json.loads(resp) 

82 raise NotFound("pdu '{}' not found".format(name)) 

83 

84 def delete(self, name, force=False): 

85 self._logger.debug("") 

86 pdud = self.get(name) 

87 querystring = "" 

88 if force: 

89 querystring = "?FORCE=True" 

90 http_code, resp = self._http.delete_cmd( 

91 "{}/{}{}".format(self._apiBase, pdud["_id"], querystring) 

92 ) 

93 # print('HTTP CODE: {}'.format(http_code)) 

94 # print('RESP: {}'.format(resp)) 

95 if http_code == 202: 

96 print("Deletion in progress") 

97 elif http_code == 204: 

98 print("Deleted") 

99 else: 

100 msg = resp or "" 

101 # if resp: 

102 # try: 

103 # msg = json.loads(resp) 

104 # except ValueError: 

105 # msg = resp 

106 raise ClientException("failed to delete pdu {} - {}".format(name, msg)) 

107 

108 def create(self, pdu, update_endpoint=None): 

109 self._logger.debug("") 

110 

111 if pdu["vim_accounts"]: 

112 vim_account_list = [] 

113 for vim_account in pdu["vim_accounts"]: 

114 vim = self._get_vim_account(vim_account) 

115 vim_account_list.append(vim["_id"]) 

116 pdu["vim_accounts"] = vim_account_list 

117 

118 self._client.get_token() 

119 headers = self._client._headers 

120 headers["Content-Type"] = "application/yaml" 

121 self._http.set_http_header(headers) 

122 if update_endpoint: 

123 http_code, resp = self._http.patch_cmd( 

124 endpoint=update_endpoint, postfields_dict=pdu 

125 ) 

126 else: 

127 endpoint = self._apiBase 

128 # endpoint = '{}{}'.format(self._apiBase,ow_string) 

129 http_code, resp = self._http.post_cmd( 

130 endpoint=endpoint, postfields_dict=pdu 

131 ) 

132 if http_code in (200, 201, 202): 

133 if resp: 

134 resp = json.loads(resp) 

135 if not resp or "id" not in resp: 

136 raise ClientException( 

137 "unexpected response from server: {}".format(resp) 

138 ) 

139 print(resp["id"]) 

140 elif http_code == 204: 

141 print("Updated") 

142 

143 def update(self, name, pdu): 

144 self._logger.debug("") 

145 pdud = self.get(name) 

146 endpoint = "{}/{}".format(self._apiBase, pdud["_id"]) 

147 self.create(pdu=pdu, update_endpoint=endpoint)