darcs.py

Go to the documentation of this file.

# Copyright 2005, 2006  Timo Savola
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.

import sys
import os
import re
import filecmp
import gobject

import filesystem

class Controller (filesystem.Controller):
        rules = ('_darcs',)

        def __init__(self, path):
                filesystem.Controller.__init__(self, path)

        def is_controlled(self, name):
                escaped = escape(os.path.join('.', name))

                re_add = re.compile('^add(file|dir) %s$' % escaped)
                re_remove = re.compile('^rm(file|dir) %s$' % escaped)
                re_move_in = re.compile('^move .* %s$' % escaped)
                re_move_out = re.compile('^move %s ' % escaped)

                current = self._get_current_path(name)
                control = os.path.exists(current)

                pending = os.path.join(self.root, '_darcs', 'patches', 'pending')
                if os.path.exists(pending):
                        file = open(pending)

                        for line in file:
                                line = line.rstrip('\n')

                                if control:
                                        if re_remove.match(line) or re_move_out.match(line):
                                                control = False
                                else:
                                        if re_add.match(line) or re_move_in.match(line):
                                                control = True

                        file.close()

                return control

        def is_modified(self, name):
                current = self._get_current_path(name)
                if os.path.exists(current):
                        if os.path.isdir(current):
                                return False
                        else:
                                work = self._get_path(name)
                                return not filecmp.cmp(work, current)
                else:
                        return True

        def get_latest(self, name):
                current = self._get_current_path(name)
                if os.path.exists(current):
                        file = open(current)
                        data = file.read()
                        file.close()
                        return data
                else:
                        return None

        def control(self, name, recursive=False):
                if recursive:
                        self._call(['add', '--recursive', name])
                else:
                        self._call(['add', name])

        def release(self, name):
                names = self._find_controlled(name)
                self._call(['remove'] + names)

        def make_file(self, name, control=True):
                filesystem.Controller.make_file(self, name)
                if control:
                        self._call(['add', name])

        def make_directory(self, name, control=True):
                filesystem.Controller.make_directory(self, name)
                if control:
                        self._call(['add', name])

        def remove(self, name, control=True):
                filesystem.Controller.remove(self, name)
                if control:
                        self._call(['remove', name])

        def rename(self, old_name, new_name, control=True):
                if control:
                        self._call(['mv', old_name, new_name])
                else:
                        filesystem.Controller.rename(self, old_name, new_name)

        def clone(self, old_name, new_name, control=True):
                filesystem.Controller.clone(self, old_name, new_name)
                if control:
                        self._call(['add', '--recursive', new_name])

        def _get_current_path(self, name):
                return os.path.join(self.root, '_darcs', 'current', name)

        def _find_controlled(self, name):
                contents = [name]

                path = self._get_path(name)
                if os.path.isdir(path):
                        for filename in os.listdir(path):
                                if filename == '.' or filename == '..':
                                        continue

                                child_name = os.path.join(name, filename)
                                if self.is_controlled(child_name):
                                        child_contents = self._find_controlled(child_name)
                                        contents = child_contents + contents

                return contents

        def _call(self, params):
                args = ['darcs'] + params
                pid, fd = call(args, self.root)

                errors = None

                try:
                        file = os.fdopen(fd, 'r')

                        while True:
                                try:
                                        errors = file.read()
                                except OSError:
                                        continue
                                break

                        file.close()
                except:
                        try:
                                os.waitpid(pid, 0)
                        except:
                                pass
                        raise

                pid, status = os.waitpid(pid, 0)

                code = status >> 8
                if code != 0:
                        if not errors:
                                errors = 'Return code: %d' % code

                        raise DarcsError(errors)

gobject.type_register(Controller)

class DarcsError (Exception):
        prefix = 'darcs failed:'

        def __init__(self, message):
                if message.startswith(self.prefix):
                        message = message[len(self.prefix):]

                message = message.strip()

                Exception.__init__(self, message)

def call(args, workdir):
        assert type(args) == list and len(args) > 0

        pipe = os.pipe()

        pid = os.fork()
        if pid == 0:
                try:
                        os.dup2(pipe[1], 2)
                        os.close(pipe[0])
                        os.close(pipe[1])

                        os.close(0)

                        null = os.open(os.devnull, os.O_RDONLY)
                        if null != 0:
                                os.dup2(null, 0)
                                os.close(null)

                        os.chdir(workdir)
                        os.execvp(args[0], args)

                except Exception, e:
                        try:
                                print >>sys.stderr, '%s:' % args[0],
                        except:
                                pass

                        try:
                                print >>sys.stderr, e
                        except:
                                pass

                os._exit(127)

        try:
                os.close(pipe[1])
        except:
                try:
                        os.waitpid(pid, 0)
                except:
                        pass
                raise

        return pid, pipe[0]

def escape(path):
        for c in '\\.':
                path = path.replace(c, '\\' + c)
        return path

---