/usr/share/pyshared/gnatpython/vcs.py is in python-gnatpython 54-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 | ############################################################################
# #
# 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 <http://www.gnu.org/licenses/> #
# #
############################################################################
"""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)
|