/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_knit.py

• browse files at revision 6744
• compare with another revision
• compare with revision 5200.3.3
• download diff
• download tarball
• view history from revision 6744

Show diffs side-by-side

added

removed

breezy/tests/test_knit.py

30

30

    knit,

31

31

    pack,

32

32

    )

33

 

from ..errors import (

34

 

    KnitHeaderError,

35

 

    NoSuchFile,

36

 

    )

37

33

from ..bzr.index import *

38

34

from ..bzr.knit import (

39

35

    AnnotatedKnitContent,

40

36

    KnitContent,

 

37

    KnitCorrupt,

 

38

    KnitDataStreamIncompatible,

 

39

    KnitDataStreamUnknown,

 

40

    KnitHeaderError,

 

41

    KnitIndexUnknownMethod,

41

42

    KnitVersionedFiles,

42

43

    PlainKnitContent,

43

44

    _VFContentMapGenerator,

75

76

    'breezy.bzr._knit_load_data_pyx')

76

77

 

77

78

 

 

79

class ErrorTests(TestCase):

 

80

 

 

81

    def test_knit_data_stream_incompatible(self):

 

82

        error = KnitDataStreamIncompatible(

 

83

            'stream format', 'target format')

 

84

        self.assertEqual('Cannot insert knit data stream of format '

 

85

                         '"stream format" into knit of format '

 

86

                         '"target format".', str(error))

 

87

 

 

88

    def test_knit_data_stream_unknown(self):

 

89

        error = KnitDataStreamUnknown(

 

90

            'stream format')

 

91

        self.assertEqual('Cannot parse knit data stream of format '

 

92

                         '"stream format".', str(error))

 

93

 

 

94

    def test_knit_header_error(self):

 

95

        error = KnitHeaderError('line foo\n', 'path/to/file')

 

96

        self.assertEqual("Knit header error: 'line foo\\n' unexpected"

 

97

                         " for file \"path/to/file\".", str(error))

 

98

 

 

99

    def test_knit_index_unknown_method(self):

 

100

        error = KnitIndexUnknownMethod('http://host/foo.kndx',

 

101

                                       ['bad', 'no-eol'])

 

102

        self.assertEqual("Knit index http://host/foo.kndx does not have a"

 

103

                         " known method in options: ['bad', 'no-eol']",

 

104

                         str(error))

 

105

 

 

106

 

78

107

class KnitContentTestsMixin(object):

79

108

 

80

109

    def test_constructor(self):

248

277

 

249

278

    def get(self, filename):

250

279

        if self.file_lines is None:

251

 

            raise NoSuchFile(filename)

 

280

            raise errors.NoSuchFile(filename)

252

281

        else:

253

282

            return BytesIO(b"\n".join(self.file_lines))

254

283

 

806

835

        access = _KnitKeyAccess(transport, ConstantMapper('filename'))

807

836

        knit = KnitVersionedFiles(None, access)

808

837

        records = [(('rev-id-1',), (('rev-id-1',), 0, len(gz_txt)))]

809

 

        self.assertRaises(errors.KnitCorrupt, list,

 

838

        self.assertRaises(KnitCorrupt, list,

810

839

            knit._read_records_iter(records))

811

840

 

812

841

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

825

854

        access = _KnitKeyAccess(transport, ConstantMapper('filename'))

826

855

        knit = KnitVersionedFiles(None, access)

827

856

        records = [(('rev-id-1',), (('rev-id-1',), 0, len(gz_txt)))]

828

 

        self.assertRaises(errors.KnitCorrupt, list,

 

857

        self.assertRaises(KnitCorrupt, list,

829

858

            knit._read_records_iter(records))

830

859

 

831

860

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

844

873

        knit = KnitVersionedFiles(None, access)

845

874

        # We are asking for rev-id-2, but the data is rev-id-1

846

875

        records = [(('rev-id-2',), (('rev-id-2',), 0, len(gz_txt)))]

847

 

        self.assertRaises(errors.KnitCorrupt, list,

 

876

        self.assertRaises(KnitCorrupt, list,

848

877

            knit._read_records_iter(records))

849

878

 

850

879

        # read_records_iter_raw detects mismatches in the header

851

 

        self.assertRaises(errors.KnitCorrupt, list,

 

880

        self.assertRaises(KnitCorrupt, list,

852

881

            knit._read_records_iter_raw(records))

853

882

 

854

883

    def test_uncompressed_data(self):

864

893

        records = [(('rev-id-1',), (('rev-id-1',), 0, len(txt)))]

865

894

 

866

895

        # We don't have valid gzip data ==> corrupt

867

 

        self.assertRaises(errors.KnitCorrupt, list,

 

896

        self.assertRaises(KnitCorrupt, list,

868

897

            knit._read_records_iter(records))

869

898

 

870

899

        # read_records_iter_raw will notice the bad data

871

 

        self.assertRaises(errors.KnitCorrupt, list,

 

900

        self.assertRaises(KnitCorrupt, list,

872

901

            knit._read_records_iter_raw(records))

873

902

 

874

903

    def test_corrupted_data(self):

884

913

        access = _KnitKeyAccess(transport, ConstantMapper('filename'))

885

914

        knit = KnitVersionedFiles(None, access)

886

915

        records = [(('rev-id-1',), (('rev-id-1',), 0, len(gz_txt)))]

887

 

        self.assertRaises(errors.KnitCorrupt, list,

 

916

        self.assertRaises(KnitCorrupt, list,

888

917

            knit._read_records_iter(records))

889

918

        # read_records_iter_raw will barf on bad gz data

890

 

        self.assertRaises(errors.KnitCorrupt, list,

 

919

        self.assertRaises(KnitCorrupt, list,

891

920

            knit._read_records_iter_raw(records))

892

921

 

893

922

 

1160

1189

 

1161

1190

        self.assertEqual("fulltext", index.get_method("a"))

1162

1191

        self.assertEqual("line-delta", index.get_method("b"))

1163

 

        self.assertRaises(errors.KnitIndexUnknownMethod, index.get_method, "c")

 

1192

        self.assertRaises(knit.KnitIndexUnknownMethod, index.get_method, "c")

1164

1193

 

1165

1194

    def test_get_options(self):

1166

1195

        transport = MockTransport([

1196

1225

            "b option 0 1 4 :"  # We don't have a 4th record

1197

1226

            ])

1198

1227

        index = self.get_knit_index(transport, 'filename', 'r')

1199

 

        self.assertRaises(errors.KnitCorrupt, index.keys)

 

1228

        self.assertRaises(KnitCorrupt, index.keys)

1200

1229

 

1201

1230

    def test_corrupted_parent(self):

1202

1231

        transport = MockTransport([

1206

1235

            "c option 0 1 1v :", # Can't have a parent of '1v'

1207

1236

            ])

1208

1237

        index = self.get_knit_index(transport, 'filename', 'r')

1209

 

        self.assertRaises(errors.KnitCorrupt, index.keys)

 

1238

        self.assertRaises(KnitCorrupt, index.keys)

1210

1239

 

1211

1240

    def test_corrupted_parent_in_list(self):

1212

1241

        transport = MockTransport([

1216

1245

            "c option 0 1 1 v :", # Can't have a parent of 'v'

1217

1246

            ])

1218

1247

        index = self.get_knit_index(transport, 'filename', 'r')

1219

 

        self.assertRaises(errors.KnitCorrupt, index.keys)

 

1248

        self.assertRaises(KnitCorrupt, index.keys)

1220

1249

 

1221

1250

    def test_invalid_position(self):

1222

1251

        transport = MockTransport([

1224

1253

            "a option 1v 1 :",

1225

1254

            ])

1226

1255

        index = self.get_knit_index(transport, 'filename', 'r')

1227

 

        self.assertRaises(errors.KnitCorrupt, index.keys)

 

1256

        self.assertRaises(KnitCorrupt, index.keys)

1228

1257

 

1229

1258

    def test_invalid_size(self):

1230

1259

        transport = MockTransport([

1232

1261

            "a option 1 1v :",

1233

1262

            ])

1234

1263

        index = self.get_knit_index(transport, 'filename', 'r')

1235

 

        self.assertRaises(errors.KnitCorrupt, index.keys)

 

1264

        self.assertRaises(KnitCorrupt, index.keys)

1236

1265

 

1237

1266

    def test_scan_unvalidated_index_not_implemented(self):

1238

1267

        transport = MockTransport()

1477

1506

        target.add_lines(basis, (), ['gam\n'])

1478

1507

        target.insert_record_stream(

1479

1508

            source.get_record_stream([broken], 'unordered', False))

1480

 

        err = self.assertRaises(errors.KnitCorrupt,

 

1509

        err = self.assertRaises(KnitCorrupt,

1481

1510

            target.get_record_stream([broken], 'unordered', True

1482

1511

            ).next().get_bytes_as, 'chunked')

1483

1512

        self.assertEqual(['gam\n', 'bar\n'], err.content)

1673

1702

 

1674

1703

    def test_add_version_delta_not_delta_index(self):

1675

1704

        index = self.two_graph_index(catch_adds=True)

1676

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1705

        self.assertRaises(KnitCorrupt, index.add_records,

1677

1706

            [(('new',), 'no-eol,line-delta', (None, 0, 100), [('parent',)])])

1678

1707

        self.assertEqual([], self.caught_entries)

1679

1708

 

1694

1723

    def test_add_version_different_dup(self):

1695

1724

        index = self.two_graph_index(deltas=True, catch_adds=True)

1696

1725

        # change options

1697

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1726

        self.assertRaises(KnitCorrupt, index.add_records,

1698

1727

            [(('tip',), 'line-delta', (None, 0, 100), [('parent',)])])

1699

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1728

        self.assertRaises(KnitCorrupt, index.add_records,

1700

1729

            [(('tip',), 'fulltext', (None, 0, 100), [('parent',)])])

1701

1730

        # parents

1702

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1731

        self.assertRaises(KnitCorrupt, index.add_records,

1703

1732

            [(('tip',), 'fulltext,no-eol', (None, 0, 100), [])])

1704

1733

        self.assertEqual([], self.caught_entries)

1705

1734

 

1727

1756

 

1728

1757

    def test_add_versions_delta_not_delta_index(self):

1729

1758

        index = self.two_graph_index(catch_adds=True)

1730

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1759

        self.assertRaises(KnitCorrupt, index.add_records,

1731

1760

            [(('new',), 'no-eol,line-delta', (None, 0, 100), [('parent',)])])

1732

1761

        self.assertEqual([], self.caught_entries)

1733

1762

 

1754

1783

    def test_add_versions_different_dup(self):

1755

1784

        index = self.two_graph_index(deltas=True, catch_adds=True)

1756

1785

        # change options

1757

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1786

        self.assertRaises(KnitCorrupt, index.add_records,

1758

1787

            [(('tip',), 'line-delta', (None, 0, 100), [('parent',)])])

1759

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1788

        self.assertRaises(KnitCorrupt, index.add_records,

1760

1789

            [(('tip',), 'fulltext', (None, 0, 100), [('parent',)])])

1761

1790

        # parents

1762

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1791

        self.assertRaises(KnitCorrupt, index.add_records,

1763

1792

            [(('tip',), 'fulltext,no-eol', (None, 0, 100), [])])

1764

1793

        # change options in the second record

1765

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1794

        self.assertRaises(KnitCorrupt, index.add_records,

1766

1795

            [(('tip',), 'fulltext,no-eol', (None, 0, 100), [('parent',)]),

1767

1796

             (('tip',), 'line-delta', (None, 0, 100), [('parent',)])])

1768

1797

        self.assertEqual([], self.caught_entries)

1895

1924

 

1896

1925

    def test_parents_deltas_incompatible(self):

1897

1926

        index = CombinedGraphIndex([])

1898

 

        self.assertRaises(errors.KnitError, _KnitGraphIndex, lambda:True,

 

1927

        self.assertRaises(knit.KnitError, _KnitGraphIndex, lambda:True,

1899

1928

            index, deltas=True, parents=False)

1900

1929

 

1901

1930

    def two_graph_index(self, catch_adds=False):

1964

1993

 

1965

1994

    def test_add_version_delta_not_delta_index(self):

1966

1995

        index = self.two_graph_index(catch_adds=True)

1967

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

1996

        self.assertRaises(KnitCorrupt, index.add_records,

1968

1997

            [(('new',), 'no-eol,line-delta', (None, 0, 100), [])])

1969

1998

        self.assertEqual([], self.caught_entries)

1970

1999

 

1983

2012

    def test_add_version_different_dup(self):

1984

2013

        index = self.two_graph_index(catch_adds=True)

1985

2014

        # change options

1986

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2015

        self.assertRaises(KnitCorrupt, index.add_records,

1987

2016

            [(('tip',), 'no-eol,line-delta', (None, 0, 100), [])])

1988

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2017

        self.assertRaises(KnitCorrupt, index.add_records,

1989

2018

            [(('tip',), 'line-delta,no-eol', (None, 0, 100), [])])

1990

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2019

        self.assertRaises(KnitCorrupt, index.add_records,

1991

2020

            [(('tip',), 'fulltext', (None, 0, 100), [])])

1992

2021

        # parents

1993

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2022

        self.assertRaises(KnitCorrupt, index.add_records,

1994

2023

            [(('tip',), 'fulltext,no-eol', (None, 0, 100), [('parent',)])])

1995

2024

        self.assertEqual([], self.caught_entries)

1996

2025

 

2006

2035

 

2007

2036

    def test_add_versions_delta_not_delta_index(self):

2008

2037

        index = self.two_graph_index(catch_adds=True)

2009

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2038

        self.assertRaises(KnitCorrupt, index.add_records,

2010

2039

            [(('new',), 'no-eol,line-delta', (None, 0, 100), [('parent',)])])

2011

2040

        self.assertEqual([], self.caught_entries)

2012

2041

 

2013

2042

    def test_add_versions_parents_not_parents_index(self):

2014

2043

        index = self.two_graph_index(catch_adds=True)

2015

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2044

        self.assertRaises(KnitCorrupt, index.add_records,

2016

2045

            [(('new',), 'no-eol,fulltext', (None, 0, 100), [('parent',)])])

2017

2046

        self.assertEqual([], self.caught_entries)

2018

2047

 

2035

2064

    def test_add_versions_different_dup(self):

2036

2065

        index = self.two_graph_index(catch_adds=True)

2037

2066

        # change options

2038

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2067

        self.assertRaises(KnitCorrupt, index.add_records,

2039

2068

            [(('tip',), 'no-eol,line-delta', (None, 0, 100), [])])

2040

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2069

        self.assertRaises(KnitCorrupt, index.add_records,

2041

2070

            [(('tip',), 'line-delta,no-eol', (None, 0, 100), [])])

2042

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2071

        self.assertRaises(KnitCorrupt, index.add_records,

2043

2072

            [(('tip',), 'fulltext', (None, 0, 100), [])])

2044

2073

        # parents

2045

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2074

        self.assertRaises(KnitCorrupt, index.add_records,

2046

2075

            [(('tip',), 'fulltext,no-eol', (None, 0, 100), [('parent',)])])

2047

2076

        # change options in the second record

2048

 

        self.assertRaises(errors.KnitCorrupt, index.add_records,

 

2077

        self.assertRaises(KnitCorrupt, index.add_records,

2049

2078

            [(('tip',), 'fulltext,no-eol', (None, 0, 100), []),

2050

2079

             (('tip',), 'no-eol,line-delta', (None, 0, 100), [])])

2051

2080

        self.assertEqual([], self.caught_entries)

• Older »

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