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
|
from datetime import datetime
from typing import Any, List
from uuid import UUID
import bson
import pytest
from bson import ObjectId
import pydantic
if not pydantic.VERSION.startswith("1"):
pytest.skip("only for pydantic v1", allow_module_level=True)
from odmantic import AIOEngine, EmbeddedModel, Model
from polyfactory.decorators import post_generated
from polyfactory.factories.odmantic_odm_factory import OdmanticModelFactory
class OtherEmbeddedDocument(EmbeddedModel): # type: ignore
name: str
serial: UUID
created_on: datetime
bson_id: bson.ObjectId
bson_int64: bson.Int64
bson_dec128: bson.Decimal128
bson_binary: bson.Binary
class MyEmbeddedDocument(EmbeddedModel): # type: ignore
name: str
serial: UUID
other_embedded_document: OtherEmbeddedDocument
created_on: datetime
bson_id: bson.ObjectId
bson_int64: bson.Int64
bson_dec128: bson.Decimal128
bson_binary: bson.Binary
class MyModel(Model): # type: ignore
created_on: datetime
bson_id: bson.ObjectId
bson_int64: bson.Int64
bson_dec128: bson.Decimal128
bson_binary: bson.Binary
name: str
embedded: MyEmbeddedDocument
embedded_list: List[MyEmbeddedDocument]
@pytest.fixture()
async def odmantic_engine(mongo_connection: Any) -> AIOEngine:
return AIOEngine(client=mongo_connection, database=mongo_connection.db_name)
def test_handles_odmantic_models() -> None:
class MyFactory(OdmanticModelFactory):
__model__ = MyModel
result = MyFactory.build()
assert isinstance(result, MyModel)
assert isinstance(result.id, bson.ObjectId)
assert isinstance(result.created_on, datetime)
assert isinstance(result.bson_id, bson.ObjectId)
assert isinstance(result.bson_int64, bson.Int64)
assert isinstance(result.bson_dec128, bson.Decimal128)
assert isinstance(result.bson_binary, bson.Binary)
assert isinstance(result.name, str)
assert isinstance(result.embedded, MyEmbeddedDocument)
assert isinstance(result.embedded_list, list)
for item in result.embedded_list:
assert isinstance(item, MyEmbeddedDocument)
assert isinstance(item.name, str)
assert isinstance(item.serial, UUID)
assert isinstance(item.created_on, datetime)
assert isinstance(item.bson_id, bson.ObjectId)
assert isinstance(item.bson_int64, bson.Int64)
assert isinstance(item.bson_dec128, bson.Decimal128)
assert isinstance(item.bson_binary, bson.Binary)
other = item.other_embedded_document
assert isinstance(other, OtherEmbeddedDocument)
assert isinstance(other.name, str)
assert isinstance(other.serial, UUID)
assert isinstance(other.created_on, datetime)
assert isinstance(other.bson_id, bson.ObjectId)
assert isinstance(other.bson_int64, bson.Int64)
assert isinstance(other.bson_dec128, bson.Decimal128)
assert isinstance(other.bson_binary, bson.Binary)
def test_post_generated_from_id() -> None:
class MyFactory(OdmanticModelFactory):
__model__ = MyModel
@post_generated
@classmethod
def name(cls, id: ObjectId) -> str:
return f"{cls.__faker__.name()} [{id.generation_time}]"
MyFactory.build()
|
