997
997
factory=factory, create=True,
998
998
delay_create=delay_create, index=index)
1000
def assertRecordContentEqual(self, knit, version_id, candidate_content):
1001
"""Assert that some raw record content matches the raw record content
1002
for a particular version_id in the given knit.
1004
index_memo = knit._index.get_position(version_id)
1005
record = (version_id, index_memo)
1006
[(_, expected_content)] = list(knit._data.read_records_iter_raw([record]))
1007
self.assertEqual(expected_content, candidate_content)
1001
1010
class BasicKnitTests(KnitTests):
1460
1469
for plan_line, expected_line in zip(plan, AB_MERGE):
1461
1470
self.assertEqual(plan_line, expected_line)
1463
def assertRecordContentEqual(self, knit, version_id, candidate_content):
1464
"""Assert that some raw record content matches the raw record content
1465
for a particular version_id in the given knit.
1467
index_memo = knit._index.get_position(version_id)
1468
record = (version_id, index_memo)
1469
[(_, expected_content)] = list(knit._data.read_records_iter_raw([record]))
1470
self.assertEqual(expected_content, candidate_content)
1472
1472
def test_get_stream_empty(self):
1473
1473
"""Get a data stream for an empty knit file."""
1474
1474
k1 = self.make_test_knit()
1615
1615
bytes = reader_callable(length)
1616
1616
self.assertRecordContentEqual(k1, version_id, bytes)
1618
def test_get_data_stream(self):
1619
# Make a simple knit
1620
k1 = self.make_test_knit()
1621
k1.add_lines('text-a', [], split_lines(TEXT_1))
1623
# Serialise it, check the output.
1624
bytes = k1.get_stream_as_bytes(['text-a'])
1625
data = bencode.bdecode(bytes)
1626
format, record = data
1627
self.assertEqual('knit-plain', format)
1628
self.assertEqual(['text-a', ['fulltext'], []], record[:3])
1629
self.assertRecordContentEqual(k1, 'text-a', record[3])
1631
def test_get_stream_as_bytes_all(self):
1632
"""Get a serialised data stream for all the records in a knit.
1634
Much like test_get_stream_all, except for get_stream_as_bytes.
1636
k1 = self.make_test_knit()
1637
# Insert the same data as test_knit_join, as they seem to cover a range
1638
# of cases (no parents, one parent, multiple parents).
1640
('text-a', [], TEXT_1),
1641
('text-b', ['text-a'], TEXT_1),
1642
('text-c', [], TEXT_1),
1643
('text-d', ['text-c'], TEXT_1),
1644
('text-m', ['text-b', 'text-d'], TEXT_1),
1646
expected_data_list = [
1647
# version, options, parents
1648
('text-a', ['fulltext'], []),
1649
('text-b', ['line-delta'], ['text-a']),
1650
('text-c', ['fulltext'], []),
1651
('text-d', ['line-delta'], ['text-c']),
1652
('text-m', ['line-delta'], ['text-b', 'text-d']),
1654
for version_id, parents, lines in test_data:
1655
k1.add_lines(version_id, parents, split_lines(lines))
1657
bytes = k1.get_stream_as_bytes(
1658
['text-a', 'text-b', 'text-c', 'text-d', 'text-m'])
1660
data = bencode.bdecode(bytes)
1661
format = data.pop(0)
1662
self.assertEqual('knit-plain', format)
1664
for expected, actual in zip(expected_data_list, data):
1665
expected_version = expected[0]
1666
expected_options = expected[1]
1667
expected_parents = expected[2]
1668
version, options, parents, bytes = actual
1669
self.assertEqual(expected_version, version)
1670
self.assertEqual(expected_options, options)
1671
self.assertEqual(expected_parents, parents)
1672
self.assertRecordContentEqual(k1, version, bytes)
1674
1618
def assertKnitFilesEqual(self, knit1, knit2):
1675
1619
"""Assert that the contents of the index and data files of two knits are