# 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
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