30541d4c51097c926f321bb4a99b61bc8f6da516
[nonrtric.git] / test / usecases / oruclosedlooprecovery / scriptversion / app / main.py
1
2 #  ============LICENSE_START===============================================
3 #  Copyright (C) 2021 Nordix Foundation. All rights reserved.
4 #  ========================================================================
5 #  Licensed under the Apache License, Version 2.0 (the "License");
6 #  you may not use this file except in compliance with the License.
7 #  You may obtain 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,
13 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 #  See the License for the specific language governing permissions and
15 #  limitations under the License.
16 #  ============LICENSE_END=================================================
17 #
18
19 import argparse
20 import ast
21 import json
22 import os
23 import requests
24 import time
25
26 MR_PATH = "/events/[TOPIC]/users/test/"
27 SDNR_PATH = "/rests/data/network-topology:network-topology/topology=topology-netconf/node=[O-DU-ID]/yang-ext:mount/o-ran-sc-du-hello-world:network-function/du-to-ru-connection=[O-RU-ID]"
28
29 UNLOCK_MESSAGE = {
30     "o-ran-sc-du-hello-world:du-to-ru-connection": [
31         {
32             "name":"",
33             "administrative-state":"UNLOCKED"
34         }
35     ]
36 }
37
38
39 def is_message_new_link_failure(message):
40     msg_as_json = json.loads(message)
41     event_headers = msg_as_json["event"]["commonEventHeader"]
42
43     link_failure = False
44     if (event_headers["domain"] == "fault"):
45         fault_fields = msg_as_json["event"]["faultFields"]
46         link_failure = fault_fields["alarmCondition"] == "30" and fault_fields["eventSeverity"] != "NORMAL"
47
48     return link_failure
49
50
51 def is_message_clear_link_failure(message):
52     msg_as_json = json.loads(message)
53     event_headers = msg_as_json["event"]["commonEventHeader"]
54
55     link_failure_clear = False
56     if (event_headers["domain"] == "fault"):
57         fault_fields = msg_as_json["event"]["faultFields"]
58         link_failure_clear = fault_fields["alarmCondition"] == "30" and fault_fields["eventSeverity"] == "NORMAL"
59
60     return link_failure_clear
61
62
63 def handle_link_failure(message, o_ru_to_o_du_map, sdnr_address):
64     verboseprint("Got a link failure: ")
65     alarm_msg_as_json = json.loads(message)
66     event_headers = alarm_msg_as_json["event"]["commonEventHeader"]
67     o_ru_id = event_headers["sourceName"]
68     verboseprint("O-RU ID: " + o_ru_id)
69     o_du_id = o_ru_to_o_du_map[o_ru_id]
70     verboseprint("O-DU ID: " + o_du_id)
71     unlock_msg = json.loads(json.dumps(UNLOCK_MESSAGE))
72     unlock_msg["o-ran-sc-du-hello-world:du-to-ru-connection"][0]["name"] = o_ru_id
73     send_path = SDNR_PATH.replace("[O-DU-ID]", o_du_id).replace("[O-RU-ID]", o_ru_id)
74     requests.post(sdnr_address + send_path, json=unlock_msg)
75
76
77 def handle_clear_link_failure(message):
78     msg_as_json = json.loads(message)
79     event_headers = msg_as_json["event"]["commonEventHeader"]
80     o_ru_id = event_headers["sourceName"]
81     verboseprint("Cleared Link Failure for O-RU ID: " + o_ru_id)
82
83
84 def read_o_ru_to_o_du_map_from_file(map_file):
85     file = open(map_file, "r")
86     contents = file.read()
87     dictionary = ast.literal_eval(contents)
88     file.close()
89     return dictionary
90
91
92 if __name__ == '__main__':
93     parser = argparse.ArgumentParser(prog='PROG')
94     parser.add_argument('--mrHost', help='The URL of the MR host (default: %(default)s)', default="http://message-router.onap")
95     parser.add_argument('--mrPort', help='The port of the MR host (default: %(default)d)', type=int, default=3904)
96     parser.add_argument('--mrTopic', help='The topic to poll messages from (default: %(default)s)', default="unauthenticated.SEC_FAULT_OUTPUT")
97     parser.add_argument('--sdnrHost', help='The URL of the SNDR host (default: %(default)s)', default="http://localhost")
98     parser.add_argument('--sdnrPort', help='The port of the SDNR host (default: %(default)d)', type=int, default=9990)
99     parser.add_argument('--oRuTooDuMapFile', help='A file with the mapping between O-RU ID and O-DU ID as a dictionary (default: %(default)s)', default="o-ru-to-o-du-map.txt")
100     parser.add_argument('--pollTime', help='The time between polls (default: %(default)d)', type=int, default=10)
101     parser.add_argument('-v', '--verbose', action='store_true', help='Turn on verbose printing')
102     parser.add_argument('--version', action='version', version='%(prog)s 1.0')
103     args = vars(parser.parse_args())
104     mr_host = args["mrHost"]
105     if os.getenv("MR-HOST") is not None:
106         mr_host = os.getenv("MR-HOST")
107         print("Using MR Host from os: " + mr_host)
108     mr_port = args["mrPort"]
109     if os.getenv("MR-PORT") is not None:
110         mr_port = os.getenv("MR-PORT")
111         print("Using MR Port from os: " + mr_port)
112     mr_topic = args["mrTopic"]
113     sdnr_host = args["sdnrHost"]
114     if os.getenv("SDNR-HOST") is not None:
115         sdnr_host = os.getenv("SDNR-HOST")
116         print("Using SNDR Host from os: " + sdnr_host)
117     sdnr_port = args["sdnrPort"]
118     if os.getenv("SDNR-PORT") is not None:
119         sdnr_port = os.getenv("SDNR-PORT")
120         print("Using SNDR Host from os: " + sdnr_port)
121     o_ru_to_o_du_map = read_o_ru_to_o_du_map_from_file(args["oRuTooDuMapFile"])
122     pollTime = args["pollTime"]
123
124     if os.getenv("VERBOSE") is not None or args["verbose"]:
125
126         def verboseprint(*args, **kwargs):
127             print(*args, **kwargs)
128
129     else:
130         verboseprint = lambda *a, **k: None  # do-nothing function
131
132     verboseprint("Using MR address: " + mr_host + ":" + str(mr_port) + " and topic: " + mr_topic)
133     verboseprint("Using SDNR address: " + sdnr_host + ":" + str(sdnr_port))
134     verboseprint("Starting with " + str(pollTime) + " seconds between polls")
135     mr_address = mr_host + ":" + str(mr_port) + MR_PATH.replace("[TOPIC]", mr_topic)
136     sdnr_address = sdnr_host + ":" + str(sdnr_port)
137
138     while True:
139         try:
140             verboseprint("Polling")
141             response = requests.get(mr_address)
142             messages = response.json()
143             for message in messages:
144                 if (is_message_new_link_failure(message)):
145                     handle_link_failure(message, o_ru_to_o_du_map, sdnr_address)
146                 elif (is_message_clear_link_failure(message)):
147                     handle_clear_link_failure(message)
148         except Exception as inst:
149             print(inst)
150
151         time.sleep(pollTime)