Compare commits

...

4 Commits

12 changed files with 156 additions and 60 deletions

View File

@ -38,7 +38,10 @@ def create(
raise ArgumentError('Lexicon prompt must be a string') raise ArgumentError('Lexicon prompt must be a string')
# Query the db to make sure the lexicon name isn't taken # Query the db to make sure the lexicon name isn't taken
if db.session.query(func.count(Lexicon.id)).filter(Lexicon.name == name).scalar() > 0: if db(
select(func.count(Lexicon.id))
.where(Lexicon.name == name)
).scalar() > 0:
raise ArgumentError('Lexicon name is already taken') raise ArgumentError('Lexicon name is already taken')
new_lexicon = Lexicon( new_lexicon = Lexicon(

View File

@ -2,6 +2,8 @@
Membership query interface Membership query interface
""" """
from sqlalchemy import select, func
from amanuensis.db import DbContext, Membership from amanuensis.db import DbContext, Membership
from amanuensis.errors import ArgumentError from amanuensis.errors import ArgumentError
@ -14,7 +16,7 @@ def create(
""" """
Create a new user membership in a lexicon. Create a new user membership in a lexicon.
""" """
# Quick argument verification # Verify argument types are correct
if not isinstance(user_id, int): if not isinstance(user_id, int):
raise ArgumentError('user_id') raise ArgumentError('user_id')
if not isinstance(lexicon_id, int): if not isinstance(lexicon_id, int):
@ -22,6 +24,14 @@ def create(
if not isinstance(is_editor, bool): if not isinstance(is_editor, bool):
raise ArgumentError('is_editor') raise ArgumentError('is_editor')
# Verify user has not already joined lexicon
if db(
select(func.count(Membership.id))
.where(Membership.user_id == user_id)
.where(Membership.lexicon_id == lexicon_id)
).scalar() > 0:
raise ArgumentError('User is already a member of lexicon')
new_membership = Membership( new_membership = Membership(
user_id=user_id, user_id=user_id,
lexicon_id=lexicon_id, lexicon_id=lexicon_id,

View File

@ -51,7 +51,10 @@ def create(
raise ArgumentError('Email must be a string') raise ArgumentError('Email must be a string')
# Query the db to make sure the username isn't taken # Query the db to make sure the username isn't taken
if db.session.query(func.count(User.id)).filter(User.username == username).scalar() > 0: if db(
select(func.count(User.id))
.where(User.username == username)
).scalar() > 0:
raise ArgumentError('Username is already taken') raise ArgumentError('Username is already taken')
new_user = User( new_user = User(

View File

@ -12,4 +12,20 @@ from .models import (
ArticleContentRuleType, ArticleContentRuleType,
ArticleContentRule, ArticleContentRule,
Post, Post,
) )
__all__ = [
'DbContext',
'User',
'Lexicon',
'Membership',
'Character',
'ArticleState',
'Article',
'IndexType',
'ArticleIndex',
'ArticleIndexRule',
'ArticleContentRuleType',
'ArticleContentRule',
'Post',
]

View File

@ -33,5 +33,10 @@ class DbContext():
# Create a thread-safe session factory # Create a thread-safe session factory
self.session = scoped_session(sessionmaker(bind=self.engine)) self.session = scoped_session(sessionmaker(bind=self.engine))
def __call__(self, *args, **kwargs):
"""Provides shortcut access to session.execute."""
return self.session.execute(*args, **kwargs)
def create_all(self): def create_all(self):
"""Initializes the database schema."""
ModelBase.metadata.create_all(self.engine) ModelBase.metadata.create_all(self.engine)

View File

@ -2,6 +2,8 @@
Data model SQL definitions Data model SQL definitions
""" """
import enum import enum
import uuid
from sqlalchemy import ( from sqlalchemy import (
Boolean, Boolean,
Column, Column,
@ -17,7 +19,7 @@ from sqlalchemy import (
TypeDecorator, TypeDecorator,
) )
from sqlalchemy.orm import relationship, backref from sqlalchemy.orm import relationship, backref
import uuid from sqlalchemy.sql.schema import UniqueConstraint
from .database import ModelBase from .database import ModelBase
@ -233,6 +235,9 @@ class Membership(ModelBase):
Represents a user's participation in a Lexicon game. Represents a user's participation in a Lexicon game.
""" """
__tablename__ = 'membership' __tablename__ = 'membership'
__table_args__ = (
UniqueConstraint('user_id', 'lexicon_id'),
)
################### ###################
# Membership keys # # Membership keys #

73
tests/conftest.py Normal file
View File

@ -0,0 +1,73 @@
"""
pytest test fixtures
"""
import pytest
from amanuensis.db import DbContext
import amanuensis.backend.lexicon as lexiq
import amanuensis.backend.membership as memq
import amanuensis.backend.user as userq
@pytest.fixture
def db():
"""Provides an initialized database in memory."""
db = DbContext('sqlite:///:memory:', debug=True)
db.create_all()
return db
@pytest.fixture
def make_user(db: DbContext):
"""Provides a factory function for creating users, with valid default values."""
def user_factory(state={'nonce': 1}, **kwargs):
default_kwargs = {
'username': f'test_user_{state["nonce"]}',
'password': 'password',
'display_name': None,
'email': 'user@example.com',
'is_site_admin': False,
}
state['nonce'] += 1
updated_kwargs = {**default_kwargs, **kwargs}
return userq.create(db, **updated_kwargs)
return user_factory
@pytest.fixture
def make_lexicon(db: DbContext):
"""Provides a factory function for creating lexicons, with valid default values."""
def lexicon_factory(state={'nonce': 1}, **kwargs):
default_kwargs = {
'name': f'Test_{state["nonce"]}',
'title': None,
'prompt': f'Test Lexicon game {state["nonce"]}'
}
state['nonce'] += 1
updated_kwargs = {**default_kwargs, **kwargs}
return lexiq.create(db, **updated_kwargs)
return lexicon_factory
@pytest.fixture
def make_membership(db: DbContext):
"""Provides a factory function for creating memberships, with valid default values."""
def membership_factory(**kwargs):
default_kwargs = {
'is_editor': False,
}
updated_kwargs = {**default_kwargs, **kwargs}
return memq.create(db, **updated_kwargs)
return membership_factory
@pytest.fixture
def lexicon_with_editor(make_user, make_lexicon, make_membership):
"""Shortcut setup for a lexicon game with an editor."""
editor = make_user()
assert editor
lexicon = make_lexicon()
assert lexicon
membership = make_membership(user_id=editor.id, lexicon_id=lexicon.id, is_editor=True)
assert membership
return (lexicon, editor)

View File

@ -1,17 +1,9 @@
import pytest
from sqlalchemy import func from sqlalchemy import func
from amanuensis.db import * from amanuensis.db import *
@pytest.fixture def test_create_db(db: DbContext):
def db():
db = DbContext('sqlite:///:memory:', debug=True)
db.create_all()
return db
def test_create(db):
"""Simple test that the database creates fine from scratch.""" """Simple test that the database creates fine from scratch."""
assert db.session.query(func.count(User.id)).scalar() == 0 assert db.session.query(func.count(User.id)).scalar() == 0
assert db.session.query(func.count(Lexicon.id)).scalar() == 0 assert db.session.query(func.count(Lexicon.id)).scalar() == 0

View File

@ -2,14 +2,12 @@ import datetime
import pytest import pytest
from amanuensis.db import * from amanuensis.db import DbContext
import amanuensis.backend.lexicon as lexiq import amanuensis.backend.lexicon as lexiq
from amanuensis.errors import ArgumentError from amanuensis.errors import ArgumentError
from .test_db import db
def test_create_lexicon(db: DbContext):
def test_create_lexicon(db):
"""Test new game creation.""" """Test new game creation."""
kwargs = { kwargs = {
'name': 'Test', 'name': 'Test',
@ -43,3 +41,4 @@ def test_create_lexicon(db):
# No duplicate lexicon names # No duplicate lexicon names
with pytest.raises(ArgumentError): with pytest.raises(ArgumentError):
duplicate = lexiq.create(db, **kwargs) duplicate = lexiq.create(db, **kwargs)
assert duplicate

View File

@ -1,17 +1,18 @@
import pytest
from sqlalchemy import select
from amanuensis.db import * from amanuensis.db import *
import amanuensis.backend.lexicon as lexiq from amanuensis.errors import ArgumentError
import amanuensis.backend.membership as memq import amanuensis.backend.membership as memq
import amanuensis.backend.user as userq
from .test_db import db
def test_create(db: DbContext): def test_create_membership(db: DbContext, make_user, make_lexicon):
"""Test joining a game.""" """Test joining a game."""
# Set up a user and a lexicon # Set up a user and a lexicon
new_user = userq.create(db, 'username', 'password', 'user', 'a@b.c', False) new_user = make_user()
assert new_user.id, 'Failed to create user' assert new_user.id, 'Failed to create user'
new_lexicon = lexiq.create(db, 'Test', None, 'prompt') new_lexicon = make_lexicon()
assert new_lexicon.id, 'Failed to create lexicon' assert new_lexicon.id, 'Failed to create lexicon'
# Add the user to the lexicon as an editor # Add the user to the lexicon as an editor
@ -19,10 +20,22 @@ def test_create(db: DbContext):
assert mem, 'Failed to create membership' assert mem, 'Failed to create membership'
# Check that the user and lexicon are mutually visible in the ORM relationships # Check that the user and lexicon are mutually visible in the ORM relationships
assert new_user.memberships, 'User memberships not updated' assert any(map(lambda mem: mem.lexicon == new_lexicon, new_user.memberships))
assert new_lexicon.memberships, 'Lexicon memberships not updated' assert any(map(lambda mem: mem.user == new_user, new_lexicon.memberships))
assert new_user.memberships[0].lexicon_id == new_lexicon.id
assert new_lexicon.memberships[0].user_id == new_user.id
# Check that the editor flag was set properly # Check that the editor flag was set properly
assert new_lexicon.memberships editor = db(
select(User)
.join(User.memberships)
.join(Membership.lexicon)
.where(Lexicon.id == new_lexicon.id)
.where(Membership.is_editor == True)
).scalar_one()
assert editor is not None
assert isinstance(editor, User)
assert editor.id == new_user.id
# Check that joining twice is not allowed
with pytest.raises(ArgumentError):
mem2 = memq.create(db, new_user.id, new_lexicon.id, False)
assert mem2

View File

@ -1,32 +1,19 @@
import pytest import pytest
from amanuensis.db import * from amanuensis.db import DbContext
import amanuensis.backend.lexicon as lexiq
import amanuensis.backend.user as userq
import amanuensis.backend.post as postq import amanuensis.backend.post as postq
import amanuensis.backend.membership as memq
from amanuensis.errors import ArgumentError from amanuensis.errors import ArgumentError
from .test_db import db
def test_create_post(db): def test_create_post(db: DbContext, lexicon_with_editor):
"""Test new post creation""" """Test new post creation"""
lexicon, editor = lexicon_with_editor
# Make user and lexicon
new_user = userq.create(db, 'username', 'password', 'user', 'a@b.c', False)
assert new_user.id, 'Failed to create user'
new_lexicon = lexiq.create(db, 'Test', None, 'prompt')
assert new_lexicon.id, 'Failed to create lexicon'
# Add the user to the lexicon as an editor
mem = memq.create(db, new_user.id, new_lexicon.id, True)
assert mem, 'Failed to create membership'
# argument dictionary for post object # argument dictionary for post object
kwargs = { kwargs = {
'lexicon_id': new_lexicon.id, 'lexicon_id': lexicon.id,
'user_id': new_user.id, 'user_id': editor.id,
'body': 'body' 'body': 'body'
} }

View File

@ -1,21 +1,11 @@
import pytest import pytest
from amanuensis.db import * from amanuensis.db import DbContext
import amanuensis.backend.lexicon as lexiq
import amanuensis.backend.user as userq import amanuensis.backend.user as userq
from amanuensis.errors import ArgumentError from amanuensis.errors import ArgumentError
from .test_db import db
def test_create_user(db: DbContext):
@pytest.fixture
def db():
db = DbContext('sqlite:///:memory:', debug=True)
db.create_all()
return db
def test_create_user(db):
"""Test new user creation.""" """Test new user creation."""
kwargs = { kwargs = {
'username': 'username', 'username': 'username',