grung-db/test/test_db.py

import tempfile
from datetime import datetime
from pathlib import Path
from pprint import pprint as print
from time import sleep

import pytest
from tinydb import Query
from tinydb.storages import MemoryStorage

from grung import examples
from grung.db import GrungDB
from grung.exceptions import (
    CircularReferenceError,
    InvalidFieldTypeError,
    InvalidLengthError,
    InvalidSizeError,
    PatternMatchError,
    UniqueConstraintError,
)


@pytest.fixture
def db():
    with tempfile.TemporaryDirectory() as path:
        _db = GrungDB.with_schema(examples, path=Path(path), storage=MemoryStorage)
        yield _db
        print(_db)


def test_crud(db):
    user = examples.User(name="john", number=23, email="john@foo")
    assert user._metadata.fields[user._metadata.primary_key].unique
    assert user._metadata.fields[user._metadata.primary_key].primary_key

    # insert
    john_something = db.save(user)
    last_insert_id = john_something.doc_id

    # read back
    assert db.User.get(doc_id=last_insert_id) == john_something
    assert john_something.name == user.name
    assert john_something.number == 23
    assert john_something.email == user.email

    # update
    john_something.name = "james?"
    before_update = db.User.get(doc_id=john_something.doc_id)
    after_update = db.save(john_something)
    assert after_update == john_something
    assert before_update != after_update

    # delete
    db.delete(john_something)
    assert len(db.User) == 0


def test_pointers(db):
    user = db.save(examples.User(name="john", email="john@foo"))
    players = db.save(examples.Group(name="players", members=[user]))

    user = db.table("User").get(doc_id=user.doc_id)
    assert user.groups.name == players.name

    assert players.members[0].groups.name == players.name


def test_subgroups(db):
    kirk = db.save(examples.User(name="James T. Kirk", email="riskybiznez@starfleet"))
    pike = db.save(examples.User(name="Christopher Pike", email="hitit@starfleet"))

    tos = db.save(examples.Group(name="The Original Series", members=[kirk]))
    snw = db.save(examples.Group(name="Strange New Worlds", members=[pike]))

    trek = db.save(examples.Group(name="trek", groups=[tos, snw]))
    tos = db.table("Group").get(doc_id=tos.doc_id)
    snw = db.table("Group").get(doc_id=snw.doc_id)
    assert tos in trek.groups
    assert snw in trek.groups

    assert trek.parent is None
    assert tos.parent.name == trek.name
    assert snw.parent.name == trek.name

    unique_users = set([user for group in trek.groups for user in group.members])

    kirk = db.table("User").get(doc_id=kirk.doc_id)
    assert kirk.reference in unique_users

    # recursion!
    with pytest.raises(CircularReferenceError):
        tos.groups = [tos]
        db.save(tos)


def test_unique(db):
    user1 = examples.User(name="john", email="john@foo")
    user2 = examples.User(name="john", email="john@foo")

    user1 = db.save(user1)
    with pytest.raises(UniqueConstraintError):
        user2 = db.save(user2)
    db.save(user1)


def test_search(db):
    # create crew members
    kirk = db.save(examples.User(name="Captain James T. Kirk", email="riskybiznez@starfleet"))
    bones = db.save(examples.User(name="Doctor McCoy", email="dammitjim@starfleet"))
    ricky = db.save(examples.User(name="Ensign Ricky Redshirt", email="invincible@starfleet"))

    # create the crew record
    crew = db.save(examples.Group(name="Crew", members=[kirk, bones, ricky]))

    User = Query()
    captains = db.User.search(User.name.matches("Captain"))
    assert len(captains) == 1

    # update the crew members so they have the backreference to crew
    kirk = db.table("User").get(doc_id=kirk.doc_id)
    bones = db.table("User").get(doc_id=bones.doc_id)
    ricky = db.table("User").get(doc_id=ricky.doc_id)

    assert kirk in crew.members
    assert bones in crew.members
    assert ricky in crew.members

    Group = Query()
    crew = db.Group.get(Group.name == "Crew", recurse=False)
    assert kirk.reference in crew.members


def test_password(db):
    user = db.save(examples.User(name="john", email="john@foo", password="fnord"))

    # make sure we don't compute the digest on an existing digest
    user = db.save(user)

    assert ":" in user.password
    assert user.password != "fnord"

    check = user._metadata.fields["password"].compare
    assert check("fnord", user.password)
    assert not check("wrong password", user.password)
    assert not check("", user.password)


def test_datetime(db):
    user = db.save(examples.User(name="john", email="john@foo", password="fnord", created=datetime.utcnow()))
    assert user.created > datetime.utcfromtimestamp(0)
    assert user.created < datetime.utcnow()
    assert user.last_updated == user.created

    sleep(1)
    user = db.save(user)
    assert user.last_updated >= user.created


def test_mapping(db):
    album = db.save(
        examples.Album(
            name="The Impossible Kid",
            credits={"Produced By": "Aesop Rock", "Lyrics By": "Aesop Rock", "Puke in the MeowMix By": "Kirby"},
            tracks=["Mystery Fish", "Rings", "Lotta Years", "Dorks"],
        )
    )
    assert album.credits["Produced By"] == "Aesop Rock"
    assert album.tracks[0] == "Mystery Fish"

    aes = db.save(
        examples.Artist(
            name="Aesop Rock",
            albums={"The Impossible Kid": album},
        )
    )

    album = db.Album.get(doc_id=album.doc_id)
    assert album.artist.uid == aes.uid
    assert album.name in aes.albums
    assert aes.albums[album.name].uid == album.uid
    assert "Kirby" in aes.albums[album.name].credits.values()


def test_file_pointers(db):
    album = db.save(
        examples.Album(
            name="The Impossible Kid",
            credits={"Produced By": "Aesop Rock", "Lyrics By": "Aesop Rock", "Puke in the MeowMix By": "Kirby"},
            tracks=["Mystery Fish", "Rings", "Lotta Years", "Dorks"],
            cover=b"some jpg data",
            review="10/10 no notes",
        )
    )
    assert album.cover == b"some jpg data"
    assert album.review == "10/10 no notes"

    db.save(
        examples.Artist(
            name="Aesop Rock",
            albums={"The Impossible Kid": album},
        )
    )

    album = db.Album.get(doc_id=album.doc_id)

    location_on_disk = db.path / album._metadata.fields["cover"].relpath(album)
    assert location_on_disk.read_bytes() == album.cover

    location_on_disk = db.path / album._metadata.fields["review"].relpath(album)
    assert location_on_disk.read_text() == album.review


@pytest.mark.parametrize(
    "updates, expected",
    [
        ({"name": ""}, InvalidLengthError),
        ({"name": "a name longer than 30 characters is what we have here"}, InvalidLengthError),
        ({"name": 23}, InvalidFieldTypeError),
        ({"number": -1}, InvalidSizeError),
        ({"number": 256}, InvalidSizeError),
        ({"email": "foo+alias@"}, PatternMatchError),
    ],
    ids=[
        "name too short",
        "name too long",
        "name is not a string",
        "number too small",
        "number too big",
        "invalid email addres",
    ],
)
def test_validators(updates, expected, db):
    user = db.save(examples.User(name="john", email="john@foo", password="fnord", created=datetime.utcnow()))
    with pytest.raises(expected):
        user.update(**updates)
        db.save(user)