/
niceSOFT
/
python3-py
Обзор
Документация
Войти
/
niceSOFT
/
python3-py
Код
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
1.4.21
py/_path/svnwc.py
1 238 строк
43 KB
holger krekel
try to fix failures related to subversion 1.7.
24 ноя 2011, 14:21
24 ноя 2011, 14:21
f6ac6b0
Код
Авторство
О чём код?
""" svn-Command based Implementation of a Subversion WorkingCopy Path. SvnWCCommandPath is the main class. """ import os, sys, time, re, calendar import py import subprocess from py._path import common #----------------------------------------------------------- # Caching latest repository revision and repo-paths # (getting them is slow with the current implementations) # # XXX make mt-safe #----------------------------------------------------------- class cache: proplist = {} info = {} entries = {} prop = {} class RepoEntry: def __init__(self, url, rev, timestamp): self.url = url self.rev = rev self.timestamp = timestamp def __str__(self): return "repo: %s;%s %s" %(self.url, self.rev, self.timestamp) class RepoCache: """ The Repocache manages discovered repository paths and their revisions. If inside a timeout the cache will even return the revision of the root. """ timeout = 20 # seconds after which we forget that we know the last revision def __init__(self): self.repos = [] def clear(self): self.repos = [] def put(self, url, rev, timestamp=None): if rev is None: return if timestamp is None: timestamp = time.time() for entry in self.repos: if url == entry.url: entry.timestamp = timestamp entry.rev = rev #print "set repo", entry break else: entry = RepoEntry(url, rev, timestamp) self.repos.append(entry) #print "appended repo", entry def get(self, url): now = time.time() for entry in self.repos: if url.startswith(entry.url): if now < entry.timestamp + self.timeout: #print "returning immediate Etrny", entry return entry.url, entry.rev return entry.url, -1 return url, -1 repositories = RepoCache() # svn support code ALLOWED_CHARS = "_ -/\\=$.~+%" #add characters as necessary when tested if sys.platform == "win32": ALLOWED_CHARS += ":" ALLOWED_CHARS_HOST = ALLOWED_CHARS + '@:' def _getsvnversion(ver=[]): try: return ver[0] except IndexError: v = py.process.cmdexec("svn -q --version") v.strip() v = '.'.join(v.split('.')[:2]) ver.append(v) return v def _escape_helper(text): text = str(text) if py.std.sys.platform != 'win32': text = str(text).replace('$', '\\$') return text def _check_for_bad_chars(text, allowed_chars=ALLOWED_CHARS): for c in str(text): if c.isalnum(): continue if c in allowed_chars: continue return True return False def checkbadchars(url): # (hpk) not quite sure about the exact purpose, guido w.? proto, uri = url.split("://", 1) if proto != "file": host, uripath = uri.split('/', 1) # only check for bad chars in the non-protocol parts if (_check_for_bad_chars(host, ALLOWED_CHARS_HOST) \ or _check_for_bad_chars(uripath, ALLOWED_CHARS)): raise ValueError("bad char in %r" % (url, )) #_______________________________________________________________ class SvnPathBase(common.PathBase): """ Base implementation for SvnPath implementations. """ sep = '/' def _geturl(self): return self.strpath url = property(_geturl, None, None, "url of this svn-path.") def __str__(self): """ return a string representation (including rev-number) """ return self.strpath def __hash__(self): return hash(self.strpath) def new(self, **kw): """ create a modified version of this path. A 'rev' argument indicates a new revision. the following keyword arguments modify various path parts:: http://host.com/repo/path/file.ext |-----------------------| dirname |------| basename |--| purebasename |--| ext """ obj = object.__new__(self.__class__) obj.rev = kw.get('rev', self.rev) obj.auth = kw.get('auth', self.auth) dirname, basename, purebasename, ext = self._getbyspec( "dirname,basename,purebasename,ext") if 'basename' in kw: if 'purebasename' in kw or 'ext' in kw: raise ValueError("invalid specification %r" % kw) else: pb = kw.setdefault('purebasename', purebasename) ext = kw.setdefault('ext', ext) if ext and not ext.startswith('.'): ext = '.' + ext kw['basename'] = pb + ext kw.setdefault('dirname', dirname) kw.setdefault('sep', self.sep) if kw['basename']: obj.strpath = "%(dirname)s%(sep)s%(basename)s" % kw else: obj.strpath = "%(dirname)s" % kw return obj def _getbyspec(self, spec): """ get specified parts of the path. 'arg' is a string with comma separated path parts. The parts are returned in exactly the order of the specification. you may specify the following parts: http://host.com/repo/path/file.ext |-----------------------| dirname |------| basename |--| purebasename |--| ext """ res = [] parts = self.strpath.split(self.sep) for name in spec.split(','): name = name.strip() if name == 'dirname': res.append(self.sep.join(parts[:-1])) elif name == 'basename': res.append(parts[-1]) else: basename = parts[-1] i = basename.rfind('.') if i == -1: purebasename, ext = basename, '' else: purebasename, ext = basename[:i], basename[i:] if name == 'purebasename': res.append(purebasename) elif name == 'ext': res.append(ext) else: raise NameError("Don't know part %r" % name) return res def __eq__(self, other): """ return true if path and rev attributes each match """ return (str(self) == str(other) and (self.rev == other.rev or self.rev == other.rev)) def __ne__(self, other): return not self == other def join(self, *args): """ return a new Path (with the same revision) which is composed of the self Path followed by 'args' path components. """ if not args: return self args = tuple([arg.strip(self.sep) for arg in args]) parts = (self.strpath, ) + args newpath = self.__class__(self.sep.join(parts), self.rev, self.auth) return newpath def propget(self, name): """ return the content of the given property. """ value = self._propget(name) return value def proplist(self): """ list all property names. """ content = self._proplist() return content def size(self): """ Return the size of the file content of the Path. """ return self.info().size def mtime(self): """ Return the last modification time of the file. """ return self.info().mtime # shared help methods def _escape(self, cmd): return _escape_helper(cmd) #def _childmaxrev(self): # """ return maximum revision number of childs (or self.rev if no childs) """ # rev = self.rev # for name, info in self._listdir_nameinfo(): # rev = max(rev, info.created_rev) # return rev #def _getlatestrevision(self): # """ return latest repo-revision for this path. """ # url = self.strpath # path = self.__class__(url, None) # # # we need a long walk to find the root-repo and revision # while 1: # try: # rev = max(rev, path._childmaxrev()) # previous = path # path = path.dirpath() # except (IOError, process.cmdexec.Error): # break # if rev is None: # raise IOError, "could not determine newest repo revision for %s" % self # return rev class Checkers(common.Checkers): def dir(self): try: return self.path.info().kind == 'dir' except py.error.Error: return self._listdirworks() def _listdirworks(self): try: self.path.listdir() except py.error.ENOENT: return False else: return True def file(self): try: return self.path.info().kind == 'file' except py.error.ENOENT: return False def exists(self): try: return self.path.info() except py.error.ENOENT: return self._listdirworks() def parse_apr_time(timestr): i = timestr.rfind('.') if i == -1: raise ValueError("could not parse %s" % timestr) timestr = timestr[:i] parsedtime = time.strptime(timestr, "%Y-%m-%dT%H:%M:%S") return time.mktime(parsedtime) class PropListDict(dict): """ a Dictionary which fetches values (InfoSvnCommand instances) lazily""" def __init__(self, path, keynames): dict.__init__(self, [(x, None) for x in keynames]) self.path = path def __getitem__(self, key): value = dict.__getitem__(self, key) if value is None: value = self.path.propget(key) dict.__setitem__(self, key, value) return value def fixlocale(): if sys.platform != 'win32': return 'LC_ALL=C ' return '' # some nasty chunk of code to solve path and url conversion and quoting issues ILLEGAL_CHARS = '* | \ / : < > ? \t \n \x0b \x0c \r'.split(' ') if os.sep in ILLEGAL_CHARS: ILLEGAL_CHARS.remove(os.sep) ISWINDOWS = sys.platform == 'win32' _reg_allow_disk = re.compile(r'^([a-z]\:\\)?[^:]+$', re.I) def _check_path(path): illegal = ILLEGAL_CHARS[:] sp = path.strpath if ISWINDOWS: illegal.remove(':') if not _reg_allow_disk.match(sp): raise ValueError('path may not contain a colon (:)') for char in sp: if char not in string.printable or char in illegal: raise ValueError('illegal character %r in path' % (char,)) def path_to_fspath(path, addat=True): _check_path(path) sp = path.strpath if addat and path.rev != -1: sp = '%s@%s' % (sp, path.rev) elif addat: sp = '%s@HEAD' % (sp,) return sp def url_from_path(path): fspath = path_to_fspath(path, False) quote = py.std.urllib.quote if ISWINDOWS: match = _reg_allow_disk.match(fspath) fspath = fspath.replace('\\', '/') if match.group(1): fspath = '/%s%s' % (match.group(1).replace('\\', '/'), quote(fspath[len(match.group(1)):])) else: fspath = quote(fspath) else: fspath = quote(fspath) if path.rev != -1: fspath = '%s@%s' % (fspath, path.rev) else: fspath = '%s@HEAD' % (fspath,) return 'file://%s' % (fspath,) class SvnAuth(object): """ container for auth information for Subversion """ def __init__(self, username, password, cache_auth=True, interactive=True): self.username = username self.password = password self.cache_auth = cache_auth self.interactive = interactive def makecmdoptions(self): uname = self.username.replace('"', '\\"') passwd = self.password.replace('"', '\\"') ret = [] if uname: ret.append('--username="%s"' % (uname,)) if passwd: ret.append('--password="%s"' % (passwd,)) if not self.cache_auth: ret.append('--no-auth-cache') if not self.interactive: ret.append('--non-interactive') return ' '.join(ret) def __str__(self): return "<SvnAuth username=%s ...>" %(self.username,) rex_blame = re.compile(r'\s*(\d+)\s*(\S+) (.*)') class SvnWCCommandPath(common.PathBase): """ path implementation offering access/modification to svn working copies. It has methods similar to the functions in os.path and similar to the commands of the svn client. """ sep = os.sep def __new__(cls, wcpath=None, auth=None): self = object.__new__(cls) if isinstance(wcpath, cls): if wcpath.__class__ == cls: return wcpath wcpath = wcpath.localpath if _check_for_bad_chars(str(wcpath), ALLOWED_CHARS): raise ValueError("bad char in wcpath %s" % (wcpath, )) self.localpath = py.path.local(wcpath) self.auth = auth return self strpath = property(lambda x: str(x.localpath), None, None, "string path") rev = property(lambda x: x.info(usecache=0).rev, None, None, "revision") def __eq__(self, other): return self.localpath == getattr(other, 'localpath', None) def _geturl(self): if getattr(self, '_url', None) is None: info = self.info() self._url = info.url #SvnPath(info.url, info.rev) assert isinstance(self._url, py.builtin._basestring) return self._url url = property(_geturl, None, None, "url of this WC item") def _escape(self, cmd): return _escape_helper(cmd) def dump(self, obj): """ pickle object into path location""" return self.localpath.dump(obj) def svnurl(self): """ return current SvnPath for this WC-item. """ info = self.info() return py.path.svnurl(info.url) def __repr__(self): return "svnwc(%r)" % (self.strpath) # , self._url) def __str__(self): return str(self.localpath) def _makeauthoptions(self): if self.auth is None: return '' return self.auth.makecmdoptions() def _authsvn(self, cmd, args=None): args = args and list(args) or [] args.append(self._makeauthoptions()) return self._svn(cmd, *args) def _svn(self, cmd, *args): l = ['svn %s' % cmd] args = [self._escape(item) for item in args] l.extend(args) l.append('"%s"' % self._escape(self.strpath)) # try fixing the locale because we can't otherwise parse string = fixlocale() + " ".join(l) try: try: key = 'LC_MESSAGES' hold = os.environ.get(key) os.environ[key] = 'C' out = py.process.cmdexec(string) finally: if hold: os.environ[key] = hold else: del os.environ[key] except py.process.cmdexec.Error: e = sys.exc_info()[1] strerr = e.err.lower() if strerr.find('not found') != -1: raise py.error.ENOENT(self) elif strerr.find("E200009:") != -1: raise py.error.ENOENT(self) if (strerr.find('file exists') != -1 or strerr.find('file already exists') != -1 or strerr.find('w150002:') != -1 or strerr.find("can't create directory") != -1): raise py.error.EEXIST(self) raise return out def switch(self, url): """ switch to given URL. """ self._authsvn('switch', [url]) def checkout(self, url=None, rev=None): """ checkout from url to local wcpath. """ args = [] if url is None: url = self.url if rev is None or rev == -1: if (py.std.sys.platform != 'win32' and _getsvnversion() == '1.3'): url += "@HEAD" else: if _getsvnversion() == '1.3': url += "@%d" % rev else: args.append('-r' + str(rev)) args.append(url) self._authsvn('co', args) def update(self, rev='HEAD', interactive=True): """ update working copy item to given revision. (None -> HEAD). """ opts = ['-r', rev] if not interactive: opts.append("--non-interactive") self._authsvn('up', opts) def write(self, content, mode='w'): """ write content into local filesystem wc. """ self.localpath.write(content, mode) def dirpath(self, *args): """ return the directory Path of the current Path. """ return self.__class__(self.localpath.dirpath(*args), auth=self.auth) def _ensuredirs(self): parent = self.dirpath() if parent.check(dir=0): parent._ensuredirs() if self.check(dir=0): self.mkdir() return self def ensure(self, *args, **kwargs): """ ensure that an args-joined path exists (by default as a file). if you specify a keyword argument 'directory=True' then the path is forced to be a directory path. """ p = self.join(*args) if p.check(): if p.check(versioned=False): p.add() return p if kwargs.get('dir', 0): return p._ensuredirs() parent = p.dirpath() parent._ensuredirs() p.write("") p.add() return p def mkdir(self, *args): """ create & return the directory joined with args. """ if args: return self.join(*args).mkdir() else: self._svn('mkdir') return self def add(self): """ add ourself to svn """ self._svn('add') def remove(self, rec=1, force=1): """ remove a file or a directory tree. 'rec'ursive is ignored and considered always true (because of underlying svn semantics. """ assert rec, "svn cannot remove non-recursively" if not self.check(versioned=True): # not added to svn (anymore?), just remove py.path.local(self).remove() return flags = [] if force: flags.append('--force') self._svn('remove', *flags) def copy(self, target): """ copy path to target.""" py.process.cmdexec("svn copy %s %s" %(str(self), str(target))) def rename(self, target): """ rename this path to target. """ py.process.cmdexec("svn move --force %s %s" %(str(self), str(target))) def lock(self): """ set a lock (exclusive) on the resource """ out = self._authsvn('lock').strip() if not out: # warning or error, raise exception raise ValueError("unknown error in svn lock command") def unlock(self): """ unset a previously set lock """ out = self._authsvn('unlock').strip() if out.startswith('svn:'): # warning or error, raise exception raise Exception(out[4:]) def cleanup(self): """ remove any locks from the resource """ # XXX should be fixed properly!!! try: self.unlock() except: pass def status(self, updates=0, rec=0, externals=0): """ return (collective) Status object for this file. """ # http://svnbook.red-bean.com/book.html#svn-ch-3-sect-4.3.1 # 2201 2192 jum test # XXX if externals: raise ValueError("XXX cannot perform status() " "on external items yet") else: #1.2 supports: externals = '--ignore-externals' externals = '' if rec: rec= '' else: rec = '--non-recursive' # XXX does not work on all subversion versions #if not externals: # externals = '--ignore-externals' if updates: updates = '-u' else: updates = '' try: cmd = 'status -v --xml --no-ignore %s %s %s' % ( updates, rec, externals) out = self._authsvn(cmd) except py.process.cmdexec.Error: cmd = 'status -v --no-ignore %s %s %s' % ( updates, rec, externals) out = self._authsvn(cmd) rootstatus = WCStatus(self).fromstring(out, self) else: rootstatus = XMLWCStatus(self).fromstring(out, self) return rootstatus def diff(self, rev=None): """ return a diff of the current path against revision rev (defaulting to the last one). """ args = [] if rev is not None: args.append("-r %d" % rev) out = self._authsvn('diff', args) return out def blame(self): """ return a list of tuples of three elements: (revision, commiter, line) """ out = self._svn('blame') result = [] blamelines = out.splitlines() reallines = py.path.svnurl(self.url).readlines() for i, (blameline, line) in enumerate( zip(blamelines, reallines)): m = rex_blame.match(blameline) if not m: raise ValueError("output line %r of svn blame does not match " "expected format" % (line, )) rev, name, _ = m.groups() result.append((int(rev), name, line)) return result _rex_commit = re.compile(r'.*Committed revision (\d+)\.$', re.DOTALL) def commit(self, msg='', rec=1): """ commit with support for non-recursive commits """ # XXX i guess escaping should be done better here?!? cmd = 'commit -m "%s" --force-log' % (msg.replace('"', '\\"'),) if not rec: cmd += ' -N' out = self._authsvn(cmd) try: del cache.info[self] except KeyError: pass if out: m = self._rex_commit.match(out) return int(m.group(1)) def propset(self, name, value, *args): """ set property name to value on this path. """ d = py.path.local.mkdtemp() try: p = d.join('value') p.write(value) self._svn('propset', name, '--file', str(p), *args) finally: d.remove() def propget(self, name): """ get property name on this path. """ res = self._svn('propget', name) return res[:-1] # strip trailing newline def propdel(self, name): """ delete property name on this path. """ res = self._svn('propdel', name) return res[:-1] # strip trailing newline def proplist(self, rec=0): """ return a mapping of property names to property values. If rec is True, then return a dictionary mapping sub-paths to such mappings. """ if rec: res = self._svn('proplist -R') return make_recursive_propdict(self, res) else: res = self._svn('proplist') lines = res.split('\n') lines = [x.strip() for x in lines[1:]] return PropListDict(self, lines) def revert(self, rec=0): """ revert the local changes of this path. if rec is True, do so recursively. """ if rec: result = self._svn('revert -R') else: result = self._svn('revert') return result def new(self, **kw): """ create a modified version of this path. A 'rev' argument indicates a new revision. the following keyword arguments modify various path parts: http://host.com/repo/path/file.ext |-----------------------| dirname |------| basename |--| purebasename |--| ext """ if kw: localpath = self.localpath.new(**kw) else: localpath = self.localpath return self.__class__(localpath, auth=self.auth) def join(self, *args, **kwargs): """ return a new Path (with the same revision) which is composed of the self Path followed by 'args' path components. """ if not args: return self localpath = self.localpath.join(*args, **kwargs) return self.__class__(localpath, auth=self.auth) def info(self, usecache=1): """ return an Info structure with svn-provided information. """ info = usecache and cache.info.get(self) if not info: try: output = self._svn('info') except py.process.cmdexec.Error: e = sys.exc_info()[1] if e.err.find('Path is not a working copy directory') != -1: raise py.error.ENOENT(self, e.err) elif e.err.find("is not under version control") != -1: raise py.error.ENOENT(self, e.err) raise # XXX SVN 1.3 has output on stderr instead of stdout (while it does # return 0!), so a bit nasty, but we assume no output is output # to stderr... if (output.strip() == '' or output.lower().find('not a versioned resource') != -1): raise py.error.ENOENT(self, output) info = InfoSvnWCCommand(output) # Can't reliably compare on Windows without access to win32api if py.std.sys.platform != 'win32': if info.path != self.localpath: raise py.error.ENOENT(self, "not a versioned resource:" + " %s != %s" % (info.path, self.localpath)) cache.info[self] = info return info def listdir(self, fil=None, sort=None): """ return a sequence of Paths. listdir will return either a tuple or a list of paths depending on implementation choices. """ if isinstance(fil, str): fil = common.FNMatcher(fil) # XXX unify argument naming with LocalPath.listdir def notsvn(path): return path.basename != '.svn' paths = [] for localpath in self.localpath.listdir(notsvn): p = self.__class__(localpath, auth=self.auth) if notsvn(p) and (not fil or fil(p)): paths.append(p) self._sortlist(paths, sort) return paths def open(self, mode='r'): """ return an opened file with the given mode. """ return open(self.strpath, mode) def _getbyspec(self, spec): return self.localpath._getbyspec(spec) class Checkers(py.path.local.Checkers): def __init__(self, path): self.svnwcpath = path self.path = path.localpath def versioned(self): try: s = self.svnwcpath.info() except (py.error.ENOENT, py.error.EEXIST): return False except py.process.cmdexec.Error: e = sys.exc_info()[1] if e.err.find('is not a working copy')!=-1: return False if e.err.lower().find('not a versioned resource') != -1: return False raise else: return True def log(self, rev_start=None, rev_end=1, verbose=False): """ return a list of LogEntry instances for this path. rev_start is the starting revision (defaulting to the first one). rev_end is the last revision (defaulting to HEAD). if verbose is True, then the LogEntry instances also know which files changed. """ assert self.check() # make it simpler for the pipe rev_start = rev_start is None and "HEAD" or rev_start rev_end = rev_end is None and "HEAD" or rev_end if rev_start == "HEAD" and rev_end == 1: rev_opt = "" else: rev_opt = "-r %s:%s" % (rev_start, rev_end) verbose_opt = verbose and "-v" or "" locale_env = fixlocale() # some blather on stderr auth_opt = self._makeauthoptions() #stdin, stdout, stderr = os.popen3(locale_env + # 'svn log --xml %s %s %s "%s"' % ( # rev_opt, verbose_opt, auth_opt, # self.strpath)) cmd = locale_env + 'svn log --xml %s %s %s "%s"' % ( rev_opt, verbose_opt, auth_opt, self.strpath) popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, ) stdout, stderr = popen.communicate() stdout = py.builtin._totext(stdout, sys.getdefaultencoding()) minidom,ExpatError = importxml() try: tree = minidom.parseString(stdout) except ExpatError: raise ValueError('no such revision') result = [] for logentry in filter(None, tree.firstChild.childNodes): if logentry.nodeType == logentry.ELEMENT_NODE: result.append(LogEntry(logentry)) return result def size(self): """ Return the size of the file content of the Path. """ return self.info().size def mtime(self): """ Return the last modification time of the file. """ return self.info().mtime def __hash__(self): return hash((self.strpath, self.__class__, self.auth)) class WCStatus: attrnames = ('modified','added', 'conflict', 'unchanged', 'external', 'deleted', 'prop_modified', 'unknown', 'update_available', 'incomplete', 'kindmismatch', 'ignored', 'locked', 'replaced' ) def __init__(self, wcpath, rev=None, modrev=None, author=None): self.wcpath = wcpath self.rev = rev self.modrev = modrev self.author = author for name in self.attrnames: setattr(self, name, []) def allpath(self, sort=True, **kw): d = {} for name in self.attrnames: if name not in kw or kw[name]: for path in getattr(self, name): d[path] = 1 l = d.keys() if sort: l.sort() return l # XXX a bit scary to assume there's always 2 spaces between username and # path, however with win32 allowing spaces in user names there doesn't # seem to be a more solid approach :( _rex_status = re.compile(r'\s+(\d+|-)\s+(\S+)\s+(.+?)\s{2,}(.*)') def fromstring(data, rootwcpath, rev=None, modrev=None, author=None): """ return a new WCStatus object from data 's' """ rootstatus = WCStatus(rootwcpath, rev, modrev, author) update_rev = None for line in data.split('\n'): if not line.strip(): continue #print "processing %r" % line flags, rest = line[:8], line[8:] # first column c0,c1,c2,c3,c4,c5,x6,c7 = flags #if '*' in line: # print "flags", repr(flags), "rest", repr(rest) if c0 in '?XI': fn = line.split(None, 1)[1] if c0 == '?': wcpath = rootwcpath.join(fn, abs=1) rootstatus.unknown.append(wcpath) elif c0 == 'X': wcpath = rootwcpath.__class__( rootwcpath.localpath.join(fn, abs=1), auth=rootwcpath.auth) rootstatus.external.append(wcpath) elif c0 == 'I': wcpath = rootwcpath.join(fn, abs=1) rootstatus.ignored.append(wcpath) continue #elif c0 in '~!' or c4 == 'S': # raise NotImplementedError("received flag %r" % c0) m = WCStatus._rex_status.match(rest) if not m: if c7 == '*': fn = rest.strip() wcpath = rootwcpath.join(fn, abs=1) rootstatus.update_available.append(wcpath) continue if line.lower().find('against revision:')!=-1: update_rev = int(rest.split(':')[1].strip()) continue if line.lower().find('status on external') > -1: # XXX not sure what to do here... perhaps we want to # store some state instead of just continuing, as right # now it makes the top-level external get added twice # (once as external, once as 'normal' unchanged item) # because of the way SVN presents external items continue # keep trying raise ValueError("could not parse line %r" % line) else: rev, modrev, author, fn = m.groups() wcpath = rootwcpath.join(fn, abs=1) #assert wcpath.check() if c0 == 'M': assert wcpath.check(file=1), "didn't expect a directory with changed content here" rootstatus.modified.append(wcpath) elif c0 == 'A' or c3 == '+' : rootstatus.added.append(wcpath) elif c0 == 'D': rootstatus.deleted.append(wcpath) elif c0 == 'C': rootstatus.conflict.append(wcpath) elif c0 == '~': rootstatus.kindmismatch.append(wcpath) elif c0 == '!': rootstatus.incomplete.append(wcpath) elif c0 == 'R': rootstatus.replaced.append(wcpath) elif not c0.strip(): rootstatus.unchanged.append(wcpath) else: raise NotImplementedError("received flag %r" % c0) if c1 == 'M': rootstatus.prop_modified.append(wcpath) # XXX do we cover all client versions here? if c2 == 'L' or c5 == 'K': rootstatus.locked.append(wcpath) if c7 == '*': rootstatus.update_available.append(wcpath) if wcpath == rootwcpath: rootstatus.rev = rev rootstatus.modrev = modrev rootstatus.author = author if update_rev: rootstatus.update_rev = update_rev continue return rootstatus fromstring = staticmethod(fromstring) class XMLWCStatus(WCStatus): def fromstring(data, rootwcpath, rev=None, modrev=None, author=None): """ parse 'data' (XML string as outputted by svn st) into a status obj """ # XXX for externals, the path is shown twice: once # with external information, and once with full info as if # the item was a normal non-external... the current way of # dealing with this issue is by ignoring it - this does make # externals appear as external items as well as 'normal', # unchanged ones in the status object so this is far from ideal rootstatus = WCStatus(rootwcpath, rev, modrev, author) update_rev = None minidom, ExpatError = importxml() try: doc = minidom.parseString(data) except ExpatError: e = sys.exc_info()[1] raise ValueError(str(e)) urevels = doc.getElementsByTagName('against') if urevels: rootstatus.update_rev = urevels[-1].getAttribute('revision') for entryel in doc.getElementsByTagName('entry'): path = entryel.getAttribute('path') statusel = entryel.getElementsByTagName('wc-status')[0] itemstatus = statusel.getAttribute('item') if itemstatus == 'unversioned': wcpath = rootwcpath.join(path, abs=1) rootstatus.unknown.append(wcpath) continue elif itemstatus == 'external': wcpath = rootwcpath.__class__( rootwcpath.localpath.join(path, abs=1), auth=rootwcpath.auth) rootstatus.external.append(wcpath) continue elif itemstatus == 'ignored': wcpath = rootwcpath.join(path, abs=1) rootstatus.ignored.append(wcpath) continue elif itemstatus == 'incomplete': wcpath = rootwcpath.join(path, abs=1) rootstatus.incomplete.append(wcpath) continue rev = statusel.getAttribute('revision') if itemstatus == 'added' or itemstatus == 'none': rev = '0' modrev = '?' author = '?' date = '' elif itemstatus == "replaced": pass else: #print entryel.toxml() commitel = entryel.getElementsByTagName('commit')[0] if commitel: modrev = commitel.getAttribute('revision') author = '' author_els = commitel.getElementsByTagName('author') if author_els: for c in author_els[0].childNodes: author += c.nodeValue date = '' for c in commitel.getElementsByTagName('date')[0]\ .childNodes: date += c.nodeValue wcpath = rootwcpath.join(path, abs=1) assert itemstatus != 'modified' or wcpath.check(file=1), ( 'did\'t expect a directory with changed content here') itemattrname = { 'normal': 'unchanged', 'unversioned': 'unknown', 'conflicted': 'conflict', 'none': 'added', }.get(itemstatus, itemstatus) attr = getattr(rootstatus, itemattrname) attr.append(wcpath) propsstatus = statusel.getAttribute('props') if propsstatus not in ('none', 'normal'): rootstatus.prop_modified.append(wcpath) if wcpath == rootwcpath: rootstatus.rev = rev rootstatus.modrev = modrev rootstatus.author = author rootstatus.date = date # handle repos-status element (remote info) rstatusels = entryel.getElementsByTagName('repos-status') if rstatusels: rstatusel = rstatusels[0] ritemstatus = rstatusel.getAttribute('item') if ritemstatus in ('added', 'modified'): rootstatus.update_available.append(wcpath) lockels = entryel.getElementsByTagName('lock') if len(lockels): rootstatus.locked.append(wcpath) return rootstatus fromstring = staticmethod(fromstring) class InfoSvnWCCommand: def __init__(self, output): # Path: test # URL: http://codespeak.net/svn/std.path/trunk/dist/std.path/test # Repository UUID: fd0d7bf2-dfb6-0310-8d31-b7ecfe96aada # Revision: 2151 # Node Kind: directory # Schedule: normal # Last Changed Author: hpk # Last Changed Rev: 2100 # Last Changed Date: 2003-10-27 20:43:14 +0100 (Mon, 27 Oct 2003) # Properties Last Updated: 2003-11-03 14:47:48 +0100 (Mon, 03 Nov 2003) d = {} for line in output.split('\n'): if not line.strip(): continue key, value = line.split(':', 1) key = key.lower().replace(' ', '') value = value.strip() d[key] = value try: self.url = d['url'] except KeyError: raise ValueError("Not a versioned resource") #raise ValueError, "Not a versioned resource %r" % path self.kind = d['nodekind'] == 'directory' and 'dir' or d['nodekind'] try: self.rev = int(d['revision']) except KeyError: self.rev = None self.path = py.path.local(d['path']) self.size = self.path.size() if 'lastchangedrev' in d: self.created_rev = int(d['lastchangedrev']) if 'lastchangedauthor' in d: self.last_author = d['lastchangedauthor'] if 'lastchangeddate' in d: self.mtime = parse_wcinfotime(d['lastchangeddate']) self.time = self.mtime * 1000000 def __eq__(self, other): return self.__dict__ == other.__dict__ def parse_wcinfotime(timestr): """ Returns seconds since epoch, UTC. """ # example: 2003-10-27 20:43:14 +0100 (Mon, 27 Oct 2003) m = re.match(r'(\d+-\d+-\d+ \d+:\d+:\d+) ([+-]\d+) .*', timestr) if not m: raise ValueError("timestring %r does not match" % timestr) timestr, timezone = m.groups() # do not handle timezone specially, return value should be UTC parsedtime = time.strptime(timestr, "%Y-%m-%d %H:%M:%S") return calendar.timegm(parsedtime) def make_recursive_propdict(wcroot, output, rex = re.compile("Properties on '(.*)':")): """ Return a dictionary of path->PropListDict mappings. """ lines = [x for x in output.split('\n') if x] pdict = {} while lines: line = lines.pop(0) m = rex.match(line) if not m: raise ValueError("could not parse propget-line: %r" % line) path = m.groups()[0] wcpath = wcroot.join(path, abs=1) propnames = [] while lines and lines[0].startswith(' '): propname = lines.pop(0).strip() propnames.append(propname) assert propnames, "must have found properties!" pdict[wcpath] = PropListDict(wcpath, propnames) return pdict def importxml(cache=[]): if cache: return cache from xml.dom import minidom from xml.parsers.expat import ExpatError cache.extend([minidom, ExpatError]) return cache class LogEntry: def __init__(self, logentry): self.rev = int(logentry.getAttribute('revision')) for lpart in filter(None, logentry.childNodes): if lpart.nodeType == lpart.ELEMENT_NODE: if lpart.nodeName == 'author': self.author = lpart.firstChild.nodeValue elif lpart.nodeName == 'msg': if lpart.firstChild: self.msg = lpart.firstChild.nodeValue else: self.msg = '' elif lpart.nodeName == 'date': #2003-07-29T20:05:11.598637Z timestr = lpart.firstChild.nodeValue self.date = parse_apr_time(timestr) elif lpart.nodeName == 'paths': self.strpaths = [] for ppart in filter(None, lpart.childNodes): if ppart.nodeType == ppart.ELEMENT_NODE: self.strpaths.append(PathEntry(ppart)) def __repr__(self): return '<Logentry rev=%d author=%s date=%s>' % ( self.rev, self.author, self.date)