173 lines
5.3 KiB
Python
173 lines
5.3 KiB
Python
from datetime import datetime
|
|
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 PointerReferenceError, UniqueConstraintError
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
_db = GrungDB.with_schema(examples, 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 = examples.User(name="john", email="john@foo")
|
|
players = examples.Group(name="players", members=[user])
|
|
|
|
with pytest.raises(PointerReferenceError):
|
|
players = db.save(players)
|
|
|
|
user = db.save(user)
|
|
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
|
|
|
|
|
|
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()
|