test_json_model.py

# type: ignore

importabc
importasyncio
importdataclasses
importdatetime
importdecimal
importuuid
fromcollectionsimportnamedtuple
fromtypingimportDict, List, Optional, Set, Union
fromunittestimportmock

importpytest
importpytest_asyncio
frommore_itertools.moreimportfirst

fromaredis_omimport (
EmbeddedJsonModel,
Field,
JsonModel,
Migrator,
NotFoundError,
QueryNotSupportedError,
RedisModel,
RedisModelError,
)

# We need to run this check as sync code (during tests) even in async mode
# because we call it in the top-level module scope.
fromredis_omimporthas_redis_json
fromtests._compatimportEmailStr, PositiveInt, ValidationError

from .conftestimportpy_test_mark_asyncio


ifnothas_redis_json():
pytestmark=pytest.mark.skip

today=datetime.date.today()


@pytest_asyncio.fixture
asyncdefm(key_prefix, redis):
classBaseJsonModel(JsonModel, abc.ABC):
classMeta:
global_key_prefix=key_prefix

classNote(EmbeddedJsonModel):
# TODO: This was going to be a full-text search example, but
# we can't index embedded documents for full-text search in
# the preview release.
description: str=Field(index=True)
created_on: datetime.datetime

classAddress(EmbeddedJsonModel):
address_line_1: str
address_line_2: Optional[str] =None
city: str=Field(index=True)
state: str
country: str
postal_code: str=Field(index=True)
note: Optional[Note] =None

classItem(EmbeddedJsonModel):
price: decimal.Decimal
name: str=Field(index=True)

classOrder(EmbeddedJsonModel):
items: List[Item]
created_on: datetime.datetime

classMember(BaseJsonModel):
first_name: str=Field(index=True, case_sensitive=True)
last_name: str=Field(index=True)
email: Optional[EmailStr] =Field(index=True, default=None)
join_date: datetime.date
age: Optional[PositiveInt] =Field(index=True, default=None)
bio: Optional[str] =Field(index=True, full_text_search=True, default="")

# Creates an embedded model.
address: Address

# Creates an embedded list of models.
orders: Optional[List[Order]] =None

awaitMigrator().run()

returnnamedtuple(
"Models", ["BaseJsonModel", "Note", "Address", "Item", "Order", "Member"]
 )(BaseJsonModel, Note, Address, Item, Order, Member)


@pytest.fixture()
defaddress(m):
try:
yieldm.Address(
address_line_1="1 Main St.",
city="Portland",
state="OR",
country="USA",
postal_code="11111",
 )
exceptExceptionase:
raisee


@pytest_asyncio.fixture()
asyncdefmembers(address, m):
member1=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
age=38,
join_date=today,
address=address,
 )

member2=m.Member(
first_name="Kim",
last_name="Brookins",
email="k@example.com",
age=34,
join_date=today,
address=address,
 )

member3=m.Member(
first_name="Andrew",
last_name="Smith",
email="as@example.com",
age=100,
join_date=today,
address=address,
 )

awaitmember1.save()
awaitmember2.save()
awaitmember3.save()

yieldmember1, member2, member3


@py_test_mark_asyncio
asyncdeftest_validate_bad_email(address, m):
# Raises ValidationError as email is malformed
withpytest.raises(ValidationError):
m.Member(
first_name="Andrew",
last_name="Brookins",
zipcode="97086",
join_date=today,
email="foobarbaz",
 )


@py_test_mark_asyncio
asyncdeftest_validate_bad_age(address, m):
# Raises ValidationError as email is malformed
withpytest.raises(ValidationError):
m.Member(
first_name="Andrew",
last_name="Brookins",
zipcode="97086",
join_date=today,
email="foo@bar.com",
address=address,
age=-5,
 )


@py_test_mark_asyncio
asyncdeftest_validates_required_fields(address, m):
# Raises ValidationError address is required
withpytest.raises(ValidationError):
m.Member(
first_name="Andrew",
last_name="Brookins",
zipcode="97086",
join_date=today,
 )


@py_test_mark_asyncio
asyncdeftest_validates_field(address, m):
# Raises ValidationError: join_date is not a date
withpytest.raises(ValidationError):
m.Member(
first_name="Andrew",
last_name="Brookins",
join_date="yesterday",
address=address,
 )


@py_test_mark_asyncio
asyncdeftest_validation_passes(address, m):
member=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
join_date=today,
age=38,
address=address,
 )
assertmember.first_name=="Andrew"


@py_test_mark_asyncio
asyncdeftest_saves_model_and_creates_pk(address, m, redis):
awaitMigrator().run()

member=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
join_date=today,
age=38,
address=address,
 )
# Save a model instance to Redis
awaitmember.save()

member2=awaitm.Member.get(member.pk)
assertmember2==member
assertmember2.address==address


@py_test_mark_asyncio
asyncdeftest_all_pks(address, m, redis):
member=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
join_date=today,
age=38,
address=address,
 )

awaitmember.save()

member1=m.Member(
first_name="Simon",
last_name="Prickett",
email="s@example.com",
join_date=today,
age=99,
address=address,
 )

awaitmember1.save()

pk_list= []
asyncforpkinawaitm.Member.all_pks():
pk_list.append(pk)

assertsorted(pk_list) ==sorted([member.pk, member1.pk])


@py_test_mark_asyncio
asyncdeftest_all_pks_with_complex_pks(key_prefix):
classCity(JsonModel):
name: str

classMeta:
global_key_prefix=key_prefix
model_key_prefix="city"

city1=City(
pk="ca:on:toronto",
name="Toronto",
 )

awaitcity1.save()

city2=City(
pk="ca:qc:montreal",
name="Montreal",
 )

awaitcity2.save()

pk_list= []
asyncforpkinawaitCity.all_pks():
pk_list.append(pk)

assertsorted(pk_list) == ["ca:on:toronto", "ca:qc:montreal"]


@py_test_mark_asyncio
asyncdeftest_delete(address, m, redis):
member=m.Member(
first_name="Simon",
last_name="Prickett",
email="s@example.com",
join_date=today,
age=38,
address=address,
 )

awaitmember.save()
response=awaitm.Member.delete(member.pk)
assertresponse==1


@py_test_mark_asyncio
asyncdeftest_saves_many_implicit_pipeline(address, m):
member1=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
join_date=today,
address=address,
age=38,
 )
member2=m.Member(
first_name="Kim",
last_name="Brookins",
email="k@example.com",
join_date=today,
address=address,
age=34,
 )
members= [member1, member2]
result=awaitm.Member.add(members)
assertresult== [member1, member2]

assertawaitm.Member.get(pk=member1.pk) ==member1
assertawaitm.Member.get(pk=member2.pk) ==member2


@py_test_mark_asyncio
asyncdeftest_saves_many_explicit_transaction(address, m):
member1=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
join_date=today,
address=address,
age=38,
 )
member2=m.Member(
first_name="Kim",
last_name="Brookins",
email="k@example.com",
join_date=today,
address=address,
age=34,
 )
members= [member1, member2]
result=awaitm.Member.add(members)
assertresult== [member1, member2]

assertawaitm.Member.get(pk=member1.pk) ==member1
assertawaitm.Member.get(pk=member2.pk) ==member2

# Test the explicit pipeline path -- here, we add multiple Members
# using a single Redis transaction, with MULTI/EXEC.
asyncwithm.Member.db().pipeline(transaction=True) aspipeline:
awaitm.Member.add(members, pipeline=pipeline)
assertresult== [member1, member2]
assertawaitpipeline.execute() == [True, True]

assertawaitm.Member.get(pk=member1.pk) ==member1
assertawaitm.Member.get(pk=member2.pk) ==member2


@py_test_mark_asyncio
asyncdeftest_delete_many_implicit_pipeline(address, m):
member1=m.Member(
first_name="Andrew",
last_name="Brookins",
email="a@example.com",
join_date=today,
address=address,
age=38,
 )
member2=m.Member(
first_name="Kim",
last_name="Brookins",
email="k@example.com",
join_date=today,
address=address,
age=34,
 )
members= [member1, member2]
result=awaitm.Member.add(members)
assertresult== [member1, member2]
result=awaitm.Member.delete_many(members)
assertresult==2
withpytest.raises(NotFoundError):
awaitm.Member.get(pk=member2.pk)


asyncdefsave(members):
forminmembers:
awaitm.save()
returnmembers


@py_test_mark_asyncio
asyncdeftest_updates_a_model(members, m):
member1, member2, member3=awaitsave(members)

# Update a field directly on the model
awaitmember1.update(last_name="Apples to oranges")
member=awaitm.Member.get(member1.pk)
assertmember.last_name=="Apples to oranges"

# Update a field in an embedded model
awaitmember2.update(address__city="Happy Valley")
member=awaitm.Member.get(member2.pk)
assertmember.address.city=="Happy Valley"


@py_test_mark_asyncio
asyncdeftest_paginate_query(members, m):
member1, member2, member3=members
actual=awaitm.Member.find().sort_by("age").all(batch_size=1)
assertactual== [member2, member1, member3]


@py_test_mark_asyncio
asyncdeftest_access_result_by_index_cached(members, m):
member1, member2, member3=members
query=m.Member.find().sort_by("age")
# Load the cache, throw away the result.
assertquery._model_cache== []
awaitquery.execute()
assertquery._model_cache== [member2, member1, member3]

# Access an item that should be in the cache.
withmock.patch.object(query.model, "db") asmock_db:
assertawaitquery.get_item(0) ==member2
assertnotmock_db.called


@py_test_mark_asyncio
asyncdeftest_access_result_by_index_not_cached(members, m):
member1, member2, member3=members
query=m.Member.find().sort_by("age")

# Assert that we don't have any models in the cache yet -- we
# haven't made any requests of Redis.
assertquery._model_cache== []
assertawaitquery.get_item(0) ==member2
assertawaitquery.get_item(1) ==member1
assertawaitquery.get_item(2) ==member3


@py_test_mark_asyncio
asyncdeftest_in_query(members, m):
member1, member2, member3=members
actual=await (
m.Member.find(m.Member.pk<< [member1.pk, member2.pk, member3.pk])
 .sort_by("age")
 .all()
 )
assertactual== [member2, member1, member3]


@py_test_mark_asyncio
asyncdeftest_not_in_query(members, m):
member1, member2, member3=members
actual=await (
m.Member.find(m.Member.pk>> [member2.pk, member3.pk]).sort_by("age").all()
 )
assertactual== [member1]


@py_test_mark_asyncio
asyncdeftest_update_query(members, m):
member1, member2, member3=members
awaitm.Member.find(m.Member.pk<< [member1.pk, member2.pk, member3.pk]).update(
first_name="Bobby"
 )
actual=await (
m.Member.find(m.Member.pk<< [member1.pk, member2.pk, member3.pk])
 .sort_by("age")
 .all()
 )
assertlen(actual) ==3
assertall([m.first_name=="Bobby"forminactual])


@py_test_mark_asyncio
asyncdeftest_exact_match_queries(members, m):
member1, member2, member3=members

actual=awaitm.Member.find(m.Member.last_name=="Brookins").sort_by("age").all()
assertactual== [member2, member1]

actual=awaitm.Member.find(
 (m.Member.last_name=="Brookins") &~(m.Member.first_name=="Andrew")
 ).all()
assertactual== [member2]

actual=awaitm.Member.find(~(m.Member.last_name=="Brookins")).all()
assertactual== [member3]

actual=awaitm.Member.find(m.Member.last_name!="Brookins").all()
assertactual== [member3]

actual=await (
m.Member.find(
 (m.Member.last_name=="Brookins") & (m.Member.first_name=="Andrew")
| (m.Member.first_name=="Kim")
 )
 .sort_by("age")
 .all()
 )
assertactual== [member2, member1]

actual=awaitm.Member.find(
m.Member.first_name=="Kim", m.Member.last_name=="Brookins"
 ).all()
assertactual== [member2]

actual= (
awaitm.Member.find(m.Member.address.city=="Portland").sort_by("age").all()
 )
assertactual== [member2, member1, member3]


@py_test_mark_asyncio
asyncdeftest_recursive_query_expression_resolution(members, m):
member1, member2, member3=members

actual=await (
m.Member.find(
 (m.Member.last_name=="Brookins")
| (m.Member.age==100) & (m.Member.last_name=="Smith")
 )
 .sort_by("age")
 .all()
 )
assertactual== [member2, member1, member3]


@py_test_mark_asyncio
asyncdeftest_recursive_query_field_resolution(members, m):
member1, _, _=members
member1.address.note=m.Note(
description="Weird house", created_on=datetime.datetime.now()
 )
awaitmember1.save()
actual=awaitm.Member.find(
m.Member.address.note.description=="Weird house"
 ).all()
assertactual== [member1]

member1.orders= [
m.Order(
items=[m.Item(price=10.99, name="Ball")],
total=10.99,
created_on=datetime.datetime.now(),
 )
 ]
awaitmember1.save()
actual=awaitm.Member.find(m.Member.orders.items.name=="Ball").all()
assertactual== [member1]
assertactual[0].orders[0].items[0].name=="Ball"


@py_test_mark_asyncio
asyncdeftest_full_text_search(members, m):
member1, member2, _=members
awaitmember1.update(bio="Hates sunsets, likes beaches")
awaitmember2.update(bio="Hates beaches, likes forests")

actual=awaitm.Member.find(m.Member.bio%"beaches").sort_by("age").all()
assertactual== [member2, member1]

actual=awaitm.Member.find(m.Member.bio%"forests").all()
assertactual== [member2]


@py_test_mark_asyncio
asyncdeftest_tag_queries_boolean_logic(members, m):
member1, member2, member3=members

actual= (
awaitm.Member.find(
 (m.Member.first_name=="Andrew") & (m.Member.last_name=="Brookins")
| (m.Member.last_name=="Smith")
 )
 .sort_by("age")
 .all()
 )
assertactual== [member1, member3]


@py_test_mark_asyncio
asyncdeftest_tag_queries_punctuation(address, m):
member1=m.Member(
first_name="Andrew, the Michael",
last_name="St. Brookins-on-Pier",
email="a|b@example.com", # NOTE: This string uses the TAG field separator.
age=38,
join_date=today,
address=address,
 )
awaitmember1.save()

member2=m.Member(
first_name="Bob",
last_name="the Villain",
email="a|villain@example.com", # NOTE: This string uses the TAG field separator.
age=38,
join_date=today,
address=address,
 )
awaitmember2.save()

assert (
awaitm.Member.find(m.Member.first_name=="Andrew, the Michael").first()
==member1
 )
assert (
awaitm.Member.find(m.Member.last_name=="St. Brookins-on-Pier").first()
==member1
 )

# Notice that when we index and query multiple values that use the internal
# TAG separator for single-value exact-match fields, like an indexed string,
# the queries will succeed. We apply a workaround that queries for the union
# of the two values separated by the tag separator.
assertawaitm.Member.find(m.Member.email=="a|b@example.com").all() == [member1]
assertawaitm.Member.find(m.Member.email=="a|villain@example.com").all() == [
member2
 ]


@py_test_mark_asyncio
asyncdeftest_tag_queries_negation(members, m):
member1, member2, member3=members

"""
 ┌first_name
 NOT EQ┤
 └Andrew

 """
query=m.Member.find(~(m.Member.first_name=="Andrew"))
assertawaitquery.all() == [member2]

"""
 ┌first_name
 ┌NOT EQ┤
 | └Andrew
 AND┤
 | ┌last_name
 └EQ┤
 └Brookins

 """
query=m.Member.find(
~(m.Member.first_name=="Andrew") & (m.Member.last_name=="Brookins")
 )
assertawaitquery.all() == [member2]

"""
 ┌first_name
 ┌NOT EQ┤
 | └Andrew
 AND┤
 | ┌last_name
 | ┌EQ┤
 | | └Brookins
 └OR┤
 | ┌last_name
 └EQ┤
 └Smith
 """
query=m.Member.find(
~(m.Member.first_name=="Andrew")
& ((m.Member.last_name=="Brookins") | (m.Member.last_name=="Smith"))
 )
assertawaitquery.all() == [member2]

"""
 ┌first_name
 ┌NOT EQ┤
 | └Andrew
 ┌AND┤
 | | ┌last_name
 | └EQ┤
 | └Brookins
 OR┤
 | ┌last_name
 └EQ┤
 └Smith
 """
query=m.Member.find(
~(m.Member.first_name=="Andrew") & (m.Member.last_name=="Brookins")
| (m.Member.last_name=="Smith")
 )
assertawaitquery.sort_by("age").all() == [member2, member3]

actual=awaitm.Member.find(
 (m.Member.first_name=="Andrew") &~(m.Member.last_name=="Brookins")
 ).all()
assertactual== [member3]


@py_test_mark_asyncio
asyncdeftest_numeric_queries(members, m):
member1, member2, member3=members

actual=awaitm.Member.find(m.Member.age==34).all()
assertactual== [member2]

actual=awaitm.Member.find(m.Member.age>34).sort_by("age").all()
assertactual== [member1, member3]

actual=awaitm.Member.find(m.Member.age<35).all()
assertactual== [member2]

actual=awaitm.Member.find(m.Member.age<=34).all()
assertactual== [member2]

actual=awaitm.Member.find(m.Member.age>=100).all()
assertactual== [member3]

actual=awaitm.Member.find(~(m.Member.age==100)).sort_by("age").all()
assertactual== [member2, member1]

actual= (
awaitm.Member.find(m.Member.age>30, m.Member.age<40).sort_by("age").all()
 )
assertactual== [member2, member1]

actual=awaitm.Member.find(m.Member.age!=34).sort_by("age").all()
assertactual== [member1, member3]


@py_test_mark_asyncio
asyncdeftest_sorting(members, m):
member1, member2, member3=members

actual=awaitm.Member.find(m.Member.age>34).sort_by("age").all()
assertactual== [member1, member3]

actual=awaitm.Member.find(m.Member.age>34).sort_by("-age").all()
assertactual== [member3, member1]

withpytest.raises(QueryNotSupportedError):
# This field does not exist.
awaitm.Member.find().sort_by("not-a-real-field").all()

withpytest.raises(QueryNotSupportedError):
# This field is not sortable.
awaitm.Member.find().sort_by("join_date").all()


@py_test_mark_asyncio
asyncdeftest_case_sensitive(members, m):
member1, member2, member3=members

actual=awaitm.Member.find(m.Member.first_name=="Andrew").all()
assertactual== [member1, member3]

actual=awaitm.Member.find(m.Member.first_name=="andrew").all()
assertactual== []


@py_test_mark_asyncio
asyncdeftest_not_found(m):
withpytest.raises(NotFoundError):
# This ID does not exist.
awaitm.Member.get(1000)


@py_test_mark_asyncio
asyncdeftest_list_field_limitations(m, redis):
withpytest.raises(RedisModelError):

classSortableTarotWitch(m.BaseJsonModel):
# We support indexing lists of strings for quality and membership
# queries. Sorting is not supported, but is planned.
tarot_cards: List[str] =Field(index=True, sortable=True)

withpytest.raises(RedisModelError):

classSortableFullTextSearchAlchemicalWitch(m.BaseJsonModel):
# We don't support indexing a list of strings for full-text search
# queries. Support for this feature is not planned.
potions: List[str] =Field(index=True, full_text_search=True)

withpytest.raises(RedisModelError):

classNumerologyWitch(m.BaseJsonModel):
# We don't support indexing a list of numbers. Support for this
# feature is To Be Determined.
lucky_numbers: List[int] =Field(index=True)

withpytest.raises(RedisModelError):

classReadingWithPrice(EmbeddedJsonModel):
gold_coins_charged: int=Field(index=True)

classTarotWitchWhoCharges(m.BaseJsonModel):
tarot_cards: List[str] =Field(index=True)

# The preview release does not support indexing numeric fields on models
# found within a list or tuple. This is the same limitation that stops
# us from indexing plain lists (or tuples) containing numeric values.
# The fate of this feature is To Be Determined.
readings: List[ReadingWithPrice]

classTarotWitch(m.BaseJsonModel):
# We support indexing lists of strings for quality and membership
# queries. Sorting is not supported, but is planned.
tarot_cards: List[str] =Field(index=True)

# We need to import and run this manually because we defined
# our model classes within a function that runs after the test
# suite's migrator has already looked for migrations to run.
awaitMigrator().run()

witch=TarotWitch(tarot_cards=["death"])
awaitwitch.save()
actual=awaitTarotWitch.find(TarotWitch.tarot_cards<<"death").all()
assertactual== [witch]


@py_test_mark_asyncio
asyncdeftest_allows_dataclasses(m):
@dataclasses.dataclass
classAddress:
address_line_1: str

classValidMember(m.BaseJsonModel):
address: Address

address=Address(address_line_1="hey")
member=ValidMember(address=address)
awaitmember.save()

member2=awaitValidMember.get(member.pk)
assertmember2==member
assertmember2.address.address_line_1=="hey"


@py_test_mark_asyncio
asyncdeftest_allows_and_serializes_dicts(m):
classValidMember(m.BaseJsonModel):
address: Dict[str, str]

member=ValidMember(address={"address_line_1": "hey"})
awaitmember.save()

member2=awaitValidMember.get(member.pk)
assertmember2==member
assertmember2.address["address_line_1"] =="hey"


@py_test_mark_asyncio
asyncdeftest_allows_and_serializes_sets(m):
classValidMember(m.BaseJsonModel):
friend_ids: Set[int]

member=ValidMember(friend_ids={1, 2})
awaitmember.save()

member2=awaitValidMember.get(member.pk)
assertmember2==member
assertmember2.friend_ids== {1, 2}


@py_test_mark_asyncio
asyncdeftest_allows_and_serializes_lists(m):
classValidMember(m.BaseJsonModel):
friend_ids: List[int]

member=ValidMember(friend_ids=[1, 2])
awaitmember.save()

member2=awaitValidMember.get(member.pk)
assertmember2==member
assertmember2.friend_ids== [1, 2]


@py_test_mark_asyncio
asyncdeftest_schema(m, key_prefix):
# We need to build the key prefix because it will differ based on whether
# these tests were copied into the tests_sync folder and unasynce'd.
key_prefix=m.Member.make_key(m.Member._meta.primary_key_pattern.format(pk=""))
assertm.Member.redisearch_schema() == (
f"ON JSON PREFIX 1 {key_prefix} SCHEMA "
"$.pk AS pk TAG SEPARATOR | "
"$.first_name AS first_name TAG SEPARATOR | CASESENSITIVE "
"$.last_name AS last_name TAG SEPARATOR | "
"$.email AS email TAG SEPARATOR | "
"$.age AS age NUMERIC "
"$.bio AS bio TAG SEPARATOR | "
"$.bio AS bio_fts TEXT "
"$.address.pk AS address_pk TAG SEPARATOR | "
"$.address.city AS address_city TAG SEPARATOR | "
"$.address.postal_code AS address_postal_code TAG SEPARATOR | "
"$.address.note.pk AS address_note_pk TAG SEPARATOR | "
"$.address.note.description AS address_note_description TAG SEPARATOR | "
"$.orders[*].pk AS orders_pk TAG SEPARATOR | "
"$.orders[*].items[*].pk AS orders_items_pk TAG SEPARATOR | "
"$.orders[*].items[*].name AS orders_items_name TAG SEPARATOR |"
 )


@py_test_mark_asyncio
asyncdeftest_count(members, m):
# member1, member2, member3 = members
actual_count=awaitm.Member.find(
 (m.Member.first_name=="Andrew") & (m.Member.last_name=="Brookins")
| (m.Member.last_name=="Smith")
 ).count()
assertactual_count==2

actual_count=awaitm.Member.find(
m.Member.first_name=="Kim", m.Member.last_name=="Brookins"
 ).count()
assertactual_count==1


@py_test_mark_asyncio
asyncdeftest_type_with_union(members, m):
classTypeWithUnion(m.BaseJsonModel):
field: Union[str, int]

twu_str=TypeWithUnion(field="hello world")
res=awaittwu_str.save()
assertres.pk==twu_str.pk
twu_str_rematerialized=awaitTypeWithUnion.get(twu_str.pk)
assert (
isinstance(twu_str_rematerialized.field, str)
andtwu_str_rematerialized.pk==twu_str.pk
 )

twu_int=TypeWithUnion(field=42)
awaittwu_int.save()
twu_int_rematerialized=awaitTypeWithUnion.get(twu_int.pk)
assert (
isinstance(twu_int_rematerialized.field, int)
andtwu_int_rematerialized.pk==twu_int.pk
 )


@py_test_mark_asyncio
asyncdeftest_type_with_uuid():
classTypeWithUuid(JsonModel):
uuid: uuid.UUID

item=TypeWithUuid(uuid=uuid.uuid4())

awaititem.save()


@py_test_mark_asyncio
asyncdeftest_xfix_queries(m):
awaitm.Member(
first_name="Steve",
last_name="Lorello",
email="s@example.com",
join_date=today,
bio="Steve is a two-bit hacker who loves Redis.",
address=m.Address(
address_line_1="42 foo bar lane",
city="Satellite Beach",
state="FL",
country="USA",
postal_code="32999",
 ),
age=34,
 ).save()

result=awaitm.Member.find(
m.Member.first_name.startswith("Ste") andm.Member.first_name=="Steve"
 ).first()
assertresult.first_name=="Steve"

result=awaitm.Member.find(
m.Member.last_name.endswith("llo") andm.Member.first_name=="Steve"
 ).first()
assertresult.first_name=="Steve"

result=awaitm.Member.find(
m.Member.address.city.contains("llite") andm.Member.first_name=="Steve"
 ).first()
assertresult.first_name=="Steve"

result=awaitm.Member.find(
m.Member.bio%"tw*"andm.Member.first_name=="Steve"
 ).first()
assertresult.first_name=="Steve"

result=awaitm.Member.find(
m.Member.bio%"*cker"andm.Member.first_name=="Steve"
 ).first()
assertresult.first_name=="Steve"

result=awaitm.Member.find(
m.Member.bio%"*ack*"andm.Member.first_name=="Steve"