Source code for ApplicationVersioningSupport.VCSQuery

# Copyright (c) Fraunhofer MEVIS, Germany. All rights reserved.
# **InsertLicense** code

# associated_tests: ApplicationVersioningSupport.tests.VCSQueryTest

import os
import subprocess
from abc import ABCMeta, abstractmethod
from builtins import str

__author__ = 'jmkuhnigk'

class VCSQuery( object ):
  __metaclass__ = ABCMeta

  @abstractmethod
  def getRevision( self, path ):
    """
    Returns the revision of path (i.e. NOT the last changed revision in the working copy).
    
    :param path: Path to a working copy file or directory
    :return: Revision of the repository the path is organized in, None if no evaluation was possible
    """
    return None

  @abstractmethod
  def getWorkingCopyRevisionInfo( self, path ):
    """
    Returns the revision info for the root of the working copy path is located in (i.e. NOT the revision of path or the \
    last changed revision in the working copy). The result format follows svnversion result format, e.g. \
    if you have a mixed checkout, you will get the minimum and maximum revision number of the working copy, separated by \
    a colon. If your working copy contains modifications, the result suffix will contain an 'M'.
    
    :param path: Path to a working copy file or directory
    :return: Revision info in svnversion format for the working copy the path is located in, None if no evaluation was possible
    """
    return None

  @abstractmethod
  def getRepositoryURL( self, path ):
    """
    Returns the repository part of the URL, i.e. for a path https://svn.lan/trunk/dir/file the string https://svn.lan is
     returned.
     
    :param path: Path to a working copy file or directory
    :return: Repository part of the URL, None if no evaluation was possible
    """
    return None

  @abstractmethod
  def getRelativeURL(self, path):
    """
    Returns the relative part of the URL, i.e. for a path "https://svn.lan/trunk/dir/file" the string "trunk/dir/file"
     is returned.
     
    :param path: Path to a working copy file or directory
    :return: Relative part of the URL, None if no evaluation was possible
    """
    return None

  @abstractmethod
  def getRelativeBranchURL( self, path, firstNonBranchSubString = "FME"):
    """
    Returns the 'branch' part of the relative part of the URL, i.e. for a path https://svn.lan/trunk/FMEwork/dir/file \
    the string "trunk" is returned.
    
    :param path: Path to a working copy file or directory
    :param firstNonBranchSubString: string to search for
    :return: Relative part of the URL, None if no evaluation was possible
    """
    return None


  @abstractmethod
  def canEvaluatePath(self, path):
    """
    Returns True if the given path can be evaluated using the VCS
    
    :param path: Path to a working copy file or directory
    :return: True if the given path can be evaluated using the VCS
    """
    return False


[docs]class SVNQuery( VCSQuery ):

  def __init__( self, pathToVCSExe = None ):
    """
    Constructor
    
    :param pathToVCSExe: Path in which svn and svnversion executable can be found (None if they are in the path anyway).
    """
    super(SVNQuery, self).__init__()
    self.__pathToVCSExe = pathToVCSExe

[docs]  def getRevision( self, path ):
    """
    Returns the revision of path (i.e. NOT the last changed revision in the working copy).
    
    :param path: Path to a working copy file or directory
    :return: Revision of the repository the path is organized in, None if no evaluation was possible
    """
    revision = None
    output = self._runSVNCommand( "info", "--show-item=revision", path )
    if output is not None:
      revision = int( output )
    return revision

[docs]  def getWorkingCopyRevisionInfo(self, path):
    """
    Returns the revision info for the root of the working copy path is located in (i.e. NOT the revision of path or the \
    last changed revision in the working copy). The result format follows svnversion result format, e.g. \
    if you have a mixed checkout, you will get the minimum and maximum revision number of the working copy, separated by \
    a colon. If your working copy contains modifications, the result suffix will contain an 'M'.
    
    :param path: Path to a working copy file or directory
    :return: Revision info in svnversion format for the working copy the path is located in, None if no evaluation was possible
    """
    revisionInfo = None
    wcRoot = self._runSVNCommand( "info", "--show-item=wc-root", path )
    if wcRoot:
      revisionInfo = self._runSVNVersionCommand( wcRoot )
    return revisionInfo

[docs]  def getRepositoryURL( self, path ):
    """
    Returns the repository part of the URL, i.e. for a path https://svn.lan/trunk/dir/file the string https://svn.lan is
     returned.
     
    :param path: Path to a working copy file or directory
    :return: Repository part of the URL, None if no evaluation was possible
    """
    return self._runSVNCommand( "info", "--show-item=repos-root-url", path )

[docs]  def getRelativeURL( self, path ):
    """
    Returns the relative part of the URL, i.e. for a path "https://svn.lan/trunk/dir/file" the string "trunk/dir/file"
     is returned.
     
    :param path: Path to a working copy file or directory
    :return: Relative part of the URL, None if no evaluation was possible
    """
    relativeURL = self._runSVNCommand( "info", "--show-item=relative-url", path )
    if relativeURL:
      relativeURL = relativeURL.lstrip( "^/" )
    return relativeURL

[docs]  def getRelativeBranchURL( self, path, firstNonBranchSubString = "FME"):
    """
    Returns the 'branch' part of the relative part of the URL, i.e. for a path https://svn.lan/trunk/FMEwork/dir/file \
    the string "trunk" is returned.
    
    :param path: Path to a working copy file or directory
    :param firstNonBranchSubString: string to search for
    :return: Relative part of the URL, None if no evaluation was possible
    """
    relativeBranchURL = self.getRelativeURL( path )
    if relativeBranchURL:
      splitIndex = relativeBranchURL.find( firstNonBranchSubString )
      relativeBranchURL = os.path.dirname( relativeBranchURL[ :splitIndex ] ) if splitIndex >= 0 else None
    return relativeBranchURL

[docs]  def canEvaluatePath(self, path):
    """
    Returns True if the given path can be evaluated using SVN
    
    :param path: Path to a working copy file or directory
    :return: True if the given path can be evaluated using SVN
    """
    try:
      output = self._runSVNCommand( "info", path )
    except:
      return True
    return output is not None

  def _runSVNCommand( self, *args ):
    return self._runCommand( "svn", *args )

  def _runSVNVersionCommand( self, *args ):
    return self._runCommand( "svnversion", *args )

  def _runCommand( self, command, *args ):
    if self.__pathToVCSExe:
      command = os.path.join( self.__pathToVCSExe, command )
    # shell=True prevents window popups, but must not be used with user-defined arguments (security issue)
    # I would use stdin=None if it weren't failing in the nightly tests with an OSError on Windows. With Python3
    # DEVNULL becomes an option. Not sure if the PIPE thing is an issue at all, though...
    p = subprocess.Popen( "{} {}".format( command, " ".join( args ) ), stdin = subprocess.PIPE,
                          stdout = subprocess.PIPE, stderr = subprocess.PIPE, shell = True )
    output, err = p.communicate( )
    if self.__getCleanedUpSVNOutput( err ):
      output = None
    return self.__getCleanedUpSVNOutput( output )

  def __getCleanedUpSVNOutput( self, output ):
    if output:
      output = str(output, 'utf-8').strip('\n\r')
    if not output:
      output = None
    return output


[docs]def GetQueryForPath( path, pathToSvnExe = None ): #, pathToGitExe = None
  """
  Creates and returns the correctly typed VCSQuery object (e.g. SVNQuery for svn repositories)
  
  :param path: Path to check for versioning system
  :param pathToSvnExe: If omitted, it is assumed that svn commandline executable is in the PATH
  :return: Correctly typed VCSQuery object or None if none is found
  """
  svn = SVNQuery( pathToVCSExe = pathToSvnExe )
  if svn.canEvaluatePath( path ):
    return svn
  #else:
  #  git = GITQuery( pathToVCSExe = pathToGitExe )
  #  if git.canEvaluatePath( path ):
  #    return git
  return None


class _VCSDummyQuery( VCSQuery ):
  """ Dummy Query always returning None/False on all requests. """
  def getRevision( self, path ):
    return None

  def getWorkingCopyRevisionInfo( self, path ):
    return None

  def getRepositoryURL( self, path ):
    return None

  def getRelativeURL(self, path):
    return None

  def getRelativeBranchURL( self, path, firstNonBranchSubString = None ):
    return None

  def canEvaluatePath(self, path):
    return False


[docs]def ExtractPathInfoToFiles( samplePath, targetFileBase  = None ):
  """
  Extracts VCS path info for the given samplePath into files a version string from a version history file and writes \
  it to another files for further processing. See PATH_INFO_FILE_SUFFIXES for the kinds of files created
  
  :param samplePath: Filename of the version history to parse for the current version number
  :param targetFileBase: Folder or file to place the result files in, see PATH_INFO_FILE_SUFFIXES for what's appended \
  for each result. If None, samplePath is used.
  :returns: None
  """
  if targetFileBase is None:
    targetFileBase = samplePath
  queryObject = GetQueryForPath( samplePath )
  if not queryObject:
    # use dummy query always returning None
    queryObject = _VCSDummyQuery()
  results = {}
  for key, suffix in PATH_INFO_FILE_SUFFIXES.items():
    results[ key ] = queryObject.__getattribute__( "get" + key )( samplePath )
  for key, value in results.items():
    targetFileName = targetFileBase + PATH_INFO_FILE_SUFFIXES[ key ]
    __writeTextToFile( str( value ), targetFileName )

PATH_INFO_FILE_SUFFIXES = {
  "RepositoryURL": "_RepositoryURL.tmp",
  "WorkingCopyRevisionInfo": "_WorkingCopyRevisionInfo.tmp",
  "RelativeBranchURL": "_RelativeBranchURL.tmp"
}


def __writeTextToFile( text, filename ):
  try:
    with open( filename, 'w' ) as f:
      f.write( text )
  except IOError as e:
    print('Failed to write file with error ({}): {}'.format( e.errno, e.strerror ))
    raise e