Coverage for osmclient/sol005/subscription.py: 19%

1# 

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

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

4# a copy of the License at 

5# 

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

7# 

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

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

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

11# License for the specific language governing permissions and limitations 

12# under the License. 

13# 

14 

15""" 

16OSM Subscription API handling 

17""" 

18 

19from osmclient.common.exceptions import NotFound 

20from osmclient.common.exceptions import ClientException 

21import yaml 

22import json 

23import logging 

24 

25 

26class Subscription(object): 

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

28 self._http = http 

29 self._client = client 

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

31 self._apiName = "/nslcm" 

32 self._apiVersion = "/v1" 

33 self._apiResource = "/subscriptions" 

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

35 self._apiName, self._apiVersion, self._apiResource 

36 ) 

37 

38 def _rebuild_apibase(self, event_type): 

39 self._logger.debug("") 

40 if event_type == "ns": 

41 self._apiName = "/nslcm" 

42 else: 

43 raise ClientException("unsupported event_type {}".format(event_type)) 

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

45 self._apiName, self._apiVersion, self._apiResource 

46 ) 

47 

48 def create(self, event_type, event): 

49 """ 

50 Creates a subscription 

51 :param event_type: event type to be subscribed (only ns is supported) 

52 :param event: yaml string defining the event to be subscribed 

53 :return: None. Exception if fail 

54 """ 

55 self._logger.debug("") 

56 self._client.get_token() 

57 self._rebuild_apibase(event_type) 

58 

59 subscription_event = yaml.safe_load(event) 

60 

61 http_code, resp = self._http.post_cmd( 

62 endpoint=self._apiBase, postfields_dict=subscription_event 

63 ) 

64 if resp: 

65 resp = json.loads(resp) 

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

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

68 print(resp["id"]) 

69 

70 def delete(self, event_type, subscription_id, force=False): 

71 """ 

72 Deletes a subscription 

73 :param event_type: event type to be subscribed (only ns is supported) 

74 :param subscription_id: identifier of the subscription 

75 :param force: forces deletion 

76 :return: None. Exception if fail 

77 """ 

78 self._logger.debug("") 

79 self._client.get_token() 

80 self._rebuild_apibase(event_type) 

81 querystring = "" 

82 if force: 

83 querystring = "?FORCE=True" 

84 http_code, resp = self._http.delete_cmd( 

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

86 ) 

87 if http_code == 202: 

88 print("Deletion in progress") 

89 elif http_code == 204: 

90 print("Deleted") 

91 else: 

92 msg = resp or "" 

93 raise ClientException( 

94 "failed to delete subscription {} - {}".format(subscription_id, msg) 

95 ) 

96 

97 def list(self, event_type, filter=None): 

98 """ 

99 Returns a list of subscriptions 

100 :param event_type: event type to be subscribed (only ns is supported) 

101 :param filter 

102 :return: list of subscriptions 

103 """ 

104 self._logger.debug("") 

105 self._client.get_token() 

106 self._rebuild_apibase(event_type) 

107 filter_string = "" 

108 if filter: 

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

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

111 if resp: 

112 return json.loads(resp) 

113 return list() 

114 

115 def get(self, event_type, subscription_id): 

116 """ 

117 Returns a subscription from a subscription id 

118 :param event_type: event type to be subscribed (only ns is supported) 

119 :param subscription_id: identifier of the subscription 

120 :return: dict with the subscription 

121 """ 

122 self._logger.debug("") 

123 self._client.get_token() 

124 self._rebuild_apibase(event_type) 

125 try: 

126 _, resp = self._http.get2_cmd( 

127 "{}/{}".format(self._apiBase, subscription_id) 

128 ) 

129 self._logger.debug(yaml.safe_dump(resp)) 

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

131 raise ClientException( 

132 "failed to get subscription info: {}".format(resp) 

133 ) 

134 return json.loads(resp) 

135 except NotFound: 

136 raise NotFound("subscription '{}' not found".format(subscription_id))