mr.developer/src/mr/developer/bazaar.py at 4b730b25331ab0463fc222743c4505f4e25b0850 · fschulze/mr.developer · GitHub

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
from mr.developer import common
import os
import subprocess

logger = common.logger


class BazaarError(common.WCError):
    pass


class BazaarWorkingCopy(common.BaseWorkingCopy):

    def __init__(self, source):
        super().__init__(source)
        self.bzr_executable = common.which('bzr')

    def bzr_branch(self, **kwargs):
        name = self.source['name']
        path = self.source['path']
        url = self.source['url']
        if os.path.exists(path):
            self.output(
                (logger.info, 'Skipped branching existing package %r.' % name))
            return
        self.output((logger.info, 'Branched %r with bazaar.' % name))
        env = dict(os.environ)
        env.pop('PYTHONPATH', None)
        cmd = subprocess.Popen(
            [self.bzr_executable, 'branch', '--quiet', url, path],
            env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = cmd.communicate()
        if cmd.returncode != 0:
            raise BazaarError(
                f'bzr branch for {name!r} failed.\n{stderr}')
        if kwargs.get('verbose', False):
            return stdout

    def bzr_pull(self, **kwargs):
        name = self.source['name']
        path = self.source['path']
        url = self.source['url']
        self.output((logger.info, 'Updated %r with bazaar.' % name))
        env = dict(os.environ)
        env.pop('PYTHONPATH', None)
        cmd = subprocess.Popen(
            [self.bzr_executable, 'pull', url], cwd=path,
            env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = cmd.communicate()
        if cmd.returncode != 0:
            raise BazaarError(
                f'bzr pull for {name!r} failed.\n{stderr}')
        if kwargs.get('verbose', False):
            return stdout

    def checkout(self, **kwargs):
        name = self.source['name']
        path = self.source['path']
        update = self.should_update(**kwargs)
        if os.path.exists(path):
            if update:
                self.update(**kwargs)
            elif self.matches():
                self.output(
                    (logger.info, 'Skipped checkout of existing package %r.' % name))
            else:
                raise BazaarError(
                    'Source URL for existing package %r differs. '
                    'Expected %r.' % (name, self.source['url']))
        else:
            return self.bzr_branch(**kwargs)

    def matches(self):
        name = self.source['name']
        path = self.source['path']
        env = dict(os.environ)
        env.pop('PYTHONPATH', None)
        cmd = subprocess.Popen(
            [self.bzr_executable, 'info'], cwd=path,
            env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = cmd.communicate()
        if cmd.returncode != 0:
            raise BazaarError(
                f'bzr info for {name!r} failed.\n{stderr}')
        return (self.source['url'] in stdout.split())

    def status(self, **kwargs):
        path = self.source['path']
        env = dict(os.environ)
        env.pop('PYTHONPATH', None)
        cmd = subprocess.Popen(
            [self.bzr_executable, 'status'], cwd=path,
            env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = cmd.communicate()
        status = stdout and 'dirty' or 'clean'
        if kwargs.get('verbose', False):
            return status, stdout
        else:
            return status

    def update(self, **kwargs):
        name = self.source['name']
        if not self.matches():
            raise BazaarError(
                "Can't update package %r because its URL doesn't match." %
                name)
        if self.status() != 'clean' and not kwargs.get('force', False):
            raise BazaarError(
                "Can't update package %r because it's dirty." % name)
        return self.bzr_pull(**kwargs)