1784
1784
knit2.transport.get_bytes(knit2._index._filename))
1786
1786
def assertKnitValuesEqual(self, left, right):
1787
"""Assert that the texts and graph of left and right are the same."""
1787
"""Assert that the texts, annotations and graph of left and right are
1788
1790
self.assertEqual(set(left.versions()), set(right.versions()))
1789
1791
for version in left.versions():
1790
1792
self.assertEqual(left.get_parents_with_ghosts(version),
2866
2868
def assertGetPosition(self, knit, versions, version, result):
2867
2869
index = self.get_index(knit, knit.get_data_stream(versions))
2868
if result[0] is None:
2869
result = (index, result[1], result[2])
2870
if result[1] is None:
2871
result = (result[0], index, result[2], result[3])
2870
2872
self.assertEqual(result, index.get_position(version))
2872
2874
def assertGetParentsWithGhosts(self, knit, versions, version, parents):
2952
2954
def test_get_position(self):
2953
2955
knit = self.make_knit_with_4_versions_2_dags()
2954
# get_position returns (index(can be None), start, end) for
2956
# get_position returns (thunk_flay, index(can be None), start, end) for
2955
2957
# _StreamAccess to use.
2956
self.assertGetPosition(knit, ['a'], 'a', (None, 0, 78))
2957
self.assertGetPosition(knit, ['a', 'c'], 'c', (None, 78, 156))
2958
self.assertGetPosition(knit, ['a'], 'a', (False, None, 0, 78))
2959
self.assertGetPosition(knit, ['a', 'c'], 'c', (False, None, 78, 156))
2958
2960
# get_position on a text that is not in the datastream (but in the
2959
# backing knit) returns ('thunk:versionid', None, None) - and then the
2961
# backing knit) returns (True, 'versionid', None, None) - and then the
2960
2962
# access object can construct the relevant data as needed.
2961
self.assertGetPosition(knit, ['a', 'c'], 'b', ('thunk:b', None, None))
2963
self.assertGetPosition(knit, ['a', 'c'], 'b', (True, 'b', None, None))
2964
2966
class Test_StreamAccess(KnitTests):
3015
3017
target_knit.join(source_knit)
3016
3018
index, access = self.get_index_access(target_knit,
3017
3019
source_knit.get_data_stream([]))
3018
raw_data = list(access.get_raw_records([("thunk:A", None, None)]))[0]
3020
raw_data = list(access.get_raw_records([(True, "A", None, None)]))[0]
3019
3021
df = GzipFile(mode='rb', fileobj=StringIO(raw_data))
3020
3022
self.assertEqual(
3021
3023
'version A 1 5d36b88bb697a2d778f024048bafabd443d74503\n'
3028
3030
index, access = self.get_index_access(knit,
3029
3031
knit.get_data_stream([]))
3030
3032
self.assertRaises(errors.KnitCorrupt,
3031
list, access.get_raw_records([("thunk:A", None, None)]))
3033
list, access.get_raw_records([(True, "A", None, None)]))