############################################################################ # # # VCS.PY # # # # Copyright (C) 2008 - 2011 Ada Core Technologies, Inc. # # # # This program is free software: you can redistribute it and/or modify # # it under the terms of the GNU General Public License as published by # # the Free Software Foundation, either version 3 of the License, or # # (at your option) any later version. # # # # This program is distributed in the hope that it will be useful, # # but WITHOUT ANY WARRANTY; without even the implied warranty of # # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # # GNU General Public License for more details. # # # # You should have received a copy of the GNU General Public License # # along with this program. If not, see # # # ############################################################################ """Version control management systems interface Currently this module provide a single class called SVN to interact with Subversion repositories. """ from gnatpython.ex import Run from xml.dom import minidom import logging import os # Set the logger for this module svnlogger = logging.getLogger('gnatpython.vcs') class SVN_Error(Exception): pass class SVN(object): """Interface to Subversion ATTRIBUTES root : the root of the subversion directory module : the module path dest : the working directory path branch : the branch rev : the revision used url : the effective Subversion url """ def __init__(self, root, module, dest, branch='trunk', rev=None, use_externals=False): """Initialize a Subversion working environment PARAMETERS root : root of the subversion repository module : module path dest : working directory branch : branch to use rev : revision to use RETURN VALUE a SVN instance REMARKS Currently if the working directory is not a current checkout of the targeted subversion repository, the initialization routine will perform a full checkout. Do not rely on this in your script as the upcoming subversion 1.5 will allow us to set a working dir without doing the checkout. If you want to perform a full checkout of a repository you must call the update method without any argument after working dir initialization. """ self.root = root self.module = module self.dest = dest self.branch = branch self.rev = rev self.cached_status = {} if not use_externals: self.externals = '--ignore-externals' else: self.externals = '' # Resolve url self.url = self.__get_url() try: # Test if the dest directory is an actual Subversion checkout info = self.info() except SVN_Error: # If not then do a checkout. Once Subversion 1.5 is out we should # do only a 'partial' checkout in order to set up the dest # directory svncheckout = Run(['svn', 'checkout', self.externals, self.url, self.dest]) if svncheckout.status: self.__error('svn checkout error:\n' + svncheckout.out) return if info['URL'] != self.url: # The dest directory is actually a checkout but not on the right # URL. So do a svn switch svnswitch = Run(['svn', 'switch', self.url, self.dest]) if svnswitch.status: self.__error('svn switch error:\n' + svnswitch.out) def __info(self, url): """Internal function""" results = {} svninfo = Run(['svn', 'info', url]) if svninfo.status: self.__error('svn info error:\n' + svninfo.out) for line in svninfo.out.splitlines(): fields = line.split(':', 1) if len(fields) > 1: results[fields[0]] = fields[1].strip() return results def info(self, path=''): """Get info on a file PARAMETERS file : a path relative to the working dir. The default '' returns the status of '.' RETURN VALUE A dictionnary containing the following keys: 'Path' 'Name' 'URL' 'Repository Root' 'Repository UUID' 'Revision' 'Node Kind' 'Schedule' 'Last Changed Author' 'Last Changed Rev' 'Last Changed Date' 'Text Last Updated' 'Checksum' key values are strings REMARKS None """ return self.__info(os.path.join(self.dest, path)) def __get_url(self): """Internal function""" return self.root + '/' + self.branch + '/' + self.module def update(self, files=None): """Update a set of files PARAMETERS files : a list of path relative to the working dir. If not set then an update of the whole working dir is done. RETURN VALUE None REMARKS None """ if files is None: files = [''] for f in files: svnupdate = Run(['svn', 'update', self.externals, f], cwd=self.dest) if svnupdate.status: self.__error('svn update error:\n' + svnupdate.out) def add(self, files): """Add a set of files PARAMETERS files : the list of files to add. RETURN VALUE None """ svnadd = Run(['svn', 'add'] + files, cwd=self.dest) if svnadd.status: self.__error('svn add error:\n' + svnadd.out) def commit(self, msg, files=None): """Commit a set of files PARAMETERS msg : the commit message (should be different from '') files : the list of files to commit. If not set then do a commit on working dir RETURN VALUE None REMARKS Before commit a check is done to see if the local copy of the files are up-to-date. If not the checkin is aborted and SVN_Error is raised. """ if not self.is_uptodate(files): svnlogger.error('svn commit error: files not up-to-date') if not self.has_diff(files, True): # There are no local modifications so just return return if files is None: files = [] svncommit = Run(['svn', 'commit', '-m', msg] + files, cwd=self.dest) if svncommit.status: self.__error('svn commit error:\n' + svncommit.out) def is_uptodate(self, files=None, use_cached_status=False): """Check if a set of files are up-to-date PARAMETERS files : the list of files we are interested in. Otherwise check if the overall working is up-to-date use_cached_status : if True use cached status. RETURN VALUE True if the files are up-to-date, False otherwise REMARKS None """ svnstatus = self.status(use_cached_status) if files is None: # If an empty list is passed check that all the files are # up-to-date for f in svnstatus: if not svnstatus[f]['uptodate']: return False return True else: # Otherwise check only the files pass by the caller for f in files: if f in svnstatus and not svnstatus[f]['uptodate']: return False return True def status(self, use_cached_status=False): """Get the status of the working directory PARAMETERS use_cached_status : if True return the cached status. RETURN VALUE A dictionnary containing a key for each file for which the status changed Each key contains a dictionnary with the following keys: - status: a character identifying the current file status. (see svn help status for more info) - uptodate: True if the file is up-to-date, False otherwise - rev: the current revision string REMARKS None """ if use_cached_status: return self.cached_status result = {} svnstatus = Run(['svn', 'status', '-u', self.dest]) for line in svnstatus.out.splitlines(): if line.startswith('Status'): break status = line[0] if line[7] == '*': uptodate = False else: uptodate = True if status == '?': rev = '' f = line[8:].lstrip() else: fields = line[8:].lstrip().split(None, 1) rev = fields[0] f = fields[1] result[f] = {'status': status, 'rev': rev, 'uptodate': uptodate} self.cached_status = result return result def has_diff(self, files=None, use_cached_status=False): """Check if there some local changes on a set of files PARAMETERS files : a list of files. If not set the overall working dir is taken into account. use_cached_status : if True use cached status. RETURN VALUE True if a least one file contains local changes. False otherwise. REMARKS None """ svnstatus = self.status(use_cached_status) if files is None: # If an empty list is passed check that all files local modifs for f in svnstatus: if svnstatus[f]['status'] in ('A', 'M'): return True return False else: # Otherwise check only the files pass by the caller for f in [self.dest + '/' + f for f in files]: if f in svnstatus and svnstatus[f]['status'] in ('A', 'M'): return True return False def log(self, rev=None, path=None): """Returns logs messages PARAMETERS rev : the revision range. If not set, gets all logs from the beginning path : the file or directory to get logs from. If not set, gets the overall working dir's logs. RETURN VALUE a list of dictionnaries containg keys : revision, author, date, msg """ cmd = ['svn', 'log', '--xml'] if rev: cmd.append('-r') cmd.append(str(rev)) if path: cmd.append(path) svnlog = Run(cmd, cwd=self.dest) if svnlog.status: self.__error('svn log error:\n' + svnlog.out) # parse log xml_log = minidom.parseString(svnlog.out) logs = [] for node in xml_log.getElementsByTagName("logentry"): entry = {} if node.getAttribute('revision'): entry['rev'] = node.getAttribute('revision') if node.getElementsByTagName('author'): entry['author'] = node.getElementsByTagName( 'author')[0].firstChild.data if node.getElementsByTagName('date'): entry['date'] = node.getElementsByTagName( 'date')[0].firstChild.data if node.getElementsByTagName('msg'): entry['msg'] = node.getElementsByTagName( 'msg')[0].firstChild.data logs.append(entry) return logs @classmethod def __error(cls, msg): """Log the message and raise SVN_Error""" svnlogger.error(msg) raise SVN_Error(msg)