forked from tortoise/tortoise-orm
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_bulk.py
92 lines (77 loc) · 3.95 KB
/
test_bulk.py
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
from uuid import UUID, uuid4
from tests.testmodels import UniqueName, UUIDPkModel
from tortoise.contrib import test
from tortoise.exceptions import IntegrityError
from tortoise.transactions import in_transaction
class TestBulk(test.TruncationTestCase):
async def test_bulk_create(self):
await UniqueName.bulk_create([UniqueName() for _ in range(1000)])
all_ = await UniqueName.all().values("id", "name")
inc = all_[0]["id"]
self.assertListSortEqual(all_, [{"id": val + inc, "name": None} for val in range(1000)])
async def test_bulk_create_with_batch_size(self):
await UniqueName.bulk_create([UniqueName() for _ in range(1000)], batch_size=100)
all_ = await UniqueName.all().values("id", "name")
inc = all_[0]["id"]
self.assertListSortEqual(all_, [{"id": val + inc, "name": None} for val in range(1000)])
async def test_bulk_create_with_specified(self):
await UniqueName.bulk_create([UniqueName(id=id_) for id_ in range(1000, 2000)])
all_ = await UniqueName.all().values("id", "name")
self.assertListSortEqual(all_, [{"id": id_, "name": None} for id_ in range(1000, 2000)])
async def test_bulk_create_mix_specified(self):
await UniqueName.bulk_create(
[UniqueName(id=id_) for id_ in range(10000, 11000)]
+ [UniqueName() for _ in range(1000)]
)
all_ = await UniqueName.all().values("id", "name")
self.assertEqual(len(all_), 2000)
self.assertListSortEqual(
all_[:1000], [{"id": id_, "name": None} for id_ in range(10000, 11000)]
)
inc = all_[1000]["id"]
self.assertListSortEqual(
all_[1000:], [{"id": val + inc, "name": None} for val in range(1000)]
)
async def test_bulk_create_uuidpk(self):
await UUIDPkModel.bulk_create([UUIDPkModel() for _ in range(1000)])
res = await UUIDPkModel.all().values_list("id", flat=True)
self.assertEqual(len(res), 1000)
self.assertIsInstance(res[0], UUID)
@test.requireCapability(supports_transactions=True)
async def test_bulk_create_in_transaction(self):
async with in_transaction():
await UniqueName.bulk_create([UniqueName() for _ in range(1000)])
all_ = await UniqueName.all().values("id", "name")
inc = all_[0]["id"]
self.assertEqual(all_, [{"id": val + inc, "name": None} for val in range(1000)])
@test.requireCapability(supports_transactions=True)
async def test_bulk_create_uuidpk_in_transaction(self):
async with in_transaction():
await UUIDPkModel.bulk_create([UUIDPkModel() for _ in range(1000)])
res = await UUIDPkModel.all().values_list("id", flat=True)
self.assertEqual(len(res), 1000)
self.assertIsInstance(res[0], UUID)
async def test_bulk_create_fail(self):
with self.assertRaises(IntegrityError):
await UniqueName.bulk_create(
[UniqueName(name=str(i)) for i in range(10)]
+ [UniqueName(name=str(i)) for i in range(10)]
)
async def test_bulk_create_uuidpk_fail(self):
val = uuid4()
with self.assertRaises(IntegrityError):
await UUIDPkModel.bulk_create([UUIDPkModel(id=val) for _ in range(10)])
@test.requireCapability(supports_transactions=True)
async def test_bulk_create_in_transaction_fail(self):
with self.assertRaises(IntegrityError):
async with in_transaction():
await UniqueName.bulk_create(
[UniqueName(name=str(i)) for i in range(10)]
+ [UniqueName(name=str(i)) for i in range(10)]
)
@test.requireCapability(supports_transactions=True)
async def test_bulk_create_uuidpk_in_transaction_fail(self):
val = uuid4()
with self.assertRaises(IntegrityError):
async with in_transaction():
await UUIDPkModel.bulk_create([UUIDPkModel(id=val) for _ in range(10)])