221 lines
5 KiB
Python
221 lines
5 KiB
Python
import abc
|
|
import decimal
|
|
import datetime
|
|
from typing import Optional, List
|
|
|
|
import pytest
|
|
import redis
|
|
from pydantic import ValidationError
|
|
|
|
from redis_developer.orm import (
|
|
JsonModel,
|
|
Field,
|
|
)
|
|
|
|
r = redis.Redis()
|
|
today = datetime.datetime.today()
|
|
|
|
|
|
class BaseJsonModel(JsonModel, abc.ABC):
|
|
class Meta:
|
|
global_key_prefix = "redis-developer"
|
|
|
|
|
|
class Address(BaseJsonModel):
|
|
address_line_1: str
|
|
address_line_2: Optional[str]
|
|
city: str
|
|
country: str
|
|
postal_code: str
|
|
|
|
|
|
class Item(BaseJsonModel):
|
|
price: decimal.Decimal
|
|
name: str
|
|
|
|
|
|
class Order(BaseJsonModel):
|
|
items: List[Item]
|
|
total: decimal.Decimal
|
|
created_on: datetime.datetime
|
|
|
|
|
|
class Member(BaseJsonModel):
|
|
first_name: str
|
|
last_name: str
|
|
email: str = Field(index=True)
|
|
join_date: datetime.date
|
|
|
|
# Creates an embedded model.
|
|
address: Address
|
|
|
|
# Creates an embedded list of models.
|
|
orders: Optional[List[Order]]
|
|
|
|
class Meta(BaseJsonModel.Meta):
|
|
model_key_prefix = "member" # This is the default
|
|
|
|
|
|
address = Address(
|
|
address_line_1="1 Main St.",
|
|
city="Happy Town",
|
|
state="WY",
|
|
postal_code=11111,
|
|
country="USA"
|
|
)
|
|
|
|
|
|
def test_validates_required_fields():
|
|
# Raises ValidationError: last_name, address are required
|
|
with pytest.raises(ValidationError):
|
|
Member(
|
|
first_name="Andrew",
|
|
zipcode="97086",
|
|
join_date=today
|
|
)
|
|
|
|
|
|
def test_validates_field():
|
|
# Raises ValidationError: join_date is not a date
|
|
with pytest.raises(ValidationError):
|
|
Member(
|
|
first_name="Andrew",
|
|
last_name="Brookins",
|
|
join_date="yesterday"
|
|
)
|
|
|
|
|
|
# Passes validation
|
|
def test_validation_passes():
|
|
member = Member(
|
|
first_name="Andrew",
|
|
last_name="Brookins",
|
|
email="a@example.com",
|
|
address=address,
|
|
join_date=today
|
|
)
|
|
assert member.first_name == "Andrew"
|
|
|
|
|
|
def test_gets_pk():
|
|
new_address = Address(
|
|
address_line_1="1 Main St.",
|
|
city="Happy Town",
|
|
state="WY",
|
|
postal_code=11111,
|
|
country="USA"
|
|
)
|
|
assert new_address.pk is not None
|
|
|
|
|
|
def test_saves_model():
|
|
# Save a model instance to Redis
|
|
address.save()
|
|
address2 = Address.get(address.pk)
|
|
assert address2 == address
|
|
|
|
|
|
def test_saves_with_embedded_models():
|
|
member = Member(
|
|
first_name="Andrew",
|
|
last_name="Brookins",
|
|
email="a@example.com",
|
|
address=address,
|
|
join_date=datetime.date.today()
|
|
)
|
|
member.save()
|
|
|
|
member2 = Member.get(member.pk)
|
|
assert member2.address == address
|
|
|
|
|
|
def test_saves_with_deeply_embedded_models():
|
|
hat = Item(
|
|
name="Cool hat",
|
|
price=2.99
|
|
)
|
|
shoe = Item(
|
|
name="Expensive shoe",
|
|
price=299.99
|
|
)
|
|
order = Order(
|
|
total=302.98,
|
|
items=[hat, shoe],
|
|
created_on=today,
|
|
)
|
|
member = Member(
|
|
first_name="Andrew",
|
|
last_name="Brookins",
|
|
email="a@example.com",
|
|
address=address,
|
|
orders=[order],
|
|
join_date=today
|
|
)
|
|
member.save()
|
|
|
|
member2 = Member.get(member.pk)
|
|
assert member2.orders[0] == order
|
|
assert member2.orders[0].items[0] == hat
|
|
|
|
|
|
# Save many model instances to Redis
|
|
@pytest.mark.skip("Not implemented yet")
|
|
def test_saves_many():
|
|
members = [
|
|
Member(
|
|
first_name="Andrew",
|
|
last_name="Brookins",
|
|
email="a@example.com",
|
|
address=address,
|
|
join_date=today
|
|
),
|
|
Member(
|
|
first_name="Kim",
|
|
last_name="Brookins",
|
|
email="k@example.com",
|
|
address=address,
|
|
join_date=today
|
|
)
|
|
]
|
|
Member.add(members)
|
|
|
|
|
|
@pytest.mark.skip("No implemented yet")
|
|
def test_updates_a_model():
|
|
member = Member(
|
|
first_name="Andrew",
|
|
last_name="Brookins",
|
|
email="a@example.com",
|
|
address=address,
|
|
join_date=today
|
|
)
|
|
|
|
# Update a model instance in Redis
|
|
member.first_name = "Brian"
|
|
member.last_name = "Sam-Bodden"
|
|
member.save()
|
|
|
|
# Or, with an implicit save:
|
|
member.update(first_name="Brian", last_name="Sam-Bodden")
|
|
|
|
# Or, affecting multiple model instances with an implicit save:
|
|
Member.filter(Member.last_name == "Brookins").update(last_name="Sam-Bodden")
|
|
|
|
|
|
@pytest.mark.skip("Not implemented yet")
|
|
def test_exact_match_queries():
|
|
# TODO: Should get() support expressions? I.e., ...
|
|
# What if the field wasn't unique and there were two "a@example.com"
|
|
# entries? This would raise a MultipleObjectsReturned error:
|
|
member = Member.get(Member.email == "a.m.brookins@gmail.com")
|
|
|
|
# What if you know there might be multiple results? Use filter():
|
|
members = Member.filter(Member.last_name == "Brookins")
|
|
|
|
# What if you want to only return values that don't match a query?
|
|
members = Member.exclude(Member.last_name == "Brookins")
|
|
|
|
# You can combine filer() and exclude():
|
|
members = Member.filter(Member.last_name == "Brookins").exclude(
|
|
Member.first_name == "Andrew")
|