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
|
import sys
from unittest.mock import MagicMock, Mock
import pytest
from odmantic.session import AIOSession, AIOTransaction, SyncSession, SyncTransaction
from ..zoo.person import PersonModel
if sys.version_info >= (3, 8):
from unittest.mock import AsyncMock
pytestmark = [
pytest.mark.asyncio,
pytest.mark.skipif(
sys.version_info < (3, 8), reason="async mock testing requires python3.8+"
),
]
@pytest.fixture(scope="function")
def mocked_driver_session():
return Mock()
@pytest.fixture(scope="function")
def mocked_aio_engine(mocked_driver_session):
engine = AsyncMock()
engine._get_session = Mock(return_value=mocked_driver_session)
return engine
@pytest.fixture(scope="function", params=["session", "transaction"])
async def mocked_aio_session(request, mocked_aio_engine):
if request.param == "session":
return AIOSession(mocked_aio_engine)
else:
session = AIOSession(mocked_aio_engine)
await session.start()
return AIOTransaction(session)
@pytest.fixture(scope="function")
def mocked_sync_engine(mocked_driver_session):
engine = MagicMock()
engine._get_session = Mock(return_value=mocked_driver_session)
return engine
@pytest.fixture(scope="function", params=["session", "transaction"])
def mocked_sync_session(request, mocked_sync_engine):
if request.param == "session":
return SyncSession(mocked_sync_engine)
else:
session = SyncSession(mocked_sync_engine)
session.start()
return SyncTransaction(session)
async def test_session_find(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.find(PersonModel)
mocked_aio_engine.find.assert_awaited_once()
assert mocked_aio_engine.find.call_args.kwargs["session"] == mocked_driver_session
async def test_session_find_one(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.find_one(PersonModel)
mocked_aio_engine.find_one.assert_awaited_once()
assert (
mocked_aio_engine.find_one.call_args.kwargs["session"] == mocked_driver_session
)
async def test_session_count(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.count(PersonModel)
mocked_aio_engine.count.assert_awaited_once()
assert mocked_aio_engine.count.call_args.kwargs["session"] == mocked_driver_session
async def test_session_save(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.save(PersonModel(first_name="John", last_name="Doe"))
mocked_aio_engine.save.assert_awaited_once()
assert mocked_aio_engine.save.call_args.kwargs["session"] == mocked_driver_session
async def test_session_save_all(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.save_all([PersonModel(first_name="John", last_name="Doe")])
mocked_aio_engine.save_all.assert_awaited_once()
assert (
mocked_aio_engine.save_all.call_args.kwargs["session"] == mocked_driver_session
)
async def test_session_delete(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.delete(PersonModel(first_name="John", last_name="Doe"))
mocked_aio_engine.delete.assert_awaited_once()
assert mocked_aio_engine.delete.call_args.kwargs["session"] == mocked_driver_session
async def test_session_remove(
mocked_aio_session, mocked_aio_engine, mocked_driver_session
):
await mocked_aio_session.remove(PersonModel, PersonModel.first_name == "John")
mocked_aio_engine.remove.assert_awaited_once()
assert mocked_aio_engine.remove.call_args.kwargs["session"] == mocked_driver_session
def test_sync_session_find(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.find(PersonModel)
mocked_sync_engine.find.assert_called_once()
assert mocked_sync_engine.find.call_args.kwargs["session"] == mocked_driver_session
def test_sync_session_find_one(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.find_one(PersonModel)
mocked_sync_engine.find_one.assert_called_once()
assert (
mocked_sync_engine.find_one.call_args.kwargs["session"] == mocked_driver_session
)
def test_sync_session_count(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.count(PersonModel)
mocked_sync_engine.count.assert_called_once()
assert mocked_sync_engine.count.call_args.kwargs["session"] == mocked_driver_session
def test_sync_session_save(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.save(PersonModel(first_name="John", last_name="Doe"))
mocked_sync_engine.save.assert_called_once()
assert mocked_sync_engine.save.call_args.kwargs["session"] == mocked_driver_session
def test_sync_session_save_all(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.save_all([PersonModel(first_name="John", last_name="Doe")])
mocked_sync_engine.save_all.assert_called_once()
assert (
mocked_sync_engine.save_all.call_args.kwargs["session"] == mocked_driver_session
)
def test_sync_session_delete(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.delete(PersonModel(first_name="John", last_name="Doe"))
mocked_sync_engine.delete.assert_called_once()
assert (
mocked_sync_engine.delete.call_args.kwargs["session"] == mocked_driver_session
)
def test_sync_session_remove(
mocked_sync_session, mocked_sync_engine, mocked_driver_session
):
mocked_sync_session.remove(PersonModel, PersonModel.first_name == "John")
mocked_sync_engine.remove.assert_called_once()
assert (
mocked_sync_engine.remove.call_args.kwargs["session"] == mocked_driver_session
)
|
