diff --git a/lib/tool_shed/grids/admin_grids.py b/lib/tool_shed/grids/admin_grids.py index e686ed16eedd..8e3ac87665ec 100644 --- a/lib/tool_shed/grids/admin_grids.py +++ b/lib/tool_shed/grids/admin_grids.py @@ -259,8 +259,8 @@ def get_value(self, trans, grid, group): class UsersColumn(grids.GridColumn): def get_value(self, trans, grid, group): - if group.members: - return len(group.members) + if group.users: + return len(group.users) return 0 title = "Groups" diff --git a/lib/tool_shed/webapp/model/__init__.py b/lib/tool_shed/webapp/model/__init__.py index 328d42ea3225..13287839fafd 100644 --- a/lib/tool_shed/webapp/model/__init__.py +++ b/lib/tool_shed/webapp/model/__init__.py @@ -7,8 +7,30 @@ ) from typing import Any, Mapping, TYPE_CHECKING +from sqlalchemy import ( + Boolean, + Column, + DateTime, + desc, + false, + ForeignKey, + Integer, + not_, + String, + Table, + TEXT, + true, + UniqueConstraint, +) +from sqlalchemy.orm import ( + registry, + relationship, +) +from sqlalchemy.orm.decl_api import DeclarativeMeta + import tool_shed.repository_types.util as rt_util from galaxy import util +from galaxy.model.custom_types import MutableJSONType, TrimmedString from galaxy.model.orm.now import now from galaxy.security.validate_user_input import validate_password_str from galaxy.util import unique_id @@ -27,22 +49,76 @@ WEAK_HG_REPO_CACHE: Mapping['Repository', Any] = weakref.WeakKeyDictionary() if TYPE_CHECKING: - from sqlalchemy.schema import Table - class _HasTable: table: Table else: _HasTable = object -class APIKeys(_HasTable): - pass - +mapper_registry = registry() + + +class Base(metaclass=DeclarativeMeta): + __abstract__ = True + registry = mapper_registry + metadata = mapper_registry.metadata + __init__ = mapper_registry.constructor + + @classmethod + def __declare_last__(cls): + cls.table = cls.__table__ + + +class APIKeys(Base, _HasTable): + __tablename__ = 'api_keys' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + user_id = Column(ForeignKey('galaxy_user.id'), index=True) + key = Column(TrimmedString(32), index=True, unique=True) + user = relationship('User', back_populates='api_keys') + + +class User(Base, Dictifiable, _HasTable): + __tablename__ = 'galaxy_user' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + email = Column(TrimmedString(255), nullable=False) + username = Column(String(255), index=True) + password = Column(TrimmedString(40), nullable=False) + external = Column(Boolean, default=False) + new_repo_alert = Column(Boolean, default=False) + deleted = Column(Boolean, index=True, default=False) + purged = Column(Boolean, index=True, default=False) + active_repositories = relationship('Repository', + primaryjoin=(lambda: (Repository.user_id == User.id) & (not_(Repository.deleted))), # type: ignore + back_populates='user', + order_by=lambda: desc(Repository.name)) # type: ignore + galaxy_sessions = relationship('GalaxySession', + back_populates='user', + order_by=lambda: desc(GalaxySession.update_time)) # type: ignore + api_keys = relationship( + 'APIKeys', + back_populates='user', + order_by=lambda: desc(APIKeys.create_time)) # type: ignore + reset_tokens = relationship('PasswordResetToken', back_populates='user') + groups = relationship('UserGroupAssociation', back_populates='user') -class User(Dictifiable, _HasTable): dict_collection_visible_keys = ['id', 'username'] dict_element_visible_keys = ['id', 'username'] bootstrap_admin_user = False + roles = relationship('UserRoleAssociation', back_populates='user') + non_private_roles = relationship( + 'UserRoleAssociation', + viewonly=True, + primaryjoin=(lambda: + (User.id == UserRoleAssociation.user_id) # type: ignore + & (UserRoleAssociation.role_id == Role.id) # type: ignore + & not_(Role.name == User.email)) # type: ignore + ) + repository_reviews = relationship('RepositoryReview', back_populates='user') def __init__(self, email=None, password=None): self.email = email @@ -50,7 +126,6 @@ def __init__(self, email=None, password=None): self.external = False self.deleted = False self.purged = False - self.username = None self.new_repo_alert = False def all_roles(self): @@ -85,7 +160,14 @@ def set_password_cleartext(self, cleartext): self.password = new_secure_hash(text_type=cleartext) -class PasswordResetToken(_HasTable): +class PasswordResetToken(Base, _HasTable): + __tablename__ = 'password_reset_token' + + token = Column(String(32), primary_key=True, unique=True, index=True) + expiration_time = Column(DateTime) + user_id = Column(ForeignKey('galaxy_user.id'), index=True) + user = relationship('User', back_populates='reset_tokens') + def __init__(self, user, token=None): if token: self.token = token @@ -95,7 +177,17 @@ def __init__(self, user, token=None): self.expiration_time = now() + timedelta(hours=24) -class Group(Dictifiable, _HasTable): +class Group(Base, Dictifiable, _HasTable): + __tablename__ = 'galaxy_group' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + name = Column(String(255), index=True, unique=True) + deleted = Column(Boolean, index=True, default=False) + roles = relationship('GroupRoleAssociation', back_populates='group') + users = relationship('UserGroupAssociation', back_populates='group') + dict_collection_visible_keys = ['id', 'name'] dict_element_visible_keys = ['id', 'name'] @@ -104,7 +196,20 @@ def __init__(self, name=None): self.deleted = False -class Role(Dictifiable, _HasTable): +class Role(Base, Dictifiable, _HasTable): + __tablename__ = 'role' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + name = Column(String(255), index=True, unique=True) + description = Column(TEXT) + type = Column(String(40), index=True) + deleted = Column(Boolean, index=True, default=False) + repositories = relationship('RepositoryRoleAssociation', back_populates='role') + groups = relationship('GroupRoleAssociation', back_populates='role') + users = relationship('UserRoleAssociation', back_populates='role') + dict_collection_visible_keys = ['id', 'name'] dict_element_visible_keys = ['id', 'name', 'description', 'type'] private_id = None @@ -114,7 +219,7 @@ class Role(Dictifiable, _HasTable): ADMIN='admin', SHARING='sharing') - def __init__(self, name="", description="", type="system", deleted=False): + def __init__(self, name=None, description=None, type=types.SYSTEM, deleted=False): self.name = name self.description = description self.type = type @@ -130,56 +235,127 @@ def is_repository_admin_role(self): return False -class UserGroupAssociation(_HasTable): +class UserGroupAssociation(Base, _HasTable): + __tablename__ = 'user_group_association' + + id = Column(Integer, primary_key=True) + user_id = Column(ForeignKey('galaxy_user.id'), index=True) + group_id = Column(ForeignKey('galaxy_group.id'), index=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + user = relationship('User', back_populates='groups') + group = relationship('Group', back_populates='users') + def __init__(self, user, group): self.user = user self.group = group -class UserRoleAssociation(_HasTable): +class UserRoleAssociation(Base, _HasTable): + __tablename__ = 'user_role_association' + + id = Column(Integer, primary_key=True) + user_id = Column(ForeignKey('galaxy_user.id'), index=True) + role_id = Column(ForeignKey('role.id'), index=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + user = relationship('User', back_populates='roles') + role = relationship('Role', back_populates='users') + def __init__(self, user, role): self.user = user self.role = role -class GroupRoleAssociation(_HasTable): +class GroupRoleAssociation(Base, _HasTable): + __tablename__ = 'group_role_association' + + id = Column(Integer, primary_key=True) + group_id = Column(ForeignKey("galaxy_group.id"), index=True) + role_id = Column(ForeignKey("role.id"), index=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + group = relationship('Group', back_populates='roles') + role = relationship('Role', back_populates='groups') + def __init__(self, group, role): self.group = group self.role = role -class RepositoryRoleAssociation(_HasTable): +class RepositoryRoleAssociation(Base, _HasTable): + __tablename__ = 'repository_role_association' + + id = Column(Integer, primary_key=True) + repository_id = Column(ForeignKey("repository.id"), index=True) + role_id = Column(ForeignKey("role.id"), index=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + repository = relationship('Repository', back_populates='roles') + role = relationship('Role', back_populates='repositories') + def __init__(self, repository, role): self.repository = repository self.role = role -class GalaxySession(_HasTable): - - def __init__(self, - id=None, - user=None, - remote_host=None, - remote_addr=None, - referer=None, - current_history=None, - session_key=None, - is_valid=False, - prev_session_id=None, - last_action=None): - self.id = id - self.user = user - self.remote_host = remote_host - self.remote_addr = remote_addr - self.referer = referer - self.current_history = current_history - self.session_key = session_key +class GalaxySession(Base, _HasTable): + __tablename__ = 'galaxy_session' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + user_id = Column(ForeignKey("galaxy_user.id"), index=True, nullable=True) + remote_host = Column(String(255)) + remote_addr = Column(String(255)) + referer = Column(TEXT) + # unique 128 bit random number coerced to a string + session_key = Column(TrimmedString(255), index=True, unique=True) + is_valid = Column(Boolean, default=False) + # saves a reference to the previous session so we have a way to chain them together + prev_session_id = Column(Integer) + last_action = Column(DateTime) + user = relationship('User', back_populates='galaxy_sessions') + + def __init__(self, is_valid=False, **kwd): + super().__init__(**kwd) self.is_valid = is_valid - self.prev_session_id = prev_session_id - self.last_action = last_action or datetime.now() - + self.last_action = self.last_action or datetime.now() + + +class Repository(Base, Dictifiable, _HasTable): + __tablename__ = 'repository' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + name = Column(TrimmedString(255), index=True) + type = Column(TrimmedString(255), index=True) + remote_repository_url = Column(TrimmedString(255)) + homepage_url = Column(TrimmedString(255)) + description = Column(TEXT) + long_description = Column(TEXT) + user_id = Column(ForeignKey("galaxy_user.id"), index=True) + private = Column(Boolean, default=False) + deleted = Column(Boolean, index=True, default=False) + email_alerts = Column(MutableJSONType, nullable=True) + times_downloaded = Column(Integer) + deprecated = Column(Boolean, default=False) + categories = relationship('RepositoryCategoryAssociation', back_populates='repository') + ratings = relationship('RepositoryRatingAssociation', + order_by=lambda: desc(RepositoryRatingAssociation.update_time), back_populates='repository') # type: ignore + user = relationship('User', back_populates='active_repositories') + downloadable_revisions = relationship('RepositoryMetadata', + primaryjoin=lambda: (Repository.id == RepositoryMetadata.repository_id) & (RepositoryMetadata.downloadable == true()), # type: ignore + viewonly=True, + order_by=lambda: desc(RepositoryMetadata.update_time)) # type: ignore + metadata_revisions = relationship('RepositoryMetadata', + order_by=lambda: desc(RepositoryMetadata.update_time), # type: ignore + back_populates='repository') + roles = relationship('RepositoryRoleAssociation', back_populates='repository') + reviews = relationship('RepositoryReview', back_populates='repository') + reviewers = relationship('User', secondary=lambda: RepositoryReview.__table__, viewonly=True) # type: ignore -class Repository(Dictifiable, _HasTable): dict_collection_visible_keys = ['id', 'name', 'type', 'remote_repository_url', 'homepage_url', 'description', 'user_id', 'private', 'deleted', 'times_downloaded', 'deprecated', 'create_time'] dict_element_visible_keys = ['id', 'name', 'type', 'remote_repository_url', 'homepage_url', 'description', 'long_description', 'user_id', 'private', @@ -190,24 +366,12 @@ class Repository(Dictifiable, _HasTable): MARKED_FOR_ADDITION='a', NOT_TRACKED='?') - def __init__(self, id=None, name=None, type=None, remote_repository_url=None, homepage_url=None, - description=None, long_description=None, user_id=None, private=False, - deleted=None, email_alerts=None, times_downloaded=0, deprecated=False, - create_time=None): - self.id = id - self.name = name or "Unnamed repository" - self.type = type - self.remote_repository_url = remote_repository_url - self.homepage_url = homepage_url - self.description = description - self.long_description = long_description - self.user_id = user_id + def __init__(self, private=False, times_downloaded=0, deprecated=False, **kwd): + super().__init__(**kwd) self.private = private - self.deleted = deleted - self.email_alerts = email_alerts self.times_downloaded = times_downloaded self.deprecated = deprecated - self.create_time = create_time + self.name = self.name or "Unnamed repository" @property def hg_repo(self): @@ -338,83 +502,83 @@ def to_dict(self, view='collection', value_mapper=None): return rval -class RepositoryMetadata(Dictifiable, _HasTable): - dict_collection_visible_keys = ['id', 'repository_id', 'numeric_revision', 'changeset_revision', 'malicious', 'downloadable', 'missing_test_components', - 'has_repository_dependencies', 'includes_datatypes', 'includes_tools', 'includes_tool_dependencies', - 'includes_tools_for_display_in_tool_panel', 'includes_workflows'] - dict_element_visible_keys = ['id', 'repository_id', 'numeric_revision', 'changeset_revision', 'malicious', 'downloadable', 'missing_test_components', - 'has_repository_dependencies', 'includes_datatypes', 'includes_tools', 'includes_tool_dependencies', - 'includes_tools_for_display_in_tool_panel', 'includes_workflows', 'repository_dependencies'] - - def __init__(self, id=None, repository_id=None, numeric_revision=None, changeset_revision=None, metadata=None, tool_versions=None, malicious=False, - downloadable=False, missing_test_components=None, tools_functionally_correct=False, test_install_error=False, - has_repository_dependencies=False, includes_datatypes=False, includes_tools=False, includes_tool_dependencies=False, - includes_workflows=False): - self.id = id - self.repository_id = repository_id - self.numeric_revision = numeric_revision - self.changeset_revision = changeset_revision - self.metadata = metadata - self.tool_versions = tool_versions - self.malicious = malicious - self.downloadable = downloadable - self.missing_test_components = missing_test_components - self.has_repository_dependencies = has_repository_dependencies - # We don't consider the special case has_repository_dependencies_only_if_compiling_contained_td here. - self.includes_datatypes = includes_datatypes - self.includes_tools = includes_tools - self.includes_tool_dependencies = includes_tool_dependencies - self.includes_workflows = includes_workflows - - @property - def includes_tools_for_display_in_tool_panel(self): - if self.metadata: - tool_dicts = self.metadata.get('tools', []) - for tool_dict in tool_dicts: - if tool_dict.get('add_to_tool_panel', True): - return True - return False - - @property - def repository_dependencies(self): - if self.has_repository_dependencies: - return [repository_dependency for repository_dependency in self.metadata['repository_dependencies']['repository_dependencies']] - return [] - +class RepositoryReview(Base, Dictifiable, _HasTable): + __tablename__ = 'repository_review' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + repository_id = Column(ForeignKey('repository.id'), index=True) + changeset_revision = Column(TrimmedString(255), index=True) + user_id = Column(ForeignKey('galaxy_user.id'), index=True, nullable=False) + approved = Column(TrimmedString(255)) + rating = Column(Integer, index=True) + deleted = Column(Boolean, index=True, default=False) + repository = relationship('Repository', back_populates='reviews') + # Take care when using the mapper below! It should be used only when a new review is being created for a repository change set revision. + # Keep in mind that repository_metadata records can be removed from the database for certain change set revisions when metadata is being + # reset on a repository! + repository_metadata = relationship('RepositoryMetadata', + viewonly=True, + foreign_keys=lambda: [RepositoryReview.repository_id, RepositoryReview.changeset_revision], # type: ignore + primaryjoin=lambda: ((RepositoryReview.repository_id == RepositoryMetadata.repository_id) # type: ignore + & (RepositoryReview.changeset_revision == RepositoryMetadata.changeset_revision)), # type: ignore + back_populates='reviews') + user = relationship('User', back_populates='repository_reviews') + + component_reviews = relationship('ComponentReview', + viewonly=True, + primaryjoin=lambda: ((RepositoryReview.id == ComponentReview.repository_review_id) # type: ignore + & (ComponentReview.deleted == false())), # type: ignore + back_populates='repository_review') + + private_component_reviews = relationship('ComponentReview', + viewonly=True, + primaryjoin=lambda: ((RepositoryReview.id == ComponentReview.repository_review_id) # type: ignore + & (ComponentReview.deleted == false()) & (ComponentReview.private == true()))) # type: ignore -class RepositoryReview(Dictifiable, _HasTable): dict_collection_visible_keys = ['id', 'repository_id', 'changeset_revision', 'user_id', 'rating', 'deleted'] dict_element_visible_keys = ['id', 'repository_id', 'changeset_revision', 'user_id', 'rating', 'deleted'] approved_states = Bunch(NO='no', YES='yes') - def __init__(self, repository_id=None, changeset_revision=None, user_id=None, rating=None, deleted=False): - self.repository_id = repository_id - self.changeset_revision = changeset_revision - self.user_id = user_id - self.rating = rating + def __init__(self, deleted=False, **kwd): + super().__init__(**kwd) self.deleted = deleted -class ComponentReview(Dictifiable, _HasTable): +class ComponentReview(Base, Dictifiable, _HasTable): + __tablename__ = 'component_review' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + repository_review_id = Column(ForeignKey("repository_review.id"), index=True) + component_id = Column(ForeignKey("component.id"), index=True) + comment = Column(TEXT) + private = Column(Boolean, default=False) + approved = Column(TrimmedString(255)) + rating = Column(Integer) + deleted = Column(Boolean, index=True, default=False) + repository_review = relationship('RepositoryReview', back_populates='component_reviews') + component = relationship('Component') + dict_collection_visible_keys = ['id', 'repository_review_id', 'component_id', 'private', 'approved', 'rating', 'deleted'] dict_element_visible_keys = ['id', 'repository_review_id', 'component_id', 'private', 'approved', 'rating', 'deleted'] approved_states = Bunch(NO='no', YES='yes', NA='not_applicable') - def __init__(self, repository_review_id=None, component_id=None, comment=None, private=False, approved=False, rating=None, deleted=False): - self.repository_review_id = repository_review_id - self.component_id = component_id - self.comment = comment + def __init__(self, private=False, approved=False, deleted=False, **kwd): + super().__init__(**kwd) self.private = private self.approved = approved - self.rating = rating self.deleted = deleted -class Component(_HasTable): +class Component(Base, _HasTable): + __tablename__ = 'component' - def __init__(self, name=None, description=None): - self.name = name - self.description = description + id = Column(Integer, primary_key=True) + name = Column(TrimmedString(255)) + description = Column(TEXT) class ItemRatingAssociation(_HasTable): @@ -430,126 +594,151 @@ def set_item(self, item): """ Set association's item. """ -class RepositoryRatingAssociation(ItemRatingAssociation, _HasTable): +class RepositoryRatingAssociation(Base, ItemRatingAssociation, _HasTable): + __tablename__ = 'repository_rating_association' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + repository_id = Column(ForeignKey("repository.id"), index=True) + user_id = Column(ForeignKey("galaxy_user.id"), index=True) + rating = Column(Integer, index=True) + comment = Column(TEXT) + repository = relationship('Repository', back_populates='ratings') + user = relationship('User') def set_item(self, repository): self.repository = repository -class Category(Dictifiable, _HasTable): +class Category(Base, Dictifiable, _HasTable): + __tablename__ = 'category' + + id = Column(Integer, primary_key=True) + create_time = Column(DateTime, default=now) + update_time = Column(DateTime, default=now, onupdate=now) + name = Column(TrimmedString(255), index=True, unique=True) + description = Column(TEXT) + deleted = Column(Boolean, index=True, default=False) + repositories = relationship('RepositoryCategoryAssociation', back_populates='category') + dict_collection_visible_keys = ['id', 'name', 'description', 'deleted'] dict_element_visible_keys = ['id', 'name', 'description', 'deleted'] - def __init__(self, name=None, description=None, deleted=False): - self.name = name - self.description = description + def __init__(self, deleted=False, **kwd): + super().__init__(**kwd) self.deleted = deleted -class RepositoryCategoryAssociation(_HasTable): +class RepositoryCategoryAssociation(Base, _HasTable): + __tablename__ = 'repository_category_association' + + id = Column(Integer, primary_key=True) + repository_id = Column(ForeignKey('repository.id'), index=True) + category_id = Column(ForeignKey('category.id'), index=True) + category = relationship('Category', back_populates='repositories') + repository = relationship('Repository', back_populates='categories') def __init__(self, repository=None, category=None): self.repository = repository self.category = category -class Tag(_HasTable): +class Tag(Base, _HasTable): + __tablename__ = 'tag' + __table_args__ = ( + UniqueConstraint('name'), + ) - def __init__(self, id=None, type=None, parent_id=None, name=None): - self.id = id - self.type = type - self.parent_id = parent_id - self.name = name + id = Column(Integer, primary_key=True) + type = Column(Integer) + parent_id = Column(ForeignKey('tag.id')) + name = Column(TrimmedString(255)) + children = relationship('Tag', back_populates='parent') + parent = relationship('Tag', back_populates='children', remote_side=[id]) def __str__(self): return "Tag(id=%s, type=%i, parent_id=%s, name=%s)" % (self.id, self.type, self.parent_id, self.name) -class ItemTagAssociation(_HasTable): - - def __init__(self, id=None, user=None, item_id=None, tag_id=None, user_tname=None, value=None): - self.id = id - self.user = user - self.item_id = item_id - self.tag_id = tag_id - self.user_tname = user_tname - self.value = None - self.user_value = None - - -class PostJobAction(_HasTable): - - def __init__(self, action_type, workflow_step, output_name=None, action_arguments=None): - self.action_type = action_type - self.output_name = output_name - self.action_arguments = action_arguments - self.workflow_step = workflow_step - - -class StoredWorkflowAnnotationAssociation(_HasTable): - pass - - -class WorkflowStepAnnotationAssociation(_HasTable): - pass - +# The RepositoryMetadata model is mapped imperatively (for details see discussion in PR #12064). +# TLDR: a declaratively-mapped class cannot have a .metadata attribute (it is used by SQLAlchemy's DeclarativeBase). -class Workflow(_HasTable): - - def __init__(self): - self.user = None - self.name = None - self.has_cycles = None - self.has_errors = None - self.steps = [] - - -class WorkflowStep(_HasTable): - - def __init__(self): - self.id = None - self.type = None - self.name = None - self.tool_id = None - self.tool_inputs = None - self.tool_errors = None - self.position = None - self.inputs = [] - self.config = None - self.label = None +class RepositoryMetadata(Dictifiable, _HasTable): + # Once the class has been mapped, all Column items in this table will be available + # as instrumented class attributes on RepositoryMetadata. + table = Table('repository_metadata', mapper_registry.metadata, + Column('id', Integer, primary_key=True), + Column('create_time', DateTime, default=now), + Column('update_time', DateTime, default=now, onupdate=now), + Column('repository_id', ForeignKey('repository.id'), index=True), + Column('changeset_revision', TrimmedString(255), index=True), + Column('numeric_revision', Integer, index=True), + Column('metadata', MutableJSONType, nullable=True), + Column('tool_versions', MutableJSONType, nullable=True), + Column('malicious', Boolean, default=False), + Column('downloadable', Boolean, default=True), + Column('missing_test_components', Boolean, default=False, index=True), + Column('has_repository_dependencies', Boolean, default=False, index=True), + Column('includes_datatypes', Boolean, default=False, index=True), + Column('includes_tools', Boolean, default=False, index=True), + Column('includes_tool_dependencies', Boolean, default=False, index=True), + Column('includes_workflows', Boolean, default=False, index=True)) - def get_or_add_input(self, input_name): - for step_input in self.inputs: - if step_input.name == input_name: - return step_input + dict_collection_visible_keys = ['id', 'repository_id', 'numeric_revision', 'changeset_revision', 'malicious', 'downloadable', 'missing_test_components', + 'has_repository_dependencies', 'includes_datatypes', 'includes_tools', 'includes_tool_dependencies', + 'includes_tools_for_display_in_tool_panel', 'includes_workflows'] + dict_element_visible_keys = ['id', 'repository_id', 'numeric_revision', 'changeset_revision', 'malicious', 'downloadable', 'missing_test_components', + 'has_repository_dependencies', 'includes_datatypes', 'includes_tools', 'includes_tool_dependencies', + 'includes_tools_for_display_in_tool_panel', 'includes_workflows', 'repository_dependencies'] - step_input = WorkflowStepInput() - step_input.workflow_step = self - step_input.name = input_name - self.inputs.append(step_input) - return step_input + def __init__(self, id=None, repository_id=None, numeric_revision=None, changeset_revision=None, metadata=None, tool_versions=None, malicious=False, + downloadable=False, missing_test_components=None, tools_functionally_correct=False, test_install_error=False, + has_repository_dependencies=False, includes_datatypes=False, includes_tools=False, includes_tool_dependencies=False, + includes_workflows=False): + self.id = id + self.repository_id = repository_id + self.numeric_revision = numeric_revision + self.changeset_revision = changeset_revision + self.metadata = metadata + self.tool_versions = tool_versions + self.malicious = malicious + self.downloadable = downloadable + self.missing_test_components = missing_test_components + self.has_repository_dependencies = has_repository_dependencies + # We don't consider the special case has_repository_dependencies_only_if_compiling_contained_td here. + self.includes_datatypes = includes_datatypes + self.includes_tools = includes_tools + self.includes_tool_dependencies = includes_tool_dependencies + self.includes_workflows = includes_workflows @property - def input_connections(self): - connections = [_ for step_input in self.inputs for _ in step_input.connections] - return connections - - -class WorkflowStepInput(_HasTable): - - def __init__(self): - self.id = None - self.name = None - self.connections = [] + def includes_tools_for_display_in_tool_panel(self): + if self.metadata: + tool_dicts = self.metadata.get('tools', []) + for tool_dict in tool_dicts: + if tool_dict.get('add_to_tool_panel', True): + return True + return False + @property + def repository_dependencies(self): + if self.has_repository_dependencies: + return [repository_dependency for repository_dependency in self.metadata['repository_dependencies']['repository_dependencies']] + return [] -class WorkflowStepConnection: - def __init__(self): - self.output_step = None - self.output_name = None - self.input_step = None - self.input_name = None +# After the map_imperatively statement has been executed, the members of the +# properties dictionary (repository, reviews) will be available as instrumented +# class attributes on RepositoryMetadata. +mapper_registry.map_imperatively(RepositoryMetadata, RepositoryMetadata.table, properties=dict( + repository=relationship(Repository, back_populates='metadata_revisions'), + reviews=relationship(RepositoryReview, + viewonly=True, + foreign_keys=lambda: [RepositoryReview.repository_id, RepositoryReview.changeset_revision], # type: ignore + primaryjoin=lambda: ((RepositoryReview.repository_id == RepositoryMetadata.repository_id) # type: ignore + & (RepositoryReview.changeset_revision == RepositoryMetadata.changeset_revision)), # type: ignore + back_populates='repository_metadata'))) # Utility methods diff --git a/lib/tool_shed/webapp/model/mapping.py b/lib/tool_shed/webapp/model/mapping.py index 54e664b53a76..e7dc76b0cb35 100644 --- a/lib/tool_shed/webapp/model/mapping.py +++ b/lib/tool_shed/webapp/model/mapping.py @@ -4,310 +4,16 @@ """ import logging -from sqlalchemy import Boolean, Column, DateTime, desc, false, ForeignKey, Integer, MetaData, not_, String, Table, TEXT, true, UniqueConstraint -from sqlalchemy.orm import backref, mapper, relation - import tool_shed.webapp.model import tool_shed.webapp.util.shed_statistics as shed_statistics from galaxy.model.base import SharedModelMapping -from galaxy.model.custom_types import MutableJSONType, TrimmedString from galaxy.model.orm.engine_factory import build_engine -from galaxy.model.orm.now import now -from tool_shed.webapp.model import APIKeys, Category, Component, ComponentReview -from tool_shed.webapp.model import GalaxySession, Group, GroupRoleAssociation -from tool_shed.webapp.model import PasswordResetToken, Repository, RepositoryCategoryAssociation -from tool_shed.webapp.model import RepositoryMetadata, RepositoryRatingAssociation -from tool_shed.webapp.model import RepositoryReview, RepositoryRoleAssociation, Role -from tool_shed.webapp.model import Tag, User, UserGroupAssociation, UserRoleAssociation +from tool_shed.webapp.model import mapper_registry from tool_shed.webapp.security import CommunityRBACAgent log = logging.getLogger(__name__) -metadata = MetaData() - - -APIKeys.table = Table("api_keys", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True), - Column("key", TrimmedString(32), index=True, unique=True)) - -User.table = Table("galaxy_user", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("email", TrimmedString(255), nullable=False), - Column("username", String(255), index=True), - Column("password", TrimmedString(40), nullable=False), - Column("external", Boolean, default=False), - Column("new_repo_alert", Boolean, default=False), - Column("deleted", Boolean, index=True, default=False), - Column("purged", Boolean, index=True, default=False)) - -PasswordResetToken.table = Table("password_reset_token", metadata, - Column("token", String(32), primary_key=True, unique=True, index=True), - Column("expiration_time", DateTime), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True)) - -Group.table = Table("galaxy_group", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("name", String(255), index=True, unique=True), - Column("deleted", Boolean, index=True, default=False)) - -Role.table = Table("role", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("name", String(255), index=True, unique=True), - Column("description", TEXT), - Column("type", String(40), index=True), - Column("deleted", Boolean, index=True, default=False)) - -UserGroupAssociation.table = Table("user_group_association", metadata, - Column("id", Integer, primary_key=True), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True), - Column("group_id", Integer, ForeignKey("galaxy_group.id"), index=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now)) - -UserRoleAssociation.table = Table("user_role_association", metadata, - Column("id", Integer, primary_key=True), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True), - Column("role_id", Integer, ForeignKey("role.id"), index=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now)) - -GroupRoleAssociation.table = Table("group_role_association", metadata, - Column("id", Integer, primary_key=True), - Column("group_id", Integer, ForeignKey("galaxy_group.id"), index=True), - Column("role_id", Integer, ForeignKey("role.id"), index=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now)) - -RepositoryRoleAssociation.table = Table("repository_role_association", metadata, - Column("id", Integer, primary_key=True), - Column("repository_id", Integer, ForeignKey("repository.id"), index=True), - Column("role_id", Integer, ForeignKey("role.id"), index=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now)) - -GalaxySession.table = Table("galaxy_session", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True, nullable=True), - Column("remote_host", String(255)), - Column("remote_addr", String(255)), - Column("referer", TEXT), - Column("session_key", TrimmedString(255), index=True, unique=True), # unique 128 bit random number coerced to a string - Column("is_valid", Boolean, default=False), - Column("prev_session_id", Integer), # saves a reference to the previous session so we have a way to chain them together - Column("last_action", DateTime)) - -Repository.table = Table("repository", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("name", TrimmedString(255), index=True), - Column("type", TrimmedString(255), index=True), - Column("remote_repository_url", TrimmedString(255)), - Column("homepage_url", TrimmedString(255)), - Column("description", TEXT), - Column("long_description", TEXT), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True), - Column("private", Boolean, default=False), - Column("deleted", Boolean, index=True, default=False), - Column("email_alerts", MutableJSONType, nullable=True), - Column("times_downloaded", Integer), - Column("deprecated", Boolean, default=False)) - -RepositoryMetadata.table = Table("repository_metadata", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("repository_id", Integer, ForeignKey("repository.id"), index=True), - Column("changeset_revision", TrimmedString(255), index=True), - Column("numeric_revision", Integer, index=True), - Column("metadata", MutableJSONType, nullable=True), - Column("tool_versions", MutableJSONType, nullable=True), - Column("malicious", Boolean, default=False), - Column("downloadable", Boolean, default=True), - Column("missing_test_components", Boolean, default=False, index=True), - Column("has_repository_dependencies", Boolean, default=False, index=True), - Column("includes_datatypes", Boolean, default=False, index=True), - Column("includes_tools", Boolean, default=False, index=True), - Column("includes_tool_dependencies", Boolean, default=False, index=True), - Column("includes_workflows", Boolean, default=False, index=True)) - -RepositoryReview.table = Table("repository_review", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("repository_id", Integer, ForeignKey("repository.id"), index=True), - Column("changeset_revision", TrimmedString(255), index=True), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True, nullable=False), - Column("approved", TrimmedString(255)), - Column("rating", Integer, index=True), - Column("deleted", Boolean, index=True, default=False)) - -ComponentReview.table = Table("component_review", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("repository_review_id", Integer, ForeignKey("repository_review.id"), index=True), - Column("component_id", Integer, ForeignKey("component.id"), index=True), - Column("comment", TEXT), - Column("private", Boolean, default=False), - Column("approved", TrimmedString(255)), - Column("rating", Integer), - Column("deleted", Boolean, index=True, default=False)) - -Component.table = Table("component", metadata, - Column("id", Integer, primary_key=True), - Column("name", TrimmedString(255)), - Column("description", TEXT)) - -RepositoryRatingAssociation.table = Table("repository_rating_association", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("repository_id", Integer, ForeignKey("repository.id"), index=True), - Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True), - Column("rating", Integer, index=True), - Column("comment", TEXT)) - -RepositoryCategoryAssociation.table = Table("repository_category_association", metadata, - Column("id", Integer, primary_key=True), - Column("repository_id", Integer, ForeignKey("repository.id"), index=True), - Column("category_id", Integer, ForeignKey("category.id"), index=True)) - -Category.table = Table("category", metadata, - Column("id", Integer, primary_key=True), - Column("create_time", DateTime, default=now), - Column("update_time", DateTime, default=now, onupdate=now), - Column("name", TrimmedString(255), index=True, unique=True), - Column("description", TEXT), - Column("deleted", Boolean, index=True, default=False)) - -Tag.table = Table("tag", metadata, - Column("id", Integer, primary_key=True), - Column("type", Integer), - Column("parent_id", Integer, ForeignKey("tag.id")), - Column("name", TrimmedString(255)), - UniqueConstraint("name")) - -# With the tables defined we can define the mappers and setup the relationships between the model objects. -mapper(User, User.table, - properties=dict(active_repositories=relation(Repository, primaryjoin=((Repository.table.c.user_id == User.table.c.id) & (not_(Repository.table.c.deleted))), order_by=(Repository.table.c.name)), - galaxy_sessions=relation(GalaxySession, order_by=desc(GalaxySession.table.c.update_time)), - api_keys=relation(APIKeys, backref="user", order_by=desc(APIKeys.table.c.create_time)))) - -mapper(PasswordResetToken, PasswordResetToken.table, - properties=dict(user=relation(User, backref="reset_tokens"))) - -mapper(APIKeys, APIKeys.table, properties={}) - -mapper(Group, Group.table, - properties=dict(users=relation(UserGroupAssociation))) - -mapper(Role, Role.table, - properties=dict( - repositories=relation(RepositoryRoleAssociation, - primaryjoin=((Role.table.c.id == RepositoryRoleAssociation.table.c.role_id) & (RepositoryRoleAssociation.table.c.repository_id == Repository.table.c.id))), - users=relation(UserRoleAssociation, - primaryjoin=((Role.table.c.id == UserRoleAssociation.table.c.role_id) & (UserRoleAssociation.table.c.user_id == User.table.c.id))), - groups=relation(GroupRoleAssociation, - primaryjoin=((Role.table.c.id == GroupRoleAssociation.table.c.role_id) & (GroupRoleAssociation.table.c.group_id == Group.table.c.id))))) - -mapper(RepositoryRoleAssociation, RepositoryRoleAssociation.table, - properties=dict( - repository=relation(Repository), - role=relation(Role))) - -mapper(UserGroupAssociation, UserGroupAssociation.table, - properties=dict(user=relation(User, backref="groups"), - group=relation(Group, backref="members"))) - -mapper(UserRoleAssociation, UserRoleAssociation.table, - properties=dict( - user=relation(User, backref="roles"), - non_private_roles=relation(User, - backref="non_private_roles", - primaryjoin=((User.table.c.id == UserRoleAssociation.table.c.user_id) & (UserRoleAssociation.table.c.role_id == Role.table.c.id) & not_(Role.table.c.name == User.table.c.email))), - role=relation(Role))) - -mapper(GroupRoleAssociation, GroupRoleAssociation.table, - properties=dict( - group=relation(Group, backref="roles"), - role=relation(Role))) - -mapper(GalaxySession, GalaxySession.table, - properties=dict(user=relation(User))) - -mapper(Tag, Tag.table, - properties=dict(children=relation(Tag, backref=backref('parent', remote_side=[Tag.table.c.id])))) - -mapper(Category, Category.table, - properties=dict(repositories=relation(RepositoryCategoryAssociation, - secondary=Repository.table, - primaryjoin=(Category.table.c.id == RepositoryCategoryAssociation.table.c.category_id), - secondaryjoin=(RepositoryCategoryAssociation.table.c.repository_id == Repository.table.c.id)))) - -mapper(Repository, Repository.table, - properties=dict( - categories=relation(RepositoryCategoryAssociation), - ratings=relation(RepositoryRatingAssociation, order_by=desc(RepositoryRatingAssociation.table.c.update_time), backref="repositories"), - user=relation(User), - downloadable_revisions=relation(RepositoryMetadata, - primaryjoin=((Repository.table.c.id == RepositoryMetadata.table.c.repository_id) & (RepositoryMetadata.table.c.downloadable == true())), - order_by=desc(RepositoryMetadata.table.c.update_time)), - metadata_revisions=relation(RepositoryMetadata, - order_by=desc(RepositoryMetadata.table.c.update_time)), - roles=relation(RepositoryRoleAssociation), - reviews=relation(RepositoryReview, - primaryjoin=(Repository.table.c.id == RepositoryReview.table.c.repository_id)), - reviewers=relation(User, - secondary=RepositoryReview.table, - primaryjoin=(Repository.table.c.id == RepositoryReview.table.c.repository_id), - secondaryjoin=(RepositoryReview.table.c.user_id == User.table.c.id)))) - -mapper(RepositoryMetadata, RepositoryMetadata.table, - properties=dict(repository=relation(Repository), - reviews=relation(RepositoryReview, - foreign_keys=[RepositoryMetadata.table.c.repository_id, RepositoryMetadata.table.c.changeset_revision], - primaryjoin=((RepositoryMetadata.table.c.repository_id == RepositoryReview.table.c.repository_id) & (RepositoryMetadata.table.c.changeset_revision == RepositoryReview.table.c.changeset_revision))))) - -mapper(RepositoryReview, RepositoryReview.table, - properties=dict(repository=relation(Repository, - primaryjoin=(RepositoryReview.table.c.repository_id == Repository.table.c.id)), - # Take care when using the mapper below! It should be used only when a new review is being created for a repository change set revision. - # Keep in mind that repository_metadata records can be removed from the database for certain change set revisions when metadata is being - # reset on a repository! - repository_metadata=relation(RepositoryMetadata, - foreign_keys=[RepositoryReview.table.c.repository_id, RepositoryReview.table.c.changeset_revision], - primaryjoin=((RepositoryReview.table.c.repository_id == RepositoryMetadata.table.c.repository_id) & (RepositoryReview.table.c.changeset_revision == RepositoryMetadata.table.c.changeset_revision)), - backref='review'), - user=relation(User, backref="repository_reviews"), - component_reviews=relation(ComponentReview, - primaryjoin=((RepositoryReview.table.c.id == ComponentReview.table.c.repository_review_id) & (ComponentReview.table.c.deleted == false()))), - private_component_reviews=relation(ComponentReview, - primaryjoin=((RepositoryReview.table.c.id == ComponentReview.table.c.repository_review_id) & (ComponentReview.table.c.deleted == false()) & (ComponentReview.table.c.private == true()))))) - -mapper(ComponentReview, ComponentReview.table, - properties=dict(repository_review=relation(RepositoryReview), - component=relation(Component, - primaryjoin=(ComponentReview.table.c.component_id == Component.table.c.id)))) - -mapper(Component, Component.table) - -mapper(RepositoryRatingAssociation, RepositoryRatingAssociation.table, - properties=dict(repository=relation(Repository), user=relation(User))) -mapper(RepositoryCategoryAssociation, RepositoryCategoryAssociation.table, - properties=dict( - category=relation(Category), - repository=relation(Repository))) +metadata = mapper_registry.metadata class ToolShedModelMapping(SharedModelMapping): @@ -331,7 +37,6 @@ def init(file_path, url, engine_options=None, create_tables=False) -> ToolShedMo result.create_tables = create_tables - # Load local tool shed security policy result.security_agent = CommunityRBACAgent(result) result.shed_counter = shed_statistics.ShedCounter(result) return result diff --git a/lib/tool_shed/webapp/security/__init__.py b/lib/tool_shed/webapp/security/__init__.py index f794ffad19b5..c5ba9486aa5d 100644 --- a/lib/tool_shed/webapp/security/__init__.py +++ b/lib/tool_shed/webapp/security/__init__.py @@ -234,7 +234,7 @@ def user_can_administer_repository(self, user, repository): # of a group that is associated with the role. for gra in role.groups: group = gra.group - for uga in group.members: + for uga in group.users: member = uga.user if member.id == user.id: return True diff --git a/test/unit/shed_unit/model/test_mapping.py b/test/unit/shed_unit/model/test_mapping.py new file mode 100644 index 000000000000..d820bc9c34a6 --- /dev/null +++ b/test/unit/shed_unit/model/test_mapping.py @@ -0,0 +1,1193 @@ +from contextlib import contextmanager +from datetime import datetime, timedelta +from uuid import uuid4 + +import pytest +from sqlalchemy import ( + delete, + select, + UniqueConstraint, +) + +import tool_shed.webapp.model.mapping as mapping + + +class BaseTest: + @pytest.fixture + def cls_(self, model): + """ + Return class under test. + Assumptions: if the class under test is Foo, then the class grouping + the tests should be a subclass of BaseTest, named TestFoo. + """ + prefix = len('Test') + class_name = self.__class__.__name__[prefix:] + return getattr(model, class_name) + + +class TestAPIKeys(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'api_keys' + + def test_columns(self, session, cls_, user): + create_time, user_id, key = datetime.now(), user.id, get_unique_value() + obj = cls_(user_id=user_id, key=key, create_time=create_time) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.user_id == user_id + assert stored_obj.key == key + + def test_relationships(self, session, cls_, user): + obj = cls_(user_id=user.id, key=get_unique_value()) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.user.id == user.id + + +class TestCategory(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'category' + + def test_columns(self, session, cls_): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + name, description, deleted = get_unique_value(), 'b', True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.name = name + obj.description = description + obj.deleted = deleted + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.name == name + assert stored_obj.description == description + assert stored_obj.deleted == deleted + + def test_relationships(self, session, cls_, repository_category_association): + obj = cls_() + obj.name = get_unique_value() + obj.repositories.append(repository_category_association) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repositories == [repository_category_association] + + +class TestComponent(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'component' + + def test_columns(self, session, cls_): + name, description = 'a', 'b' + obj = cls_(name=name, description=description) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.name == name + assert stored_obj.description == description + + +class TestComponentReview(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'component_review' + + def test_columns(self, session, cls_, repository_review, component): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + comment = 'a' + private = True + approved = 'b' + rating = 1 + deleted = True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.repository_review = repository_review + obj.component = component + obj.comment = comment + obj.private = private + obj.approved = approved + obj.rating = rating + obj.deleted = deleted + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.repository_review_id == repository_review.id + assert stored_obj.component_id == component.id + assert stored_obj.comment == comment + assert stored_obj.private == private + assert stored_obj.approved == approved + assert stored_obj.rating == rating + assert stored_obj.deleted == deleted + + def test_relationships(self, session, cls_, repository_review, component): + obj = cls_() + obj.repository_review = repository_review + obj.component = component + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repository_review.id == repository_review.id + assert stored_obj.component.id == component.id + + +class TestGalaxySession(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'galaxy_session' + + def test_columns(self, session, cls_, user, galaxy_session): + + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + remote_host = 'a' + remote_addr = 'b' + referer = 'c' + session_key = get_unique_value() + is_valid = True + last_action = update_time + timedelta(hours=1) + + obj = cls_(user=user, prev_session_id=galaxy_session.id) + + obj.create_time = create_time + obj.update_time = update_time + obj.remote_host = remote_host + obj.remote_addr = remote_addr + obj.referer = referer + obj.session_key = session_key + obj.is_valid = is_valid + obj.last_action = last_action + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.user_id == user.id + assert stored_obj.remote_host == remote_host + assert stored_obj.remote_addr == remote_addr + assert stored_obj.referer == referer + assert stored_obj.session_key == session_key + assert stored_obj.is_valid == is_valid + assert stored_obj.prev_session_id == galaxy_session.id + assert stored_obj.last_action == last_action + + def test_relationships(self, session, cls_, user): + obj = cls_(user=user) + obj.session_key = get_unique_value() + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.user.id == user.id + + +class TestGroup(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'galaxy_group' + + def test_columns(self, session, cls_): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + name = get_unique_value() + deleted = True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.name = name + obj.deleted = deleted + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.name == name + assert stored_obj.deleted == deleted + + def test_relationships( + self, + session, + cls_, + group_role_association, + user_group_association, + ): + obj = cls_(name=get_unique_value()) + obj.roles.append(group_role_association) + obj.users.append(user_group_association) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.roles == [group_role_association] + assert stored_obj.users == [user_group_association] + + +class TestGroupRoleAssociation(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'group_role_association' + + def test_columns(self, session, cls_, group, role): + obj = cls_(group, role) + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + obj.create_time = create_time + obj.update_time = update_time + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.group_id == group.id + assert stored_obj.role_id == role.id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + + def test_relationships(self, session, cls_, group, role): + obj = cls_(group, role) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.group.id == group.id + assert stored_obj.role.id == role.id + + +class TestPasswordResetToken(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'password_reset_token' + + def test_columns_and_relationships(self, session, cls_, user): + token = get_unique_value() + expiration_time = datetime.now() + obj = cls_(user, token) + obj.expiration_time = expiration_time + + where_clause = cls_.token == token + + with dbcleanup(session, obj, where_clause): + stored_obj = get_stored_obj(session, cls_, where_clause=where_clause) + # test columns + assert stored_obj.token == token + assert stored_obj.expiration_time == expiration_time + assert stored_obj.user_id == user.id + # test relationships + assert stored_obj.user.id == user.id + + +class TestRepository(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'repository' + + def test_columns(self, session, cls_, user): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + name = 'a' + type = 'b' + remote_repository_url = 'c' + homepage_url = 'd' + description = 'e' + long_description = 'f' + private = True + deleted = True + email_alerts = False + times_downloaded = 1 + deprecated = True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.name = name + obj.type = type + obj.remote_repository_url = remote_repository_url + obj.homepage_url = homepage_url + obj.description = description + obj.long_description = long_description + obj.user = user + obj.private = private + obj.deleted = deleted + obj.email_alerts = email_alerts + obj.times_downloaded = times_downloaded + obj.deprecated = deprecated + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.name == name + assert stored_obj.type == type + assert stored_obj.remote_repository_url == remote_repository_url + assert stored_obj.homepage_url == homepage_url + assert stored_obj.description == description + assert stored_obj.long_description == long_description + assert stored_obj.user_id == user.id + assert stored_obj.private == private + assert stored_obj.deleted == deleted + assert stored_obj.email_alerts == email_alerts + assert stored_obj.times_downloaded == times_downloaded + assert stored_obj.deprecated == deprecated + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.name == name + assert stored_obj.description == description + assert stored_obj.deleted == deleted + + def test_relationships( + self, + session, + cls_, + repository_category_association, + repository_rating_association, + repository_metadata_factory, + repository_role_association, + repository_review_factory, + user, + user_factory, + ): + obj = cls_() + obj.user = user + obj.categories.append(repository_category_association) + obj.ratings.append(repository_rating_association) + obj.roles.append(repository_role_association) + + reviewer1 = user_factory() + review1 = repository_review_factory() + review1.user = reviewer1 + review1.repository = obj + + reviewer2 = user_factory() + review2 = repository_review_factory() + review2.user = reviewer2 + review2.repository = obj + + metadata1 = repository_metadata_factory() + metadata1.repository = obj + metadata1.downloadable = False + + metadata2 = repository_metadata_factory() + metadata2.repository = obj + metadata2.downloadable = True + + session.add_all([metadata1, metadata2]) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.user.id == user.id + assert stored_obj.categories == [repository_category_association] + assert stored_obj.ratings == [repository_rating_association] + assert stored_obj.roles == [repository_role_association] + assert are_same_entity_collections(stored_obj.reviews, [review1, review2]) + assert are_same_entity_collections(stored_obj.reviewers, [reviewer1, reviewer2]) + assert are_same_entity_collections(stored_obj.metadata_revisions, [metadata1, metadata2]) + assert stored_obj.downloadable_revisions == [metadata2] + + delete_from_database(session, [reviewer1, reviewer2, review1, review2, metadata1, metadata2]) + + +class TestRepositoryCategoryAssociation(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'repository_category_association' + + def test_columns(self, session, cls_, repository, category): + obj = cls_(repository=repository, category=category) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.repository_id == repository.id + assert stored_obj.category_id == category.id + + def test_relationships(self, session, cls_, repository, category): + obj = cls_(repository=repository, category=category) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repository.id == repository.id + assert stored_obj.category.id == category.id + + +class TestRepositoryMetadata(BaseTest): + + def test_table(self, cls_): + assert cls_.table.name == 'repository_metadata' + + def test_columns(self, session, cls_, repository): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + changeset_revision = 'a' + numeric_revision = 1 + metadata = 'b' + tool_versions = 'c' + malicious = True + downloadable = False + missing_test_components = True + has_repository_dependencies = True + includes_datatypes = True + includes_tools = True + includes_tool_dependencies = True + includes_workflows = True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.repository = repository + obj.changeset_revision = changeset_revision + obj.numeric_revision = numeric_revision + obj.metadata = metadata + obj.tool_versions = tool_versions + obj.malicious = malicious + obj.downloadable = downloadable + obj.missing_test_components = missing_test_components + obj.has_repository_dependencies = has_repository_dependencies + obj.includes_datatypes = includes_datatypes + obj.includes_tools = includes_tools + obj.includes_tool_dependencies = includes_tool_dependencies + obj.includes_workflows = includes_workflows + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.repository_id == repository.id + assert stored_obj.changeset_revision == changeset_revision + assert stored_obj.numeric_revision == numeric_revision + assert stored_obj.metadata == metadata + assert stored_obj.tool_versions == tool_versions + assert stored_obj.malicious == malicious + assert stored_obj.downloadable == downloadable + assert stored_obj.missing_test_components == missing_test_components + assert stored_obj.has_repository_dependencies == has_repository_dependencies + assert stored_obj.includes_datatypes == includes_datatypes + assert stored_obj.includes_tools == includes_tools + assert stored_obj.includes_tool_dependencies == includes_tool_dependencies + assert stored_obj.includes_workflows == includes_workflows + + def test_relationships( + self, + session, + cls_, + repository, + repository_review_factory, + user, + ): + + obj = cls_() + obj.repository = repository + obj.changeset_revision = 'nonempty' + + # create 3 reviews + review1 = repository_review_factory() + review2 = repository_review_factory() + review3 = repository_review_factory() + + # set the same user for all reviews + review1.user = user + review2.user = user + review3.user = user + + # set the same repository for all reviews + review1.repository = obj.repository + review2.repository = obj.repository + review3.repository = obj.repository + + # set the same changeset for reviews 1,2 + review1.changeset_revision = obj.changeset_revision + review2.changeset_revision = obj.changeset_revision + review3.changeset_revision = 'something else' # this won't be in reviews for this metadata + + # add to session + session.add(review1) + session.add(review2) + session.add(review3) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repository.id == repository.id + assert are_same_entity_collections(stored_obj.reviews, [review1, review2]) + + +class TestRepositoryRatingAssociation(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'repository_rating_association' + + def test_columns(self, session, cls_, repository, user): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + rating = 1 + comment = 'a' + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.repository = repository + obj.user = user + obj.rating = rating + obj.comment = comment + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.repository_id == repository.id + assert stored_obj.user_id == user.id + assert stored_obj.rating == rating + assert stored_obj.comment == comment + + def test_relationships(self, session, cls_, repository, user): + obj = cls_() + obj.repository = repository + obj.user = user + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repository.id == repository.id + assert stored_obj.user.id == user.id + + +class TestRepositoryReview(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'repository_review' + + def test_columns(self, session, cls_, repository, user): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + changeset_revision = 'a' + approved = 'b' + rating = 1 + deleted = True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.repository = repository + obj.changeset_revision = changeset_revision + obj.user = user + obj.approved = approved + obj.rating = rating + obj.deleted = deleted + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.repository_id == repository.id + assert stored_obj.changeset_revision == changeset_revision + assert stored_obj.user_id == user.id + assert stored_obj.approved == approved + assert stored_obj.rating == rating + assert stored_obj.deleted == deleted + + def test_relationships( + self, + session, + cls_, + repository, + user, + repository_metadata_factory, + component_review_factory + ): + obj = cls_() + changeset = 'nonempty' + obj.changeset_revision = changeset + obj.repository = repository + obj.user = user + + metadata1 = repository_metadata_factory() + metadata2 = repository_metadata_factory() + metadata1.repository = repository + metadata2.repository = repository + metadata1.changeset_revision = changeset + metadata2.changeset_revision = 'something else' + + component_review1 = component_review_factory() + component_review1.repository_review = obj + component_review1.deleted = False + + component_review2 = component_review_factory() + component_review2.repository_review = obj + component_review2.deleted = False + component_review2.private = True + + session.add_all([metadata1, metadata2, component_review1, component_review2]) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repository.id == repository.id + assert stored_obj.user.id == user.id + assert stored_obj.repository_metadata == metadata1 + assert are_same_entity_collections(stored_obj.component_reviews, [component_review1, component_review2]) + assert stored_obj.private_component_reviews == [component_review2] + + delete_from_database(session, [component_review1, component_review2, metadata1, metadata2]) + + +class TestRepositoryRoleAssociation(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'repository_role_association' + + def test_columns(self, session, cls_, repository, role): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + obj = cls_(repository, role) + obj.create_time = create_time + obj.update_time = update_time + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.repository_id == repository.id + assert stored_obj.role_id == role.id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + + def test_relationships(self, session, cls_, repository, role): + obj = cls_(repository, role) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repository.id == repository.id + assert stored_obj.role.id == role.id + + +class TestRole(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'role' + + def test_columns(self, session, cls_): + name, description, type_, deleted = get_unique_value(), 'b', cls_.types.SYSTEM, True + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + obj = cls_(name, description, type_, deleted) + obj.create_time = create_time + obj.update_time = update_time + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.name == name + assert stored_obj.description == description + assert stored_obj.type == type_ + assert stored_obj.deleted == deleted + + def test_relationships( + self, + session, + cls_, + repository_role_association, + user_role_association, + group_role_association_factory, + group, + ): + name, description, type_ = get_unique_value(), 'b', cls_.types.SYSTEM + obj = cls_(name, description, type_) + obj.repositories.append(repository_role_association) + obj.users.append(user_role_association) + + gra = group_role_association_factory(group, obj) + obj.groups.append(gra) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.repositories == [repository_role_association] + assert stored_obj.users == [user_role_association] + assert stored_obj.groups == [gra] + + delete_from_database(session, gra) + + +class TestTag(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'tag' + assert has_unique_constraint(cls_.table, ('name',)) + + def test_columns(self, session, cls_): + parent_tag = cls_() + type_, name = 1, get_unique_value() + obj = cls_(type=type_, name=name) + obj.parent = parent_tag + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.type == type_ + assert stored_obj.parent_id == parent_tag.id + assert stored_obj.name == name + + def test_relationships( + self, + session, + cls_, + ): + obj = cls_() + parent_tag = cls_() + child_tag = cls_() + obj.parent = parent_tag + obj.children.append(child_tag) + + def add_association(assoc_object, assoc_attribute): + assoc_object.tag = obj + getattr(obj, assoc_attribute).append(assoc_object) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.parent.id == parent_tag.id + assert stored_obj.children == [child_tag] + + +class TestUser(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'galaxy_user' + + def test_columns(self, session, cls_): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + email = get_unique_value() + username = get_unique_value() + password = 'c' + external = True + new_repo_alert = True + deleted = True + purged = True + + obj = cls_() + obj.create_time = create_time + obj.update_time = update_time + obj.email = email + obj.username = username + obj.password = password + obj.external = external + obj.new_repo_alert = new_repo_alert + obj.deleted = deleted + obj.purged = purged + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + assert stored_obj.email == email + assert stored_obj.username == username + assert stored_obj.password == password + assert stored_obj.external == external + assert stored_obj.new_repo_alert == new_repo_alert + assert stored_obj.deleted == deleted + assert stored_obj.purged == purged + + def test_relationships( + self, + session, + cls_, + repository, + galaxy_session, + api_keys, + repository_review, + role, + group, + password_reset_token, + user_group_association, + user_role_association, + role_factory, + user_role_association_factory + ): + obj = cls_() + obj.email = get_unique_value() + obj.password = 'a' + obj.active_repositories.append(repository) + obj.galaxy_sessions.append(galaxy_session) + obj.api_keys.append(api_keys) + obj.reset_tokens.append(password_reset_token) + obj.groups.append(user_group_association) + obj.repository_reviews.append(repository_review) + + _private_role = role_factory(name=obj.email) + private_user_role = user_role_association_factory(obj, _private_role) + obj.roles.append(private_user_role) + + _non_private_role = role_factory(name='a') + non_private_user_role = user_role_association_factory(obj, _non_private_role) + obj.roles.append(non_private_user_role) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.active_repositories == [repository] + assert stored_obj.galaxy_sessions == [galaxy_session] + assert stored_obj.api_keys == [api_keys] + assert stored_obj.reset_tokens == [password_reset_token] + assert stored_obj.groups == [user_group_association] + assert stored_obj.repository_reviews == [repository_review] + assert are_same_entity_collections(stored_obj.roles, [private_user_role, non_private_user_role]) + assert stored_obj.non_private_roles == [non_private_user_role] + + +class TestUserGroupAssociation(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'user_group_association' + + def test_columns(self, session, cls_, user, group): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + obj = cls_(user, group) + obj.create_time = create_time + obj.update_time = update_time + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.user_id == user.id + assert stored_obj.group_id == group.id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + + def test_relationships(self, session, cls_, user, group): + obj = cls_(user, group) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.user.id == user.id + assert stored_obj.group.id == group.id + + +class TestUserRoleAssociation(BaseTest): + + def test_table(self, cls_): + assert cls_.__tablename__ == 'user_role_association' + + def test_columns(self, session, cls_, user, role): + create_time = datetime.now() + update_time = create_time + timedelta(hours=1) + obj = cls_(user, role) + obj.create_time = create_time + obj.update_time = update_time + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.id == obj_id + assert stored_obj.user_id == user.id + assert stored_obj.role_id == role.id + assert stored_obj.create_time == create_time + assert stored_obj.update_time == update_time + + def test_relationships(self, session, cls_, user, role): + obj = cls_(user, role) + + with dbcleanup(session, obj) as obj_id: + stored_obj = get_stored_obj(session, cls_, obj_id) + assert stored_obj.user.id == user.id + assert stored_obj.role.id == role.id + + +# Misc. helper fixtures. + +@pytest.fixture(scope='module') +def model(): + db_uri = 'sqlite:///:memory:' + return mapping.init('/tmp', db_uri, create_tables=True) + + +@pytest.fixture +def session(model): + Session = model.session + yield Session() + Session.remove() # Ensures we get a new session for each test + + +@pytest.fixture +def api_keys(model, session): + instance = model.APIKeys(key=get_unique_value()) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def category(model, session): + instance = model.Category(name=get_unique_value()) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def component(model, session): + instance = model.Component() + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def galaxy_session(model, session): + instance = model.GalaxySession(session_key=get_unique_value()) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def group(model, session): + instance = model.Group(name=get_unique_value()) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def group_role_association(model, session): + instance = model.GroupRoleAssociation(None, None) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def password_reset_token(model, session, user): + token = get_unique_value() + instance = model.PasswordResetToken(user, token) + where_clause = model.PasswordResetToken.token == token + yield from dbcleanup_wrapper(session, instance, where_clause) + + +@pytest.fixture +def repository(model, session): + instance = model.Repository() + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def repository_metadata(model, session): + instance = model.RepositoryMetadata() + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def repository_review(model, session, user): + instance = model.RepositoryReview() + instance.user = user + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def repository_category_association(model, session, repository, category): + instance = model.RepositoryCategoryAssociation(repository, category) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def repository_rating_association(model, session): + instance = model.RepositoryRatingAssociation() + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def repository_role_association(model, session, repository, role): + instance = model.RepositoryRoleAssociation(repository, role) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def role(model, session): + instance = model.Role(name=get_unique_value()) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def user(model, session): + instance = model.User(email=get_unique_value(), password='password') + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def user_group_association(model, session, user, group): + instance = model.UserGroupAssociation(user, group) + yield from dbcleanup_wrapper(session, instance) + + +@pytest.fixture +def user_role_association(model, session, user, role): + instance = model.UserRoleAssociation(user, role) + yield from dbcleanup_wrapper(session, instance) + + +# Fixtures yielding factory functions. + +@pytest.fixture +def component_review_factory(model): + def make_instance(*args, **kwds): + return model.ComponentReview(*args, **kwds) + return make_instance + + +@pytest.fixture +def group_role_association_factory(model): + def make_instance(*args, **kwds): + return model.GroupRoleAssociation(*args, **kwds) + return make_instance + + +@pytest.fixture +def repository_metadata_factory(model): + def make_instance(*args, **kwds): + return model.RepositoryMetadata(*args, **kwds) + return make_instance + + +@pytest.fixture +def repository_review_factory(model): + def make_instance(*args, **kwds): + return model.RepositoryReview(*args, **kwds) + return make_instance + + +@pytest.fixture +def role_factory(model): + def make_instance(*args, **kwds): + return model.Role(*args, **kwds) + return make_instance + + +@pytest.fixture +def user_factory(model): + def make_instance(*args, **kwds): + user = model.User(*args, **kwds) + user.email = get_unique_value() + user.password = 'a' + return user + return make_instance + + +@pytest.fixture +def user_role_association_factory(model): + def make_instance(*args, **kwds): + return model.UserRoleAssociation(*args, **kwds) + return make_instance + + +# Test utilities + +def dbcleanup_wrapper(session, obj, where_clause=None): + with dbcleanup(session, obj, where_clause): + yield obj + + +@contextmanager +def dbcleanup(session, obj, where_clause=None): + """ + Use the session to store obj in database; delete from database on exit, bypassing the session. + + If obj does not have an id field, a SQLAlchemy WHERE clause should be provided to construct + a custom select statement. + """ + return_id = where_clause is None + + try: + obj_id = persist(session, obj, return_id) + yield obj_id + finally: + table = obj.table + if where_clause is None: + where_clause = _get_default_where_clause(type(obj), obj_id) + stmt = delete(table).where(where_clause) + session.execute(stmt) + + +def persist(session, obj, return_id=True): + """ + Use the session to store obj in database, then remove obj from session, + so that on a subsequent load from the database we get a clean instance. + """ + session.add(obj) + session.flush() + obj_id = obj.id if return_id else None # save this before obj is expunged + session.expunge(obj) + return obj_id + + +def delete_from_database(session, objects): + """ + Delete each object in objects from database. + May be called at the end of a test if use of a context manager is impractical. + (Assume all objects have the id field as their primary key.) + """ + # Ensure we have a list of objects (check for list explicitly: a model can be iterable) + if not isinstance(objects, list): + objects = [objects] + + for obj in objects: + table = obj.__table__ + stmt = delete(table).where(table.c.id == obj.id) + session.execute(stmt) + + +def get_stored_obj(session, cls, obj_id=None, where_clause=None, unique=False): + # Either obj_id or where_clause must be provided, but not both + assert bool(obj_id) ^ (where_clause is not None) + if where_clause is None: + where_clause = _get_default_where_clause(cls, obj_id) + stmt = select(cls).where(where_clause) + result = session.execute(stmt) + # unique() is required if result contains joint eager loads against collections + # https://gerrit.sqlalchemy.org/c/sqlalchemy/sqlalchemy/+/2253 + if unique: + result = result.unique() + return result.scalar_one() + + +def _get_default_where_clause(cls, obj_id): + where_clause = cls.table.c.id == obj_id + return where_clause + + +def has_unique_constraint(table, fields): + for constraint in table.constraints: + if isinstance(constraint, UniqueConstraint): + col_names = {c.name for c in constraint.columns} + if set(fields) == col_names: + return True + + +def get_unique_value(): + """Generate unique values to accommodate unique constraints.""" + return uuid4().hex + + +def are_same_entity_collections(collection1, collection2): + """ + The 2 arguments are collections of instances of models that have an `id` + attribute as their primary key. + Returns `True` if collections are the same size and contain the same + instances. Instance equality is determined by the object's `id` attribute, + not its Python object identifier. + """ + if len(collection1) != len(collection2): + return False + + collection1.sort(key=lambda item: item.id) + collection2.sort(key=lambda item: item.id) + + for item1, item2 in zip(collection1, collection2): + if item1.id != item2.id: + return False + return True