Source code for nxpy.svn.svn

# nxpy.svn package -----------------------------------------------------------

# Copyright Nicola Musatti 2010 - 2016
# Use, modification, and distribution are subject to the Boost Software
# License, Version 1.0. (See accompanying file LICENSE.txt or copy at
# http://www.boost.org/LICENSE_1_0.txt)

# See http://nxpy.sourceforge.net for library home page. ---------------------

r"""
Subversion client wrapper.

Only supports versions 1.6, 1.7 and 1.8, others might work but have not been
tested.
Requires at least Python 2.6.

"""

# Ensures that tests are skipped rather than broken when run with Python 2.5
from __future__ import with_statement
from __future__ import absolute_import
 
import logging
import os
import os.path
import re

import six

import nxpy.command.command
import nxpy.command.option
import nxpy.core.error
import nxpy.core.past
import nxpy.core.temp_file
import nxpy.svn.url

import nxpy.core._impl.log

_log = logging.getLogger(__name__)


[docs]class Info(object): r"""Represents the output of the ``svn info`` command in a structured way.""" _outRe = re.compile(r"([^:]+):\s+(.*)")
[docs] def __init__(self, out): for l in out.split("\n"): match = Info._outRe.match(l) if match: if match.group(1) == "Path": self.path = os.path.realpath(match.group(2)) elif match.group(1) == "URL": self.url = match.group(2) elif match.group(1) == "Repository Root": self.root = match.group(2)
[docs] def __str__(self): return " - ".join((self.path, str(self.url), self.root))
[docs]class Status(object): r"""Represents the output of one line of the ``svn status`` command in a structured way."""
[docs] def __init__(self, line): self.item = line[8:] self.state = line[0] self.properties = line[1] self.locked = line[2] == 'L' self.add_with_history = line[3] == '+' self.switched = line[4] == 'S' self.lock = line[5] self.tree_conflict = line[6] == 'C'
[docs] def __str__(self): return ( self.state + self.properties + ( 'L' if self.locked else ' ') + ( '+' if self.add_with_history else ' ' ) + ( 'S' if self.switched else ' ' ) + self.lock + ( 'C' if self.tree_conflict else ' ' ) + ' ' + self.item )
_config = nxpy.command.option.Config( bool_opts = ( "ignore_externals", "keep_local", "parents", "quiet", "stop_on_copy", "summarize", "verbose", "version", "xml" ), value_opts = ( "file", "message", "password", "revision", "username", "value" ), opposite_opts = { "ignore" : "--no-ignore" }, mapped_opts = { "ignore_externals" : "--ignore-externals", "keep_local" : "--keep-local", "stop_on_copy" : "--stop-on-copy", "value" : "", "version" : "--version" } )
[docs]class Parser(nxpy.command.option.Parser): r"""Allows passing *nxpy.svn.url.Url* instances as arguments to *Svn*'s methods."""
[docs] def __init__(self, command, arguments, options, **defaults): super(Parser, self).__init__(_config, command, [ self._pathtostring(a) for a in arguments ], options, **defaults)
def _pathtostring(self, path): if isinstance(path, nxpy.svn.url.Url): return str(path) elif isinstance(path, six.string_types): return path else: raise nxpy.core.error.ArgumentError()
[docs]class Svn(nxpy.command.command.Command): r"""The actual wrapper."""
[docs] def __init__(self, debug=None): super(Svn, self).__init__("svn --non-interactive", debug) self._version = None
_versionRe = re.compile(r"(\d+)\.(\d+)\.(\d+)")
[docs] def version(self): op = Parser("", (), {}, version=True) out = self.run(op)[0] match = Svn._versionRe.search(out) version = ( int(match.group(1)), int(match.group(2)), int(match.group(3))) return version
[docs] def info(self, *targets): op = Parser("info", targets, {}) out = self.run(op)[0] return Info(out)
[docs] def list(self, *targets): op = Parser("list", targets, {}) out = self.run(op)[0] return out.split()
[docs] def mkdir(self, *targets, **options): op = Parser("mkdir", targets, options, parents=False, username="", password="", message="\"[" + __name__ + "] Make dir(s) " + ", ".join(targets) + "\"") self.run(op)
[docs] def status(self, *targets, **options): op = Parser("status", targets, options, ignore=True, quiet=True, ignore_externals=True) out = self.run(op)[0] ignore_externals = options.get('ignore_externals', True) res = [] for l in out.split("\n"): if len(l) > 0 and ( not ignore_externals or l[0] != "X" ): res.append(Status(l)) return res
[docs] def import_(self, src, dest, debug=None, **options): op = Parser("import", ( src, dest ), {}, username="", password="", message="\"[" + __name__ + "] Import from " + src + "\"") self.run(op, debug)
[docs] def checkout(self, src, dest, debug=None, **options): op = Parser("checkout", ( src, dest ), options, username="", password="", ignore_externals=True) self.run(op, debug)
[docs] def update(self, *targets, **options): op = Parser("update", targets, options, ignore_externals=True) self.run(op)
[docs] def commit(self, src, debug=None, **options): op = Parser("commit", ( src, ), options, username="", password="", message="\"[" + __name__ + "] Commit from " + src + "\"") self.run(op, debug)
[docs] def copy(self, src, dest, debug=None, **options): op = Parser("copy", ( src, dest ), options, ignore_externals=True, username="", password="", message="\"[" + __name__ + "] Copy from " + src + "\"") self.run(op, debug)
def _delete_paths(self, debug, *paths, **options): op = Parser("delete", paths, options, keep_local=False) self.run(op, debug) def _delete_urls(self, debug, *urls, **options): op = Parser("delete", urls, options, username="", password="", message="\"[" + __name__ + "] Delete " + ", ".join([ str(u) for u in urls ]) + "\"") self.run(op, debug)
[docs] def delete(self, *targets, **options): debug = options.get("debug") if debug is not None: del options["debug"] paths = [] urls = [] for t in targets: if not isinstance(t, nxpy.svn.url.Url) and os.access(t, os.R_OK): paths.append(t) else: urls.append(t) if len(paths) > 0: self._delete_paths(debug, *paths, **options) if len(urls) > 0: self._delete_urls(debug, *urls, **options)
[docs] def propget(self, name, *targets): op = Parser("propget", ( name, ) + targets, {}) out = self.run(op)[0] return out.strip()
_extRe = re.compile(r"([\S]+)\s+(.*)")
[docs] def getexternals(self, d): out = self.propget("svn:externals", d) res = {} for l in out.split("\n"): match = Svn._extRe.match(l) if match: res[match.group(1)] = match.group(2) return res
[docs] def getignore(self, d): out = self.propget("svn:ignore", d) return list(out.split("\n"))
[docs] def propset(self, name, *targets, **options): debug = options.get("debug") value = options.get("value") if value: del options["value"] op = Parser("propset", ( name, value ) + targets, options, username="", password="") else: op = Parser("propset", ( name, ) + targets, options, file="", username="", password="") self.run(op, debug)
[docs] def setexternals(self, externals, d, username="", password=""): nxpy.core.past.enforce_at_least(nxpy.core.past.V_2_6) with nxpy.core.temp_file.TempFile(mode="w+", prefix="svn_setexternals") as f: for k, v in six.iteritems(externals): f.write(k + "\t" + v + "\n") f.seek(0, os.SEEK_SET) _log.debug(f.read()) f.close() self.propset("svn:externals", d, file=f.name, username=username, password=password)
[docs] def setignore(self, ignore, d, username="", password=""): nxpy.core.past.enforce_at_least(nxpy.core.past.V_2_6) with nxpy.core.temp_file.TempFile(mode="w+", prefix="svn_setignore") as f: for v in ignore: f.write(v + "\n") f.seek(0, os.SEEK_SET) _log.debug(f.read()) f.close() self.propset("svn:ignore", d, file=f.name, username=username, password=password)
[docs] def diff(self, *targets, **options): op = Parser("diff", targets, options, summarize=False, revision=None) out = self.run(op)[0] return out
[docs] def cat(self, *targets, **options): op = Parser("cat", targets, options, revision=None) out = self.run(op)[0] return out
[docs] def log(self, src, **options): op = Parser("log", ( src, ), options, verbose=False, stop_on_copy=False, revision=None, xml=False) out = self.run(op)[0] return out