diff --git a/baselayer b/baselayer index 9fa312e..81f6f22 160000 --- a/baselayer +++ b/baselayer @@ -1 +1 @@ -Subproject commit 9fa312e4320c651fa7f9c9cdcbb2ceab0bef7d28 +Subproject commit 81f6f22631c39c6c174a116675fade95b0134fb0 diff --git a/requirements.txt b/requirements.txt index e69de29..a3b91c4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -0,0 +1 @@ +factory-boy==3.2.0 diff --git a/template_app/app_server.py b/template_app/app_server.py index 677f17f..6d61883 100644 --- a/template_app/app_server.py +++ b/template_app/app_server.py @@ -1,6 +1,7 @@ import tornado.web from baselayer.app import models, model_util +from . import models as blt_models from .handlers.example_computation import ExampleComputationHandler from .handlers.push_notification import PushNotificationHandler diff --git a/template_app/models.py b/template_app/models.py new file mode 100644 index 0000000..276bef2 --- /dev/null +++ b/template_app/models.py @@ -0,0 +1,84 @@ +import sqlalchemy as sa +from sqlalchemy.orm import relationship + +from baselayer.app.models import ( + AccessibleIfRelatedRowsAreAccessible, + AccessibleIfUserMatches, + public, + restricted, + Base, + User, + Token +) + +class PublicModel(Base): + update = delete = public + + +class RestrictedModel(Base): + create = read = restricted + + +class UserAccessibleModel(Base): + create = read = update = delete = AccessibleIfUserMatches('user') + + user_id = sa.Column( + sa.Integer, sa.ForeignKey('users.id', ondelete='CASCADE'), + nullable=False, + index=True + ) + user = relationship('User') + + +class RelatedModel(Base): + create = read = update = delete = AccessibleIfRelatedRowsAreAccessible(user="delete") + + user_id = sa.Column( + sa.Integer, sa.ForeignKey('users.id', ondelete='CASCADE'), + nullable=False, + index=True + ) + user = relationship('User') + + +class CompositeAndModel(Base): + create = read = update = delete = ( + AccessibleIfUserMatches('user') & + AccessibleIfUserMatches('last_modified_by') + ) + + user_id = sa.Column( + sa.Integer, sa.ForeignKey('users.id', ondelete='CASCADE'), + nullable=False, + index=True + ) + user = relationship('User', foreign_keys=[user_id]) + + last_modified_by_id = sa.Column( + sa.Integer, sa.ForeignKey('users.id', ondelete='CASCADE'), + nullable=False, + index=True + ) + last_modified_by = relationship('User', foreign_keys=[last_modified_by_id]) + + +class CompositeOrModel(Base): + create = read = update = delete = ( + AccessibleIfUserMatches('user') | + AccessibleIfUserMatches('last_modified_by') + ) + + user_id = sa.Column( + sa.Integer, sa.ForeignKey('users.id', ondelete='CASCADE'), + nullable=False, + index=True + ) + user = relationship('User', foreign_keys=[user_id]) + + last_modified_by_id = sa.Column( + sa.Integer, sa.ForeignKey('users.id', ondelete='CASCADE'), + nullable=False, + index=True + ) + last_modified_by = relationship('User', foreign_keys=[last_modified_by_id]) + diff --git a/template_app/tests/conftest.py b/template_app/tests/conftest.py index fd4ecd9..8c2d0de 100644 --- a/template_app/tests/conftest.py +++ b/template_app/tests/conftest.py @@ -16,8 +16,66 @@ from baselayer.app.test_util import (driver, MyCustomWebDriver, reset_state, set_server_url) +from baselayer.app import models, model_util +from baselayer.app.models import User, ACL, DBSession +from baselayer.app.env import load_env + +from .fixtures import ( + PublicModelFactory, + RestrictedModelFactory, + CompositeAndModelFactory, + CompositeOrModelFactory, + UserAccessibleModelFactory, + RelatedModelFactory, + UserFactory +) + print('Loading test configuration from test_config.yaml') basedir = pathlib.Path(os.path.dirname(__file__))/'../..' cfg = load_config([basedir/'test_config.yaml']) set_server_url(f'http://localhost:{cfg["ports.app"]}') +models.init_db(**cfg['database']) + +acl = ACL.create_or_get('System admin') +DBSession().add(acl) +DBSession().commit() + +@pytest.fixture() +def super_admin_user(): + return UserFactory(acls=[acl]) + + +@pytest.fixture() +def user(): + return UserFactory() + + +@pytest.fixture() +def public_model(): + return PublicModelFactory() + + +@pytest.fixture() +def restricted_model(): + return RestrictedModelFactory() + + +@pytest.fixture() +def related_model(user): + return RelatedModelFactory(user=user) + + +@pytest.fixture() +def user_accessible_model(super_admin_user): + return UserAccessibleModelFactory(user=super_admin_user) + + +@pytest.fixture() +def composite_and_model(super_admin_user, user): + return CompositeAndModelFactory(last_modified_by=super_admin_user, user=user) + + +@pytest.fixture() +def composite_or_model(super_admin_user, user): + return CompositeOrModelFactory(last_modified_by=super_admin_user, user=user) diff --git a/template_app/tests/fixtures.py b/template_app/tests/fixtures.py new file mode 100644 index 0000000..1672097 --- /dev/null +++ b/template_app/tests/fixtures.py @@ -0,0 +1,80 @@ +from template_app.models import ( + PublicModel, + RestrictedModel, + CompositeAndModel, + CompositeOrModel, + UserAccessibleModel, + RelatedModel, + User +) + +from baselayer.app.models import DBSession + +import factory +import uuid + + +class BaseMeta: + sqlalchemy_session = DBSession() + sqlalchemy_session_persistence = 'commit' + + +class UserFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = User + + username = factory.LazyFunction(lambda: str(uuid.uuid4())) + contact_email = factory.LazyFunction(lambda: f'{uuid.uuid4().hex[:10]}@gmail.com') + first_name = factory.LazyFunction(lambda: f'{uuid.uuid4().hex[:4]}') + last_name = factory.LazyFunction(lambda: f'{uuid.uuid4().hex[:4]}') + + @factory.post_generation + def acls(obj, create, extracted, **kwargs): + if not create: + return + + if extracted: + for acl in extracted: + obj.acls.append(acl) + DBSession().add(obj) + DBSession().commit() + + +class PublicModelFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = PublicModel + + +class RestrictedModelFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = RestrictedModel + + +class CompositeAndModelFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = CompositeAndModel + + user = factory.SubFactory(UserFactory) + last_modified_by = factory.SubFactory(UserFactory) + + +class CompositeOrModelFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = CompositeOrModel + + user = factory.SubFactory(UserFactory) + last_modified_by = factory.SubFactory(UserFactory) + + +class UserAccessibleModelFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = UserAccessibleModel + + user = factory.SubFactory(UserFactory) + + +class RelatedModelFactory(factory.alchemy.SQLAlchemyModelFactory): + class Meta(BaseMeta): + model = RelatedModel + + user = factory.SubFactory(UserFactory) diff --git a/template_app/tests/models/test_models.py b/template_app/tests/models/test_models.py new file mode 100644 index 0000000..7b09b40 --- /dev/null +++ b/template_app/tests/models/test_models.py @@ -0,0 +1,48 @@ +def test_user_can_read_public_model(user, public_model): + assert public_model.is_accessible_by(user) + + +def test_super_user_can_read_public_model(super_admin_user, public_model): + assert public_model.is_accessible_by(super_admin_user) + + +def test_user_can_read_restricted_model(user, restricted_model): + assert not restricted_model.is_accessible_by(user) # need System admin + + +def test_super_user_can_read_restricted_model(super_admin_user, restricted_model): + assert restricted_model.is_accessible_by(super_admin_user) + + +def test_user_can_read_related_model(user, related_model): + # user must be able to delete related record + assert not related_model.is_accessible_by(user) + + +def test_super_user_can_read_related_model(super_admin_user, related_model): + assert related_model.is_accessible_by(super_admin_user) + + +def test_user_can_read_user_accessible_model(user, user_accessible_model): + assert not user_accessible_model.is_accessible_by(user) + + +def test_super_admin_user_can_read_user_accessible_model(super_admin_user, user_accessible_model): + assert user_accessible_model.is_accessible_by(super_admin_user) + + +def test_user_can_read_composite_and_model(user, composite_and_model): + # last_modified_by must be the same as user, but is super_admin_user + assert not composite_and_model.is_accessible_by(user) + + +def test_super_admin_user_can_read_composite_and_model(super_admin_user, composite_and_model): + assert composite_and_model.is_accessible_by(super_admin_user) + + +def test_user_can_read_composite_or_model(user, composite_or_model): + assert composite_or_model.is_accessible_by(user) + + +def test_super_admin_user_can_read_composite_or_model(super_admin_user, composite_or_model): + assert composite_or_model.is_accessible_by(super_admin_user)