1
|
|
|
import os |
2
|
|
|
import pytest |
3
|
|
|
from glob import glob |
4
|
|
|
from sqlalchemy import Column |
5
|
|
|
from sqlalchemy.orm import Session |
6
|
|
|
from sqlalchemy.exc import NoSuchTableError |
7
|
|
|
from unittest import TestCase |
8
|
|
|
from .seed_db import create_table |
9
|
|
|
from lazy_alchemy import get_lazy_class, CustomTable |
10
|
|
|
|
11
|
|
|
|
12
|
|
|
class TestLazyOrmSuite(TestCase): |
13
|
|
|
def setUp(self) -> None: |
14
|
|
|
self.engine = create_table() |
15
|
|
|
self.session = Session(self.engine) |
16
|
|
|
self.lazy_class = get_lazy_class(self.engine) |
17
|
|
|
|
18
|
|
|
def test_lazy_db_connection(self): |
19
|
|
|
table = self.lazy_class.user |
20
|
|
|
self.assertIsInstance(table, CustomTable) |
21
|
|
|
self.assertIsInstance(table.username, Column) |
22
|
|
|
self.assertIsInstance(table.age, Column) |
23
|
|
|
self.assertEqual(table.name, "user") |
24
|
|
|
self.assertEqual(len(table.constraints), 1) |
25
|
|
|
self.assertTrue( |
26
|
|
|
any(i for i in table.indexes if i.name == "idx_user_username")) |
27
|
|
|
|
28
|
|
|
def test_insert_operation_on_dynamic_class(self): |
29
|
|
|
user: CustomTable = self.lazy_class.user |
30
|
|
|
self.assertTrue(user) |
31
|
|
|
insert_statement = user.insert().values(username="fake_user", age=21) |
32
|
|
|
self.session.execute(insert_statement) |
33
|
|
|
self.session.commit() |
34
|
|
|
obj = self.session.query(user).first() |
35
|
|
|
self.assertEqual(obj.username, "fake_user") |
36
|
|
|
self.assertEqual(obj.age, 21) |
37
|
|
|
|
38
|
|
|
def test_invalid_table_and_column(self): |
39
|
|
|
with pytest.raises(NoSuchTableError): |
40
|
|
|
self.lazy_class.foo_bar |
41
|
|
|
with pytest.raises(AttributeError): |
42
|
|
|
self.lazy_class.user.abc |
43
|
|
|
|
44
|
|
|
def test_overriding_table(self): |
45
|
|
|
_ = self.lazy_class.user |
46
|
|
|
self.lazy_class.user = CustomTable("userabc", self.lazy_class.metadata) |
47
|
|
|
self.assertNotEqual(_, self.lazy_class.user) |
48
|
|
|
self.assertEqual(self.lazy_class.user.name, "userabc") |
49
|
|
|
|
50
|
|
|
def tearDown(self) -> None: |
51
|
|
|
self.session.close() |
52
|
|
|
print(f"cleaning files {glob('*.db')}") |
53
|
|
|
os.unlink("test.db") |
54
|
|
|
|