redis-om-python/tests/test_json_model.py

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()


class BaseJsonModel(JsonModel):
    class Meta(JsonModel.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 Order(BaseJsonModel):
    total: decimal.Decimal
    currency: str
    created_on: datetime.datetime


class Member(BaseJsonModel):
    first_name: str
    last_name: str
    email: str = Field(unique=True, index=True)
    join_date: datetime.date

    # Creates an embedded model: stored as hash fields or JSON document.
    address: Address

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

    class Meta(BaseJsonModel.Meta):
        model_key_prefix = "member"
        primary_key_pattern = ""


def test_validates_required_fields():
    # Raises ValidationError: last_name, address are required
    with pytest.raises(ValidationError):
        Member(
            first_name="Andrew",
            zipcode="97086",
            join_date=datetime.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():
    address = Address(
        address_line_1="1 Main St.",
        city="Happy Town",
        state="WY",
        postal_code=11111,
        country="USA"
    )
    member = Member(
        first_name="Andrew",
        last_name="Brookins",
        email="a@example.com",
        address=address,
        join_date=datetime.date.today()
    )
    assert member.first_name == "Andrew"


def test_gets_pk():
    address = Address(
        address_line_1="1 Main St.",
        city="Happy Town",
        state="WY",
        postal_code=11111,
        country="USA"
    )
    assert address.pk is not None


def test_saves_model():
    address = Address(
        address_line_1="1 Main St.",
        city="Happy Town",
        state="WY",
        postal_code=11111,
        country="USA"
    )
    # Save a model instance to Redis
    address.save()

    address2 = Address.get(address.pk)
    assert address2 == address


# Saves a model with embedded models
def test_saves_with_relationships():
    address = Address(
        address_line_1="1 Main St.",
        city="Happy Town",
        state="WY",
        postal_code=11111,
        country="USA"
    )
    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


# Save many model instances to Redis
@pytest.mark.skip("Not implemented yet")
def test_saves_many():
    address = Address(
        address_line_1="1 Main St.",
        city="Happy Town",
        state="WY",
        postal_code=11111,
        country="USA"
    )
    today = datetime.date.today()
    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():
    address = Address(
        address_line_1="1 Main St.",
        city="Happy Town",
        state="WY",
        postal_code=11111,
        country="USA"
    )
    today = datetime.date.today()
    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():
    # 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")