/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_knit.py

• browse files at revision 2900.3.13
• compare with another revision
• compare with revision 0.200.155
• download diff
• download tarball
• view history from revision 2900.3.13

Show diffs side-by-side

added

removed

bzrlib/tests/test_knit.py

370

370

        self.assertEqual({'rev-id-1':(['foo\n', 'bar\n'], sha1sum)}, contents)

371

371

 

372

372

        raw_contents = list(data.read_records_iter_raw(records))

373

 

        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)

 

373

        self.assertEqual([('rev-id-1', gz_txt, sha1sum)], raw_contents)

374

374

 

375

375

    def test_not_enough_lines(self):

376

376

        sha1sum = sha.new('foo\n').hexdigest()

387

387

 

388

388

        # read_records_iter_raw won't detect that sort of mismatch/corruption

389

389

        raw_contents = list(data.read_records_iter_raw(records))

390

 

        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)

 

390

        self.assertEqual([('rev-id-1', gz_txt, sha1sum)], raw_contents)

391

391

 

392

392

    def test_too_many_lines(self):

393

393

        sha1sum = sha.new('foo\nbar\n').hexdigest()

405

405

 

406

406

        # read_records_iter_raw won't detect that sort of mismatch/corruption

407

407

        raw_contents = list(data.read_records_iter_raw(records))

408

 

        self.assertEqual([('rev-id-1', gz_txt)], raw_contents)

 

408

        self.assertEqual([('rev-id-1', gz_txt, sha1sum)], raw_contents)

409

409

 

410

410

    def test_mismatched_version_id(self):

411

411

        sha1sum = sha.new('foo\nbar\n').hexdigest()

1056

1056

        """

1057

1057

        index_memo = knit._index.get_position(version_id)

1058

1058

        record = (version_id, index_memo)

1059

 

        [(_, expected_content)] = list(knit._data.read_records_iter_raw([record]))

 

1059

        [(_, expected_content, _)] = list(knit._data.read_records_iter_raw([record]))

1060

1060

        self.assertEqual(expected_content, candidate_content)

1061

1061

 

1062

1062

 

1293

1293

        self.assertEquals(origins[1], ('text-1', 'b\n'))

1294

1294

        self.assertEquals(origins[2], ('text-1', 'c\n'))

1295

1295

 

1296

 

    def _test_join_with_factories(self, k1_factory, k2_factory):

1297

 

        k1 = make_file_knit('test1', get_transport('.'), factory=k1_factory, create=True)

1298

 

        k1.add_lines('text-a', [], ['a1\n', 'a2\n', 'a3\n'])

1299

 

        k1.add_lines('text-b', ['text-a'], ['a1\n', 'b2\n', 'a3\n'])

1300

 

        k1.add_lines('text-c', [], ['c1\n', 'c2\n', 'c3\n'])

1301

 

        k1.add_lines('text-d', ['text-c'], ['c1\n', 'd2\n', 'd3\n'])

1302

 

        k1.add_lines('text-m', ['text-b', 'text-d'], ['a1\n', 'b2\n', 'd3\n'])

1303

 

        k2 = make_file_knit('test2', get_transport('.'), factory=k2_factory, create=True)

1304

 

        count = k2.join(k1, version_ids=['text-m'])

1305

 

        self.assertEquals(count, 5)

1306

 

        self.assertTrue(k2.has_version('text-a'))

1307

 

        self.assertTrue(k2.has_version('text-c'))

1308

 

        origins = k2.annotate('text-m')

1309

 

        self.assertEquals(origins[0], ('text-a', 'a1\n'))

1310

 

        self.assertEquals(origins[1], ('text-b', 'b2\n'))

1311

 

        self.assertEquals(origins[2], ('text-d', 'd3\n'))

1312

 

 

1313

 

    def test_knit_join_plain_to_plain(self):

1314

 

        """Test joining a plain knit with a plain knit."""

1315

 

        self._test_join_with_factories(KnitPlainFactory(), KnitPlainFactory())

1316

 

 

1317

 

    def test_knit_join_anno_to_anno(self):

1318

 

        """Test joining an annotated knit with an annotated knit."""

1319

 

        self._test_join_with_factories(None, None)

1320

 

 

1321

 

    def test_knit_join_anno_to_plain(self):

1322

 

        """Test joining an annotated knit with a plain knit."""

1323

 

        self._test_join_with_factories(None, KnitPlainFactory())

1324

 

 

1325

 

    def test_knit_join_plain_to_anno(self):

1326

 

        """Test joining a plain knit with an annotated knit."""

1327

 

        self._test_join_with_factories(KnitPlainFactory(), None)

1328

 

 

1329

1296

    def test_reannotate(self):

1330

1297

        k1 = make_file_knit('knit1', get_transport('.'),

1331

1298

                               factory=KnitAnnotateFactory(), create=True)

1336

1303

 

1337

1304

        k2 = make_file_knit('test2', get_transport('.'),

1338

1305

                               factory=KnitAnnotateFactory(), create=True)

1339

 

        k2.join(k1, version_ids=['text-b'])

 

1306

        k2.insert_record_stream(k1.get_record_stream(k1.versions(),

 

1307

            'unordered', False))

1340

1308

 

1341

1309

        # 2

1342

1310

        k1.add_lines('text-X', ['text-b'], ['a\n', 'b\n'])

1346

1314

        k2.add_lines('text-Y', ['text-b'], ['b\n', 'c\n'])

1347

1315

 

1348

1316

        # test-c will have index 3

1349

 

        k1.join(k2, version_ids=['text-c'])

 

1317

        k1.insert_record_stream(k2.get_record_stream(['text-c'],

 

1318

            'unordered', False))

1350

1319

 

1351

1320

        lines = k1.get_lines('text-c')

1352

1321

        self.assertEquals(lines, ['z\n', 'c\n'])

1505

1474

        for plan_line, expected_line in zip(plan, AB_MERGE):

1506

1475

            self.assertEqual(plan_line, expected_line)

1507

1476

 

 

1477

 

 

1478

class GetDataStreamTests(KnitTests):

 

1479

    """Tests for get_data_stream."""

 

1480

 

1508

1481

    def test_get_stream_empty(self):

1509

1482

        """Get a data stream for an empty knit file."""

1510

1483

        k1 = self.make_test_knit()

1696

1669

            bytes = reader_callable(length)

1697

1670

            self.assertRecordContentEqual(k1, version_id, bytes)

1698

1671

 

 

1672

 

 

1673

class InsertDataStreamTests(KnitTests):

 

1674

    """Tests for insert_data_stream."""

 

1675

 

1699

1676

    def assertKnitFilesEqual(self, knit1, knit2):

1700

1677

        """Assert that the contents of the index and data files of two knits are

1701

1678

        equal.

1720

1697

            self.assertEqual(left.annotate(version),

1721

1698

                right.annotate(version))

1722

1699

 

1723

 

    def test_insert_data_stream_empty(self):

 

1700

    def test_empty_stream(self):

1724

1701

        """Inserting a data stream with no records should not put any data into

1725

1702

        the knit.

1726

1703

        """

1733

1710

                         k1.transport.get_bytes(k1._index._filename),

1734

1711

                         "The .kndx should have nothing apart from the header.")

1735

1712

 

1736

 

    def test_insert_data_stream_one_record(self):

 

1713

    def test_one_record(self):

1737

1714

        """Inserting a data stream with one record from a knit with one record

1738

1715

        results in byte-identical files.

1739

1716

        """

1744

1721

        target.insert_data_stream(data_stream)

1745

1722

        self.assertKnitFilesEqual(source, target)

1746

1723

 

1747

 

    def test_insert_data_stream_annotated_unannotated(self):

 

1724

    def test_annotated_stream_into_unannotated_knit(self):

1748

1725

        """Inserting an annotated datastream to an unannotated knit works."""

1749

1726

        # case one - full texts.

1750

1727

        source = self.make_test_knit(name='source', annotate=True)

1757

1734

        target.insert_data_stream(source.get_data_stream(['text-b']))

1758

1735

        self.assertKnitValuesEqual(source, target)

1759

1736

 

1760

 

    def test_insert_data_stream_unannotated_annotated(self):

 

1737

    def test_unannotated_stream_into_annotated_knit(self):

1761

1738

        """Inserting an unannotated datastream to an annotated knit works."""

1762

1739

        # case one - full texts.

1763

1740

        source = self.make_test_knit(name='source', annotate=False)

1770

1747

        target.insert_data_stream(source.get_data_stream(['text-b']))

1771

1748

        self.assertKnitValuesEqual(source, target)

1772

1749

 

1773

 

    def test_insert_data_stream_records_already_present(self):

 

1750

    def test_records_already_present(self):

1774

1751

        """Insert a data stream where some records are alreday present in the

1775

1752

        target, and some not.  Only the new records are inserted.

1776

1753

        """

1788

1765

        # record was not added a second time.

1789

1766

        self.assertKnitFilesEqual(source, target)

1790

1767

 

1791

 

    def test_insert_data_stream_multiple_records(self):

 

1768

    def test_multiple_records(self):

1792

1769

        """Inserting a data stream of all records from a knit with multiple

1793

1770

        records results in byte-identical files.

1794

1771

        """

1803

1780

        

1804

1781

        self.assertKnitFilesEqual(source, target)

1805

1782

 

1806

 

    def test_insert_data_stream_ghost_parent(self):

 

1783

    def test_ghost_parent(self):

1807

1784

        """Insert a data stream with a record that has a ghost parent."""

1808

1785

        # Make a knit with a record, text-a, that has a ghost parent.

1809

1786

        source = self.make_test_knit(name='source')

1824

1801

            target.get_parent_map(['text-a', 'text-ghost']))

1825

1802

        self.assertEqual(split_lines(TEXT_1), target.get_lines('text-a'))

1826

1803

 

1827

 

    def test_insert_data_stream_inconsistent_version_lines(self):

 

1804

    def test_inconsistent_version_lines(self):

1828

1805

        """Inserting a data stream which has different content for a version_id

1829

1806

        than already exists in the knit will raise KnitCorrupt.

1830

1807

        """

1838

1815

        self.assertRaises(

1839

1816

            errors.KnitCorrupt, target.insert_data_stream, data_stream)

1840

1817

 

1841

 

    def test_insert_data_stream_inconsistent_version_parents(self):

 

1818

    def test_inconsistent_version_parents(self):

1842

1819

        """Inserting a data stream which has different parents for a version_id

1843

1820

        than already exists in the knit will raise KnitCorrupt.

1844

1821

        """

1853

1830

        self.assertRaises(

1854

1831

            errors.KnitCorrupt, target.insert_data_stream, data_stream)

1855

1832

 

1856

 

    def test_insert_data_stream_unknown_format(self):

 

1833

    def test_unknown_stream_format(self):

1857

1834

        """A data stream in a different format to the target knit cannot be

1858

1835

        inserted.

1859

1836

 

1867

1844

            errors.KnitDataStreamUnknown,

1868

1845

            target.insert_data_stream, data_stream)

1869

1846

 

1870

 

    def test_insert_data_stream_bug_208418(self):

 

1847

    def test_bug_208418(self):

1871

1848

        """You can insert a stream with an incompatible format, even when:

1872

1849

          * the stream has a line-delta record,

1873

1850

          * whose parent is in the target, also stored as a line-delta

1897

1874

        target.insert_data_stream(data_stream)

1898

1875

        # No errors should have been raised.

1899

1876

 

1900

 

 

1901

 

    #  * test that a stream of "already present version, then new version"

1902

 

    #    inserts correctly.

1903

 

 

 

1877

    def test_line_delta_record_into_non_delta_knit(self):

 

1878

        # Make a data stream with a line-delta record

 

1879

        source = self.make_test_knit(name='source', delta=True)

 

1880

        base_lines = split_lines(TEXT_1)

 

1881

        source.add_lines('version-1', [], base_lines)

 

1882

        source.add_lines('version-2', ['version-1'], base_lines + ['a\n'])

 

1883

        # The second record should be a delta.

 

1884

        self.assertEqual('line-delta', source._index.get_method('version-2'))

 

1885

        data_stream = source.get_data_stream(['version-1', 'version-2'])

 

1886

 

 

1887

        # Insert the stream into a non-delta knit.

 

1888

        target = self.make_test_knit(name='target', delta=False)

 

1889

        target.insert_data_stream(data_stream)

 

1890

        

 

1891

        # Both versions are fulltexts in the target

 

1892

        self.assertEqual('fulltext', target._index.get_method('version-1'))

 

1893

        self.assertEqual('fulltext', target._index.get_method('version-2'))

 

1894

 

 

1895

 

 

1896

class DataStreamTests(KnitTests):

1904

1897

 

1905

1898

    def assertMadeStreamKnit(self, source_knit, versions, target_knit):

1906

1899

        """Assert that a knit made from a stream is as expected."""

2793

2786

        target_knit = self.make_test_knit(name='annotated', annotate=True)

2794

2787

        source_knit.add_lines("A", [], ["Foo\n"])

2795

2788

        # Give the target A, so we can try to thunk across to it.

2796

 

        target_knit.join(source_knit)

 

2789

        target_knit.insert_record_stream(source_knit.get_record_stream(['A'],

 

2790

            'unordered', False))

2797

2791

        index, access = self.get_index_access(target_knit,

2798

2792

            source_knit.get_data_stream([]))

2799

2793

        raw_data = list(access.get_raw_records([(True, "A", None, None)]))[0]

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0