# Copyright (C) 2007 Canonical Ltd # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA """Tests for indices.""" from bzrlib import errors from bzrlib.index import * from bzrlib.tests import TestCaseWithMemoryTransport class TestGraphIndexBuilder(TestCaseWithMemoryTransport): def test_build_index_empty(self): builder = GraphIndexBuilder() stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=0\n\n", contents) def test_build_index_one_reference_list_empty(self): builder = GraphIndexBuilder(reference_lists=1) stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=1\n\n", contents) def test_build_index_two_reference_list_empty(self): builder = GraphIndexBuilder(reference_lists=2) stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=2\n\n", contents) def test_build_index_one_node(self): builder = GraphIndexBuilder() builder.add_node('akey', (), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=0\n" "akey\x00\x00\x00data\n\n", contents) def test_add_node_empty_value(self): builder = GraphIndexBuilder() builder.add_node('akey', (), '') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=0\n" "akey\x00\x00\x00\n\n", contents) def test_build_index_two_nodes_sorted(self): # the highest sorted node comes first. builder = GraphIndexBuilder() # use three to have a good chance of glitching dictionary hash # lookups etc. Insert in randomish order that is not correct # and not the reverse of the correct order. builder.add_node('2002', (), 'data') builder.add_node('2000', (), 'data') builder.add_node('2001', (), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=0\n" "2000\x00\x00\x00data\n" "2001\x00\x00\x00data\n" "2002\x00\x00\x00data\n" "\n", contents) def test_build_index_reference_lists_are_included_one(self): builder = GraphIndexBuilder(reference_lists=1) builder.add_node('key', ([], ), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=1\n" "key\x00\x00\x00data\n" "\n", contents) def test_build_index_reference_lists_are_included_two(self): builder = GraphIndexBuilder(reference_lists=2) builder.add_node('key', ([], []), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=2\n" "key\x00\x00\t\x00data\n" "\n", contents) def test_node_references_are_byte_offsets(self): builder = GraphIndexBuilder(reference_lists=1) builder.add_node('reference', ([], ), 'data') builder.add_node('key', (['reference'], ), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=1\n" "key\x00\x0051\x00data\n" "reference\x00\x00\x00data\n" "\n", contents) def test_node_references_are_cr_delimited(self): builder = GraphIndexBuilder(reference_lists=1) builder.add_node('reference', ([], ), 'data') builder.add_node('reference2', ([], ), 'data') builder.add_node('key', (['reference', 'reference2'], ), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=1\n" "key\x00\x0054\r71\x00data\n" "reference\x00\x00\x00data\n" "reference2\x00\x00\x00data\n" "\n", contents) def test_multiple_reference_lists_are_tab_delimited(self): builder = GraphIndexBuilder(reference_lists=2) builder.add_node('keference', ([], []), 'data') builder.add_node('rey', (['keference'], ['keference']), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=2\n" "keference\x00\x00\t\x00data\n" "rey\x00\x0038\t38\x00data\n" "\n", contents) def test_add_node_referencing_missing_key_makes_absent(self): builder = GraphIndexBuilder(reference_lists=1) builder.add_node('rey', (['beference', 'aeference2'], ), 'data') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=1\n" "aeference2\x00a\x00\x00\n" "beference\x00a\x00\x00\n" "rey\x00\x0053\r38\x00data\n" "\n", contents) def test_node_references_three_digits(self): # test the node digit expands as needed. builder = GraphIndexBuilder(reference_lists=1) references = map(str, reversed(range(9))) builder.add_node('2-key', (references, ), '') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=1\n" "0\x00a\x00\x00\n" "1\x00a\x00\x00\n" "2\x00a\x00\x00\n" "2-key\x00\x00130\r124\r118\r112\r106\r100\r050\r044\r038\x00\n" "3\x00a\x00\x00\n" "4\x00a\x00\x00\n" "5\x00a\x00\x00\n" "6\x00a\x00\x00\n" "7\x00a\x00\x00\n" "8\x00a\x00\x00\n" "\n", contents) def test_absent_has_no_reference_overhead(self): # the offsets after an absent record should be correct when there are # >1 reference lists. builder = GraphIndexBuilder(reference_lists=2) builder.add_node('parent', (['aail', 'zther'], []), '') stream = builder.finish() contents = stream.read() self.assertEqual("Bazaar Graph Index 1\nnode_ref_lists=2\n" "aail\x00a\x00\x00\n" "parent\x00\x0038\r63\t\x00\n" "zther\x00a\x00\x00\n" "\n", contents) def test_add_node_bad_key(self): builder = GraphIndexBuilder() for bad_char in '\t\n\x0b\x0c\r\x00 ': self.assertRaises(errors.BadIndexKey, builder.add_node, 'a%skey' % bad_char, (), 'data') self.assertRaises(errors.BadIndexKey, builder.add_node, '', (), 'data') def test_add_node_bad_data(self): builder = GraphIndexBuilder() self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', (), 'data\naa') self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', (), 'data\x00aa') def test_add_node_bad_mismatched_ref_lists_length(self): builder = GraphIndexBuilder() self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', ([], ), 'data aa') builder = GraphIndexBuilder(reference_lists=1) self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', (), 'data aa') self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', ([], []), 'data aa') builder = GraphIndexBuilder(reference_lists=2) self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', (), 'data aa') self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', ([], ), 'data aa') self.assertRaises(errors.BadIndexValue, builder.add_node, 'akey', ([], [], []), 'data aa') def test_add_node_bad_key_in_reference_lists(self): # first list, first key - trivial builder = GraphIndexBuilder(reference_lists=1) self.assertRaises(errors.BadIndexKey, builder.add_node, 'akey', (['a key'], ), 'data aa') # need to check more than the first key in the list self.assertRaises(errors.BadIndexKey, builder.add_node, 'akey', (['agoodkey', 'this is a bad key'], ), 'data aa') # and if there is more than one list it should be getting checked # too builder = GraphIndexBuilder(reference_lists=2) self.assertRaises(errors.BadIndexKey, builder.add_node, 'akey', ([], ['a bad key']), 'data aa') def test_add_duplicate_key(self): builder = GraphIndexBuilder() builder.add_node('key', (), 'data') self.assertRaises(errors.BadIndexDuplicateKey, builder.add_node, 'key', (), 'data') def test_add_key_after_referencing_key(self): builder = GraphIndexBuilder(reference_lists=1) builder.add_node('key', (['reference'], ), 'data') builder.add_node('reference', ([],), 'data') class TestGraphIndex(TestCaseWithMemoryTransport): def make_index(self, ref_lists=0, nodes=[]): builder = GraphIndexBuilder(ref_lists) for node, references, value in nodes: builder.add_node(node, references, value) stream = builder.finish() trans = self.get_transport() trans.put_file('index', stream) return GraphIndex(trans, 'index') def test_open_bad_index_no_error(self): trans = self.get_transport() trans.put_bytes('name', "not an index\n") index = GraphIndex(trans, 'name') def test_iter_all_entries_empty(self): index = self.make_index() self.assertEqual([], list(index.iter_all_entries())) def test_iter_all_entries_simple(self): index = self.make_index(nodes=[('name', (), 'data')]) self.assertEqual([('name', (), 'data')], list(index.iter_all_entries())) def test_iter_all_entries_references_resolved(self): index = self.make_index(1, nodes=[ ('name', (['ref'], ), 'data'), ('ref', ([], ), 'refdata')]) self.assertEqual(set([('name', (('ref',),), 'data'), ('ref', ((), ), 'refdata')]), set(index.iter_all_entries())) def test_iteration_absent_skipped(self): index = self.make_index(1, nodes=[ ('name', (['ref'], ), 'data')]) self.assertEqual(set([('name', (('ref',),), 'data')]), set(index.iter_all_entries())) self.assertEqual(set([('name', (('ref',),), 'data')]), set(index.iter_entries(['name']))) self.assertEqual([], list(index.iter_entries(['ref']))) def test_iter_all_keys(self): index = self.make_index(1, nodes=[ ('name', (['ref'], ), 'data'), ('ref', ([], ), 'refdata')]) self.assertEqual(set([('name', (('ref',),), 'data'), ('ref', ((), ), 'refdata')]), set(index.iter_entries(['name', 'ref']))) def test_iter_nothing_empty(self): index = self.make_index() self.assertEqual([], list(index.iter_entries([]))) def test_iter_missing_entry_empty(self): index = self.make_index() self.assertEqual([], list(index.iter_entries(['a']))) def test_validate_bad_index_errors(self): trans = self.get_transport() trans.put_bytes('name', "not an index\n") index = GraphIndex(trans, 'name') self.assertRaises(errors.BadIndexFormatSignature, index.validate) def test_validate_bad_node_refs(self): index = self.make_index(2) trans = self.get_transport() content = trans.get_bytes('index') # change the options line to end with a rather than a parseable number new_content = content[:-2] + 'a\n\n' trans.put_bytes('index', new_content) self.assertRaises(errors.BadIndexOptions, index.validate) def test_validate_missing_end_line_empty(self): index = self.make_index(2) trans = self.get_transport() content = trans.get_bytes('index') # truncate the last byte trans.put_bytes('index', content[:-1]) self.assertRaises(errors.BadIndexData, index.validate) def test_validate_missing_end_line_nonempty(self): index = self.make_index(2, [('key', ([], []), '')]) trans = self.get_transport() content = trans.get_bytes('index') # truncate the last byte trans.put_bytes('index', content[:-1]) self.assertRaises(errors.BadIndexData, index.validate) def test_validate_empty(self): index = self.make_index() index.validate() def test_validate_no_refs_content(self): index = self.make_index(nodes=[('key', (), 'value')]) index.validate() class TestCombinedGraphIndex(TestCaseWithMemoryTransport): def make_index(self, name, ref_lists=0, nodes=[]): builder = GraphIndexBuilder(ref_lists) for node, references, value in nodes: builder.add_node(node, references, value) stream = builder.finish() trans = self.get_transport() trans.put_file(name, stream) return GraphIndex(trans, name) def test_open_missing_index_no_error(self): trans = self.get_transport() index1 = GraphIndex(trans, 'missing') index = CombinedGraphIndex([index1]) def test_add_index(self): index = CombinedGraphIndex([]) index1 = self.make_index('name', nodes=[('key', (), '')]) index.insert_index(0, index1) self.assertEqual([('key', (), '')], list(index.iter_all_entries())) def test_iter_all_entries_empty(self): index = CombinedGraphIndex([]) self.assertEqual([], list(index.iter_all_entries())) def test_iter_all_entries_children_empty(self): index1 = self.make_index('name') index = CombinedGraphIndex([index1]) self.assertEqual([], list(index.iter_all_entries())) def test_iter_all_entries_simple(self): index1 = self.make_index('name', nodes=[('name', (), 'data')]) index = CombinedGraphIndex([index1]) self.assertEqual([('name', (), 'data')], list(index.iter_all_entries())) def test_iter_all_entries_two_indices(self): index1 = self.make_index('name1', nodes=[('name', (), 'data')]) index2 = self.make_index('name2', nodes=[('2', (), '')]) index = CombinedGraphIndex([index1, index2]) self.assertEqual([('name', (), 'data'), ('2', (), '')], list(index.iter_all_entries())) def test_iter_entries_two_indices_dup_key(self): index1 = self.make_index('name1', nodes=[('name', (), 'data')]) index2 = self.make_index('name2', nodes=[('name', (), 'data')]) index = CombinedGraphIndex([index1, index2]) self.assertEqual([('name', (), 'data')], list(index.iter_entries(['name']))) def test_iter_all_entries_two_indices_dup_key(self): index1 = self.make_index('name1', nodes=[('name', (), 'data')]) index2 = self.make_index('name2', nodes=[('name', (), 'data')]) index = CombinedGraphIndex([index1, index2]) self.assertEqual([('name', (), 'data')], list(index.iter_all_entries())) def test_iter_nothing_empty(self): index = CombinedGraphIndex([]) self.assertEqual([], list(index.iter_entries([]))) def test_iter_nothing_children_empty(self): index1 = self.make_index('name') index = CombinedGraphIndex([index1]) self.assertEqual([], list(index.iter_entries([]))) def test_iter_all_keys(self): index1 = self.make_index('1', 1, nodes=[ ('name', (['ref'], ), 'data')]) index2 = self.make_index('2', 1, nodes=[ ('ref', ([], ), 'refdata')]) index = CombinedGraphIndex([index1, index2]) self.assertEqual(set([('name', (('ref',),), 'data'), ('ref', ((), ), 'refdata')]), set(index.iter_entries(['name', 'ref']))) def test_iter_all_keys_dup_entry(self): index1 = self.make_index('1', 1, nodes=[ ('name', (['ref'], ), 'data'), ('ref', ([], ), 'refdata')]) index2 = self.make_index('2', 1, nodes=[ ('ref', ([], ), 'refdata')]) index = CombinedGraphIndex([index1, index2]) self.assertEqual(set([('name', (('ref',),), 'data'), ('ref', ((), ), 'refdata')]), set(index.iter_entries(['name', 'ref']))) def test_iter_missing_entry_empty(self): index = CombinedGraphIndex([]) self.assertEqual([], list(index.iter_entries(['a']))) def test_iter_missing_entry_one_index(self): index1 = self.make_index('1') index = CombinedGraphIndex([index1]) self.assertEqual([], list(index.iter_entries(['a']))) def test_iter_missing_entry_two_index(self): index1 = self.make_index('1') index2 = self.make_index('2') index = CombinedGraphIndex([index1, index2]) self.assertEqual([], list(index.iter_entries(['a']))) def test_iter_entry_present_one_index_only(self): index1 = self.make_index('1', nodes=[('key', (), '')]) index2 = self.make_index('2', nodes=[]) index = CombinedGraphIndex([index1, index2]) self.assertEqual([('key', (), '')], list(index.iter_entries(['key']))) # and in the other direction index = CombinedGraphIndex([index2, index1]) self.assertEqual([('key', (), '')], list(index.iter_entries(['key']))) def test_validate_bad_child_index_errors(self): trans = self.get_transport() trans.put_bytes('name', "not an index\n") index1 = GraphIndex(trans, 'name') index = CombinedGraphIndex([index1]) self.assertRaises(errors.BadIndexFormatSignature, index.validate) def test_validate_empty(self): index = CombinedGraphIndex([]) index.validate() class TestInMemoryGraphIndex(TestCaseWithMemoryTransport): def make_index(self, ref_lists=0, nodes=[]): result = InMemoryGraphIndex(ref_lists) result.add_nodes(nodes) return result def test_add_nodes(self): index = self.make_index(1) index.add_nodes([('name', ([],), 'data')]) index.add_nodes([('name2', ([],), ''), ('name3', (['r'],), '')]) self.assertEqual(set([ ('name', ((),), 'data'), ('name2', ((),), ''), ('name3', (('r',),), ''), ]), set(index.iter_all_entries())) def test_iter_all_entries_empty(self): index = self.make_index() self.assertEqual([], list(index.iter_all_entries())) def test_iter_all_entries_simple(self): index = self.make_index(nodes=[('name', (), 'data')]) self.assertEqual([('name', (), 'data')], list(index.iter_all_entries())) def test_iter_all_entries_references(self): index = self.make_index(1, nodes=[ ('name', (['ref'], ), 'data'), ('ref', ([], ), 'refdata')]) self.assertEqual(set([('name', (('ref',),), 'data'), ('ref', ((), ), 'refdata')]), set(index.iter_all_entries())) def test_iteration_absent_skipped(self): index = self.make_index(1, nodes=[ ('name', (['ref'], ), 'data')]) self.assertEqual(set([('name', (('ref',),), 'data')]), set(index.iter_all_entries())) self.assertEqual(set([('name', (('ref',),), 'data')]), set(index.iter_entries(['name']))) self.assertEqual([], list(index.iter_entries(['ref']))) def test_iter_all_keys(self): index = self.make_index(1, nodes=[ ('name', (['ref'], ), 'data'), ('ref', ([], ), 'refdata')]) self.assertEqual(set([('name', (('ref',),), 'data'), ('ref', ((), ), 'refdata')]), set(index.iter_entries(['name', 'ref']))) def test_iter_nothing_empty(self): index = self.make_index() self.assertEqual([], list(index.iter_entries([]))) def test_iter_missing_entry_empty(self): index = self.make_index() self.assertEqual([], list(index.iter_entries(['a']))) def test_validate_empty(self): index = self.make_index() index.validate() def test_validate_no_refs_content(self): index = self.make_index(nodes=[('key', (), 'value')]) index.validate()