Coverage for utils_aiosqlite/tests/unit/test_myaiosqlite.py: 26%

1"""Unit tests for MyAioSQLite class.""" 

2 

3import tempfile 

4from collections.abc import AsyncGenerator 

5from pathlib import Path 

6from unittest.mock import AsyncMock, patch 

7 

8import aiosqlite 

9import pandas as pd 

10import pytest 

11 

12from utils_aiosqlite import MyAioSQLite 

13 

14 

15class TestMyAioSQLite: 

16 """Test MyAioSQLite class functionality.""" 

17 

18 @pytest.fixture 

19 async def db_instance(self) -> AsyncGenerator[MyAioSQLite, None]: 

20 """Create a MyAioSQLite instance for testing.""" 

21 with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: 

22 db_path = Path(tmp.name) 

23 

24 instance = await MyAioSQLite.async_init( 

25 pth_db=db_path, timeout=10, auto_commit=False, readonly=False, prefetch=500 

26 ) 

27 yield instance 

28 

29 # Cleanup 

30 if db_path.exists(): 

31 db_path.unlink() 

32 

33 @pytest.mark.asyncio 

34 async def test_async_init(self): 

35 """Test MyAioSQLite initialization.""" 

36 with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: 

37 db_path = Path(tmp.name) 

38 

39 try: 

40 instance = await MyAioSQLite.async_init( 

41 pth_db=db_path, timeout=15, auto_commit=True, readonly=True, prefetch=2000 

42 ) 

43 

44 assert instance.pth_db == db_path 

45 assert instance.default_timeout == 15 

46 assert instance.auto_commit is True 

47 assert instance.readonly is True 

48 assert instance.prefetch == 2000 

49 finally: 

50 if db_path.exists(): 

51 db_path.unlink() 

52 

53 @pytest.mark.asyncio 

54 async def test_async_init_defaults(self): 

55 """Test MyAioSQLite initialization with default values.""" 

56 with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: 

57 db_path = Path(tmp.name) 

58 

59 try: 

60 instance = await MyAioSQLite.async_init(pth_db=db_path) 

61 

62 assert instance.pth_db == db_path 

63 assert instance.default_timeout == 5 

64 assert instance.auto_commit is False 

65 assert instance.readonly is False 

66 assert instance.prefetch == 1000 

67 finally: 

68 if db_path.exists(): 

69 db_path.unlink() 

70 

71 @pytest.mark.asyncio 

72 async def test_call_select_query(self, db_instance: MyAioSQLite): 

73 """Test __call__ method with SELECT queries.""" 

74 # Create a test table and insert data 

75 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

76 await db_instance.execute("INSERT INTO test_table VALUES (1, 'Alice'), (2, 'Bob')") 

77 

78 # Test SELECT query 

79 df = await db_instance("SELECT * FROM test_table ORDER BY id") 

80 

81 assert isinstance(df, pd.DataFrame) 

82 assert len(df) == 2 

83 assert list(df.columns) == ["id", "name"] 

84 assert df.iloc[0]["id"] == 1 

85 assert df.iloc[0]["name"] == "Alice" 

86 assert df.iloc[1]["id"] == 2 

87 assert df.iloc[1]["name"] == "Bob" 

88 

89 @pytest.mark.asyncio 

90 async def test_call_select_with_params(self, db_instance: MyAioSQLite): 

91 """Test __call__ method with parameterized SELECT queries.""" 

92 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

93 await db_instance.execute("INSERT INTO test_table VALUES (1, 'Alice'), (2, 'Bob')") 

94 

95 df = await db_instance("SELECT * FROM test_table WHERE id = ?", 1) 

96 

97 assert len(df) == 1 

98 assert df.iloc[0]["name"] == "Alice" 

99 

100 @pytest.mark.asyncio 

101 async def test_call_select_no_results(self, db_instance: MyAioSQLite): 

102 """Test __call__ method with SELECT query returning no results.""" 

103 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

104 

105 df = await db_instance("SELECT * FROM test_table") 

106 

107 assert isinstance(df, pd.DataFrame) 

108 assert len(df) == 0 

109 

110 @pytest.mark.asyncio 

111 async def test_call_non_select_query_raises_error(self, db_instance: MyAioSQLite): 

112 """Test __call__ method raises error for non-SELECT queries.""" 

113 with pytest.raises(ValueError, match="Only SELECT queries are allowed"): 

114 await db_instance("INSERT INTO test_table VALUES (1, 'Alice')") 

115 

116 with pytest.raises(ValueError, match="Only SELECT queries are allowed"): 

117 await db_instance("UPDATE test_table SET name = 'Charlie' WHERE id = 1") 

118 

119 with pytest.raises(ValueError, match="Only SELECT queries are allowed"): 

120 await db_instance("DELETE FROM test_table WHERE id = 1") 

121 

122 @pytest.mark.asyncio 

123 async def test_execute_create_table(self, db_instance: MyAioSQLite): 

124 """Test execute method with CREATE TABLE.""" 

125 await db_instance.execute("CREATE TABLE test_table (id INTEGER PRIMARY KEY, name TEXT)") 

126 

127 # Verify table was created 

128 df = await db_instance("SELECT name FROM sqlite_master WHERE type='table' AND name='test_table'") 

129 assert len(df) == 1 

130 

131 @pytest.mark.asyncio 

132 async def test_execute_insert_update_delete(self, db_instance: MyAioSQLite): 

133 """Test execute method with INSERT, UPDATE, DELETE operations.""" 

134 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

135 

136 # Test INSERT 

137 await db_instance.execute("INSERT INTO test_table VALUES (?, ?)", 1, "Alice") 

138 await db_instance.execute("INSERT INTO test_table VALUES (?, ?)", 2, "Bob") 

139 

140 # Verify INSERT 

141 df = await db_instance("SELECT COUNT(*) as count FROM test_table") 

142 assert df.iloc[0]["count"] == 2 

143 

144 # Test UPDATE 

145 await db_instance.execute("UPDATE test_table SET name = ? WHERE id = ?", "Charlie", 1) 

146 

147 # Verify UPDATE 

148 df = await db_instance("SELECT name FROM test_table WHERE id = 1") 

149 assert df.iloc[0]["name"] == "Charlie" 

150 

151 # Test DELETE 

152 await db_instance.execute("DELETE FROM test_table WHERE id = ?", 2) 

153 

154 # Verify DELETE 

155 df = await db_instance("SELECT COUNT(*) as count FROM test_table") 

156 assert df.iloc[0]["count"] == 1 

157 

158 @pytest.mark.asyncio 

159 async def test_execute_readonly_mode(self, db_instance: MyAioSQLite): 

160 """Test execute method in readonly mode.""" 

161 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

162 await db_instance.execute("INSERT INTO test_table VALUES (1, 'Alice')") 

163 

164 # Test that modification queries raise error in readonly mode 

165 with pytest.raises(ValueError, match="Cannot execute modification query in readonly mode"): 

166 await db_instance.execute("INSERT INTO test_table VALUES (2, 'Bob')", readonly=True) 

167 

168 with pytest.raises(ValueError, match="Cannot execute modification query in readonly mode"): 

169 await db_instance.execute("UPDATE test_table SET name = 'Charlie'", readonly=True) 

170 

171 with pytest.raises(ValueError, match="Cannot execute modification query in readonly mode"): 

172 await db_instance.execute("DELETE FROM test_table", readonly=True) 

173 

174 @pytest.mark.asyncio 

175 async def test_executemany(self, db_instance: MyAioSQLite): 

176 """Test executemany method.""" 

177 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

178 

179 data = [(1, "Alice"), (2, "Bob"), (3, "Charlie")] 

180 await db_instance.executemany("INSERT INTO test_table VALUES (?, ?)", data) 

181 

182 df = await db_instance("SELECT COUNT(*) as count FROM test_table") 

183 assert df.iloc[0]["count"] == 3 

184 

185 @pytest.mark.asyncio 

186 async def test_iterate(self, db_instance: MyAioSQLite): 

187 """Test iterate method.""" 

188 await db_instance.execute("CREATE TABLE test_table (id INTEGER, name TEXT)") 

189 await db_instance.executemany("INSERT INTO test_table VALUES (?, ?)", [(i, f"Name{i}") for i in range(1, 6)]) 

190 

191 results = [] 

192 async for row in db_instance.iterate("SELECT * FROM test_table ORDER BY id"): 

193 results.append(row) 

194 

195 assert len(results) == 5 

196 assert results[0] == {"id": 1, "name": "Name1"} 

197 assert results[4] == {"id": 5, "name": "Name5"} 

198 

199 @pytest.mark.asyncio 

200 async def test_iterate_with_prefetch(self, db_instance: MyAioSQLite): 

201 """Test iterate method with custom prefetch size.""" 

202 await db_instance.execute("CREATE TABLE test_table (id INTEGER)") 

203 await db_instance.executemany("INSERT INTO test_table VALUES (?)", [(i,) for i in range(1, 11)]) 

204 

205 results = [] 

206 async for row in db_instance.iterate("SELECT * FROM test_table ORDER BY id", prefetch=3): 

207 results.append(row) 

208 

209 assert len(results) == 10 

210 assert all("id" in row for row in results) 

211 

212 @pytest.mark.asyncio 

213 async def test_create_function(self, db_instance: MyAioSQLite): 

214 """Test create_function method.""" 

215 

216 def double_value(x): 

217 return x * 2 

218 

219 await db_instance.create_function("double", 1, double_value) 

220 await db_instance.execute("CREATE TABLE test_table (value INTEGER)") 

221 await db_instance.execute("INSERT INTO test_table VALUES (5)") 

222 

223 # Test the function within the same connection context 

224 async with aiosqlite.connect(db_instance.pth_db) as conn: 

225 await conn.create_function("double", 1, double_value) 

226 cursor = await conn.execute("SELECT double(value) as doubled FROM test_table") 

227 row = await cursor.fetchone() 

228 assert row[0] == 10 

229 

230 @pytest.mark.asyncio 

231 async def test_create_table_if_not_exists(self, db_instance: MyAioSQLite): 

232 """Test create_table_if_not_exists method.""" 

233 await db_instance.create_table_if_not_exists("test_table", "id INTEGER PRIMARY KEY, name TEXT NOT NULL") 

234 

235 # Verify table was created 

236 df = await db_instance("SELECT name FROM sqlite_master WHERE type='table' AND name='test_table'") 

237 assert len(df) == 1 

238 

239 # Test that calling again doesn't raise error 

240 await db_instance.create_table_if_not_exists("test_table", "id INTEGER PRIMARY KEY, name TEXT NOT NULL") 

241 

242 @pytest.mark.asyncio 

243 async def test_table_exists(self, db_instance: MyAioSQLite): 

244 """Test table_exists method.""" 

245 # Test non-existent table 

246 exists = await db_instance.table_exists("nonexistent_table") 

247 assert exists is False 

248 

249 # Create table and test again 

250 await db_instance.execute("CREATE TABLE test_table (id INTEGER)") 

251 exists = await db_instance.table_exists("test_table") 

252 assert exists is True 

253 

254 @pytest.mark.asyncio 

255 async def test_vacuum(self, db_instance: MyAioSQLite): 

256 """Test vacuum method.""" 

257 await db_instance.execute("CREATE TABLE test_table (id INTEGER)") 

258 await db_instance.executemany("INSERT INTO test_table VALUES (?)", [(i,) for i in range(100)]) 

259 await db_instance.execute("DELETE FROM test_table WHERE id < 50") 

260 

261 # VACUUM should not raise any errors 

262 await db_instance.vacuum() 

263 

264 @pytest.mark.asyncio 

265 async def test_get_table_info(self, db_instance: MyAioSQLite): 

266 """Test get_table_info method.""" 

267 await db_instance.execute(""" 

268 CREATE TABLE test_table ( 

269 id INTEGER PRIMARY KEY, 

270 name TEXT NOT NULL, 

271 age INTEGER DEFAULT 0 

272 ) 

273 """) 

274 

275 info_df = await db_instance.get_table_info("test_table") 

276 

277 assert len(info_df) == 3 

278 assert "name" in info_df.columns 

279 assert "type" in info_df.columns 

280 

281 # Check column names 

282 column_names = info_df["name"].tolist() 

283 assert "id" in column_names 

284 assert "name" in column_names 

285 assert "age" in column_names 

286 

287 def test_dict_to_blob(self): 

288 """Test dict_to_blob static method.""" 

289 test_dict = {"key": "value", "number": 42, "list": [1, 2, 3]} 

290 blob = MyAioSQLite.dict_to_blob(test_dict) 

291 

292 assert isinstance(blob, bytes) 

293 assert len(blob) > 0 

294 

295 def test_dict_to_blob_invalid_data(self): 

296 """Test dict_to_blob with unserializable data.""" 

297 

298 class UnserializableClass: 

299 pass 

300 

301 invalid_dict = {"obj": UnserializableClass()} 

302 blob = MyAioSQLite.dict_to_blob(invalid_dict) 

303 

304 # Should return empty JSON object as bytes 

305 assert blob == b"{}" 

306 

307 def test_blob_to_dict(self): 

308 """Test blob_to_dict static method.""" 

309 test_dict = {"key": "value", "number": 42, "list": [1, 2, 3]} 

310 blob = MyAioSQLite.dict_to_blob(test_dict) 

311 result_dict = MyAioSQLite.blob_to_dict(blob) 

312 

313 assert result_dict == test_dict 

314 

315 def test_blob_to_dict_invalid_blob(self): 

316 """Test blob_to_dict with invalid blob data.""" 

317 invalid_blob = b"not valid json" 

318 result = MyAioSQLite.blob_to_dict(invalid_blob) 

319 

320 # Should return empty dict 

321 assert result == {} 

322 

323 @pytest.mark.asyncio 

324 async def test_error_handling_in_call(self, db_instance: MyAioSQLite): 

325 """Test error handling in __call__ method.""" 

326 with pytest.raises(Exception): # noqa: B017 Should raise some database error 

327 await db_instance("SELECT * FROM nonexistent_table") 

328 

329 @pytest.mark.asyncio 

330 async def test_error_handling_in_execute(self, db_instance: MyAioSQLite): 

331 """Test error handling in execute method.""" 

332 with pytest.raises(Exception): # noqa: B017 Should raise some database error 

333 await db_instance.execute("INVALID SQL SYNTAX") 

334 

335 @pytest.mark.asyncio 

336 async def test_auto_commit_mode(self): 

337 """Test auto_commit mode.""" 

338 with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: 

339 db_path = Path(tmp.name) 

340 

341 try: 

342 instance = await MyAioSQLite.async_init(pth_db=db_path, auto_commit=True) 

343 

344 await instance.execute("CREATE TABLE test_table (id INTEGER)") 

345 await instance.execute("INSERT INTO test_table VALUES (1)", auto_commit=True) 

346 

347 # Verify data was committed 

348 df = await instance("SELECT COUNT(*) as count FROM test_table") 

349 assert df.iloc[0]["count"] == 1 

350 finally: 

351 if db_path.exists(): 

352 db_path.unlink()