def get_latest_downloadable_repository_metadata( trans, repository ):
"""
Return the latest downloadable repository_metadata record for the received repository. This will
return repositories of type unrestricted as well as types repository_suite_definition and
tool_dependency_definition.
"""
encoded_repository_id = trans.security.encode_id( repository.id )
repo = hg_util.get_repo_for_repository( trans.app, repository=repository, repo_path=None, create=False )
tip_ctx = str( repo.changectx( repo.changelog.tip() ) )
repository_metadata = None
try:
repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans.app, encoded_repository_id, tip_ctx )
if repository_metadata is not None and repository_metadata.downloadable:
return repository_metadata
return None
except:
latest_downloadable_revision = metadata_util.get_previous_metadata_changeset_revision( repository,
repo,
tip_ctx,
downloadable=True )
if latest_downloadable_revision == hg_util.INITIAL_CHANGELOG_HASH:
return None
repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans.app,
encoded_repository_id,
latest_downloadable_revision )
if repository_metadata is not None and repository_metadata.downloadable:
return repository_metadata
return None
def get_version_lineage_for_tool( self, repository_id, repository_metadata, guid ):
"""
Return the tool version lineage chain in descendant order for the received
guid contained in the received repsitory_metadata.tool_versions. This function
is called only from the Tool Shed.
"""
repository = suc.get_repository_by_id( self.app, repository_id )
repo = hg_util.get_repo_for_repository( self.app, repository=repository, repo_path=None, create=False )
# Initialize the tool lineage
version_lineage = [ guid ]
# Get all ancestor guids of the received guid.
current_child_guid = guid
for changeset in hg_util.reversed_upper_bounded_changelog( repo, repository_metadata.changeset_revision ):
ctx = repo.changectx( changeset )
rm = suc.get_repository_metadata_by_changeset_revision( self.app, repository_id, str( ctx ) )
if rm:
parent_guid = rm.tool_versions.get( current_child_guid, None )
if parent_guid:
version_lineage.append( parent_guid )
current_child_guid = parent_guid
# Get all descendant guids of the received guid.
current_parent_guid = guid
for changeset in hg_util.reversed_lower_upper_bounded_changelog( repo,
repository_metadata.changeset_revision,
repository.tip( self.app ) ):
ctx = repo.changectx( changeset )
rm = suc.get_repository_metadata_by_changeset_revision( self.app, repository_id, str( ctx ) )
if rm:
tool_versions = rm.tool_versions
for child_guid, parent_guid in tool_versions.items():
if parent_guid == current_parent_guid:
version_lineage.insert( 0, child_guid )
current_parent_guid = child_guid
break
return version_lineage
def get_repository_metadata_by_repository_id_changeset_revision( app, id, changeset_revision, metadata_only=False ):
"""Get a specified metadata record for a specified repository in the tool shed."""
if metadata_only:
repository_metadata = suc.get_repository_metadata_by_changeset_revision( app, id, changeset_revision )
if repository_metadata and repository_metadata.metadata:
return repository_metadata.metadata
return None
return suc.get_repository_metadata_by_changeset_revision( app, id, changeset_revision )
def get_repo_info_dict( app, user, repository_id, changeset_revision ):
repository = suc.get_repository_in_tool_shed( app, repository_id )
repo = hg_util.get_repo_for_repository( app, repository=repository, repo_path=None, create=False )
repository_clone_url = common_util.generate_clone_url_for_repository_in_tool_shed( user, repository )
repository_metadata = suc.get_repository_metadata_by_changeset_revision( app,
repository_id,
changeset_revision )
if not repository_metadata:
# The received changeset_revision is no longer installable, so get the next changeset_revision
# in the repository's changelog. This generally occurs only with repositories of type
# repository_suite_definition or tool_dependency_definition.
next_downloadable_changeset_revision = \
suc.get_next_downloadable_changeset_revision( repository, repo, changeset_revision )
if next_downloadable_changeset_revision:
repository_metadata = suc.get_repository_metadata_by_changeset_revision( app,
repository_id,
next_downloadable_changeset_revision )
if repository_metadata:
# For now, we'll always assume that we'll get repository_metadata, but if we discover our assumption
# is not valid we'll have to enhance the callers to handle repository_metadata values of None in the
# returned repo_info_dict.
metadata = repository_metadata.metadata
if 'tools' in metadata:
includes_tools = True
else:
includes_tools = False
includes_tools_for_display_in_tool_panel = repository_metadata.includes_tools_for_display_in_tool_panel
repository_dependencies_dict = metadata.get( 'repository_dependencies', {} )
repository_dependencies = repository_dependencies_dict.get( 'repository_dependencies', [] )
has_repository_dependencies, has_repository_dependencies_only_if_compiling_contained_td = \
suc.get_repository_dependency_types( repository_dependencies )
if 'tool_dependencies' in metadata:
includes_tool_dependencies = True
else:
includes_tool_dependencies = False
else:
# Here's where we may have to handle enhancements to the callers. See above comment.
includes_tools = False
has_repository_dependencies = False
has_repository_dependencies_only_if_compiling_contained_td = False
includes_tool_dependencies = False
includes_tools_for_display_in_tool_panel = False
ctx = hg_util.get_changectx_for_changeset( repo, changeset_revision )
repo_info_dict = create_repo_info_dict( app=app,
repository_clone_url=repository_clone_url,
changeset_revision=changeset_revision,
ctx_rev=str( ctx.rev() ),
repository_owner=repository.user.username,
repository_name=repository.name,
repository=repository,
repository_metadata=repository_metadata,
tool_dependencies=None,
repository_dependencies=None )
return repo_info_dict, includes_tools, includes_tool_dependencies, includes_tools_for_display_in_tool_panel, \
has_repository_dependencies, has_repository_dependencies_only_if_compiling_contained_td
def check_status_and_reset_downloadable( trans, import_results_tups ):
"""Check the status of each imported repository and set downloadable to False if errors."""
flush = False
for import_results_tup in import_results_tups:
ok, name_owner, message = import_results_tup
name, owner = name_owner
if not ok:
repository = suc.get_repository_by_name_and_owner( trans.app, name, owner )
# Do not allow the repository to be automatically installed if population resulted in errors.
tip_changeset_revision = repository.tip( trans.app )
repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans,
trans.security.encode_id( repository.id ),
tip_changeset_revision )
if repository_metadata:
if repository_metadata.downloadable:
repository_metadata.downloadable = False
trans.sa_session.add( repository_metadata )
if not flush:
flush = True
# Do not allow dependent repository revisions to be automatically installed if population
# resulted in errors.
dependent_downloadable_revisions = suc.get_dependent_downloadable_revisions( trans, repository_metadata )
for dependent_downloadable_revision in dependent_downloadable_revisions:
if dependent_downloadable_revision.downloadable:
dependent_downloadable_revision.downloadable = False
trans.sa_session.add( dependent_downloadable_revision )
if not flush:
flush = True
if flush:
trans.sa_session.flush()
def get_certified_level_one_tuple( self, repository ):
"""
Return True if the latest installable changeset_revision of the received repository is level one certified.
"""
if repository is None:
return ( None, False )
if repository.deleted or repository.deprecated:
return ( None, False )
repo = hg_util.get_repo_for_repository( self.app, repository=repository, repo_path=None, create=False )
# Get the latest installable changeset revision since that is all that is currently configured for testing.
latest_installable_changeset_revision = suc.get_latest_downloadable_changeset_revision( self.app, repository, repo )
if latest_installable_changeset_revision not in [ None, hg_util.INITIAL_CHANGELOG_HASH ]:
encoded_repository_id = self.app.security.encode_id( repository.id )
repository_metadata = suc.get_repository_metadata_by_changeset_revision( self.app,
encoded_repository_id,
latest_installable_changeset_revision )
if repository_metadata:
# Filter out repository revisions that have not been tested.
if repository_metadata.time_last_tested is not None and repository_metadata.tool_test_results is not None:
if repository.type in [ rt_util.REPOSITORY_SUITE_DEFINITION, rt_util.TOOL_DEPENDENCY_DEFINITION ]:
# Look in the tool_test_results dictionary for installation errors.
try:
tool_test_results_dict = repository_metadata.tool_test_results[ 0 ]
except Exception, e:
message = 'Error attempting to retrieve install and test results for repository %s:\n' % str( repository.name )
message += '%s' % str( e )
log.exception( message )
return ( latest_installable_changeset_revision, False )
if 'installation_errors' in tool_test_results_dict:
return ( latest_installable_changeset_revision, False )
return ( latest_installable_changeset_revision, True )
else:
# We have a repository with type Unrestricted.
if repository_metadata.includes_tools:
if repository_metadata.tools_functionally_correct:
return ( latest_installable_changeset_revision, True )
return ( latest_installable_changeset_revision, False )
else:
# Look in the tool_test_results dictionary for installation errors.
try:
tool_test_results_dict = repository_metadata.tool_test_results[ 0 ]
except Exception, e:
message = 'Error attempting to retrieve install and test results for repository %s:\n' % str( repository.name )
message += '%s' % str( e )
log.exception( message )
return ( latest_installable_changeset_revision, False )
if 'installation_errors' in tool_test_results_dict:
return ( latest_installable_changeset_revision, False )
return ( latest_installable_changeset_revision, True )
else:
# No test results.
return ( latest_installable_changeset_revision, False )
def load_viewable_repositories_and_suites_by_category( self ):
# Clear all dictionaries just in case they were previously loaded.
self.certified_level_one_viewable_repositories_and_suites_by_category = {}
self.certified_level_one_viewable_suites_by_category = {}
self.certified_level_two_viewable_repositories_and_suites_by_category = {}
self.certified_level_two_viewable_suites_by_category = {}
self.viewable_repositories_and_suites_by_category = {}
self.viewable_suites_by_category = {}
self.viewable_valid_repositories_and_suites_by_category = {}
self.viewable_valid_suites_by_category = {}
for category in self.sa_session.query( model.Category ):
category_name = str( category.name )
if category not in self.certified_level_one_viewable_repositories_and_suites_by_category:
self.certified_level_one_viewable_repositories_and_suites_by_category[ category_name ] = 0
if category not in self.certified_level_one_viewable_suites_by_category:
self.certified_level_one_viewable_suites_by_category[ category_name ] = 0
if category not in self.viewable_repositories_and_suites_by_category:
self.viewable_repositories_and_suites_by_category[ category_name ] = 0
if category not in self.viewable_suites_by_category:
self.viewable_suites_by_category[ category_name ] = 0
if category not in self.viewable_valid_repositories_and_suites_by_category:
self.viewable_valid_repositories_and_suites_by_category[ category_name ] = 0
if category not in self.viewable_valid_suites_by_category:
self.viewable_valid_suites_by_category[ category_name ] = 0
for rca in category.repositories:
repository = rca.repository
if not repository.deleted and not repository.deprecated:
is_valid = self.is_valid( repository )
encoded_repository_id = self.app.security.encode_id( repository.id )
tip_changeset_hash = repository.tip( self.app )
repository_metadata = suc.get_repository_metadata_by_changeset_revision( self.app,
encoded_repository_id,
tip_changeset_hash )
self.viewable_repositories_and_suites_by_category[ category_name ] += 1
if is_valid:
self.viewable_valid_repositories_and_suites_by_category[ category_name ] += 1
if repository.type in [ rt_util.REPOSITORY_SUITE_DEFINITION ]:
self.viewable_suites_by_category[ category_name ] += 1
if is_valid:
self.viewable_valid_suites_by_category[ category_name ] += 1
if self.is_level_one_certified( repository_metadata ):
self.certified_level_one_viewable_repositories_and_suites_by_category[ category_name ] += 1
if repository.type in [ rt_util.REPOSITORY_SUITE_DEFINITION ]:
self.certified_level_one_viewable_suites_by_category[ category_name ] += 1
def create_review( self, trans, **kwd ):
# The value of the received id is the encoded repository id.
message = escape( kwd.get( 'message', '' ) )
status = kwd.get( 'status', 'done' )
repository_id = kwd.get( 'id', None )
changeset_revision = kwd.get( 'changeset_revision', None )
previous_review_id = kwd.get( 'previous_review_id', None )
create_without_copying = 'create_without_copying' in kwd
if repository_id:
if changeset_revision:
# Make sure there is not already a review of the revision by the user.
repository = suc.get_repository_in_tool_shed( trans.app, repository_id )
if review_util.get_review_by_repository_id_changeset_revision_user_id( app=trans.app,
repository_id=repository_id,
changeset_revision=changeset_revision,
user_id=trans.security.encode_id( trans.user.id ) ):
message = "You have already created a review for revision <b>%s</b> of repository <b>%s</b>." % ( changeset_revision, escape( repository.name ) )
status = "error"
else:
# See if there are any reviews for previous changeset revisions that the user can copy.
if not create_without_copying and \
not previous_review_id and \
review_util.has_previous_repository_reviews( trans.app, repository, changeset_revision ):
return trans.response.send_redirect( web.url_for( controller='repository_review',
action='select_previous_review',
**kwd ) )
# A review can be initially performed only on an installable revision of a repository, so make sure we have metadata associated
# with the received changeset_revision.
repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans.app, repository_id, changeset_revision )
if repository_metadata:
metadata = repository_metadata.metadata
if metadata:
review = trans.app.model.RepositoryReview( repository_id=repository_metadata.repository_id,
changeset_revision=changeset_revision,
user_id=trans.user.id,
rating=None,
deleted=False )
trans.sa_session.add( review )
trans.sa_session.flush()
if previous_review_id:
review_to_copy = review_util.get_review( trans.app, previous_review_id )
self.copy_review( trans, review_to_copy, review )
review_id = trans.security.encode_id( review.id )
message = "Begin your review of revision <b>%s</b> of repository <b>%s</b>." \
% ( changeset_revision, repository.name )
status = 'done'
trans.response.send_redirect( web.url_for( controller='repository_review',
action='edit_review',
id=review_id,
message=message,
status=status ) )
else:
message = "A new review cannot be created for revision <b>%s</b> of repository <b>%s</b>. Select a valid revision and try again." \
% ( changeset_revision, escape( repository.name ) )
kwd[ 'message' ] = message
kwd[ 'status' ] = 'error'
else:
return trans.response.send_redirect( web.url_for( controller='repository_review',
action='manage_repository_reviews',
**kwd ) )
return trans.response.send_redirect( web.url_for( controller='repository_review',
action='view_or_manage_repository',
**kwd ) )
def get_repository_revision_install_info( self, trans, name, owner, changeset_revision, **kwd ):
"""
GET /api/repositories/get_repository_revision_install_info
:param name: the name of the Repository
:param owner: the owner of the Repository
:param changeset_revision: the changeset_revision of the RepositoryMetadata object associated with the Repository
Returns a list of the following dictionaries::
- a dictionary defining the Repository. For example:
{
"deleted": false,
"deprecated": false,
"description": "add_column hello",
"id": "f9cad7b01a472135",
"long_description": "add_column hello",
"name": "add_column",
"owner": "test",
"private": false,
"times_downloaded": 6,
"url": "/api/repositories/f9cad7b01a472135",
"user_id": "f9cad7b01a472135"
}
- a dictionary defining the Repository revision (RepositoryMetadata). For example:
{
"changeset_revision": "3a08cc21466f",
"downloadable": true,
"has_repository_dependencies": false,
"has_repository_dependencies_only_if_compiling_contained_td": false,
"id": "f9cad7b01a472135",
"includes_datatypes": false,
"includes_tool_dependencies": false,
"includes_tools": true,
"includes_tools_for_display_in_tool_panel": true,
"includes_workflows": false,
"malicious": false,
"repository_id": "f9cad7b01a472135",
"url": "/api/repository_revisions/f9cad7b01a472135"
}
- a dictionary including the additional information required to install the repository. For example:
{
"add_column": [
"add_column hello",
"http://[email protected]:9009/repos/test/add_column",
"3a08cc21466f",
"1",
"test",
{},
{}
]
}
"""
# Example URL:
# http://<xyz>/api/repositories/get_repository_revision_install_info?name=<n>&owner=<o>&changeset_revision=<cr>
if name and owner and changeset_revision:
# Get the repository information.
repository = suc.get_repository_by_name_and_owner( trans.app, name, owner )
if repository is None:
log.debug( 'Cannot locate repository %s owned by %s' % ( str( name ), str( owner ) ) )
return {}, {}, {}
encoded_repository_id = trans.security.encode_id( repository.id )
repository_dict = repository.to_dict( view='element',
value_mapper=self.__get_value_mapper( trans ) )
repository_dict[ 'url' ] = web.url_for( controller='repositories',
action='show',
id=encoded_repository_id )
# Get the repository_metadata information.
repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans.app,
encoded_repository_id,
changeset_revision )
if repository_metadata is None:
# The changeset_revision column in the repository_metadata table has been updated with a new
# value value, so find the changeset_revision to which we need to update.
repo = hg_util.get_repo_for_repository( trans.app, repository=repository, repo_path=None, create=False )
new_changeset_revision = suc.get_next_downloadable_changeset_revision( repository, repo, changeset_revision )
repository_metadata = suc.get_repository_metadata_by_changeset_revision( trans.app,
encoded_repository_id,
new_changeset_revision )
changeset_revision = new_changeset_revision
if repository_metadata is not None:
encoded_repository_metadata_id = trans.security.encode_id( repository_metadata.id )
repository_metadata_dict = repository_metadata.to_dict( view='collection',
value_mapper=self.__get_value_mapper( trans ) )
repository_metadata_dict[ 'url' ] = web.url_for( controller='repository_revisions',
action='show',
id=encoded_repository_metadata_id )
# Get the repo_info_dict for installing the repository.
repo_info_dict, \
includes_tools, \
includes_tool_dependencies, \
includes_tools_for_display_in_tool_panel, \
has_repository_dependencies, \
has_repository_dependencies_only_if_compiling_contained_td = \
repository_util.get_repo_info_dict( trans.app,
trans.user,
encoded_repository_id,
changeset_revision )
return repository_dict, repository_metadata_dict, repo_info_dict
else:
log.debug( "Unable to locate repository_metadata record for repository id %s and changeset_revision %s" % \
#.........这里部分代码省略.........
def repository_dependencies( self, trans, id, **kwd ):
"""
GET /api/repository_revisions/{encoded repository_metadata id}/repository_dependencies
Returns a list of dictionaries that each define a specific downloadable revision of a
repository in the Tool Shed. This method returns dictionaries with more information in
them than other methods in this controller. The information about repository_metdata is
enhanced to include information about the repository (e.g., name, owner, etc) associated
with the repository_metadata record.
:param id: the encoded id of the `RepositoryMetadata` object
"""
# Example URL: http://localhost:9009/api/repository_revisions/repository_dependencies/bb125606ff9ea620
repository_dependencies_dicts = []
repository_metadata = metadata_util.get_repository_metadata_by_id( trans, id )
if repository_metadata is None:
log.debug( 'Invalid repository_metadata id received: %s' % str( id ) )
return repository_dependencies_dicts
metadata = repository_metadata.metadata
if metadata is None:
log.debug( 'The repository_metadata record with id %s has no metadata.' % str ( id ) )
return repository_dependencies_dicts
if 'repository_dependencies' in metadata:
rd_tups = metadata[ 'repository_dependencies' ][ 'repository_dependencies' ]
for rd_tup in rd_tups:
tool_shed, name, owner, changeset_revision = rd_tup[ 0:4 ]
repository_dependency = suc.get_repository_by_name_and_owner( trans.app, name, owner )
if repository_dependency is None:
log.dbug( 'Cannot locate repository dependency %s owned by %s.' % ( name, owner ) )
continue
repository_dependency_id = trans.security.encode_id( repository_dependency.id )
repository_dependency_repository_metadata = \
suc.get_repository_metadata_by_changeset_revision( trans, repository_dependency_id, changeset_revision )
if repository_dependency_repository_metadata is None:
# The changeset_revision column in the repository_metadata table has been updated with a new
# value value, so find the changeset_revision to which we need to update.
repo_dir = repository_dependency.repo_path( trans.app )
repo = hg.repository( suc.get_configured_ui(), repo_dir )
new_changeset_revision = suc.get_next_downloadable_changeset_revision( repository_dependency,
repo,
changeset_revision )
repository_dependency_repository_metadata = \
suc.get_repository_metadata_by_changeset_revision( trans,
repository_dependency_id,
new_changeset_revision )
if repository_dependency_repository_metadata is None:
decoded_repository_dependency_id = trans.security.decode_id( repository_dependency_id )
debug_msg = 'Cannot locate repository_metadata with id %d for repository dependency %s owned by %s ' % \
( decoded_repository_dependency_id, str( name ), str( owner ) )
debug_msg += 'using either of these changeset_revisions: %s, %s.' % \
( str( changeset_revision ), str( new_changeset_revision ) )
log.debug( debug_msg )
continue
else:
changeset_revision = new_changeset_revision
repository_dependency_repository_metadata_id = trans.security.encode_id( repository_dependency_repository_metadata.id )
repository_dependency_metadata_dict = \
repository_dependency_repository_metadata.to_dict( view='element',
value_mapper=self.__get_value_mapper( trans ) )
repository_dependency_dict = repository_dependency.to_dict( view='element',
value_mapper=self.__get_value_mapper( trans ) )
# We need to be careful with the entries in our repository_dependency_dict here since this Tool Shed API
# controller is working with repository_metadata records. The above to_dict() method returns a dictionary
# with an id entry for the repository record. However, all of the other methods in this controller have
# the id entry associated with a repository_metadata record id. To avoid confusion, we'll update the
# repository_dependency_metadata_dict with entries from the repository_dependency_dict without using the
# Python dictionary update() method because we do not want to overwrite existing entries.
for k, v in repository_dependency_dict.items():
if k not in repository_dependency_metadata_dict:
repository_dependency_metadata_dict[ k ] = v
repository_dependency_metadata_dict[ 'url' ] = web.url_for( controller='repositories',
action='show',
id=repository_dependency_id )
repository_dependencies_dicts.append( repository_dependency_metadata_dict )
return repository_dependencies_dicts
客服电话
APP下载
官方微信
请发表评论