13336313523

Committed 14 Feb 2025 07:45PM UTC coverage: 55.747% (+0.4%) from 55.374%

Build # 13336313523

Build Type

Pull #416

github

Committed by

web-flow

Commit Message

Merge ddbed1d43 into 5c5ba2915

Pull Request Pull Request #416: Change strings to consts

Run Details

37 of 48 new or added lines in 8 files covered. (77.08%)

4 existing lines in 2 files now uncovered.

1135 of 2036 relevant lines covered (55.75%)

1.11 hits per line

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

88.57

/sdx_controller/utils/db_utils.py

import logging
import os
from urllib.parse import urlparse
from sdx_controller.utils.constants import MongoCollections

import pymongo


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 key, collection in MongoCollections.__dict__.items():
            if not key.startswith("__") and collection not in self.sdxdb.list_collection_names():
                self.sdxdb.create_collection(collection)

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

atlanticwave-sdx / sdx-controller / 13336313523

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