1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
"""SO Element module."""
# Copyright 2022 Orange, Deutsche Telekom AG
#
# 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
from abc import ABC
from dataclasses import dataclass
from enum import Enum
from typing import Dict
from onapsdk.configuration import settings
from onapsdk.sdc.service import Service
from onapsdk.sdc.vf import Vf
from onapsdk.onap_service import OnapService
from onapsdk.utils.headers_creator import headers_so_creator
from onapsdk.utils.jinja import jinja_env
from onapsdk.utils.mixins import WaitForFinishMixin
from onapsdk.utils.tosca_file_handler import get_modules_list_from_tosca_file
from onapsdk.utils.gui import GuiItem, GuiList
@dataclass
class SoElement(OnapService):
"""Mother Class of all SO elements."""
name: str = None
_server: str = "SO"
base_url = settings.SO_URL
api_version = settings.SO_API_VERSION
_status: str = None
@property
def headers(self):
"""Create headers for SO request.
It is used as a property because x-transactionid header should be unique for each request.
"""
return headers_so_creator(OnapService.headers)
@classmethod
def get_subscription_service_type(cls, vf_name):
"""Retrieve the model info of the VFs."""
vf_object = Vf(name=vf_name)
return vf_object.name
@classmethod
def get_service_model_info(cls, service_name):
"""Retrieve Service Model info."""
service = Service(name=service_name)
template_service = jinja_env().get_template("service_instance_model_info.json.j2")
# Get service instance model
parsed = json.loads(
template_service.render(
model_invariant_id=service.unique_uuid,
model_name_version_id=service.identifier,
model_name=service.name,
model_version=service.version,
)
)
return json.dumps(parsed, indent=4)
@classmethod
def get_vnf_model_info(cls, vf_name):
"""Retrieve the model info of the VFs."""
vf_object = Vf(name=vf_name)
template_service = jinja_env().get_template("vnf_model_info.json.j2")
parsed = json.loads(
template_service.render(
vnf_model_invariant_uuid=vf_object.unique_uuid,
vnf_model_customization_id="????",
vnf_model_version_id=vf_object.identifier,
vnf_model_name=vf_object.name,
vnf_model_version=vf_object.version,
vnf_model_instance_name=(vf_object.name + " 0"),
)
)
# we need also a vnf instance Name
# Usually it is found like that
# name: toto
# instance name: toto 0
# it can be retrieved from the toscafrom onapsdk.configuration import settings
return json.dumps(parsed, indent=4)
@classmethod
def get_vf_model_info(cls, vf_model: str) -> str:
"""Retrieve the VF model info From Tosca?."""
modules: Dict = get_modules_list_from_tosca_file(vf_model)
template_service = jinja_env().get_template("vf_model_info.json.j2")
parsed = json.loads(template_service.render(modules=modules))
return json.dumps(parsed, indent=4)
@classmethod
def _base_create_url(cls) -> str:
"""
Give back the base url of SO.
Returns:
str: the base url
"""
return "{}/onap/so/infra/serviceInstantiation/{}/serviceInstances".format(
cls.base_url, cls.api_version
)
@classmethod
def get_guis(cls) -> GuiItem:
"""Retrieve the status of the SO GUIs.
Only one GUI is referenced for SO: SO monitor
Return the list of GUIs
"""
gui_url = settings.SO_MONITOR_GUI_SERVICE
so_gui_response = cls.send_message(
"GET", "Get SO GUI Status", gui_url)
guilist = GuiList([])
guilist.add(GuiItem(
gui_url,
so_gui_response.status_code))
return guilist
class OrchestrationRequest(SoElement, WaitForFinishMixin, ABC):
"""Base SO orchestration request class."""
WAIT_FOR_SLEEP_TIME = 10
def __init__(self,
request_id: str) -> None:
"""Instantiate object initialization.
Initializator used by classes inherited from this abstract class.
Args:
request_id (str): request ID
"""
super().__init__()
self.request_id: str = request_id
class StatusEnum(Enum):
"""Status enum.
Store possible statuses for instantiation:
- IN_PROGRESS,
- FAILED,
- COMPLETE.
If instantiation has status which is not covered by these values
UNKNOWN value is used.
"""
IN_PROGRESS = "IN_PROGRESS"
FAILED = "FAILED"
COMPLETED = "COMPLETE"
UNKNOWN = "UNKNOWN"
@property
def status(self) -> "StatusEnum":
"""Object instantiation status.
It's populated by call SO orchestation request endpoint.
Returns:
StatusEnum: Instantiation status.
"""
response: dict = self.send_message_json(
"GET",
f"Check {self.request_id} orchestration request status",
(f"{self.base_url}/onap/so/infra/"
f"orchestrationRequests/{self.api_version}/{self.request_id}"),
headers=headers_so_creator(OnapService.headers)
)
try:
return self.StatusEnum(response["request"]["requestStatus"]["requestState"])
except (KeyError, ValueError):
self._logger.exception("Invalid status")
return self.StatusEnum.UNKNOWN
@property
def finished(self) -> bool:
"""Store an information if instantion is finished or not.
Instantiation is finished if it's status is COMPLETED or FAILED.
Returns:
bool: True if instantiation is finished, False otherwise.
"""
return self.status in [self.StatusEnum.COMPLETED, self.StatusEnum.FAILED]
@property
def completed(self) -> bool:
"""Store an information if instantion is completed or not.
Instantiation is completed if it's status is COMPLETED.
Returns:
bool: True if instantiation is completed, False otherwise.
"""
return self.finished and self.status == self.StatusEnum.COMPLETED
@property
def failed(self) -> bool:
"""Store an information if instantion is failed or not.
Instantiation is failed if it's status is FAILED.
Returns:
bool: True if instantiation is failed, False otherwise.
"""
return self.finished and self.status == self.StatusEnum.FAILED
|
