-
-
Notifications
You must be signed in to change notification settings - Fork 98
Expand file tree
/
Copy pathconftest.py
More file actions
108 lines (76 loc) · 2.92 KB
/
Copy pathconftest.py
File metadata and controls
108 lines (76 loc) · 2.92 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
import asyncio
import random
import string
import nest_asyncio
import pytest
import pytest_asyncio
from tests.lifespan import init_tests
from tests.settings import create_config
import ormar
base_ormar_config = create_config()
nest_asyncio.apply()
pytestmark = pytest.mark.asyncio
class Author(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="authors")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
score: float = ormar.Integer(minimum=0, maximum=100)
class AuthorWithManyFields(Author):
year_born: int = ormar.Integer()
year_died: int = ormar.Integer(nullable=True)
birthplace: str = ormar.String(max_length=255)
class Publisher(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="publishers")
id: int = ormar.Integer(primary_key=True)
name: str = ormar.String(max_length=100)
prestige: int = ormar.Integer(minimum=0, maximum=10)
class Book(ormar.Model):
ormar_config = base_ormar_config.copy(tablename="books")
id: int = ormar.Integer(primary_key=True)
author: Author = ormar.ForeignKey(Author, index=True)
publisher: Publisher = ormar.ForeignKey(Publisher, index=True)
title: str = ormar.String(max_length=100)
year: int = ormar.Integer(nullable=True)
create_test_database = init_tests(base_ormar_config, scope="function")
@pytest_asyncio.fixture(autouse=True, scope="function")
async def connect_database(create_test_database):
if not base_ormar_config.database.is_connected:
await base_ormar_config.database.connect()
yield
if base_ormar_config.database.is_connected:
await base_ormar_config.database.disconnect()
@pytest_asyncio.fixture
async def author():
author = await Author(name="Author", score=10).save()
return author
@pytest_asyncio.fixture
async def publisher():
publisher = await Publisher(name="Publisher", prestige=random.randint(0, 10)).save()
return publisher
@pytest_asyncio.fixture
async def authors_in_db(num_models: int):
authors = [
Author(
name="".join(random.sample(string.ascii_letters, 5)),
score=int(random.random() * 100),
)
for i in range(0, num_models)
]
await Author.objects.bulk_create(authors)
return await Author.objects.all()
@pytest_asyncio.fixture
async def aio_benchmark(benchmark):
def _fixture_wrapper(func):
def _func_wrapper(*args, **kwargs):
if asyncio.iscoroutinefunction(func):
# Get the running event loop instead of requesting it as a fixture
loop = asyncio.get_running_loop()
@benchmark
def benchmarked_func():
a = loop.run_until_complete(func(*args, **kwargs))
return a
return benchmarked_func
else:
return benchmark(func, *args, **kwargs)
return _func_wrapper
return _fixture_wrapper