# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-from subprocess import Popen, PIPE
import os
-import config
-
-def invoke(args):
- q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE)
- output = q.stdout.read()
- error = q.stderr.read()
- status = q.wait()
- if status >= 0:
- return status, output, error
- raise Exception("Command failed: %s" % error)
+import tempfile
+import config
+from rcs import invoke, CommandError
def invoke_client(*args, **kwargs):
+ directory = kwargs.get('directory')
+ expect = kwargs.get('expect', (0, 1))
cl_args = ["bzr"]
cl_args.extend(args)
- status,output,error = invoke(cl_args)
- if status not in (0,):
- raise Exception("Command failed: %s" % error)
- return output
+ if directory:
+ old_dir = os.getcwd()
+ os.chdir(directory)
+ try:
+ status,output,error = invoke(cl_args, expect)
+ finally:
+ if directory:
+ os.chdir(old_dir)
+ return status, output
def add_id(filename, paranoid=False):
invoke_client("add", filename)
old_path = path
path = os.path.dirname(path)
+def precommit(directory):
+ pass
+def commit(directory, summary, body=None):
+ if body is not None:
+ summary += '\n' + body
+ descriptor, filename = tempfile.mkstemp()
+ try:
+ temp_file = os.fdopen(descriptor, 'wb')
+ temp_file.write(summary)
+ temp_file.close()
+ try:
+ invoke_client('commit', '--unchanged', '--file', filename,
+ directory=directory)
+ except:
+ raise invoke_client('status', directory=directory)[1]
+ finally:
+ os.unlink(filename)
+
+def postcommit(directory):
+ pass
name = "bzr"
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+from subprocess import Popen, PIPE
def rcs_by_name(rcs_name):
"""Return the module for the RCS with the given name"""
if rcs_name == "Arch":
return bzr
import no_rcs
return no_rcs
+
+class CommandError(Exception):
+ def __init__(self, err_str, status):
+ Exception.__init__(self, "Command failed (%d): %s" % (status, err_str))
+ self.err_str = err_str
+ self.status = status
+
+def invoke(args, expect=(0,)):
+ q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE)
+ output = q.stdout.read()
+ error = q.stderr.read()
+ status = q.wait()
+ if status not in expect:
+ raise CommandError(error, status)
+ return status, output, error