1
# Copyright (C) 2006 by 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 version 2 as published by
5
# the Free Software Foundation.
7
# This program is distributed in the hope that it will be useful,
8
# but WITHOUT ANY WARRANTY; without even the implied warranty of
9
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10
# GNU General Public License for more details.
12
# You should have received a copy of the GNU General Public License
13
# along with this program; if not, write to the Free Software
14
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
Revision stores are responsible for storing a group of revisions
19
and returning some interesting data about them such as ancestors
20
and ghosts information.
24
from copy import deepcopy
25
from cStringIO import StringIO
26
from unittest import TestSuite
30
import bzrlib.errors as errors
31
from bzrlib.trace import mutter
34
class RevisionStoreTestProviderAdapter(object):
35
"""A tool to generate a suite testing multiple repository stores.
37
This is done by copying the test once for each repository store
38
and injecting the transport_server, transport_readonly_server,
39
and revision-store-factory into each copy.
40
Each copy is also given a new id() to make it easy to identify.
43
def __init__(self, transport_server, transport_readonly_server, factories):
44
self._transport_server = transport_server
45
self._transport_readonly_server = transport_readonly_server
46
self._factories = factories
48
def adapt(self, test):
50
for factory in self._factories:
51
new_test = deepcopy(test)
52
new_test.transport_server = self._transport_server
53
new_test.transport_readonly_server = self._transport_readonly_server
54
new_test.store_factory = factory
55
def make_new_test_id():
56
new_id = "%s(%s)" % (new_test.id(), factory)
58
new_test.id = make_new_test_id()
59
result.addTest(new_test)
63
def default_test_list():
64
"""Generate the default list of revision store permutations to test."""
65
from bzrlib.store.revision.text import TextRevisionStoreTestFactory
66
from bzrlib.store.revision.knit import KnitRevisionStoreFactory
68
# test the fallback InterVersionedFile from weave to annotated knits
69
result.append(TextRevisionStoreTestFactory())
70
result.append(KnitRevisionStoreFactory())
74
class RevisionStore(object):
75
"""A revision store stores revisions."""
77
def __init__(self, serializer=None):
78
if serializer is None:
79
serializer = bzrlib.xml5.serializer_v5
80
self._serializer = serializer
82
def add_revision(self, revision, transaction):
83
"""Add revision to the revision store.
85
:param rev: The revision object.
87
# serialisation : common to all at the moment.
89
self._serializer.write_revision(revision, rev_tmp)
91
self._add_revision(revision, rev_tmp, transaction)
92
mutter('added revision_id {%s}', revision.revision_id)
94
def _add_revision(self, revision, revision_as_file, transaction):
95
"""Template method helper to store revision in this store."""
96
raise NotImplementedError(self._add_revision)
98
def add_revision_signature_text(self, revision_id, signature_text, transaction):
99
"""Add signature_text as a signature for revision_id."""
100
raise NotImplementedError(self.add_revision_signature_text)
102
def all_revision_ids(self, transaction):
103
"""Returns a list of all the revision ids in the revision store.
105
:return: list of revision_ids in topological order.
107
raise NotImplementedError(self.all_revision_ids)
109
def get_revision(self, revision_id, transaction):
110
"""Return the Revision object for a named revision."""
111
raise NotImplementedError(self.get_revision)
113
def get_signature_text(self, revision_id, transaction):
114
"""Get the signature text for the digital signature of revision_id.
116
:return: a signature text.
118
self._guard_revision(revision_id, transaction)
119
return self._get_signature_text(revision_id, transaction)
121
def _get_signature_text(self, revision_id, transaction):
122
"""Helper method for get_signature_text to return the text itself."""
123
raise NotImplementedError(self.get_signature_text)
125
def _guard_revision(self, revision_id, transaction):
126
"""Guard method for testing the presence of a revision."""
127
if not self.has_revision_id(revision_id, transaction):
128
raise errors.NoSuchRevision(self, revision_id)
130
def has_revision_id(self, revision_id, transaction):
131
"""True if the store contains revision_id."""
132
raise NotImplementedError(self.has_revision_id)
134
def has_signature(self, revision_id, transaction):
135
"""True if the store has a signature for revision_id."""
136
self._guard_revision(revision_id, transaction)
137
return self._has_signature(revision_id, transaction)
139
def _has_signature(self, revision_id, transaction):
140
"""Return the presence of a signature for revision_id.
142
A worker memrthod for has_signature, this can assume the
145
return NotImplementedError(self._has_signature)
147
def total_size(self, transaction):
148
"""How big is the store?
150
:return: (count, bytes) tuple.
152
raise NotImplementedError(self.total_size)
added
removed
bzrlib/store/revision/__init__.py
