8181484903

Committed 07 Mar 2024 01:30AM UTC coverage: 50.85% (+12.2%) from 38.63%

Build # 8181484903

Build Type

Pull #225

github

Committed by

web-flow

Commit Message

Merge c53829de8 into 23d4f8c2a

Pull Request Pull Request #225: Handle connection deletions

Run Details

288 of 535 branches covered (53.83%)

Branch coverage included in aggregate %.

22 of 28 new or added lines in 6 files covered. (78.57%)

34 existing lines in 5 files now uncovered.

699 of 1406 relevant lines covered (49.72%)

1.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

22.22

/sdx_controller/handlers/lc_message_handler.py

import json
import logging

from sdx_controller.handlers.connection_handler import ConnectionHandler
from sdx_controller.utils.parse_helper import ParseHelper

logger = logging.getLogger(__name__)


class LcMessageHandler:
    def __init__(self, db_instance, te_manager):
        self.db_instance = db_instance
        self.te_manager = te_manager
        self.parse_helper = ParseHelper()
        self.connection_handler = ConnectionHandler(db_instance)

    def process_lc_json_msg(
        self,
        msg,
        latest_topo,
        domain_list,
        num_domain_topos,
    ):
        logger.info("MQ received message:" + str(msg))
        msg_json = json.loads(msg)
        msg_id = msg_json["id"]
        msg_version = msg_json["version"]

        lc_queue_name = msg_json["lc_queue_name"]
        logger.debug("Processing LC message: lc_queue_name:")
        logger.debug(lc_queue_name)

        domain_name = self.parse_helper.find_between(msg_id, "topology:", ".net")
        msg_json["domain_name"] = domain_name

        db_msg_id = str(msg_id) + "-" + str(msg_version)
        # add message to db
        self.db_instance.add_key_value_pair_to_db(db_msg_id, msg)
        logger.info("Save to database complete.")
        logger.info("message ID:" + str(db_msg_id))

        # Update existing topology
        if domain_name in domain_list:
            logger.info("Updating topo")
            logger.debug(msg_json)
            self.te_manager.update_topology(msg_json)
            if "link_failure" in msg_json:
                logger.info("Processing link failure.")
                self.connection_handler.handle_link_failure(msg_json)
        # Add new topology
        else:
            domain_list.append(domain_name)
            self.db_instance.add_key_value_pair_to_db("domain_list", domain_list)

            logger.info("Adding topo")
            self.te_manager.add_topology(msg_json)

            if self.db_instance.read_from_db("num_domain_topos") is None:
                num_domain_topos = 1
                self.db_instance.add_key_value_pair_to_db(
                    "num_domain_topos", num_domain_topos
                )
            else:
                num_domain_topos = len(domain_list)
                num_domain_topos = int(num_domain_topos) + 1
                self.db_instance.add_key_value_pair_to_db(
                    "num_domain_topos", num_domain_topos
                )

        logger.info("Adding topo to db.")
        db_key = "LC-" + str(num_domain_topos)
        self.db_instance.add_key_value_pair_to_db(db_key, json.dumps(msg_json))

        # TODO: use TEManager API directly; but TEManager does not
        # expose a `get_topology()` method yet.
        latest_topo = json.dumps(
            self.te_manager.topology_manager.get_topology().to_dict()
        )
        # use 'latest_topo' as PK to save latest topo to db
        self.db_instance.add_key_value_pair_to_db("latest_topo", latest_topo)
        logger.info("Save to database complete.")

1	import json	4✔
2	import logging	4✔
3
4	from sdx_controller.handlers.connection_handler import ConnectionHandler	4✔
5	from sdx_controller.utils.parse_helper import ParseHelper	4✔
6
7	logger = logging.getLogger(__name__)	4✔
8
9
10	class LcMessageHandler:	4✔
11	def __init__(self, db_instance, te_manager):	4✔
12	self.db_instance = db_instance	4✔
13	self.te_manager = te_manager	4✔
14	self.parse_helper = ParseHelper()	4✔
15	self.connection_handler = ConnectionHandler(db_instance)	4✔
16
17	def process_lc_json_msg(	4✔
18	self,
19	msg,
20	latest_topo,
21	domain_list,
22	num_domain_topos,
23	):
24	logger.info("MQ received message:" + str(msg))	×
25	msg_json = json.loads(msg)	×
26	msg_id = msg_json["id"]	×
27	msg_version = msg_json["version"]	×
28
29	lc_queue_name = msg_json["lc_queue_name"]	×
30	logger.debug("Processing LC message: lc_queue_name:")	×
31	logger.debug(lc_queue_name)	×
32
33	domain_name = self.parse_helper.find_between(msg_id, "topology:", ".net")	×
34	msg_json["domain_name"] = domain_name	×
35
36	db_msg_id = str(msg_id) + "-" + str(msg_version)	×
37	# add message to db
38	self.db_instance.add_key_value_pair_to_db(db_msg_id, msg)	×
39	logger.info("Save to database complete.")	×
40	logger.info("message ID:" + str(db_msg_id))	×
41
42	# Update existing topology
43	if domain_name in domain_list:	×
44	logger.info("Updating topo")	×
45	logger.debug(msg_json)	×
NEW 46	self.te_manager.update_topology(msg_json)	×
47	if "link_failure" in msg_json:	×
48	logger.info("Processing link failure.")	×
49	self.connection_handler.handle_link_failure(msg_json)	×
50	# Add new topology
51	else:
52	domain_list.append(domain_name)	×
53	self.db_instance.add_key_value_pair_to_db("domain_list", domain_list)	×
54
55	logger.info("Adding topo")	×
NEW 56	self.te_manager.add_topology(msg_json)	×
57
58	if self.db_instance.read_from_db("num_domain_topos") is None:	×
59	num_domain_topos = 1	×
60	self.db_instance.add_key_value_pair_to_db(	×
61	"num_domain_topos", num_domain_topos
62	)
63	else:
64	num_domain_topos = len(domain_list)	×
65	num_domain_topos = int(num_domain_topos) + 1	×
66	self.db_instance.add_key_value_pair_to_db(	×
67	"num_domain_topos", num_domain_topos
68	)
69
70	logger.info("Adding topo to db.")	×
71	db_key = "LC-" + str(num_domain_topos)	×
72	self.db_instance.add_key_value_pair_to_db(db_key, json.dumps(msg_json))	×
73
74	# TODO: use TEManager API directly; but TEManager does not
75	# expose a `get_topology()` method yet.
NEW 76	latest_topo = json.dumps(	×
77	self.te_manager.topology_manager.get_topology().to_dict()
78	)
79	# use 'latest_topo' as PK to save latest topo to db
UNCOV 80	self.db_instance.add_key_value_pair_to_db("latest_topo", latest_topo)	×
UNCOV 81	logger.info("Save to database complete.")	×

atlanticwave-sdx / sdx-controller / 8181484903

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous