-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_sqliter.py
More file actions
1091 lines (873 loc) · 37.6 KB
/
test_sqliter.py
File metadata and controls
1091 lines (873 loc) · 37.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Test suite for the 'sqliter' library."""
import gc
import warnings
from pathlib import Path
import pytest
from pytest_mock import MockerFixture
from sqliter import SqliterDB
from sqliter.exceptions import (
RecordFetchError,
RecordNotFoundError,
TableCreationError,
)
from sqliter.model import BaseDBModel
from sqliter.orm import BaseDBModel as ORMBaseDBModel
from sqliter.orm.registry import ModelRegistry
from tests.conftest import ComplexModel, DetailedPersonModel, ExampleModel
class ExistOkModel(BaseDBModel):
"""Just used to test table creation with an existing table."""
name: str
age: int
class Meta:
"""Meta class for the model."""
table_name = "exist_ok_table"
class TestSqliterDB:
"""Test class to test the SqliterDB class."""
def test_auto_commit_default(self) -> None:
"""Test that auto_commit is enabled by default."""
db = SqliterDB(":memory:")
assert db.auto_commit
def test_auto_commit_disabled(self) -> None:
"""Test that auto_commit can be disabled."""
db = SqliterDB(":memory:", auto_commit=False)
assert not db.auto_commit
@pytest.mark.skip(reason="This does not test the behavour correctly.")
def test_data_lost_when_auto_commit_disabled(self) -> None:
"""Test that data is lost when auto_commit is disabled.
The other cases when auto_commit is enabled are tested in all the other
tests.
"""
db = SqliterDB(":memory:", auto_commit=False)
db.create_table(ExampleModel)
# Insert a record
test_model = ExampleModel(
slug="test", name="Test License", content="Test Content"
)
result = db.insert(test_model)
# Ensure the record exists
fetched_license = db.get(ExampleModel, result.pk)
assert fetched_license is not None
# Close the connection
db.close()
# Re-open the connection
db.connect()
# Ensure the data is lost
with pytest.raises(RecordFetchError):
db.get(ExampleModel, result.pk)
def test_create_table(self, db_mock: SqliterDB) -> None:
"""Test table creation."""
with db_mock.connect() as conn:
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
assert len(tables) == 2
assert ("test_table",) in tables
def test_close_connection(self, db_mock: SqliterDB) -> None:
"""Test closing the connection."""
db_mock.close()
assert db_mock.conn is None
def test_finalizer_closes_unclosed_connection(self) -> None:
"""Finalization should release abandoned SQLite connections."""
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", ResourceWarning)
db = SqliterDB(":memory:")
db.create_table(ExampleModel)
del db
gc.collect()
resource_warnings = [
warning
for warning in caught
if issubclass(warning.category, ResourceWarning)
and "unclosed database" in str(warning.message)
]
assert resource_warnings == []
def test_commit_changes(self, mocker: MockerFixture) -> None:
"""Test committing changes to the database."""
db = SqliterDB(":memory:", auto_commit=False)
db.create_table(ExampleModel)
db.insert(
ExampleModel(slug="test", name="Test License", content="Content")
)
mock_conn = mocker.Mock()
mocker.patch.object(db, "conn", mock_conn)
db.commit()
assert mock_conn.commit.called
def test_create_table_with_default_auto_increment(
self, db_mock: SqliterDB
) -> None:
"""Test table creation with auto-incrementing primary key."""
class AutoIncrementModel(BaseDBModel):
name: str
class Meta:
table_name: str = "auto_increment_table"
# Create the table
db_mock.create_table(AutoIncrementModel)
# Verify that the table was created with an auto-incrementing PK
with db_mock.connect() as conn:
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(auto_increment_table);")
table_info = cursor.fetchall()
# Check that the first column is 'id' and it's an auto-incrementing int
assert table_info[0][1] == "pk" # Column name
assert table_info[0][2] == "INTEGER" # Column type
assert table_info[0][5] == 1 # Primary key flag
def test_default_table_name(self, db_mock: SqliterDB) -> None:
"""Test the default table name generation.
It should default to the class name in lowercase, plural form.
"""
class DefaultNameModel(BaseDBModel):
name: str
class Meta:
table_name = None # Explicitly set to None to test the default
# Verify that get_table_name defaults to class name in lowercase
assert DefaultNameModel.get_table_name() == "default_names"
def test_get_table_name_fallback_without_inflect(
self, mocker: MockerFixture
) -> None:
"""Test get_table_name falls back to manual plural without 'inflect."""
# Mock the inflect import to raise ImportError for `inflect`
mocker.patch.dict("sys.modules", {"inflect": None})
class UserModel(BaseDBModel):
pass
table_name = UserModel.get_table_name()
assert table_name == "users" # Fallback logic should add 's'
def test_get_table_name_no_double_s_without_inflect(
self, mocker: MockerFixture
) -> None:
"""Test get_table_name doesn't add extra 's' if already there."""
# Mock the sys.modules to simulate 'inflect' being unavailable
mocker.patch.dict("sys.modules", {"inflect": None})
class UsersModel(BaseDBModel):
pass
table_name = UsersModel.get_table_name()
assert table_name == "users" # Should not add an extra 's'
def test_get_table_name_with_inflect(self) -> None:
"""Test get_table_name uses 'inflect' for pluralization if available."""
class PersonModel(BaseDBModel):
pass
table_name = PersonModel.get_table_name()
# Here, we assume that inflect will pluralize 'person' to 'people'
assert table_name == "people", (
f"Expected table name to be 'people', but got '{table_name}' - "
"Make sure you installed the 'extras' using 'uv sync --all-extras'"
", or ignore this failure."
)
def test_insert_license(self, db_mock: SqliterDB) -> None:
"""Test inserting a license into the database."""
test_model = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
db_mock.insert(test_model)
with db_mock.connect() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM test_table WHERE slug = ?", ("mit",))
result = cursor.fetchone()
assert result[0] == 1
assert result[3] == "mit"
assert result[4] == "MIT License"
assert result[5] == "MIT License Content"
def test_insert_updates_supplied_instance(self, db_mock: SqliterDB) -> None:
"""Insert should mark the supplied instance as saved."""
test_model = ExampleModel(
slug="apache",
name="Apache License",
content="Apache License Content",
)
inserted = db_mock.insert(test_model)
assert inserted is test_model
assert test_model.pk == inserted.pk
assert test_model.pk > 0
def test_insert_updates_orm_instance_context(self) -> None:
"""Insert should attach db_context to supplied ORM instances."""
state = ModelRegistry.snapshot()
try:
class SavedORMModel(ORMBaseDBModel):
"""ORM model for insert context tests."""
name: str
db = SqliterDB(memory=True)
db.create_table(SavedORMModel)
model = SavedORMModel(name="saved")
inserted = db.insert(model)
assert inserted is model
assert model.pk > 0
assert model.db_context is db
db.close()
finally:
ModelRegistry.restore(state)
def test_create_instance_from_data_applies_pk(
self,
db_mock: SqliterDB,
) -> None:
"""create_instance_from_data should apply the supplied primary key."""
instance = db_mock.create_instance_from_data(
ExampleModel,
{
"slug": "bsd",
"name": "BSD License",
"content": "BSD License Content",
},
pk=42,
)
assert instance.pk == 42
assert instance.slug == "bsd"
def test_build_insert_plan_binds_none_values(self) -> None:
"""Insert plans keep placeholders stable when values are None."""
db = SqliterDB(":memory:")
model = ComplexModel(
name="Alice",
age=30.5,
is_active=True,
score=85,
nullable_field=None,
)
plan = db._build_insert_plan(model, timestamp_override=False)
assert '"nullable_field"' in plan.sql
assert "NULL" not in plan.sql
assert plan.sql.count("?") == len(plan.values)
assert plan.values[-1] is None
def test_crud_quotes_reserved_table_name(self) -> None:
"""Core CRUD methods handle reserved table names."""
state = ModelRegistry.snapshot()
try:
class ReservedCrudModel(BaseDBModel):
name: str
class Meta:
table_name = "order"
db = SqliterDB(":memory:")
db.create_table(ReservedCrudModel)
inserted = db.insert(ReservedCrudModel(name="Initial"))
fetched = db.get(ReservedCrudModel, inserted.pk)
assert fetched is not None
assert fetched.name == "Initial"
fetched.name = "Updated"
db.update(fetched)
updated = db.get(ReservedCrudModel, inserted.pk, bypass_cache=True)
assert updated is not None
assert updated.name == "Updated"
db.delete(ReservedCrudModel, inserted.pk)
assert db.get(ReservedCrudModel, inserted.pk) is None
db.close()
finally:
ModelRegistry.restore(state)
def test_fetch_license(self, db_mock: SqliterDB) -> None:
"""Test fetching a license by primary key."""
test_model = ExampleModel(
slug="gpl", name="GPL License", content="GPL License Content"
)
result = db_mock.insert(test_model)
fetched_license = db_mock.get(ExampleModel, result.pk)
assert fetched_license is not None
assert fetched_license.slug == "gpl"
assert fetched_license.name == "GPL License"
assert fetched_license.content == "GPL License Content"
def test_update(self, db_mock: SqliterDB) -> None:
"""Test updating an existing license."""
test_model = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
result = db_mock.insert(test_model)
# Update license content
result.content = "Updated MIT License Content"
db_mock.update(result)
# Fetch and check if updated
fetched_license = db_mock.get(ExampleModel, result.pk)
assert fetched_license is not None
assert fetched_license.content == "Updated MIT License Content"
def test_delete(self, db_mock: SqliterDB) -> None:
"""Test deleting a license."""
test_model = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
result = db_mock.insert(test_model)
# Delete the record
db_mock.delete(ExampleModel, result.pk)
# Ensure it no longer exists
fetched_license = db_mock.get(ExampleModel, result.pk)
assert fetched_license is None
def test_select_filter(self, db_mock: SqliterDB) -> None:
"""Test filtering licenses using the QueryBuilder."""
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
license2 = ExampleModel(
slug="gpl", name="GPL License", content="GPL License Content"
)
db_mock.insert(license1)
db_mock.insert(license2)
# Query and filter licenses
filtered = (
db_mock.select(ExampleModel).filter(name="GPL License").fetch_all()
)
assert len(filtered) == 1
assert filtered[0].slug == "gpl"
def test_query_fetch_first(self, db_mock: SqliterDB) -> None:
"""Test fetching the first record."""
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
license2 = ExampleModel(
slug="gpl", name="GPL License", content="GPL License Content"
)
db_mock.insert(license1)
db_mock.insert(license2)
first_record = db_mock.select(ExampleModel).fetch_first()
assert first_record is not None
assert first_record.slug == "mit"
def test_query_fetch_last(self, db_mock: SqliterDB) -> None:
"""Test fetching the last record."""
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
license2 = ExampleModel(
slug="gpl", name="GPL License", content="GPL License Content"
)
db_mock.insert(license1)
db_mock.insert(license2)
last_record = db_mock.select(ExampleModel).fetch_last()
assert last_record is not None
assert last_record.slug == "gpl"
def test_count_records(self, db_mock: SqliterDB) -> None:
"""Test counting records in the database."""
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
license2 = ExampleModel(
slug="gpl", name="GPL License", content="GPL License Content"
)
db_mock.insert(license1)
db_mock.insert(license2)
count = db_mock.select(ExampleModel).count()
assert count == 2
def test_exists_record(self, db_mock: SqliterDB) -> None:
"""Test checking if a record exists."""
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
db_mock.insert(license1)
exists = db_mock.select(ExampleModel).filter(slug="mit").exists()
assert exists
def test_transaction_commit(
self, db_mock: SqliterDB, mocker: MockerFixture
) -> None:
"""Test if auto_commit works correctly when enabled."""
# Mock the commit method on the connection
mock_conn = mocker.MagicMock()
# Close the fixture-owned real connection before replacing it.
db_mock.close()
# Manually reset the connection to ensure our mock is used.
db_mock.conn = mock_conn
# Patch connect method to return the mock connection
mocker.patch.object(db_mock, "connect", return_value=mock_conn)
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
with db_mock:
db_mock.create_table(ExampleModel)
db_mock.insert(license1)
# Ensure commit was called only once, when the context manager exited.
assert mock_conn.commit.call_count == 1
def test_transaction_manual_commit(self, mocker: MockerFixture) -> None:
"""Test context-manager commit when auto_commit is set to False.
Regardless of the auto_commit setting, the context manager should commit
changes when exiting the context.
"""
db_manual = SqliterDB(":memory:", auto_commit=True)
# Mock the connection and commit
mock_conn = mocker.MagicMock()
mocker.patch.object(db_manual, "connect", return_value=mock_conn)
db_manual.conn = (
mock_conn # Ensure the db_manual uses the mock connection
)
license1 = ExampleModel(
slug="mit", name="MIT License", content="MIT License Content"
)
with db_manual:
db_manual.create_table(ExampleModel)
db_manual.insert(license1)
# Ensure commit hasn't been called yet
mock_conn.commit.assert_not_called()
# After leaving the context, commit should now be called
mock_conn.commit.assert_called_once()
def test_update_existing_record(self, db_mock: SqliterDB) -> None:
"""Test that updating an existing record works correctly."""
# Insert an example record
example_model = ExampleModel(
slug="test", name="Test License", content="Test Content"
)
result = db_mock.insert(example_model)
# Update the record's content
result.content = "Updated Content"
db_mock.update(result)
# Fetch the updated record and verify the changes
updated_record = db_mock.get(ExampleModel, result.pk)
assert updated_record is not None
assert updated_record.content == "Updated Content"
def test_update_non_existing_record(self, db_mock: SqliterDB) -> None:
"""Test updating a non-existing record raises RecordNotFoundError."""
# Create an example record that is not inserted into the DB
example_model = ExampleModel(
slug="nonexistent",
name="Nonexistent License",
content="Nonexistent Content",
)
# Try updating the non-existent record
with pytest.raises(RecordNotFoundError) as exc_info:
db_mock.update(example_model)
# Check that the correct error message is raised
assert "Failed to find that record in the table (key '0')" in str(
exc_info.value
)
def test_get_non_existent_table(self, db_mock: SqliterDB) -> None:
"""Test fetching from a non-existent table raises RecordFetchError."""
class NonExistentModel(ExampleModel):
class Meta:
table_name = "non_existent_table" # A table that doesn't exist
with pytest.raises(RecordFetchError):
db_mock.get(NonExistentModel, -1)
def test_get_record_no_result(self, db_mock: SqliterDB) -> None:
"""Test fetching a non-existent record returns None."""
result = db_mock.get(ExampleModel, -1)
assert result is None
def test_get_does_not_cache_default_negative_result(
self, tmp_path: Path
) -> None:
"""Default negative cache entries do not mask inserts from other DBs."""
db_path = tmp_path / "negative-cache.db"
db_reader = SqliterDB(str(db_path), cache_enabled=True)
db_writer = SqliterDB(str(db_path), cache_enabled=True)
db_reader.create_table(ExampleModel)
assert db_reader.get(ExampleModel, 1) is None
db_writer.insert(
ExampleModel(pk=1, slug="later", name="Later", content="Inserted")
)
fetched = db_reader.get(ExampleModel, 1)
assert fetched is not None
assert fetched.slug == "later"
db_reader.close()
db_writer.close()
def test_get_caches_negative_result_with_explicit_ttl(
self, tmp_path: Path
) -> None:
"""Explicit cache_ttl keeps negative cache entries for later lookups."""
db_path = tmp_path / "negative-cache-ttl.db"
db_reader = SqliterDB(str(db_path), cache_enabled=True)
db_writer = SqliterDB(str(db_path), cache_enabled=True)
db_reader.create_table(ExampleModel)
assert db_reader.get(ExampleModel, 1, cache_ttl=60) is None
db_writer.insert(
ExampleModel(pk=1, slug="later", name="Later", content="Inserted")
)
assert db_reader.get(ExampleModel, 1, cache_ttl=60) is None
db_reader.close()
db_writer.close()
def test_delete_non_existent_record(self, db_mock: SqliterDB) -> None:
"""Test that trying to delete a non-existent record raises exception."""
with pytest.raises(RecordNotFoundError):
db_mock.delete(ExampleModel, -1)
def test_delete_existing_record(self, db_mock: SqliterDB) -> None:
"""Test that a record is deleted successfully."""
# Insert a record first
test_model = ExampleModel(
slug="test", name="Test License", content="Test Content"
)
result = db_mock.insert(test_model)
# Now delete the record
db_mock.delete(ExampleModel, result.pk)
# Fetch the deleted record to confirm it's gone
deleted_result = db_mock.get(ExampleModel, result.pk)
assert deleted_result is None
def test_select_with_exclude_single_field(
self,
db_mock_detailed: SqliterDB,
) -> None:
"""Test selecting with exclude parameter to remove a single field."""
results = db_mock_detailed.select(
DetailedPersonModel, exclude=["email"]
).fetch_all()
assert len(results) == 3
for result in results:
assert hasattr(result, "name")
assert hasattr(result, "age")
assert not hasattr(result, "email")
def test_select_with_exclude_multiple_fields(
self,
db_mock_detailed: SqliterDB,
) -> None:
"""Test selecting with exclude parameter to remove multiple fields."""
results = db_mock_detailed.select(
DetailedPersonModel, exclude=["email", "phone"]
).fetch_all()
assert len(results) == 3
for result in results:
assert hasattr(result, "name")
assert hasattr(result, "age")
assert not hasattr(result, "email")
assert not hasattr(result, "phone")
def test_select_with_exclude_all_fields_error(
self,
db_mock_detailed: SqliterDB,
) -> None:
"""Test select with excluding all fields raises error."""
with pytest.raises(
ValueError, match=r"Exclusion results in no fields being selected."
):
db_mock_detailed.select(
DetailedPersonModel,
exclude=[
"created_at",
"updated_at",
"name",
"age",
"email",
"address",
"phone",
"occupation",
],
).fetch_all()
def test_select_with_exclude_and_filter(
self,
db_mock_detailed: SqliterDB,
) -> None:
"""Test selecting with exclude and filter combined."""
results = (
db_mock_detailed.select(DetailedPersonModel, exclude=["phone"])
.filter(age__gte=30)
.fetch_all()
)
assert len(results) == 2
for result in results:
assert hasattr(result, "name")
assert hasattr(result, "email")
assert not hasattr(result, "phone")
def test_in_memory_db_initialization(self) -> None:
"""Test that an in-memory database is initialized correctly."""
db = SqliterDB(memory=True)
assert db.db_filename == ":memory:"
def test_file_based_db_initialization(self) -> None:
"""Test that a file-based database is initialized correctly."""
db = SqliterDB(db_filename="test.db")
assert db.db_filename == "test.db"
def test_error_when_no_db_name_and_not_memory(self) -> None:
"""Error is raised when no db_filename is provided and memory=False."""
with pytest.raises(ValueError, match="Database name must be provided"):
SqliterDB(memory=False)
def test_file_is_created_when_filename_is_provided(
self, mocker: MockerFixture
) -> None:
"""Test that sqlite3.connect is called with the correct file path."""
mock_connect = mocker.patch("sqlite3.connect")
db_filename = "/fakepath/test.db"
db = SqliterDB(db_filename=db_filename)
db.connect()
# Check if sqlite3.connect was called with the correct filename
mock_connect.assert_called_with(db_filename)
def test_memory_database_no_file_created(
self, mocker: MockerFixture
) -> None:
"""Test sqlite3.connect is called with ':memory:' when memory=True."""
mock_connect = mocker.patch("sqlite3.connect")
db = SqliterDB(memory=True)
db.connect()
# Check if sqlite3.connect was called with ':memory:' for the in-memory
# DB
mock_connect.assert_called_with(":memory:")
def test_memory_db_ignores_filename(self, mocker: MockerFixture) -> None:
"""Test memory=True igores any filename, creating an in-memory DB."""
mock_connect = mocker.patch("sqlite3.connect")
db_filename = "/fakepath/test.db"
db = SqliterDB(db_filename=db_filename, memory=True)
db.connect()
# Check that sqlite3.connect was called with ':memory:', ignoring the
# filename
mock_connect.assert_called_with(":memory:")
def test_complex_model_field_types(self, db_mock: SqliterDB) -> None:
"""Test that the table is created with the correct field types."""
# Create table based on ComplexModel
db_mock.create_table(ComplexModel)
# Get table info for ComplexModel
with db_mock.connect() as conn:
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(complex_model);")
table_info = cursor.fetchall()
# Expected types in SQLite (INTEGER, REAL, TEXT, etc.)
expected_types = {
"pk": "INTEGER",
"created_at": "INTEGER",
"updated_at": "INTEGER",
"name": "TEXT",
"age": "REAL",
"price": "REAL",
"is_active": "INTEGER", # Boolean stored as INTEGER
"nullable_field": "TEXT", # Optional fields default to TEXT
"score": "TEXT",
}
# Assert each field has the correct SQLite type
for column in table_info:
column_name = column[
1
] # Column name is the second element in table_info
column_type = column[2] # Column type is the third element
assert expected_types[column_name] == column_type, (
f"Field {column_name} expected {expected_types[column_name]} "
f"but got {column_type}"
)
def test_complex_model_primary_key(self, db_mock: SqliterDB) -> None:
"""Test that the primary key is correctly created for ComplexModel."""
# Create table based on ComplexModel
db_mock.create_table(ComplexModel)
# Get table info for ComplexModel
with db_mock.connect() as conn:
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(complex_model);")
table_info = cursor.fetchall()
# Find the primary key column
primary_key_column = None
for column in table_info:
if (
column[5] == 1
): # The 6th element in table_info is '1' if column is the pk
primary_key_column = column
# Assert that the primary key is the 'id' field and is an INTEGER
assert primary_key_column is not None, "Primary key not found"
assert primary_key_column[1] == "pk", (
f"Expected 'id' as primary key, but got {primary_key_column[1]}"
)
assert primary_key_column[2] == "INTEGER", (
f"Expected 'INTEGER' type for primary key, but got "
f"{primary_key_column[2]}"
)
def test_reset_database_on_init(self, temp_db_path: str) -> None:
"""Test that the database is reset when reset=True is passed."""
class TestModel(BaseDBModel):
name: str
class Meta:
table_name = "test_reset_table"
# Create a database and add some data
db = SqliterDB(temp_db_path)
db.create_table(TestModel)
db.insert(TestModel(name="Test Data"))
db.close()
# Create a new connection with reset=True
db_reset = SqliterDB(temp_db_path, reset=True)
# Verify the table no longer exists
with pytest.raises(RecordFetchError):
db_reset.select(TestModel).fetch_all()
def test_reset_database_preserves_connection(
self, temp_db_path: str
) -> None:
"""Test that resetting the database doesn't break the connection."""
class TestModel(BaseDBModel):
name: str
class Meta:
table_name = "test_reset_table"
db = SqliterDB(temp_db_path, reset=True)
# Create a table after reset
db.create_table(TestModel)
db.insert(TestModel(name="New Data"))
# Verify data exists
result = db.select(TestModel).fetch_all()
assert len(result) == 1
def test_reset_database_with_multiple_tables(
self, temp_db_path: str
) -> None:
"""Test that reset drops all tables in the database."""
class TestModel1(BaseDBModel):
name: str
class Meta:
table_name = "test_reset_table1"
class TestModel2(BaseDBModel):
age: int
class Meta:
table_name = "test_reset_table2"
# Create a database and add some data
db = SqliterDB(temp_db_path)
db.create_table(TestModel1)
db.create_table(TestModel2)
db.insert(TestModel1(name="Test Data"))
db.insert(TestModel2(age=25))
db.close()
# Reset the database
db_reset = SqliterDB(temp_db_path, reset=True)
# Verify both tables no longer exist
with pytest.raises(RecordFetchError):
db_reset.select(TestModel1).fetch_all()
with pytest.raises(RecordFetchError):
db_reset.select(TestModel2).fetch_all()
def test_reset_database_quotes_reserved_table_name(
self, temp_db_path: str
) -> None:
"""reset=True handles reserved table names."""
state = ModelRegistry.snapshot()
try:
class ReservedResetModel(BaseDBModel):
name: str
class Meta:
table_name = "order"
db = SqliterDB(temp_db_path)
db.create_table(ReservedResetModel)
db.close()
db_reset = SqliterDB(temp_db_path, reset=True)
assert "order" not in db_reset.table_names
db_reset.close()
finally:
ModelRegistry.restore(state)
def test_create_table_exists_ok_true(self, db_mock: SqliterDB) -> None:
"""Test creating a table with exists_ok=True (default behavior)."""
# First creation should succeed
db_mock.create_table(ExistOkModel)
# Second creation should not raise an error
try:
db_mock.create_table(ExistOkModel)
except TableCreationError as e:
pytest.fail(f"create_table raised {type(e).__name__} unexpectedly!")
def test_create_table_exists_ok_false(self, db_mock: SqliterDB) -> None:
"""Test creating a table with exists_ok=False."""
# First creation should succeed
db_mock.create_table(ExistOkModel)
# Second creation should raise an error
with pytest.raises(TableCreationError):
db_mock.create_table(ExistOkModel, exists_ok=False)
def test_create_table_exists_ok_false_new_table(self) -> None:
"""Test creating a new table with exists_ok=False."""
# Create a new database connection
new_db = SqliterDB(":memory:")
# Define a new model class specifically for this test
class UniqueTestModel(BaseDBModel):
name: str
age: int
class Meta:
table_name = "unique_test_table"
# Creation of a new table should succeed with exists_ok=False
try:
new_db.create_table(UniqueTestModel, exists_ok=False)
except TableCreationError as e:
pytest.fail(f"create_table raised {type(e).__name__} unexpectedly!")
# Clean up
new_db.close()
def test_create_table_sql_generation(
self, db_mock: SqliterDB, mocker: MockerFixture
) -> None:
"""Test SQL generation for table creation based on exists_ok value."""
mock_cursor = mocker.MagicMock()
mocker.patch.object(
db_mock, "connect"
).return_value.cursor.return_value = mock_cursor
# Test with exists_ok=True
db_mock.create_table(ExistOkModel, exists_ok=True)
mock_cursor.execute.assert_called()
sql = mock_cursor.execute.call_args[0][0]
assert "CREATE TABLE IF NOT EXISTS" in sql
# Reset the mock
mock_cursor.reset_mock()
# Test with exists_ok=False
db_mock.create_table(ExistOkModel, exists_ok=False)
mock_cursor.execute.assert_called()
sql = mock_cursor.execute.call_args[0][0]
assert "CREATE TABLE" in sql
assert "IF NOT EXISTS" not in sql
def test_create_table_force(self) -> None:
"""Test creating a table with force=True."""
# Create a new database
db = SqliterDB(":memory:")
# Define initial model
class InitialTestModel(BaseDBModel):
name: str
age: int
class Meta:
table_name = "force_test_table"
# First creation
db.create_table(InitialTestModel)
# Insert a record
initial_record = InitialTestModel(name="Alice", age=30)
db.insert(initial_record)
# Define modified model
class ModifiedTestModel(BaseDBModel):
name: str
email: str # New field instead of age
class Meta:
table_name = "force_test_table"
# Recreate with force=True
db.create_table(ModifiedTestModel, force=True)
# Try to insert a record with the new structure
new_record = ModifiedTestModel(name="Bob", email="bob@example.com")
db.insert(new_record)
# Fetch and check if the new structure is in place
result = db.select(ModifiedTestModel).fetch_one()
assert result is not None
assert hasattr(result, "email")
assert not hasattr(result, "age")
# Verify that the old structure is gone by checking table info
with db.connect() as conn:
cursor = conn.cursor()