10727285533

Committed 05 Sep 2024 07:58PM UTC coverage: 57.725% (+0.08%) from 57.645%

Build # 10727285533

Build Type

Pull #326

github

Committed by

web-flow

Commit Message

Merge 86e5acb98 into def24445a

Pull Request Pull Request #326: Reuse service id for patch

Run Details

467 of 745 branches covered (62.68%)

Branch coverage included in aggregate %.

18 of 26 new or added lines in 4 files covered. (69.23%)

22 existing lines in 2 files now uncovered.

1050 of 1883 relevant lines covered (55.76%)

2.23 hits per line

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

85.23

/sdx_controller/utils/db_utils.py

import logging
import os
from urllib.parse import urlparse

import pymongo

COLLECTION_NAMES = ["topologies", "connections", "breakdowns", "domains", "links", "historical_connections"]

pymongo_logger = logging.getLogger("pymongo")
pymongo_logger.setLevel(logging.INFO)


def obfuscate_password_in_uri(uri: str) -> str:
    """
    Replace password field in URIs with a `*`, for logging.
    """
    parts = urlparse(uri)
    if parts.password:
        return uri.replace(parts.password, "*")
    else:
        return uri


class DbUtils(object):
    def __init__(self):
        self.db_name = os.environ.get("DB_NAME")
        if self.db_name is None:
            raise Exception("DB_NAME environment variable is not set")

        # self.config_table_name = os.environ.get("DB_CONFIG_TABLE_NAME")
        # if self.config_table_name is None:
        #     raise Exception("DB_CONFIG_TABLE_NAME environ variable is not set")

        mongo_user = os.getenv("MONGO_USER") or "guest"
        mongo_pass = os.getenv("MONGO_PASS") or "guest"
        mongo_host = os.getenv("MONGO_HOST")
        mongo_port = os.getenv("MONGO_PORT") or 27017

        if mongo_host is None:
            mongo_connstring = os.getenv("MONGODB_CONNSTRING")
            if mongo_connstring is None:
                raise Exception("Neither MONGO_HOST nor MONGODB_CONNSTRING is set")
        else:
            mongo_connstring = (
                f"mongodb://{mongo_user}:{mongo_pass}@{mongo_host}:{mongo_port}/"
            )

        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.DEBUG)

        # Log DB URI, without a password.
        self.logger.info(f"[DB] Using {obfuscate_password_in_uri(mongo_connstring)}")

        self.mongo_client = pymongo.MongoClient(mongo_connstring)

    def initialize_db(self):
        self.logger.debug(f"Trying to load {self.db_name} from DB")

        if self.db_name not in self.mongo_client.list_database_names():
            self.logger.debug(f"No existing {self.db_name} from DB, creating table")
            self.sdxdb = self.mongo_client[self.db_name]
            self.logger.debug(f"DB {self.db_name} initialized")

        self.sdxdb = self.mongo_client[self.db_name]
        # config_col = self.sdxdb[self.config_table_name]
        for name in COLLECTION_NAMES:
            if name not in self.sdxdb.list_collection_names():
                self.sdxdb.create_collection(name)

        self.logger.debug(f"DB {self.db_name} initialized")

    def add_key_value_pair_to_db(self, collection, key, value):
        key = str(key)
        obj = self.read_from_db(collection, key)
        if obj is None:
            return self.sdxdb[collection].insert_one({key: value})

        query = {"_id": obj["_id"]}
        result = self.sdxdb[collection].replace_one(query, {key: value})
        return result

    def read_from_db(self, collection, key):
        key = str(key)
        return self.sdxdb[collection].find_one(
            {key: {"$exists": 1}, "deleted": {"$ne": True}}
        )

    def get_all_entries_in_collection(self, collection):
        db_collection = self.sdxdb[collection]
        # MongoDB has an ObjectId for each item, so need to exclude the ObjectIds
        all_entries = db_collection.find({"deleted": {"$ne": True}}, {"_id": 0})
        return all_entries

    def mark_deleted(self, collection, key):
        db_collection = self.sdxdb[collection]
        key = str(key)
        item_to_delete = self.read_from_db(collection, key)
        if item_to_delete is None:
            return False
        filter = {"_id": item_to_delete["_id"]}
        update = {"$set": {"deleted": True}}
        db_collection.update_one(filter, update)
        return True

    def delete_one_entry(self, collection, key):
        key = str(key)
        db_collection = self.sdxdb[collection]
        db_collection.delete_one({key: {"$exists": True}})

1	import logging	4✔
2	import os	4✔
3	from urllib.parse import urlparse	4✔
4
5	import pymongo	4✔
6
7	COLLECTION_NAMES = ["topologies", "connections", "breakdowns", "domains", "links", "historical_connections"]	4✔
8
9	pymongo_logger = logging.getLogger("pymongo")	4✔
10	pymongo_logger.setLevel(logging.INFO)	4✔
11
12
13	def obfuscate_password_in_uri(uri: str) -> str:	4✔
14	"""
15	Replace password field in URIs with a `*`, for logging.
16	"""
17	parts = urlparse(uri)	4✔
18	if parts.password:	4✔
19	return uri.replace(parts.password, "*")	4✔
20	else:
UNCOV 21	return uri	×
22
23
24	class DbUtils(object):	4✔
25	def __init__(self):	4✔
26	self.db_name = os.environ.get("DB_NAME")	4✔
27	if self.db_name is None:	4✔
28	raise Exception("DB_NAME environment variable is not set")	4✔
29
30	# self.config_table_name = os.environ.get("DB_CONFIG_TABLE_NAME")
31	# if self.config_table_name is None:
32	# raise Exception("DB_CONFIG_TABLE_NAME environ variable is not set")
33
34	mongo_user = os.getenv("MONGO_USER") or "guest"	4✔
35	mongo_pass = os.getenv("MONGO_PASS") or "guest"	4✔
36	mongo_host = os.getenv("MONGO_HOST")	4✔
37	mongo_port = os.getenv("MONGO_PORT") or 27017	4✔
38
39	if mongo_host is None:	4✔
UNCOV 40	mongo_connstring = os.getenv("MONGODB_CONNSTRING")	×
UNCOV 41	if mongo_connstring is None:	×
UNCOV 42	raise Exception("Neither MONGO_HOST nor MONGODB_CONNSTRING is set")	×
43	else:
44	mongo_connstring = (	4✔
45	f"mongodb://{mongo_user}:{mongo_pass}@{mongo_host}:{mongo_port}/"
46	)
47
48	self.logger = logging.getLogger(__name__)	4!
49	self.logger.setLevel(logging.DEBUG)	4✔
50
51	# Log DB URI, without a password.
52	self.logger.info(f"[DB] Using {obfuscate_password_in_uri(mongo_connstring)}")	4✔
53
54	self.mongo_client = pymongo.MongoClient(mongo_connstring)	4✔
55
56	def initialize_db(self):	4✔
57	self.logger.debug(f"Trying to load {self.db_name} from DB")	4✔
58
59	if self.db_name not in self.mongo_client.list_database_names():	4✔
60	self.logger.debug(f"No existing {self.db_name} from DB, creating table")	4✔
61	self.sdxdb = self.mongo_client[self.db_name]	4✔
62	self.logger.debug(f"DB {self.db_name} initialized")	4✔
63
64	self.sdxdb = self.mongo_client[self.db_name]	4✔
65	# config_col = self.sdxdb[self.config_table_name]
66	for name in COLLECTION_NAMES:	4✔
67	if name not in self.sdxdb.list_collection_names():	4✔
68	self.sdxdb.create_collection(name)	4✔
69
70	self.logger.debug(f"DB {self.db_name} initialized")	4✔
71
72	def add_key_value_pair_to_db(self, collection, key, value):	4✔
73	key = str(key)	4✔
74	obj = self.read_from_db(collection, key)	4✔
75	if obj is None:	4✔
76	return self.sdxdb[collection].insert_one({key: value})	4✔
77
78	query = {"_id": obj["_id"]}	4✔
79	result = self.sdxdb[collection].replace_one(query, {key: value})	4✔
80	return result	4✔
81
82	def read_from_db(self, collection, key):	4✔
83	key = str(key)	4✔
84	return self.sdxdb[collection].find_one(	4✔
85	{key: {"$exists": 1}, "deleted": {"$ne": True}}
86	)
87
88	def get_all_entries_in_collection(self, collection):	4✔
89	db_collection = self.sdxdb[collection]	4✔
90	# MongoDB has an ObjectId for each item, so need to exclude the ObjectIds
91	all_entries = db_collection.find({"deleted": {"$ne": True}}, {"_id": 0})	4✔
92	return all_entries	4✔
93
94	def mark_deleted(self, collection, key):	4✔
95	db_collection = self.sdxdb[collection]	4✔
96	key = str(key)	4✔
97	item_to_delete = self.read_from_db(collection, key)	4✔
98	if item_to_delete is None:	4✔
99	return False	4✔
UNCOV 100	filter = {"_id": item_to_delete["_id"]}	×
UNCOV 101	update = {"$set": {"deleted": True}}	×
UNCOV 102	db_collection.update_one(filter, update)	×
UNCOV 103	return True	×
104
105	def delete_one_entry(self, collection, key):	4✔
106	key = str(key)	4✔
107	db_collection = self.sdxdb[collection]	4✔
108	db_collection.delete_one({key: {"$exists": True}})	4✔

atlanticwave-sdx / sdx-controller / 10727285533

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