Compare commits

..

2 Commits

Author SHA1 Message Date
Tim Van Baak 8a4d666248 Add basic database test 2021-05-04 23:05:14 -07:00
Tim Van Baak 9f2c9d14d3 Refactor db infra to require instantiating object 2021-05-04 23:05:03 -07:00
3 changed files with 61 additions and 39 deletions

View File

@ -6,10 +6,6 @@ import sqlite3
import uuid import uuid
# Register GUID as a known type with sqlite
sqlite3.register_converter('GUID', lambda h: uuid.UUID(hex=h))
sqlite3.register_adapter(uuid.UUID, lambda u: u.hex)
class Uuid(TypeDecorator): class Uuid(TypeDecorator):
""" """
A uuid backed by a char(32) field in sqlite. A uuid backed by a char(32) field in sqlite.
@ -32,17 +28,9 @@ class Uuid(TypeDecorator):
else: else:
return value return value
engine = create_engine('sqlite:///:memory:', connect_args={'detect_types': sqlite3.PARSE_DECLTYPES})
# Enable foreign key constraints
@event.listens_for(engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
# Define naming conventions for generated constraints # Define naming conventions for generated constraints
metadata = MetaData(bind=engine, naming_convention={ metadata = MetaData(naming_convention={
"ix": "ix_%(column_0_label)s", "ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s", "uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s", "ck": "ck_%(table_name)s_%(constraint_name)s",
@ -50,8 +38,22 @@ metadata = MetaData(bind=engine, naming_convention={
"pk": "pk_%(table_name)s" "pk": "pk_%(table_name)s"
}) })
# Thread-safe db session
session = scoped_session(sessionmaker(bind=engine))
# Base class for ORM models # Base class for ORM models
ModelBase = declarative_base(metadata=metadata) ModelBase = declarative_base(metadata=metadata)
class DbContext():
def __init__(self, db_uri, debug=False):
# Create an engine and enable foreign key constraints in sqlite
self.engine = create_engine(db_uri, echo=debug)
@event.listens_for(self.engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
# Create a thread-safe session factory
self.session = scoped_session(sessionmaker(bind=self.engine))
def create_all(self):
ModelBase.metadata.create_all(self.engine)

View File

@ -1,5 +0,0 @@
from amanuensis import __version__
def test_version():
assert __version__ == '0.1.0'

25
tests/test_db.py Normal file
View File

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