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
|
from sys import version_info
from typing import List
import pymongo
import pytest
from beanie import Document, Link, init_beanie
from beanie.odm.fields import Indexed, PydanticObjectId
from mongomock_motor import AsyncMongoMockClient
from polyfactory.factories.beanie_odm_factory import BeanieDocumentFactory
@pytest.fixture()
def mongo_connection() -> AsyncMongoMockClient:
return AsyncMongoMockClient()
class MyDocument(Document):
id: PydanticObjectId
name: str
index: Indexed(str, pymongo.DESCENDING) # type: ignore
siblings: List[PydanticObjectId]
class MyOtherDocument(Document):
id: PydanticObjectId
document: Link[MyDocument]
class MyFactory(BeanieDocumentFactory):
__model__ = MyDocument
class MyOtherFactory(BeanieDocumentFactory):
__model__ = MyOtherDocument
@pytest.fixture(autouse=True)
async def beanie_init(mongo_connection: AsyncMongoMockClient) -> None:
await init_beanie(database=mongo_connection.db_name, document_models=[MyDocument, MyOtherDocument])
async def test_handling_of_beanie_types() -> None:
result = MyFactory.build()
assert result.name
assert result.index
assert isinstance(result.index, str)
async def test_beanie_persistence_of_single_instance() -> None:
result = await MyFactory.create_async()
assert result.id
assert result.name
assert result.index
assert isinstance(result.index, str)
async def test_beanie_persistence_of_multiple_instances() -> None:
result = await MyFactory.create_batch_async(size=3)
assert len(result) == 3
for instance in result:
assert instance.id
assert instance.name
assert instance.index
assert isinstance(instance.index, str)
@pytest.mark.skipif(version_info < (3, 11), reason="test isolation issues on lower versions")
async def test_beanie_links() -> None:
result = await MyOtherFactory.create_async()
assert isinstance(result.document, MyDocument)
|
