From 2108563705a2ec8bb80029d36122c69fa4d06df5 Mon Sep 17 00:00:00 2001 From: "Hansen, Tony (th1395)" Date: Wed, 1 Dec 2021 22:01:56 +0000 Subject: run the black formatting tool on python code also fix up some copyright & license block lines Change-Id: Ifb628e2ef1e5f13fed0a29964eec387d3982d605 Signed-off-by: Hansen, Tony (th1395) Issue-ID: DCAEGEN2-2995 Signed-off-by: Hansen, Tony (th1395) --- miss_htbt_service/mod/trapd_vnf_table.py | 193 ++++++++++++++++++++----------- 1 file changed, 125 insertions(+), 68 deletions(-) (limited to 'miss_htbt_service/mod/trapd_vnf_table.py') diff --git a/miss_htbt_service/mod/trapd_vnf_table.py b/miss_htbt_service/mod/trapd_vnf_table.py index db6bd17..1b00336 100644 --- a/miss_htbt_service/mod/trapd_vnf_table.py +++ b/miss_htbt_service/mod/trapd_vnf_table.py @@ -1,11 +1,9 @@ # ============LICENSE_START======================================================= -# org.onap.dcae -# ================================================================================ -# Copyright (c) 2017-2020 AT&T Intellectual Property. All rights reserved. +# Copyright (c) 2017-2021 AT&T Intellectual Property. All rights reserved. # Copyright (c) 2019 Pantheon.tech. All rights reserved. -# Copyright 2020 Deutsche Telekom. All rights reserved. -# Copyright 2021 Samsung Electronics. All rights reserved. -# Copyright 2021 Fujitsu Ltd. +# Copyright (c) 2020 Deutsche Telekom. All rights reserved. +# Copyright (c) 2021 Samsung Electronics. All rights reserved. +# Copyright (c) 2021 Fujitsu Ltd. # ================================================================================ # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -44,45 +42,47 @@ _logger = logging.getLogger(__name__) def hb_properties(): - #Read the hbproperties.yaml for postgress and CBS related data - s=open(hb_properties_file, 'r') - a=yaml.full_load(s) - ip_address = a['pg_ipAddress'] - port_num = a['pg_portNum'] - user_name = a['pg_userName'] - password = a['pg_passwd'] - dbName = a['pg_dbName'] - db_name = dbName.lower() - cbs_polling_required = a['CBS_polling_allowed'] - cbs_polling_interval = a['CBS_polling_interval'] - s.close() - return ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval - - -def verify_DB_creation_1(user_name,password,ip_address,port_num,db_name): - connection_db = pm.postgres_db_open(user_name,password,ip_address,port_num,db_name) + # Read the hbproperties.yaml for postgress and CBS related data + s = open(hb_properties_file, "r") + a = yaml.full_load(s) + ip_address = a["pg_ipAddress"] + port_num = a["pg_portNum"] + user_name = a["pg_userName"] + password = a["pg_passwd"] + dbName = a["pg_dbName"] + db_name = dbName.lower() + cbs_polling_required = a["CBS_polling_allowed"] + cbs_polling_interval = a["CBS_polling_interval"] + s.close() + return ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval + + +def verify_DB_creation_1(user_name, password, ip_address, port_num, db_name): + connection_db = pm.postgres_db_open(user_name, password, ip_address, port_num, db_name) try: - _db_status=pm.db_table_creation_check(connection_db,"vnf_table_1") + _db_status = pm.db_table_creation_check(connection_db, "vnf_table_1") except Exception as e: return None return _db_status -def verify_DB_creation_2(user_name,password,ip_address,port_num,db_name): - connection_db = pm.postgres_db_open(user_name,password,ip_address,port_num,db_name) +def verify_DB_creation_2(user_name, password, ip_address, port_num, db_name): + + connection_db = pm.postgres_db_open(user_name, password, ip_address, port_num, db_name) try: - _db_status=pm.db_table_creation_check(connection_db,"vnf_table_2") + _db_status = pm.db_table_creation_check(connection_db, "vnf_table_2") except Exception as e: return None return _db_status -def verify_DB_creation_hb_common(user_name,password,ip_address,port_num,db_name): - connection_db = pm.postgres_db_open(user_name,password,ip_address,port_num,db_name) +def verify_DB_creation_hb_common(user_name, password, ip_address, port_num, db_name): + + connection_db = pm.postgres_db_open(user_name, password, ip_address, port_num, db_name) try: - _db_status=pm.db_table_creation_check(connection_db,"hb_common") + _db_status = pm.db_table_creation_check(connection_db, "hb_common") except Exception as e: return None @@ -90,35 +90,36 @@ def verify_DB_creation_hb_common(user_name,password,ip_address,port_num,db_name) def verify_cbspolling(): - os.environ['pytest']='test' - os.environ['SERVICE_NAME']='mvp-dcaegen2-heartbeat-static' + os.environ["pytest"] = "test" + os.environ["SERVICE_NAME"] = "mvp-dcaegen2-heartbeat-static" cbs.poll_cbs(10) - os.unsetenv('pytest') - os.unsetenv('SERVICE_NAME') + os.unsetenv("pytest") + os.unsetenv("SERVICE_NAME") def verify_fetch_json_file(): - os.environ['pytest']='test' - os.environ['SERVICE_NAME']='mvp-dcaegen2-heartbeat-static' - os.environ['CONSUL_HOST']='localhost' - os.environ['HOSTNAME']='mvp-dcaegen2-heartbeat-static' + os.environ["pytest"] = "test" + os.environ["SERVICE_NAME"] = "mvp-dcaegen2-heartbeat-static" + os.environ["CONSUL_HOST"] = "localhost" + os.environ["HOSTNAME"] = "mvp-dcaegen2-heartbeat-static" try: - db.fetch_json_file() - result = True + db.fetch_json_file() + result = True except Exception as e: - result = False + result = False print(result) - os.unsetenv('pytest') - os.unsetenv('SERVICE_NAME') - os.unsetenv('CONSUL_HOST') - os.unsetenv('HOSTNAME') + os.unsetenv("pytest") + os.unsetenv("SERVICE_NAME") + os.unsetenv("CONSUL_HOST") + os.unsetenv("HOSTNAME") return result + def verify_misshtbtdmain(): - os.environ['pytest']='test' - os.environ['SERVICE_NAME']='mvp-dcaegen2-heartbeat-static' - os.environ['CONSUL_HOST']='localhost' - os.environ['HOSTNAME']='mvp-dcaegen2-heartbeat-static' + os.environ["pytest"] = "test" + os.environ["SERVICE_NAME"] = "mvp-dcaegen2-heartbeat-static" + os.environ["CONSUL_HOST"] = "localhost" + os.environ["HOSTNAME"] = "mvp-dcaegen2-heartbeat-static" try: db.main() @@ -126,69 +127,125 @@ def verify_misshtbtdmain(): except Exception as e: result = False print(result) - os.unsetenv('pytest') - os.unsetenv('SERVICE_NAME') - os.unsetenv('CONSUL_HOST') - os.unsetenv('HOSTNAME') + os.unsetenv("pytest") + os.unsetenv("SERVICE_NAME") + os.unsetenv("CONSUL_HOST") + os.unsetenv("HOSTNAME") return result + def verify_dbmonitoring(): - os.environ['pytest']='test' - os.environ['SERVICE_NAME']='mvp-dcaegen2-heartbeat-static' - os.environ['CONSUL_HOST']='localhost' - os.environ['HOSTNAME']='mvp-dcaegen2-heartbeat-static' + os.environ["pytest"] = "test" + os.environ["SERVICE_NAME"] = "mvp-dcaegen2-heartbeat-static" + os.environ["CONSUL_HOST"] = "localhost" + os.environ["HOSTNAME"] = "mvp-dcaegen2-heartbeat-static" try: jsfile = db.fetch_json_file() ip_address, port_num, user_name, password, db_name, cbs_polling_required, cbs_polling_interval = hb_properties() - hbc_pid, hbc_state, hbc_srcName, hbc_time = db.read_hb_common(user_name,password,ip_address,port_num,db_name) - dbmon.db_monitoring(hbc_pid,jsfile,user_name,password,ip_address,port_num,db_name) + hbc_pid, hbc_state, hbc_srcName, hbc_time = db.read_hb_common( + user_name, password, ip_address, port_num, db_name + ) + dbmon.db_monitoring(hbc_pid, jsfile, user_name, password, ip_address, port_num, db_name) result = True except Exception as e: print("Message process error - %s" % e) result = False print(result) - os.unsetenv('pytest') - os.unsetenv('SERVICE_NAME') - os.unsetenv('CONSUL_HOST') - os.unsetenv('HOSTNAME') + os.unsetenv("pytest") + os.unsetenv("SERVICE_NAME") + os.unsetenv("CONSUL_HOST") + os.unsetenv("HOSTNAME") return result + def verify_dbmon_startup(): try: - p = subprocess.Popen(['./miss_htbt_service/db_monitoring.py'], stdout=subprocess.PIPE,shell=True) + p = subprocess.Popen(["./miss_htbt_service/db_monitoring.py"], stdout=subprocess.PIPE, shell=True) time.sleep(1) except Exception as e: return None return True + def verify_sendControlLoop_VNF_ONSET(): try: pol_url = "http://10.12.5.252:3904/events/unauthenticated.DCAE_CL_OUTPUT/" - _CL_return = dbmon.sendControlLoopEvent("ONSET", pol_url, "1.0", "vFireWall", "pscope", "VNF", "srcname1", 1541234567, "SampleCLName", "1.0", "genVnfName") + _CL_return = dbmon.sendControlLoopEvent( + "ONSET", + pol_url, + "1.0", + "vFireWall", + "pscope", + "VNF", + "srcname1", + 1541234567, + "SampleCLName", + "1.0", + "genVnfName", + ) except Exception as e: return None return _CL_return + def verify_sendControlLoop_VM_ONSET(): try: pol_url = "http://10.12.5.252:3904/events/unauthenticated.DCAE_CL_OUTPUT/" - _CL_return = dbmon.sendControlLoopEvent("ONSET", pol_url, "1.0", "vFireWall", "pscope", "VM", "srcname1", 1541234567, "SampleCLName", "1.0", "genVnfName") + _CL_return = dbmon.sendControlLoopEvent( + "ONSET", + pol_url, + "1.0", + "vFireWall", + "pscope", + "VM", + "srcname1", + 1541234567, + "SampleCLName", + "1.0", + "genVnfName", + ) except Exception as e: return None return _CL_return + def verify_sendControlLoop_VNF_ABATED(): try: pol_url = "http://10.12.5.252:3904/events/unauthenticated.DCAE_CL_OUTPUT/" - _CL_return = dbmon.sendControlLoopEvent("ABATED", pol_url, "1.0", "vFireWall", "pscope", "VNF", "srcname1", 1541234567, "SampleCLName", "1.0", "genVnfName") + _CL_return = dbmon.sendControlLoopEvent( + "ABATED", + pol_url, + "1.0", + "vFireWall", + "pscope", + "VNF", + "srcname1", + 1541234567, + "SampleCLName", + "1.0", + "genVnfName", + ) except Exception as e: return None return _CL_return + def verify_sendControlLoop_VM_ABATED(): try: pol_url = "http://10.12.5.252:3904/events/unauthenticated.DCAE_CL_OUTPUT/" - _CL_return = dbmon.sendControlLoopEvent("ABATED", pol_url, "1.0", "vFireWall", "pscope", "VM", "srcname1", 1541234567, "SampleCLName", "1.0", "genVnfName") + _CL_return = dbmon.sendControlLoopEvent( + "ABATED", + pol_url, + "1.0", + "vFireWall", + "pscope", + "VM", + "srcname1", + 1541234567, + "SampleCLName", + "1.0", + "genVnfName", + ) except Exception as e: return None return _CL_return -- cgit 1.2.3-korg