9518334494

Committed 14 Jun 2024 03:09PM UTC coverage: 52.459% (+0.1%) from 52.322%

Build # 9518334494

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #292 from atlanticwave-sdx/291.use-tox-expose-directive

Use tox-docker `expose` directive

Run Details

301 of 549 branches covered (54.83%)

Branch coverage included in aggregate %.

10 of 12 new or added lines in 2 files covered. (83.33%)

755 of 1464 relevant lines covered (51.57%)

2.06 hits per line

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

92.0

/sdx_controller/utils/db_utils.py

import json
import logging
import os

import pymongo

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


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")

        if mongo_host is None:
            raise Exception("MONGO_HOST environment variable is not set")

        if mongo_port is None:
            raise Exception("MONGO_PORT environment variable is not set")

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

        mongo_connstring = (
            f"mongodb://{mongo_user}:{mongo_pass}@{mongo_host}:{mongo_port}/"
        )

        # Log DB URI, without a password.
        self.logger.info(
            f"[DB] Using mongodb://{mongo_user}@{mongo_host}:{mongo_port}/"
        )

        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

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

atlanticwave-sdx / sdx-controller / 9518334494

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