11131304124

Committed 01 Oct 2024 06:30PM UTC coverage: 58.396% (+0.1%) from 58.299%

Build # 11131304124

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #328 from atlanticwave-sdx/add-topo-when-initialize-te-manager

Use topology from database when initialize te manager

Run Details

472 of 751 branches covered (62.85%)

Branch coverage included in aggregate %.

3 of 9 new or added lines in 3 files covered. (33.33%)

1 existing line in 1 file now uncovered.

1086 of 1917 relevant lines covered (56.65%)

2.26 hits per line

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

24.0

/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"]

        domain_name = self.parse_helper.find_domain_name(msg_id, ":")
        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("topologies", 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)
            failed_links = self.te_manager.get_failed_links()
            if failed_links:
                logger.info("Processing link failure.")
                self.connection_handler.handle_link_failure(
                    self.te_manager, failed_links
                )
        # Add new topology
        else:
            domain_list.append(domain_name)
            self.db_instance.add_key_value_pair_to_db(
                "domains", "domain_list", domain_list
            )

            logger.info("Adding topology to TE manager")
            self.te_manager.add_topology(msg_json)

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

        logger.info("Adding topology to db: " + domain_name)

        self.db_instance.add_key_value_pair_to_db(
            "topologies", domain_name, 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(
            "topologies", "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	domain_name = self.parse_helper.find_domain_name(msg_id, ":")	×
30	msg_json["domain_name"] = domain_name	×
31
32	db_msg_id = str(msg_id) + "-" + str(msg_version)	×
33	# add message to db
34	self.db_instance.add_key_value_pair_to_db("topologies", db_msg_id, msg)	×
35	logger.info("Save to database complete.")	×
36	logger.info("message ID:" + str(db_msg_id))	×
37
38	# Update existing topology
39	if domain_name in domain_list:	×
40	logger.info("Updating topo")	×
41	logger.debug(msg_json)	×
42	self.te_manager.update_topology(msg_json)	×
43	failed_links = self.te_manager.get_failed_links()	×
44	if failed_links:	×
45	logger.info("Processing link failure.")	×
46	self.connection_handler.handle_link_failure(	×
47	self.te_manager, failed_links
48	)
49	# Add new topology
50	else:
51	domain_list.append(domain_name)	×
52	self.db_instance.add_key_value_pair_to_db(	×
53	"domains", "domain_list", domain_list
54	)
55
NEW 56	logger.info("Adding topology to TE manager")	×
57	self.te_manager.add_topology(msg_json)	×
58
59	if self.db_instance.read_from_db("topologies", "num_domain_topos") is None:	×
60	num_domain_topos = 1	×
61	self.db_instance.add_key_value_pair_to_db(	×
62	"topologies", "num_domain_topos", num_domain_topos
63	)
64	else:
65	num_domain_topos = len(domain_list)	×
66	self.db_instance.add_key_value_pair_to_db(	×
67	"topologies", "num_domain_topos", num_domain_topos
68	)
69
NEW 70	logger.info("Adding topology to db: " + domain_name)	×
71
UNCOV 72	self.db_instance.add_key_value_pair_to_db(	×
73	"topologies", domain_name, json.dumps(msg_json)
74	)
75
76	# TODO: use TEManager API directly; but TEManager does not
77	# expose a `get_topology()` method yet.
78	latest_topo = json.dumps(	×
79	self.te_manager.topology_manager.get_topology().to_dict()
80	)
81	# use 'latest_topo' as PK to save latest topo to db
82	self.db_instance.add_key_value_pair_to_db(	×
83	"topologies", "latest_topo", latest_topo
84	)
85	logger.info("Save to database complete.")	×

atlanticwave-sdx / sdx-controller / 11131304124

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