python
14 hours, 48 minutes ago
import unittest
from unittest.mock import patch, MagicMock
from mongoservice import MongoService
from mongomock import MongoClient as MockMongoClient
class TestMongoService(unittest.TestCase):
def setUp(self):
self.config = {
"mongo-db": {
"EVENTHUB": {
"MONGO_URL": "mock://localhost:27017",
"MONGO_DB": "test_db"
}
}
}
self.collection_name = "test_collection"
self.sample_data = {"_id": 1, "name": "test"}
@patch("mongomock.MongoClient", MockMongoClient)
def test_init_regular_mode(self):
service = MongoService(self.config, mode="regular")
self.assertEqual(service.db.name, "test_db")
self.assertEqual(service.db.client.address, ("localhost", 27017))
@patch("mongomock.MongoClient", MockMongoClient)
def test_init_mock_mode(self):
service = MongoService(self.config, mode="mock")
self.assertIsInstance(service.client, MockMongoClient)
@patch("mongomock.MongoClient", MockMongoClient)
def test_insert_all_query(self):
service = MongoService(self.config, mode="mock")
data = [{"_id": i, "value": i} for i in range(5)]
service.insert_all_query(self.collection_name, data)
collection = service.client["test_db"][self.collection_name]
self.assertEqual(collection.count_documents({}), 5)
@patch("mongomock.MongoClient", MockMongoClient)
def test_insert_one_query(self):
service = MongoService(self.config, mode="mock")
service.insert_one_query(self.collection_name, self.sample_data)
collection = service.client["test_db"][self.collection_name]
self.assertEqual(collection.count_documents({}), 1)
self.assertEqual(collection.find_one({"_id": 1})["name"], "test")
@patch("mongomock.MongoClient", MockMongoClient)
def test_update_one_query(self):
service = MongoService(self.config, mode="mock")
collection = service.client["test_db"][self.collection_name]
collection.insert_one(self.sample_data)
service.update_one_query(self.collection_name, {"_id": 1}, {"$set": {"name": "updated"}})
updated_doc = collection.find_one({"_id": 1})
self.assertEqual(updated_doc["name"], "updated")
@patch("mongomock.MongoClient", MockMongoClient)
def test_update_many_query(self):
service = MongoService(self.config, mode="mock")
data = [{"_id": i, "value": i} for i in range(5)]
collection = service.client["test_db"][self.collection_name]
collection.insert_many(data)
service.update_many_query(self.collection_name, {"value": {"$gte": 2}}, {"$set": {"status": "updated"}})
updated_count = collection.count_documents({"status": "updated"})
self.assertEqual(updated_count, 3)
@patch("mongomock.MongoClient", MockMongoClient)
def test_upsert_update_one(self):
service = MongoService(self.config, mode="mock")
# Record doesn't exist initially
service.update_one_query(self.collection_name, {"_id": 1}, {"$set": {"name": "upserted"}}, upsert=True)
collection = service.client["test_db"][self.collection_name]
self.assertEqual(collection.count_documents({}), 1)
self.assertEqual(collection.find_one({"_id": 1})["name"], "upserted")
@patch("mongomock.MongoClient", MockMongoClient)
def test_upsert_update_many(self):
service = MongoService(self.config, mode="mock")
# Record doesn't exist initially
service.update_many_query(self.collection_name, {"_id": {"$gte": 1}}, {"$set": {"name": "upserted"}}, upsert=True)
collection = service.client["test_db"][self.collection_name]
self.assertGreaterEqual(collection.count_documents({}), 1)
if __name__ == "__main__":
unittest.main()
0 Comments
Please Login to Comment Here