1
# Copyright (C) 2010 Canonical Ltd
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.
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.
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Matchers for breezy.
19
Primarily test support, Matchers are used by self.assertThat in the breezy
20
test suite. A matcher is a stateful test helper which can be used to determine
21
if a passed object 'matches', much like a regex. If the object does not match
22
the mismatch can be described in a human readable fashion. assertThat then
23
raises if a mismatch occurs, showing the description as the assertion error.
25
Matchers are designed to be more reusable and composable than layered
26
assertions in Test Case objects, so they are recommended for new testing work.
35
'RevisionHistoryMatches',
40
revision as _mod_revision,
42
from .. import lazy_import
43
lazy_import.lazy_import(globals(),
45
from breezy.bzr.smart.request import request_handlers as smart_request_handlers
46
from breezy.bzr.smart import vfs
48
from ..sixish import (
51
from ..tree import find_previous_path
53
from testtools.matchers import Equals, Mismatch, Matcher
56
class ReturnsUnlockable(Matcher):
57
"""A matcher that checks for the pattern we want lock* methods to have:
59
They should return an object with an unlock() method.
60
Calling that method should unlock the original object.
62
:ivar lockable_thing: The object which can be locked that will be
66
def __init__(self, lockable_thing):
67
Matcher.__init__(self)
68
self.lockable_thing = lockable_thing
71
return ('ReturnsUnlockable(lockable_thing=%s)' %
74
def match(self, lock_method):
75
lock_method().unlock()
76
if self.lockable_thing.is_locked():
77
return _IsLocked(self.lockable_thing)
81
class _IsLocked(Mismatch):
82
"""Something is locked."""
84
def __init__(self, lockable_thing):
85
self.lockable_thing = lockable_thing
88
return "%s is locked" % self.lockable_thing
91
class _AncestryMismatch(Mismatch):
92
"""Ancestry matching mismatch."""
94
def __init__(self, tip_revision, got, expected):
95
self.tip_revision = tip_revision
97
self.expected = expected
100
return "mismatched ancestry for revision %r was %r, expected %r" % (
101
self.tip_revision, self.got, self.expected)
104
class MatchesAncestry(Matcher):
105
"""A matcher that checks the ancestry of a particular revision.
107
:ivar graph: Graph in which to check the ancestry
108
:ivar revision_id: Revision id of the revision
111
def __init__(self, repository, revision_id):
112
Matcher.__init__(self)
113
self.repository = repository
114
self.revision_id = revision_id
117
return ('MatchesAncestry(repository=%r, revision_id=%r)' % (
118
self.repository, self.revision_id))
120
def match(self, expected):
121
with self.repository.lock_read():
122
graph = self.repository.get_graph()
123
got = [r for r, p in graph.iter_ancestry([self.revision_id])]
124
if _mod_revision.NULL_REVISION in got:
125
got.remove(_mod_revision.NULL_REVISION)
126
if sorted(got) != sorted(expected):
127
return _AncestryMismatch(self.revision_id, sorted(got),
131
class HasLayout(Matcher):
132
"""A matcher that checks if a tree has a specific layout.
134
:ivar entries: List of expected entries, as (path, file_id) pairs.
137
def __init__(self, entries):
138
Matcher.__init__(self)
139
self.entries = entries
141
def get_tree_layout(self, tree):
142
"""Get the (path, file_id) pairs for the current tree."""
143
with tree.lock_read():
144
for path, ie in tree.iter_entries_by_dir():
145
if ie.parent_id is None:
146
yield (u"", ie.file_id)
148
yield (path+ie.kind_character(), ie.file_id)
151
def _strip_unreferenced_directories(entries):
152
"""Strip all directories that don't (in)directly contain any files.
154
:param entries: List of path strings or (path, ie) tuples to process
157
for entry in entries:
158
if isinstance(entry, (str, text_type)):
162
if not path or path[-1] == "/":
164
directories.append((path, entry))
166
# Yield the referenced parent directories
167
for dirpath, direntry in directories:
168
if osutils.is_inside(dirpath, path):
174
return 'HasLayout(%r)' % self.entries
176
def match(self, tree):
177
actual = list(self.get_tree_layout(tree))
178
if self.entries and isinstance(self.entries[0], (str, text_type)):
179
actual = [path for (path, fileid) in actual]
180
if not tree.has_versioned_directories():
181
entries = list(self._strip_unreferenced_directories(self.entries))
183
entries = self.entries
184
return Equals(entries).match(actual)
187
class HasPathRelations(Matcher):
188
"""Matcher verifies that paths have a relation to those in another tree.
190
:ivar previous_tree: tree to compare to
191
:ivar previous_entries: List of expected entries, as (path, previous_path) pairs.
194
def __init__(self, previous_tree, previous_entries):
195
Matcher.__init__(self)
196
self.previous_tree = previous_tree
197
self.previous_entries = previous_entries
199
def get_path_map(self, tree):
200
"""Get the (path, previous_path) pairs for the current tree."""
201
with tree.lock_read(), self.previous_tree.lock_read():
202
for path, ie in tree.iter_entries_by_dir():
203
if tree.supports_rename_tracking():
204
previous_path = find_previous_path(tree, self.previous_tree, path)
206
if self.previous_tree.is_versioned(path):
211
kind = self.previous_tree.kind(previous_path)
212
if kind == 'directory':
214
if ie.parent_id is None:
215
yield (u"", previous_path)
217
yield (path+ie.kind_character(), previous_path)
220
def _strip_unreferenced_directories(entries):
221
"""Strip all directories that don't (in)directly contain any files.
223
:param entries: List of path strings or (path, previous_path) tuples to process
225
directory_used = set()
227
for (path, previous_path) in entries:
228
if not path or path[-1] == "/":
230
directories.append((path, previous_path))
232
# Yield the referenced parent directories
233
for direntry in directories:
234
if osutils.is_inside(direntry[0], path):
235
directory_used.add(direntry[0])
236
for (path, previous_path) in entries:
237
if (not path.endswith("/")) or path in directory_used:
238
yield (path, previous_path)
241
return 'HasPathRelations(%r, %r)' % (self.previous_tree, self.previous_entries)
243
def match(self, tree):
244
actual = list(self.get_path_map(tree))
245
if not tree.has_versioned_directories():
246
entries = list(self._strip_unreferenced_directories(self.previous_entries))
248
entries = self.previous_entries
249
if not tree.supports_rename_tracking():
251
(path, path if self.previous_tree.is_versioned(path) else None)
252
for (path, previous_path) in entries]
253
return Equals(entries).match(actual)
256
class RevisionHistoryMatches(Matcher):
257
"""A matcher that checks if a branch has a specific revision history.
259
:ivar history: Revision history, as list of revisions. Oldest first.
262
def __init__(self, history):
263
Matcher.__init__(self)
264
self.expected = history
267
return 'RevisionHistoryMatches(%r)' % self.expected
269
def match(self, branch):
270
with branch.lock_read():
271
graph = branch.repository.get_graph()
272
history = list(graph.iter_lefthand_ancestry(
273
branch.last_revision(), [_mod_revision.NULL_REVISION]))
275
return Equals(self.expected).match(history)
278
class _NoVfsCallsMismatch(Mismatch):
279
"""Mismatch describing a list of HPSS calls which includes VFS requests."""
281
def __init__(self, vfs_calls):
282
self.vfs_calls = vfs_calls
285
return "no VFS calls expected, got: %s" % ",".join([
286
"%s(%s)" % (c.method,
287
", ".join([repr(a) for a in c.args])) for c in self.vfs_calls])
290
class ContainsNoVfsCalls(Matcher):
291
"""Ensure that none of the specified calls are HPSS calls."""
294
return 'ContainsNoVfsCalls()'
297
def match(cls, hpss_calls):
299
for call in hpss_calls:
301
request_method = smart_request_handlers.get(call.call.method)
303
# A method we don't know about doesn't count as a VFS method.
305
if issubclass(request_method, vfs.VfsRequest):
306
vfs_calls.append(call.call)
307
if len(vfs_calls) == 0:
309
return _NoVfsCallsMismatch(vfs_calls)