Coverage for tests/core_tests/test_mongodb_manager.py: 100%

1"""Test the MongoDB Manager class.""" 

2 

3import os 

4import sys 

5from dotenv import load_dotenv 

6import pytest 

7from pymongo.errors import ( 

8 ConfigurationError, 

9 OperationFailure, 

10 ServerSelectionTimeoutError, 

11 DuplicateKeyError, 

12) 

13 

14 

15# flake8: noqa: F811 

16 

17# Add the root directory to the Python path 

18sys.path.append( 

19 os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) 

20) 

21from core import shared 

22from core.database_mongo_manager import DatabaseMongoManager 

23 

24os.environ["IS_TEST"] = "True" 

25 

26load_dotenv() 

27 

28 

29@pytest.fixture() 

30def database(): 

31 """Fixture to create a test database.""" 

32 DATABASE = DatabaseMongoManager( 

33 shared.getenv("MONGO_URI"), 

34 shared.getenv("MONGO_DB_TEST", "cs3528_testing"), 

35 ) 

36 DATABASE.delete_all("test_collection") # Cleanup after tests 

37 DATABASE.delete_collection("test_collection") 

38 yield DATABASE 

39 DATABASE.delete_all("test_collection") # Cleanup after tests 

40 DATABASE.delete_collection("test_collection") 

41 DATABASE.connection.close() 

42 

43 

44def test_insert_and_get_by_id(database): 

45 test_data = {"_id": "test1", "name": "Test Entry"} 

46 database.insert("test_collection", test_data) 

47 result = database.get_one_by_id("test_collection", "test1") 

48 assert result is not None 

49 assert result["name"] == "Test Entry" 

50 database.delete_by_id("test_collection", "test1") 

51 

52 

53def test_update_one_by_id(database): 

54 test_data = {"_id": "test2", "name": "Initial Name"} 

55 database.insert("test_collection", test_data) 

56 database.update_one_by_id("test_collection", "test2", {"name": "Updated Name"}) 

57 updated = database.get_one_by_id("test_collection", "test2") 

58 assert updated["name"] == "Updated Name" 

59 database.delete_by_id("test_collection", "test2") 

60 

61 

62def test_delete_by_id(database): 

63 test_data = {"_id": "test3", "name": "To be deleted"} 

64 database.insert("test_collection", test_data) 

65 database.delete_by_id("test_collection", "test3") 

66 deleted = database.get_one_by_id("test_collection", "test3") 

67 assert deleted is None 

68 

69 

70def test_get_all(database): 

71 test_data1 = {"_id": "test4", "name": "Entry 1"} 

72 test_data2 = {"_id": "test5", "name": "Entry 2"} 

73 database.insert("test_collection", test_data1) 

74 database.insert("test_collection", test_data2) 

75 results = database.get_all("test_collection") 

76 assert len(results) >= 2 

77 database.delete_all("test_collection") 

78 

79 

80def test_increment(database): 

81 test_data = {"_id": "test6", "count": 0} 

82 database.insert("test_collection", test_data) 

83 database.increment("test_collection", "test6", "count", 5) 

84 updated = database.get_one_by_id("test_collection", "test6") 

85 assert updated["count"] == 5 

86 database.delete_by_id("test_collection", "test6") 

87 

88 

89def test_get_by_email(database): 

90 test_data = {"_id": "test7", "email": "test@example.com"} 

91 database.insert("test_collection", test_data) 

92 result = database.get_by_email("test_collection", "test@example.com") 

93 assert result is not None 

94 assert result["email"] == "test@example.com" 

95 database.delete_by_id("test_collection", "test7") 

96 

97 

98def test_connection_success(): 

99 db = DatabaseMongoManager( 

100 shared.getenv("MONGO_URI"), 

101 shared.getenv("MONGO_DB_TEST", "cs3528_testing"), 

102 ) 

103 assert db.connection is not None 

104 db.close_connection() 

105 

106 

107def test_insert_duplicate_id(database): 

108 test_data = {"_id": "duplicate_test", "name": "First Entry"} 

109 database.insert("test_collection", test_data) 

110 

111 with pytest.raises(DuplicateKeyError): 

112 database.insert("test_collection", test_data) 

113 

114 database.delete_by_id("test_collection", "duplicate_test") 

115 

116 

117def test_update_non_existent_record(database): 

118 result = database.update_one_by_id( 

119 "test_collection", "non_existent", {"name": "Updated"} 

120 ) 

121 assert result.matched_count == 0 

122 

123 

124def test_delete_non_existent_record(database): 

125 result = database.delete_by_id("test_collection", "non_existent") 

126 assert result.deleted_count == 0 

127 

128 

129def test_get_all_empty(database): 

130 results = database.get_all("empty_collection") 

131 assert results == [] 

132 

133 

134def test_increment_non_existent_field(database): 

135 test_data = {"_id": "increment_test"} 

136 database.insert("test_collection", test_data) 

137 

138 database.increment("test_collection", "increment_test", "counter", 10) 

139 updated = database.get_one_by_id("test_collection", "increment_test") 

140 

141 assert updated["counter"] == 10 

142 

143 database.delete_by_id("test_collection", "increment_test") 

144 

145 

146def test_case_sensitivity_strict_search(database): 

147 test_data = {"_id": "case_test", "email": "TEST@EXAMPLE.COM"} 

148 database.insert("test_collection", test_data) 

149 

150 result = database.get_one_by_field_strict( 

151 "test_collection", "email", "test@example.com" 

152 ) 

153 

154 assert result is not None # Should match despite case difference 

155 database.delete_by_id("test_collection", "case_test") 

156 

157 

158def test_large_data_insertion(database): 

159 large_text = "A" * 16777216 

160 test_data = {"_id": "large_data_test", "text": large_text} 

161 

162 with pytest.raises(Exception): 

163 database.insert("test_collection", test_data) 

164 

165 

166def test_text_search_on_non_indexed_field(database): 

167 test_data = {"_id": "text_test", "description": "MongoDB text search test"} 

168 database.insert("test_collection", test_data) 

169 

170 with pytest.raises(Exception): 

171 database.get_all_by_text_search("test_collection", "MongoDB") 

172 

173 database.delete_by_id("test_collection", "text_test") 

174 

175 

176def test_get_all_by_field(database): 

177 """Test fetching all entries by a specific field value.""" 

178 database.delete_all("test_collection") 

179 database.delete_collection("test_collection") 

180 test_data1 = {"_id": "test8", "category": "A", "name": "Entry 1"} 

181 test_data2 = {"_id": "test9", "category": "B", "name": "Entry 2"} 

182 test_data3 = {"_id": "test10", "category": "A", "name": "Entry 3"} 

183 database.insert("test_collection", test_data1) 

184 database.insert("test_collection", test_data2) 

185 database.insert("test_collection", test_data3) 

186 

187 results = database.get_all_by_field("test_collection", "category", "A") 

188 assert len(results) == 2 

189 assert results[0]["category"] == "A" 

190 assert results[0]["name"] == "Entry 1" 

191 assert results[1]["category"] == "A" 

192 assert results[1]["name"] == "Entry 3" 

193 

194 database.delete_by_id("test_collection", "test8") 

195 database.delete_by_id("test_collection", "test9") 

196 database.delete_by_id("test_collection", "test10") 

197 

198 

199def test_is_table(database): 

200 assert database.is_table("test_collection") is False 

201 database.add_table("test_collection") 

202 assert database.is_table("test_collection") is True 

203 assert database.is_table("non_existent_collection") is False 

204 

205 

206def test_get_all_by_two_fields(database): 

207 test_data1 = {"_id": "test10", "category": "A", "status": "active"} 

208 test_data2 = {"_id": "test11", "category": "A", "status": "inactive"} 

209 test_data3 = {"_id": "test12", "category": "B", "status": "active"} 

210 test_data4 = {"_id": "test13", "category": "B", "status": "inactive"} 

211 test_data5 = {"_id": "test14", "category": "A", "status": "active"} 

212 database.insert("test_collection", test_data1) 

213 database.insert("test_collection", test_data2) 

214 database.insert("test_collection", test_data3) 

215 database.insert("test_collection", test_data4) 

216 database.insert("test_collection", test_data5) 

217 

218 results = database.get_all_by_two_fields( 

219 "test_collection", "category", "A", "status", "active" 

220 ) 

221 assert len(results) == 2 

222 assert results[0]["status"] == "active" 

223 assert results[1]["status"] == "active" 

224 assert results[0]["_id"] == "test10" 

225 assert results[1]["_id"] == "test14" 

226 

227 database.delete_by_id("test_collection", "test10") 

228 database.delete_by_id("test_collection", "test11") 

229 database.delete_by_id("test_collection", "test12") 

230 database.delete_by_id("test_collection", "test13") 

231 database.delete_by_id("test_collection", "test14") 

232 

233 

234def test_update_by_field(database): 

235 test_data = {"_id": "test15", "category": "A", "name": "Old Name"} 

236 database.insert("test_collection", test_data) 

237 database.update_by_field("test_collection", "category", "A", {"name": "New Name"}) 

238 

239 updated = database.get_one_by_id("test_collection", "test15") 

240 assert updated["name"] == "New Name" 

241 

242 database.delete_by_id("test_collection", "test15") 

243 

244 

245def test_create_index(database): 

246 result = database.create_index("test_collection", "name") 

247 assert isinstance(result, str) 

248 

249 

250def test_delete_one_by_field(database): 

251 test_data = {"_id": "test20", "category": "A", "name": "Entry to delete"} 

252 database.insert("test_collection", test_data) 

253 

254 database.delete_one_by_field("test_collection", "category", "A") 

255 deleted = database.get_one_by_id("test_collection", "test20") 

256 assert deleted is None 

257 

258 

259def test_get_all_by_in_list(database): 

260 test_data1 = {"_id": "test21", "category": "A", "name": "Entry 1"} 

261 test_data2 = {"_id": "test22", "category": "B", "name": "Entry 2"} 

262 test_data3 = {"_id": "test23", "category": "C", "name": "Entry 3"} 

263 test_data4 = {"_id": "test24", "category": "A", "name": "Entry 4"} 

264 

265 database.insert("test_collection", test_data1) 

266 database.insert("test_collection", test_data2) 

267 database.insert("test_collection", test_data3) 

268 database.insert("test_collection", test_data4) 

269 

270 # Fetch entries with category "A" or "C" 

271 results = database.get_all_by_in_list("test_collection", "category", ["A", "C"]) 

272 

273 assert len(results) == 3 

274 assert any(item["_id"] == "test21" for item in results) 

275 assert any(item["_id"] == "test23" for item in results) 

276 assert any(item["_id"] == "test24" for item in results) 

277 

278 database.delete_by_id("test_collection", "test21") 

279 database.delete_by_id("test_collection", "test22") 

280 database.delete_by_id("test_collection", "test23") 

281 database.delete_by_id("test_collection", "test24") 

282 

283 

284def test_invalid_connection_string(monkeypatch): 

285 """Test that an invalid connection string raises ConfigurationError.""" 

286 

287 def mock_mongo_client(*args, **kwargs): 

288 raise ConfigurationError("Invalid MongoDB connection string") 

289 

290 monkeypatch.setattr("pymongo.MongoClient", mock_mongo_client) 

291 

292 with pytest.raises(ConfigurationError, match="Invalid MongoDB connection string"): 

293 DatabaseMongoManager("invalid_connection_string", "cs3528_testing") 

294 

295 

296def test_operation_failure(monkeypatch): 

297 """Test that an operation failure raises SystemExit.""" 

298 

299 class MockClient: 

300 class MockAdmin: 

301 def command(self, cmd): 

302 raise OperationFailure("Authentication failed") 

303 

304 def __init__(self, *args, **kwargs): 

305 self.admin = self.MockAdmin() 

306 

307 monkeypatch.setattr( 

308 "pymongo.MongoClient", MockClient 

309 ) # Correctly patch inside pymongo 

310 

311 with pytest.raises(SystemExit) as exc_info: 

312 DatabaseMongoManager("invalid_connection_string", "cs3528_testing") 

313 

314 assert exc_info.value.code == 1 

315 

316 

317def test_server_selection_timeout(monkeypatch): 

318 """Test that a server selection timeout raises SystemExit.""" 

319 

320 class MockClient: 

321 class MockAdmin: 

322 def command(self, cmd): 

323 raise ServerSelectionTimeoutError("Server selection timed out") 

324 

325 def __init__(self, *args, **kwargs): 

326 self.admin = self.MockAdmin() 

327 

328 monkeypatch.setattr( 

329 "pymongo.MongoClient", MockClient 

330 ) # Correctly mock inside pymongo 

331 

332 with pytest.raises(SystemExit) as exc_info: 

333 DatabaseMongoManager("invalid_connection_string", "cs3528_testing") 

334 

335 assert exc_info.value.code == 1 

336 

337 

338def test_default_database_if_empty(): 

339 """Test that an empty database name defaults to 'cs3528_testing'.""" 

340 db_manager = DatabaseMongoManager(shared.getenv("MONGO_URI"), "") 

341 assert db_manager.database.name == "cs3528_testing"