/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to tests/__init__.py

  • Committer: Martin Pool
  • Date: 2005-08-24 08:59:32 UTC
  • Revision ID: mbp@sourcefrog.net-20050824085932-c61f1f1f1c930e13
- Add a simple UIFactory 

  The idea of this is to let a client of bzrlib set some 
  policy about how output is displayed.

  In this revision all that's done is that progress bars
  are constructed by a policy established by the application
  rather than being randomly constructed in the library 
  or passed down the calls.  This avoids progress bars
  popping up while running the test suite and cleans up
  some code.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/__init__.py
1
 
# Copyright (C) 2006, 2007 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
"""The basic test suite for bzr-git."""
18
 
 
19
 
import subprocess
20
 
import time
21
 
 
22
 
from bzrlib import (
23
 
    osutils,
24
 
    tests,
25
 
    trace,
26
 
    )
27
 
from  bzrlib.plugins.git import errors
28
 
 
29
 
TestCase = tests.TestCase
30
 
TestCaseInTempDir = tests.TestCaseInTempDir
31
 
TestCaseWithTransport = tests.TestCaseWithTransport
32
 
TestCaseWithMemoryTransport = tests.TestCaseWithMemoryTransport
33
 
 
34
 
class _GitCommandFeature(tests.Feature):
35
 
 
36
 
    def _probe(self):
37
 
        try:
38
 
            p = subprocess.Popen(['git', '--version'], stdout=subprocess.PIPE)
39
 
        except IOError:
40
 
            return False
41
 
        out, err = p.communicate()
42
 
        trace.mutter('Using: %s', out.rstrip('\n'))
43
 
        return True
44
 
 
45
 
    def feature_name(self):
46
 
        return 'git'
47
 
 
48
 
GitCommandFeature = _GitCommandFeature()
49
 
 
50
 
 
51
 
def run_git(*args):
52
 
    cmd = ['git'] + list(args)
53
 
    p = subprocess.Popen(cmd,
54
 
                         stdout=subprocess.PIPE,
55
 
                         stderr=subprocess.PIPE)
56
 
    out, err = p.communicate()
57
 
    if p.returncode != 0:
58
 
        raise AssertionError('Bad return code: %d for %s:\n%s'
59
 
                             % (p.returncode, ' '.join(cmd), err))
60
 
    return out
61
 
 
62
 
 
63
 
class GitBranchBuilder(object):
64
 
 
65
 
    def __init__(self, stream=None):
66
 
        self.commit_info = []
67
 
        self.stream = stream
68
 
        self._process = None
69
 
        self._counter = 0
70
 
        self._branch = 'refs/heads/master'
71
 
        if stream is None:
72
 
            # Write the marks file into the git sandbox.
73
 
            self._marks_file_name = osutils.abspath('marks')
74
 
            self._process = subprocess.Popen(
75
 
                ['git', 'fast-import', '--quiet',
76
 
                 # GIT doesn't support '--export-marks foo'
77
 
                 # it only supports '--export-marks=foo'
78
 
                 # And gives a 'unknown option' otherwise.
79
 
                 '--export-marks=' + self._marks_file_name,
80
 
                ],
81
 
                stdout=subprocess.PIPE,
82
 
                stderr=subprocess.PIPE,
83
 
                stdin=subprocess.PIPE,
84
 
                )
85
 
            self.stream = self._process.stdin
86
 
        else:
87
 
            self._process = None
88
 
 
89
 
    def set_branch(self, branch):
90
 
        """Set the branch we are committing."""
91
 
        self._branch = branch
92
 
 
93
 
    def _write(self, text):
94
 
        try:
95
 
            self.stream.write(text)
96
 
        except IOError, e:
97
 
            if self._process is None:
98
 
                raise
99
 
            raise errors.GitCommandError(self._process.returncode,
100
 
                                         'git fast-import',
101
 
                                         self._process.stderr.read())
102
 
 
103
 
    def _writelines(self, lines):
104
 
        try:
105
 
            self.stream.writelines(lines)
106
 
        except IOError, e:
107
 
            if self._process is None:
108
 
                raise
109
 
            raise errors.GitCommandError(self._process.returncode,
110
 
                                         'git fast-import',
111
 
                                         self._process.stderr.read())
112
 
 
113
 
    def _create_blob(self, content):
114
 
        self._counter += 1
115
 
        self._write('blob\n')
116
 
        self._write('mark :%d\n' % (self._counter,))
117
 
        self._write('data %d\n' % (len(content),))
118
 
        self._write(content)
119
 
        self._write('\n')
120
 
        return self._counter
121
 
 
122
 
    def set_file(self, path, content, executable):
123
 
        """Create or update content at a given path."""
124
 
        mark = self._create_blob(content)
125
 
        if executable:
126
 
            mode = '100755'
127
 
        else:
128
 
            mode = '100644'
129
 
        self.commit_info.append('M %s :%d %s\n'
130
 
                                % (mode, mark, self._encode_path(path)))
131
 
 
132
 
    def set_link(self, path, link_target):
133
 
        """Create or update a link at a given path."""
134
 
        mark = self._create_blob(link_target)
135
 
        self.commit_info.append('M 120000 :%d %s\n'
136
 
                                % (mark, self._encode_path(path)))
137
 
 
138
 
    def delete_entry(self, path):
139
 
        """This will delete files or symlinks at the given location."""
140
 
        self.commit_info.append('D %s\n' % (self._encode_path(path),))
141
 
 
142
 
    @staticmethod
143
 
    def _encode_path(path):
144
 
        if '\n' in path or path[0] == '"':
145
 
            path = path.replace('\\', '\\\\')
146
 
            path = path.replace('\n', '\\n')
147
 
            path = path.replace('"', '\\"')
148
 
            path = '"' + path + '"'
149
 
        return path.encode('utf-8')
150
 
 
151
 
    # TODO: Author
152
 
    # TODO: Author timestamp+timezone
153
 
    def commit(self, committer, message, timestamp=None,
154
 
               timezone='+0000', author=None,
155
 
               merge=None, base=None):
156
 
        """Commit the new content.
157
 
 
158
 
        :param committer: The name and address for the committer
159
 
        :param message: The commit message
160
 
        :param timestamp: The timestamp for the commit
161
 
        :param timezone: The timezone of the commit, such as '+0000' or '-1000'
162
 
        :param author: The name and address of the author (if different from
163
 
            committer)
164
 
        :param merge: A list of marks if this should merge in another commit
165
 
        :param base: An id for the base revision (primary parent) if that
166
 
            is not the last commit.
167
 
        :return: A mark which can be used in the future to reference this
168
 
            commit.
169
 
        """
170
 
        self._counter += 1
171
 
        mark = self._counter
172
 
        if timestamp is None:
173
 
            timestamp = int(time.time())
174
 
        self._write('commit %s\n' % (self._branch,))
175
 
        self._write('mark :%d\n' % (mark,))
176
 
        self._write('committer %s %s %s\n'
177
 
                    % (committer, timestamp, timezone))
178
 
        message = message.encode('UTF-8')
179
 
        self._write('data %d\n' % (len(message),))
180
 
        self._write(message)
181
 
        self._write('\n')
182
 
        if base is not None:
183
 
            self._write('from :%d\n' % (base,))
184
 
        if merge is not None:
185
 
            for m in merge:
186
 
                self._write('merge :%d\n' % (m,))
187
 
        self._writelines(self.commit_info)
188
 
        self._write('\n')
189
 
        self.commit_info = []
190
 
        return mark
191
 
 
192
 
    def reset(self, ref=None, mark=None):
193
 
        """Create or recreate the named branch.
194
 
 
195
 
        :param ref: branch name, defaults to the current branch.
196
 
        :param mark: commit the branch will point to.
197
 
        """
198
 
        if ref is None:
199
 
            ref = self._branch
200
 
        self._write('reset %s\n' % (ref,))
201
 
        if mark is not None:
202
 
            self._write('from :%d\n' % mark)
203
 
        self._write('\n')
204
 
 
205
 
    def finish(self):
206
 
        """We are finished building, close the stream, get the id mapping"""
207
 
        self.stream.close()
208
 
        if self._process is None:
209
 
            return {}
210
 
        if self._process.wait() != 0:
211
 
            raise errors.GitCommandError(self._process.returncode,
212
 
                                         'git fast-import',
213
 
                                         self._process.stderr.read())
214
 
        marks_file = open(self._marks_file_name)
215
 
        mapping = {}
216
 
        for line in marks_file:
217
 
            mark, shasum = line.split()
218
 
            assert mark.startswith(':')
219
 
            mapping[int(mark[1:])] = shasum
220
 
        marks_file.close()
221
 
        return mapping
222
 
 
223
 
 
224
 
def test_suite():
225
 
    loader = tests.TestLoader()
226
 
 
227
 
    suite = tests.TestSuite()
228
 
 
229
 
    testmod_names = [
230
 
        'test_builder',
231
 
        'test_branch',
232
 
        'test_dir',
233
 
        'test_repository',
234
 
        'test_ids',
235
 
        'test_blackbox',
236
 
        ]
237
 
    testmod_names = ['%s.%s' % (__name__, t) for t in testmod_names]
238
 
    suite.addTests(loader.loadTestsFromModuleNames(testmod_names))
239
 
 
240
 
    return suite