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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
|
"""Pnf instance 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.
from typing import Iterator, Optional, TYPE_CHECKING
from onapsdk.exceptions import ResourceNotFound
from onapsdk.so.deletion import PnfDeletionRequest
from .instance import Instance
if TYPE_CHECKING:
from .service import ServiceInstance # pylint: disable=cyclic-import
class PnfInstance(Instance): # pylint: disable=too-many-instance-attributes
"""Pnf instance class."""
def __init__(self, # pylint: disable=too-many-arguments, too-many-locals
service_instance: "ServiceInstance",
pnf_name: str,
in_maint: bool,
selflink: str = None,
pnf_id: str = None,
equip_type: str = None,
equip_vendor: str = None,
equip_model: str = None,
management_option: str = None,
orchestration_status: str = None,
ipaddress_v4_oam: str = None,
sw_version: str = None,
frame_id: str = None,
serial_number: str = None,
ipaddress_v4_loopback_0: str = None,
ipaddress_v6_loopback_0: str = None,
ipaddress_v4_aim: str = None,
ipaddress_v6_aim: str = None,
ipaddress_v6_oam: str = None,
inv_status: str = None,
resource_version: str = None,
prov_status: str = None,
nf_role: str = None,
admin_status: str = None,
operational_status: str = None,
model_customization_id: str = None,
model_invariant_id: str = None,
model_version_id: str = None,
pnf_ipv4_address: str = None,
pnf_ipv6_address: str = None) -> None:
"""Pnf instance object initialization.
Args:
service_instance (ServiceInstance): Service instance object
pnf_name (str): unique name of Physical Network Function
in_maint (bool): Used to indicate whether or not this object is in maintenance mode
(maintenance mode = True). This field (in conjunction with prov_status)
is used to suppress alarms and vSCL on VNFs/VMs.
selflink (str, optional): URL to endpoint where AAI can get more details.
Defaults to None.
pnf_id (str, optional): id of pnf. Defaults to None.
equip_type (str, optional): Equipment type. Source of truth should define valid values.
Defaults to None.
equip_vendor (str, optional): Equipment vendor. Source of truth should define
valid values. Defaults to None.
equip_model (str, optional): Equipment model. Source of truth should define
valid values. Defaults to None.
management_option (str, optional): identifier of managed customer. Defaults to None.
orchestration_status (str, optional): Orchestration status of this pnf.
Defaults to None.
ipaddress_v4_oam (str, optional): ipv4-oam-address with new naming
convention for IP addresses. Defaults to None.
sw_version (str, optional): sw-version is the version of SW for the hosted
application on the PNF. Defaults to None.
frame_id (str, optional): ID of the physical frame (relay rack) where pnf is installed.
Defaults to None.
serial_number (str, optional): Serial number of the device. Defaults to None.
ipaddress_v4_loopback_0 (str, optional): IPV4 Loopback 0 address. Defaults to None.
ipaddress_v6_loopback_0 (str, optional): IPV6 Loopback 0 address. Defaults to None.
ipaddress_v4_aim (str, optional): IPV4 AIM address. Defaults to None.
ipaddress_v6_aim (str, optional): IPV6 AIM address. Defaults to None.
ipaddress_v6_oam (str, optional): IPV6 OAM address. Defaults to None.
inv_status (str, optional): CANOPI's inventory status. Only set with values exactly
as defined by CANOPI. Defaults to None.
resource_version (str, optional): Used for optimistic concurrency.
Must be empty on create, valid on update and delete. Defaults to None.
prov_status (str, optional): Prov Status of this device (not under canopi control)
Valid values [PREPROV/NVTPROV/PROV]. Defaults to None.
nf_role (str, optional): Nf Role is the role performed by this instance in the network.
Defaults to None.
admin_status (str, optional): admin Status of this PNF. Defaults to None.
operational_status (str, optional): Store the operational-status for this object.
Defaults to None.
model_customization_id (str, optional): Store the model-customization-id
for this object. Defaults to None.
model_invariant_id (str, optional): The ASDC model id for this resource model.
Defaults to None.
model_version_id (str, optional): The ASDC model version for this resource model.
Defaults to None.
pnf_ipv4_address (str, optional): This is the IP address (IPv4) for the PNF itself.
This is the IPv4 address that the PNF iself can be accessed at. Defaults to None.
pnf_ipv6_address (str, optional): This is the IP address (IPv6) for the PNF itself.
This is the IPv6 address that the PNF iself can be accessed at. Defaults to None.
"""
super().__init__(resource_version=resource_version,
model_invariant_id=model_invariant_id,
model_version_id=model_version_id)
self.service_instance: "ServiceInstance" = service_instance
self.pnf_name: str = pnf_name
self.in_maint: bool = in_maint
self.selflink: Optional[str] = selflink
self.pnf_id: Optional[str] = pnf_id
self.equip_type: Optional[str] = equip_type
self.equip_vendor: Optional[str] = equip_vendor
self.equip_model: Optional[str] = equip_model
self.management_option: Optional[str] = management_option
self.orchestration_status: Optional[str] = orchestration_status
self.ipaddress_v4_oam: Optional[str] = ipaddress_v4_oam
self.sw_version: Optional[str] = sw_version
self.frame_id: Optional[str] = frame_id
self.serial_number: Optional[str] = serial_number
self.ipaddress_v4_loopback_0: Optional[str] = ipaddress_v4_loopback_0
self.ipaddress_v6_loopback_0: Optional[str] = ipaddress_v6_loopback_0
self.ipaddress_v4_aim: Optional[str] = ipaddress_v4_aim
self.ipaddress_v6_aim: Optional[str] = ipaddress_v6_aim
self.ipaddress_v6_oam: Optional[str] = ipaddress_v6_oam
self.inv_status: Optional[str] = inv_status
self.prov_status: Optional[str] = prov_status
self.nf_role: Optional[str] = nf_role
self.admin_status: Optional[str] = admin_status
self.operational_status: Optional[str] = operational_status
self.model_customization_id: Optional[str] = model_customization_id
self.pnf_ipv4_address: Optional[str] = pnf_ipv4_address
self.pnf_ipv6_address: Optional[str] = pnf_ipv6_address
self._pnf: "Pnf" = None
def __repr__(self) -> str:
"""Pnf instance object representation.
Returns:
str: Human readable pnf instance representation
"""
return f"PnfInstance(pnf_name={self.pnf_name})"
@classmethod
def get_all_url(cls) -> str: # pylint: disable=arguments-differ
"""Return an url to get all pnfs.
Returns:
str: Url to get all pnfs
"""
return f"{cls.base_url}{cls.api_version}/network/pnfs/"
@classmethod
def get_all(cls) -> Iterator["PnfInstance"]:
"""Get all PNF instances.
Yields:
PnfInstance: Pnf instance
"""
for pnf_data in cls.send_message_json( \
"GET", \
"Get all pnf instances", \
cls.get_all_url() \
).get("pnf", []):
yield cls.create_from_api_response(pnf_data, None)
@property
def url(self) -> str:
"""Network instance url.
Returns:
str: NetworkInstance url
"""
return f"{self.base_url}{self.api_version}/network/pnfs/pnf/{self.pnf_name}"
@property
def pnf(self) -> "Pnf":
"""Pnf associated with that pnf instance.
Raises:
ResourceNotFound: Could not find PNF for that PNF instance
Returns:
Pnf: Pnf object associated with Pnf instance
"""
if not self._pnf:
for pnf in self.service_instance.sdc_service.pnfs:
if pnf.model_version_id == self.model_version_id:
self._pnf = pnf
return self._pnf
msg = (
f'Could not find PNF for the PNF instance'
f' with model version ID "{self.model_version_id}"'
)
raise ResourceNotFound(msg)
return self._pnf
@classmethod
def create_from_api_response(cls, api_response: dict,
service_instance: "ServiceInstance") -> "PnfInstance":
"""Create pnf instance object using HTTP API response dictionary.
Args:
api_response (dict): A&AI API response dictionary
service_instance (ServiceInstance): Service instance with which network is related
Returns:
PnfInstance: PnfInstance object
"""
return cls(service_instance=service_instance,
pnf_name=api_response["pnf-name"],
in_maint=api_response["in-maint"],
selflink=api_response.get("selflink"),
pnf_id=api_response.get("pnf-id"),
equip_type=api_response.get("equip-type"),
equip_vendor=api_response.get("equip-vendor"),
equip_model=api_response.get("equip-model"),
management_option=api_response.get("management-option"),
orchestration_status=api_response.get("orchestration-status"),
ipaddress_v4_oam=api_response.get("ipaddress-v4-oam"),
sw_version=api_response.get("sw-version"),
frame_id=api_response.get("frame-id"),
serial_number=api_response.get("serial-number"),
ipaddress_v4_loopback_0=api_response.get("ipaddress-v4-loopback-0"),
ipaddress_v6_loopback_0=api_response.get("ipaddress-v6-loopback-0"),
ipaddress_v4_aim=api_response.get("ipaddress-v4-aim"),
ipaddress_v6_aim=api_response.get("ipaddress-v6-aim"),
ipaddress_v6_oam=api_response.get("ipaddress-v6-oam"),
inv_status=api_response.get("inv-status"),
resource_version=api_response.get("resource-version"),
prov_status=api_response.get("prov-status"),
nf_role=api_response.get("nf-role"),
admin_status=api_response.get("admin-status"),
operational_status=api_response.get("operational-status"),
model_customization_id=api_response.get("model-customization-id"),
model_invariant_id=api_response.get("model-invariant-id"),
model_version_id=api_response.get("model-version-id"),
pnf_ipv4_address=api_response.get("pnf-ipv4-address"),
pnf_ipv6_address=api_response.get("pnf-ipv6-address"))
def delete(self, a_la_carte: bool = True) -> "PnfDeletionRequest":
"""Create PNF deletion request.
Send request to delete PNF instance
Args:
a_la_carte (boolean): deletion mode
Returns:
PnfDeletionRequest: Deletion request
"""
self._logger.debug("Delete %s pnf", self.pnf_name)
return PnfDeletionRequest.send_request(self, a_la_carte)
|
