blob: 31c16d21bae100152d3a83b1091e652465f261dc (
plain)
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
|
# http://www.apache.org/licenses/LICENSE-2.0
"""PNF simulator registration module."""
import time
import requests
from jinja2 import Environment, PackageLoader, select_autoescape
from onapsdk.configuration import settings
from onapsdk.ves.ves import Ves
from onaptests.steps.base import BaseStep
from onaptests.steps.cloud.expose_service_node_port import \
ExposeServiceNodePortStep
from onaptests.utils.exceptions import OnapTestException
class SendPnfRegisterVesEvent(BaseStep):
"""PNF VES registration step."""
def __init__(self) -> None:
"""Initialize step."""
super().__init__(cleanup=BaseStep.HAS_NO_CLEANUP)
if settings.EXPOSE_SERVICES_NODE_PORTS:
self.add_step(ExposeServiceNodePortStep(component="VES-Collector",
service_name="dcae-ves-collector",
port=8080,
node_port=settings.VES_NODE_PORT))
@property
def description(self) -> str:
"""Step description."""
return "Register PNF with VES."
@property
def component(self) -> str:
"""Component name."""
return "Environment"
@BaseStep.store_state
def execute(self) -> None:
"""Send PNF registration event."""
super().execute()
registration_number: int = 0
source_name = settings.SERVICE_INSTANCE_NAME
jinja_env = Environment(autoescape=select_autoescape(['jinja']),
loader=PackageLoader('onaptests.templates',
'artifacts'))
template = jinja_env.get_template("pnf_register_ves_message.jinja")
event_data = template.render(
source_name=source_name)
registered_successfully: bool = False
while (registration_number < settings.PNF_REGISTRATION_NUMBER_OF_TRIES and
not registered_successfully):
try:
response = Ves.send_event(version="v7", json_event=event_data,
basic_auth=settings.VES_BASIC_AUTH)
if response is None:
raise OnapTestException("Failed to send event to VES SERVER")
response.raise_for_status()
registered_successfully = True
self._logger.info(f"PNF registered with {settings.SERVICE_INSTANCE_NAME} "
"source name")
except (requests.ConnectionError, requests.HTTPError) as http_error:
self._logger.debug(f"Can't send to ves: {str(http_error)}")
registration_number += 1
time.sleep(settings.PNF_WAIT_TIME)
if not registered_successfully:
raise OnapTestException("PNF not registered successfully")
|
