jkrauss/sigl
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add Tests

Browse Source
This commit is contained in:
Jonathan Krauss
2022-07-12 07:20:54 -07:00
parent daadf7d4b4
commit 4102eecd22
10 changed files with 473 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
Makefile
View File

@@ -30,4 +30,15 @@ serve :
shell :
poetry run flask shell
.PHONY : lint shell serve
test :
poetry run python -m pytest tests
test-x :
poetry run python -m pytest tests -x
test-wip :
poetry run python -m pytest tests -m wip
.PHONY : db-init db-migrate db-upgrad db-downgrade \
lint shell serve \
test test-wip test-x

24
migrations/versions/938218f911e8_.py → migrations/versions/04f3fe65d40b_.py
View File

@@ -1,8 +1,8 @@
"""empty message
Revision ID: 938218f911e8
Revision ID: 04f3fe65d40b
Revises:
Create Date: 2022-07-12 06:31:18.835124
Create Date: 2022-07-12 07:03:12.713616
"""
from alembic import op
@@ -10,7 +10,7 @@ import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '938218f911e8'
revision = '04f3fe65d40b'
down_revision = None
branch_labels = None
depends_on = None
@@ -31,8 +31,14 @@ def upgrade():
sa.Column('restoredAt', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('key')
)
op.create_table('sigl_config',
sa.Column('key', sa.String(), nullable=False),
sa.Column('value', sa.String(length=128), nullable=True),
sa.PrimaryKeyConstraint('key')
)
op.create_table('access_tokens',
sa.Column('token', sa.String(length=64), nullable=False),
sa.Column('key', sa.String(length=64), nullable=True),
sa.Column('clientIP', sa.String(length=46), nullable=False),
sa.Column('userAgent', sa.String(length=255), nullable=False),
sa.Column('expired', sa.Boolean(), nullable=False),
@@ -40,6 +46,7 @@ def upgrade():
sa.Column('issuedAt', sa.DateTime(), nullable=True),
sa.Column('expiresAt', sa.DateTime(), nullable=True),
sa.Column('revokedAt', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['key'], ['access_keys.key'], ),
sa.PrimaryKeyConstraint('token')
)
op.create_table('lists',
@@ -49,24 +56,21 @@ def upgrade():
sa.Column('notes', sa.String(), nullable=True),
sa.Column('createdAt', sa.DateTime(), nullable=True),
sa.Column('modifiedAt', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['accessKey'], ['access_keys.key'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('products',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('accessKey', sa.String(length=64), nullable=True),
sa.Column('key', sa.String(length=64), nullable=True),
sa.Column('name', sa.String(length=128), nullable=False),
sa.Column('category', sa.String(length=128), nullable=False),
sa.Column('notes', sa.String(), nullable=True),
sa.Column('createdAt', sa.DateTime(), nullable=True),
sa.Column('modifiedAt', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['key'], ['access_keys.key'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_products_category'), 'products', ['category'], unique=False)
op.create_table('sigl_config',
sa.Column('key', sa.String(), nullable=False),
sa.Column('value', sa.String(length=128), nullable=True),
sa.PrimaryKeyConstraint('key')
)
op.create_table('list_entries',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('list_id', sa.Integer(), nullable=False),
@@ -99,10 +103,10 @@ def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('product_locations')
op.drop_table('list_entries')
op.drop_table('sigl_config')
op.drop_index(op.f('ix_products_category'), table_name='products')
op.drop_table('products')
op.drop_table('lists')
op.drop_table('access_tokens')
op.drop_table('sigl_config')
op.drop_table('access_keys')
# ### end Alembic commands ###

57
sigl/database/orm.py
View File

@@ -27,43 +27,48 @@ __all__ = ('init_orm', )
def init_orm():
"""Initialize the Sigl ORM."""
# Access Keys
db.mapper(AccessKey, access_keys, properties={
'tokens': db.relationship(
AccessToken,
backref='accessKey',
cascade='all, delete-orphan',
)
})
# # Access Keys
# db.mapper(AccessKey, access_keys, properties={
# 'tokens': db.relationship(
# AccessToken,
# backref='accessKey',
# cascade='all, delete-orphan',
# )
# })
# Access Tokens
db.mapper(AccessToken, access_tokens)
# # Access Tokens
# db.mapper(AccessToken, access_tokens)
# List Entries
db.mapper(ListEntry, list_entries)
# # List Entries
# db.mapper(ListEntry, list_entries)
# Products
db.mapper(Product, products, properties={
'locations': db.relationship(
ProductLocation,
backref='product',
cascade='all, delete-orphan',
),
'shoppingLists': db.relationship(
ShoppingList,
backref='product',
back_populates='product',
cascade='all, delete-orphan',
),
# 'shoppingLists': db.relationship(
# ShoppingList,
# backref='product',
# cascade='all, delete-orphan',
# ),
})
# Product Locations
db.mapper(ProductLocation, product_locations)
# Shopping Lists
db.mapper(ShoppingList, lists, properties={
'entries': db.relationship(
ListEntry,
backref='shoppingList',
cascade='all, delete-orphan',
db.mapper(ProductLocation, product_locations, properties={
'product': db.relationship(
Product,
back_populates='locations',
)
})
# # Shopping Lists
# db.mapper(ShoppingList, lists, properties={
# 'entries': db.relationship(
# ListEntry,
# backref='shoppingList',
# cascade='all, delete-orphan',
# )
# })

7
sigl/database/tables.py
View File

@@ -45,7 +45,8 @@ access_tokens = db.Table(
# Primary Key
db.Column('token', db.String(64), primary_key=True),
# Client Attributes
# Token Attributes
db.Column('key', db.ForeignKey('access_keys.key'), default=None),
db.Column('clientIP', db.String(46), nullable=False),
db.Column('userAgent', db.String(255), nullable=False),
@@ -67,7 +68,7 @@ lists = db.Table(
db.Column('id', db.Integer, primary_key=True),
# Access Key
db.Column('accessKey', db.String(64), default=None),
db.Column('accessKey', db.ForeignKey('access_keys.key'), default=None),
# List Attributes
db.Column('name', db.String(128), nullable=False),
@@ -108,7 +109,7 @@ products = db.Table(
db.Column('id', db.Integer, primary_key=True),
# Access Key
db.Column('accessKey', db.String(64), default=None),
db.Column('key', db.ForeignKey('access_keys.key'), default=None),
# Product Attributes
db.Column('name', db.String(128), nullable=False),

2
sigl/domain/models/accessKey.py
View File

@@ -56,7 +56,7 @@ class AccessToken:
User Agent string are logged for future auditing purposes.
"""
token: str = None
accessKey: str = None
accessKey: 'AccessKey' = None
clientIP: str = None
userAgent: str = None

5
tests/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Sigl Test Package.
Simple Grocery List (Sigl) | sigl.app
Copyright (c) 2022 Asymworks, LLC. All Rights Reserved.
"""

137
tests/conftest.py Normal file
View File

@@ -0,0 +1,137 @@
"""Sigl Test Fixtures.
Simple Grocery List (Sigl) | sigl.app
Copyright (c) 2022 Asymworks, LLC. All Rights Reserved.
"""
import os
from alembic.command import upgrade
from alembic.config import Config
import flask
import pytest
import sigl
from sigl.database import db as _db
from sigl.factory import create_app
DATA_DIR = '.pytest-data'
@pytest.fixture(scope='session')
def app_config(request):
"""Application Configuration for the Test Session.
Loads configuration from config/test.py, patches the database file with
a session-global temporary file, and returns the new configuration data
to pass to create_app().
"""
test_dir = os.path.dirname(__file__)
test_cfg = os.path.join(test_dir, '../config/test.py')
cfg_file = os.environ.get('SIGL_TEST_CONFIG', test_cfg)
cfg_data = flask.Config(test_dir)
cfg_data.from_pyfile(cfg_file)
# Patch Test Configuration
cfg_data['TESTING'] = True
cfg_data['DB_DRIVER'] = 'sqlite'
cfg_data['DB_FILE'] = os.path.join(test_dir, DATA_DIR, 'test.db')
# Ensure Database Path exists
db_dir = os.path.dirname(cfg_data['DB_FILE'])
if not os.path.isdir(db_dir):
os.mkdir(db_dir)
if not os.path.isdir(db_dir):
raise Exception('Database path "%s" does not exist' % (db_dir))
db_file = cfg_data['DB_FILE']
if os.path.exists(db_file):
os.unlink(db_file)
def teardown():
if os.path.exists(db_file):
os.unlink(db_file)
if os.path.exists(db_dir) and len(os.listdir(db_dir)) == 0:
os.rmdir(db_dir)
request.addfinalizer(teardown)
return cfg_data
@pytest.fixture(scope='module')
def app(request, app_config):
"""Module-Wide Sigl Flask Application with Database.
Loads configuration from config/test.py, patches the database file with
a session-global temporary file, and initializes the application.
"""
# Apply Database Migrations
_db.clear_mappers()
_app = create_app(app_config, __name__)
with _app.app_context():
alembic_config = Config('migrations/alembic.ini')
alembic_config.set_main_option('script_location', 'migrations')
upgrade(alembic_config, 'head')
_db.clear_mappers()
# Initialize Application and Context
app = create_app(app_config, __name__)
ctx = app.app_context()
ctx.push()
# Add Finalizers
def teardown():
_db.drop_all()
_db.clear_mappers()
if os.path.exists(app.config['DB_FILE']):
os.unlink(app.config['DB_FILE'])
ctx.pop()
request.addfinalizer(teardown)
return app
@pytest.fixture(scope='module')
def app_without_database(request, app_config):
"""Module-Wide Sigl Flask Application without Database.
Loads configuration from config/test.py and initializes the application,
but does not create the database.
"""
_db.clear_mappers()
# Initialize Application and Context
app = create_app(app_config, __name__)
ctx = app.app_context()
ctx.push()
# Add Finalizers
def teardown():
_db.clear_mappers()
_db.drop_all()
ctx.pop()
request.addfinalizer(teardown)
return app
@pytest.fixture(scope='function')
def session(request, monkeypatch, app):
"""Create a new Database Session for the Request."""
with app.app_context():
connection = _db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds={})
session = _db.create_scoped_session(options=options)
# Patch sigl.db with the current session
monkeypatch.setattr(_db, 'session', session)
def teardown():
transaction.rollback()
connection.close()
session.remove()
request.addfinalizer(teardown)
return session

180
tests/test_01_db_makeuri.py Normal file
View File

@@ -0,0 +1,180 @@
"""Test Database URI Helper.
Simple Grocery List (Sigl) | sigl.app
Copyright (c) 2022 Asymworks, LLC. All Rights Reserved.
"""
import pytest
from sigl.database.util import make_uri
from sigl.exc import ConfigError
BASE_TEST_CONFIG = {
'APP_SESSION_KEY': 'test',
'APP_TOKEN_KEY': 'test',
'APP_TOKEN_SALT': 'test',
'DB_DRIVER': 'sqlite',
'DB_FILE': '/dev/null',
'MAIL_SERVER': 'localhost',
'MAIL_SENDER': 'test@localhost',
}
class MockApp(object):
def __init__(self, config):
self.config = config
def test_dbconfig_no_driver():
'''
Application startup should fail if DB_DRIVER is not in the configuration
'''
test_cfg = dict(BASE_TEST_CONFIG)
del test_cfg['DB_DRIVER']
app = MockApp(test_cfg)
with pytest.raises(ConfigError, match='application configuration') as excinfo:
make_uri(app)
# Ensure that the offending key is DB_DRIVER
assert excinfo.value.config_key == 'DB_DRIVER'
def test_dbconfig_sqlite_no_file():
'''
Application Startup should fail if an SQlite database is specified
without a DB_FILE key
'''
test_cfg = dict(BASE_TEST_CONFIG)
del test_cfg['DB_FILE']
app = MockApp(test_cfg)
with pytest.raises(ConfigError, match='application configuration') as excinfo:
make_uri(app)
# Ensure that the offending key is DB_FILE
assert excinfo.value.config_key == 'DB_FILE'
def test_dbconfig_sqlite_with_file():
'''
Application Startup should succeed if an SQlite database is specified
with a DB_FILE key
'''
test_cfg = dict(BASE_TEST_CONFIG)
app = MockApp(test_cfg)
uri = make_uri(app)
# Ensure that the URI was correctly configured and logged
assert uri == 'sqlite:////dev/null'
def test_dbconfig_mysql_no_host():
'''
Application Startup should fail if a MySQL database is specified
without a DB_HOST key
'''
test_cfg = dict(BASE_TEST_CONFIG)
test_cfg['DB_DRIVER'] = 'mysql'
app = MockApp(test_cfg)
with pytest.raises(ConfigError, match='application configuration') as excinfo:
make_uri(app)
# Ensure that the offending key is DB_FILE
assert excinfo.value.config_key == 'DB_HOST'
def test_dbconfig_mysql_no_name():
'''
Application Startup should fail if a MySQL database is specified
without a DB_HOST key
'''
test_cfg = dict(BASE_TEST_CONFIG)
test_cfg['DB_DRIVER'] = 'mysql'
test_cfg['DB_HOST'] = 'localhost'
app = MockApp(test_cfg)
with pytest.raises(ConfigError, match='application configuration') as excinfo:
make_uri(app)
# Ensure that the offending key is DB_NAME
assert excinfo.value.config_key == 'DB_NAME'
def test_dbconfig_mysql_no_username():
'''
Application Startup should fail if a MySQL database is specified
with a DB_USERNAME key but without a DB_PASSWORD key
'''
test_cfg = dict(BASE_TEST_CONFIG)
test_cfg['DB_DRIVER'] = 'mysql'
test_cfg['DB_HOST'] = 'localhost'
test_cfg['DB_NAME'] = 'test'
test_cfg['DB_PASSWORD'] = 'hunter2'
app = MockApp(test_cfg)
with pytest.raises(ConfigError, match='application configuration') as excinfo:
make_uri(app)
# Ensure that the offending key is DB_USERNAME
assert excinfo.value.config_key == 'DB_USERNAME'
def test_dbconfig_mysql_without_port():
'''
Application Startup should succeed when a MySQL database is specified
with a DB_PORT key
'''
test_cfg = dict(BASE_TEST_CONFIG)
test_cfg['DB_DRIVER'] = 'mysql'
test_cfg['DB_HOST'] = 'localhost'
test_cfg['DB_NAME'] = 'test'
test_cfg['DB_USERNAME'] = 'root'
test_cfg['DB_PASSWORD'] = 'hunter^2'
app = MockApp(test_cfg)
uri = make_uri(app)
# Ensure that the application startup succeeded
assert uri == 'mysql://root:hunter%5E2@localhost/test'
def test_dbconfig_mysql_with_port():
'''
Application Startup should succeed when a MySQL database is specified
with a DB_PORT key
'''
test_cfg = dict(BASE_TEST_CONFIG)
test_cfg['DB_DRIVER'] = 'mysql'
test_cfg['DB_HOST'] = 'localhost'
test_cfg['DB_PORT'] = 1234
test_cfg['DB_NAME'] = 'test'
test_cfg['DB_USERNAME'] = 'root'
test_cfg['DB_PASSWORD'] = 'hunter^2'
app = MockApp(test_cfg)
uri = make_uri(app)
# Ensure that the application startup succeeded
assert uri == 'mysql://root:hunter%5E2@localhost:1234/test'
def test_dbconfig_mysql_with_username():
'''
Application Startup should succeed when a MySQL database is specified
with a DB_USERNAME key and without a DB_PASSWORD key
'''
test_cfg = dict(BASE_TEST_CONFIG)
test_cfg['DB_DRIVER'] = 'mysql'
test_cfg['DB_HOST'] = 'localhost'
test_cfg['DB_PORT'] = 1234
test_cfg['DB_NAME'] = 'test'
test_cfg['DB_USERNAME'] = 'root'
app = MockApp(test_cfg)
uri = make_uri(app)
# Ensure that the application startup succeeded
assert uri == 'mysql://root@localhost:1234/test'

78
tests/test_11_product_model.py Normal file
View File

@@ -0,0 +1,78 @@
"""Test the Product and ProductLocation Models.
Simple Grocery List (Sigl) | sigl.app
Copyright (c) 2022 Asymworks, LLC. All Rights Reserved.
"""
import pytest
from sqlalchemy.exc import IntegrityError
from sigl.domain.models import Product, ProductLocation
# Always use 'app' fixture so ORM gets initialized
pytestmark = pytest.mark.usefixtures('app')
@pytest.mark.unit
def test_product_model_init(session):
"""Test newly created Products have no Locations."""
p = Product(name='Eggs', category='Dairy')
session.add(p)
session.commit()
assert p.id is not None
assert p.name == 'Eggs'
assert p.category == 'Dairy'
assert not p.locations
@pytest.mark.unit
def test_product_model_can_add_location(session):
"""Test that a Location can be added to a Product."""
p = Product(name='Eggs', category='Dairy')
l = ProductLocation(product=p, store='Pavilions', aisle='Back Wall')
session.add(p)
session.add(l)
session.commit()
assert l.aisle == 'Back Wall'
assert l.bin is None
assert l.product == p
assert l in p.locations
@pytest.mark.unit
def test_product_model_can_add_multiple_stores(session):
"""Test that multiple Locations can be added to a Product."""
p = Product(name='Eggs', category='Dairy')
l1 = ProductLocation(product=p, store='Pavilions', aisle='Back Wall')
l2 = ProductLocation(product=p, store='Stater Brothers', aisle='Left Wall')
session.add(p)
session.add(l1)
session.add(l2)
session.commit()
assert l1.product == p
assert l1 in p.locations
assert l2.product == p
assert l2 in p.locations
@pytest.mark.unit
def test_product_model_same_store_fails(session):
"""Test that two Locations with the same Store cannot be added to a Product."""
p = Product(name='Eggs', category='Dairy')
l1 = ProductLocation(product=p, store='Pavilions', aisle='Back Wall')
l2 = ProductLocation(product=p, store='Pavilions', aisle='Left Wall')
session.add(p)
session.add(l1)
session.add(l2)
with pytest.raises(IntegrityError):
session.commit()

11
tests/tox.ini Normal file
View File

@@ -0,0 +1,11 @@
[flake8]
ignore = D300,D400,E201,E202,E203,E241,E501,E712
[pytest]
addopts = --strict-markers
markers =
spike: marks test as "spikes" that are intended to develop algorithms but not test behavior (deselected by default, select with '-m spike')
unit: marks tests as domain-level unit tests (deselect with '-m "not unit"')
wip: marks tests as works in progress and likely to fail (deselect with '-m "not wip"')
filterwarnings =
ignore::DeprecationWarning:eventlet.*