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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
|
import struct
from mercurial.node import (
bin as node_bin,
hex,
)
from mercurial import error
try:
from mercurial import rustext
rustext.__name__ # trigger immediate actual import
except ImportError:
rustext = None
else:
# this would fail already without appropriate ancestor.__package__
from mercurial.rustext.ancestor import LazyAncestors
from mercurial.testing import revlog as revlogtesting
header = struct.unpack(">I", revlogtesting.data_non_inlined[:4])[0]
class RustInnerRevlogTestMixin:
"""Common tests for both Rust Python bindings."""
node_hex0 = b'd1f4bbb0befc13bd8cd39d0fcdd93b8c078c4a2f'
node0 = node_bin(node_hex0)
bogus_node_hex = b'cafe' * 10
bogus_node = node_bin(bogus_node_hex)
node_hex2 = b"020a0ec626a192ae360b0269fe2de5ba6f05d1e7"
node2 = node_bin(node_hex2)
def test_index_nodemap(self):
idx = self.parserustindex()
self.assertTrue(idx.has_node(self.node0))
self.assertFalse(idx.has_node(self.bogus_node))
self.assertEqual(idx.get_rev(self.node0), 0)
self.assertEqual(idx.get_rev(self.node0), 0)
self.assertEqual(idx.rev(self.node0), 0)
with self.assertRaises(error.RevlogError) as exc_info:
idx.rev(self.bogus_node)
self.assertEqual(exc_info.exception.args, (None,))
self.assertEqual(idx.partialmatch(self.node_hex0[:3]), self.node0)
self.assertIsNone(idx.partialmatch(self.bogus_node_hex[:3]))
self.assertEqual(idx.shortest(self.node0), 1)
def test_len(self):
idx = self.parserustindex()
self.assertEqual(len(idx), 4)
def test_getitem(self):
idx = self.parserustindex()
as_tuple = (0, 82969, 484626, 0, 0, -1, -1, self.node0, 0, 0, 2, 2, -1)
self.assertEqual(idx[0], as_tuple)
self.assertEqual(idx[self.node0], 0)
def test_heads(self):
idx = self.parserustindex()
self.assertEqual(idx.headrevs(), [3])
def test_index_append(self):
idx = self.parserustindex(data=b'')
self.assertEqual(len(idx), 0)
self.assertIsNone(idx.get_rev(self.node0))
non_empty_index = self.parserustindex()
idx.append(non_empty_index[0])
self.assertEqual(len(idx), 1)
self.assertEqual(idx.get_rev(self.node0), 0)
def test_index_delitem_single(self):
idx = self.parserustindex()
del idx[2]
self.assertEqual(len(idx), 2)
# the nodetree is consistent
self.assertEqual(idx.get_rev(self.node0), 0)
self.assertIsNone(idx.get_rev(self.node2))
# not an error and does nothing
del idx[-1]
self.assertEqual(len(idx), 2)
for bogus in (-2, 17):
try:
del idx[bogus]
except ValueError as exc:
# this underlines that we should do better with this message
assert exc.args[0] == (
f"Inconsistency: Revision {bogus} found in nodemap "
"is not in revlog index"
)
else:
raise AssertionError(
f"an exception was expected for `del idx[{bogus}]`"
)
def test_index_delitem_slice(self):
idx = self.parserustindex()
del idx[2:3]
self.assertEqual(len(idx), 2)
# not an error and not equivalent to `del idx[0::]` but to
# `del idx[-1]` instead and thus does nothing.
del idx[-1::]
self.assertEqual(len(idx), 2)
for start, stop in (
(-2, None),
(17, None),
):
try:
del idx[start:stop]
except ValueError as exc:
# this underlines that we should do better with this message
assert exc.args[0] == (
f"Inconsistency: Revision {start} found in nodemap "
"is not in revlog index"
)
else:
raise AssertionError(
f"an exception was expected for `del idx[{start}:{stop}]`"
)
# although the upper bound is way too big, this is not an error:
del idx[0::17]
self.assertEqual(len(idx), 0)
def test_standalone_nodetree(self):
idx = self.parserustindex()
nt = self.nodetree(idx)
for i in range(4):
nt.insert(i)
# invalidation is upon mutation *of the index*
self.assertFalse(nt.is_invalidated())
bin_nodes = [entry[7] for entry in idx]
hex_nodes = [hex(n) for n in bin_nodes]
for i, node in enumerate(hex_nodes):
self.assertEqual(nt.prefix_rev_lookup(node), i)
self.assertEqual(nt.prefix_rev_lookup(node[:5]), i)
# all 4 revisions in idx (standard data set) have different
# first nybbles in their Node IDs,
# hence `nt.shortest()` should return 1 for them, except when
# the leading nybble is 0 (ambiguity with NULL_NODE)
for i, (bin_node, hex_node) in enumerate(zip(bin_nodes, hex_nodes)):
shortest = nt.shortest(bin_node)
expected = 2 if hex_node[0] == ord('0') else 1
self.assertEqual(shortest, expected)
self.assertEqual(nt.prefix_rev_lookup(hex_node[:shortest]), i)
# test invalidation (generation poisoning) detection
del idx[3]
self.assertTrue(nt.is_invalidated())
def test_reading_context_manager(self):
irl = self.make_inner_revlog()
try:
with irl.reading():
# not much to do yet
pass
except error.RevlogError as exc:
# well our data file does not even exist
self.assertTrue(b"when reading Just a path/test.d" in exc.args[0])
# Conditional skipping done by the base class
class RustInnerRevlogTest(
revlogtesting.RustRevlogBasedTestBase, RustInnerRevlogTestMixin
):
"""For reference"""
def test_ancestors(self):
rustidx = self.parserustindex()
lazy = LazyAncestors(rustidx, [3], 0, True)
# we have two more references to the index:
# - in its inner iterator for __contains__ and __bool__
# - in the LazyAncestors instance itself (to spawn new iterators)
self.assertTrue(2 in lazy)
self.assertTrue(bool(lazy))
self.assertEqual(list(lazy), [3, 2, 1, 0])
# a second time to validate that we spawn new iterators
self.assertEqual(list(lazy), [3, 2, 1, 0])
# let's check bool for an empty one
self.assertFalse(LazyAncestors(rustidx, [0], 0, False))
def test_canonical_index_file(self):
irl = self.make_inner_revlog()
self.assertEqual(irl.canonical_index_file, b'test.i')
# Conditional skipping done by the base class
class PyO3InnerRevlogTest(
revlogtesting.PyO3RevlogBasedTestBase, RustInnerRevlogTestMixin
):
"""Testing new PyO3 bindings, by comparison with rust-cpython bindings."""
if __name__ == '__main__':
import silenttestrunner
silenttestrunner.main(__name__)
|
