64
70
load_tests = scenarios.load_tests_apply_scenarios
66
# We need adapters that can build a config store in a test context. Test
67
# classes, based on TestCaseWithTransport, can use the registry to parametrize
68
# themselves. The builder will receive a test instance and should return a
69
# ready-to-use store. Plugins that defines new stores can also register
70
# themselves here to be tested against the tests defined below.
72
# FIXME: plugins should *not* need to import test_config to register their
73
# helpers (or selftest -s xxx will be broken), the following registry should be
74
# moved to bzrlib.config instead so that selftest -s bt.test_config also runs
75
# the plugin specific tests (selftest -s bp.xxx won't, that would be against
76
# the spirit of '-s') -- vila 20110503
77
test_store_builder_registry = registry.Registry()
78
test_store_builder_registry.register(
72
# Register helpers to build stores
73
config.test_store_builder_registry.register(
79
74
'configobj', lambda test: config.IniFileStore(test.get_transport(),
81
test_store_builder_registry.register(
76
config.test_store_builder_registry.register(
82
77
'bazaar', lambda test: config.GlobalStore())
83
test_store_builder_registry.register(
78
config.test_store_builder_registry.register(
84
79
'location', lambda test: config.LocationStore())
85
test_store_builder_registry.register(
86
'branch', lambda test: config.BranchStore(test.branch))
88
# FIXME: Same remark as above for the following registry -- vila 20110503
89
test_stack_builder_registry = registry.Registry()
90
test_stack_builder_registry.register(
82
def build_backing_branch(test, relpath,
83
transport_class=None, server_class=None):
84
"""Test helper to create a backing branch only once.
86
Some tests needs multiple stores/stacks to check concurrent update
87
behaviours. As such, they need to build different branch *objects* even if
88
they share the branch on disk.
90
:param relpath: The relative path to the branch. (Note that the helper
91
should always specify the same relpath).
93
:param transport_class: The Transport class the test needs to use.
95
:param server_class: The server associated with the ``transport_class``
98
Either both or neither of ``transport_class`` and ``server_class`` should
101
if transport_class is not None and server_class is not None:
102
test.transport_class = transport_class
103
test.transport_server = server_class
104
elif not (transport_class is None and server_class is None):
105
raise AssertionError('Specify both ``transport_class`` and '
106
'``server_class`` or neither of them')
107
if getattr(test, 'backing_branch', None) is None:
108
# First call, let's build the branch on disk
109
test.backing_branch = test.make_branch(relpath)
112
def build_branch_store(test):
113
build_backing_branch(test, 'branch')
114
b = branch.Branch.open('branch')
115
return config.BranchStore(b)
116
config.test_store_builder_registry.register('branch', build_branch_store)
119
def build_remote_branch_store(test):
120
# There is only one permutation (but we won't be able to handle more with
121
# this design anyway)
123
server_class) = transport_remote.get_test_permutations()[0]
124
build_backing_branch(test, 'branch', transport_class, server_class)
125
b = branch.Branch.open(test.get_url('branch'))
126
return config.BranchStore(b)
127
config.test_store_builder_registry.register('remote_branch',
128
build_remote_branch_store)
131
config.test_stack_builder_registry.register(
91
132
'bazaar', lambda test: config.GlobalStack())
92
test_stack_builder_registry.register(
133
config.test_stack_builder_registry.register(
93
134
'location', lambda test: config.LocationStack('.'))
94
test_stack_builder_registry.register(
95
'branch', lambda test: config.BranchStack(test.branch))
137
def build_branch_stack(test):
138
build_backing_branch(test, 'branch')
139
b = branch.Branch.open('branch')
140
return config.BranchStack(b)
141
config.test_stack_builder_registry.register('branch', build_branch_stack)
144
def build_remote_branch_stack(test):
145
# There is only one permutation (but we won't be able to handle more with
146
# this design anyway)
148
server_class) = transport_remote.get_test_permutations()[0]
149
build_backing_branch(test, 'branch', transport_class, server_class)
150
b = branch.Branch.open(test.get_url('branch'))
151
return config.BranchStack(b)
152
config.test_stack_builder_registry.register('remote_branch',
153
build_remote_branch_stack)
98
156
sample_long_alias="log -r-15..-1 --line"
1846
1966
self.assertIs(None, bzrdir_config.get_default_stack_on())
1969
class TestOldConfigHooks(tests.TestCaseWithTransport):
1972
super(TestOldConfigHooks, self).setUp()
1973
create_configs_with_file_option(self)
1975
def assertGetHook(self, conf, name, value):
1979
config.OldConfigHooks.install_named_hook('get', hook, None)
1981
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1982
self.assertLength(0, calls)
1983
actual_value = conf.get_user_option(name)
1984
self.assertEquals(value, actual_value)
1985
self.assertLength(1, calls)
1986
self.assertEquals((conf, name, value), calls[0])
1988
def test_get_hook_bazaar(self):
1989
self.assertGetHook(self.bazaar_config, 'file', 'bazaar')
1991
def test_get_hook_locations(self):
1992
self.assertGetHook(self.locations_config, 'file', 'locations')
1994
def test_get_hook_branch(self):
1995
# Since locations masks branch, we define a different option
1996
self.branch_config.set_user_option('file2', 'branch')
1997
self.assertGetHook(self.branch_config, 'file2', 'branch')
1999
def assertSetHook(self, conf, name, value):
2003
config.OldConfigHooks.install_named_hook('set', hook, None)
2005
config.OldConfigHooks.uninstall_named_hook, 'set', None)
2006
self.assertLength(0, calls)
2007
conf.set_user_option(name, value)
2008
self.assertLength(1, calls)
2009
# We can't assert the conf object below as different configs use
2010
# different means to implement set_user_option and we care only about
2012
self.assertEquals((name, value), calls[0][1:])
2014
def test_set_hook_bazaar(self):
2015
self.assertSetHook(self.bazaar_config, 'foo', 'bazaar')
2017
def test_set_hook_locations(self):
2018
self.assertSetHook(self.locations_config, 'foo', 'locations')
2020
def test_set_hook_branch(self):
2021
self.assertSetHook(self.branch_config, 'foo', 'branch')
2023
def assertRemoveHook(self, conf, name, section_name=None):
2027
config.OldConfigHooks.install_named_hook('remove', hook, None)
2029
config.OldConfigHooks.uninstall_named_hook, 'remove', None)
2030
self.assertLength(0, calls)
2031
conf.remove_user_option(name, section_name)
2032
self.assertLength(1, calls)
2033
# We can't assert the conf object below as different configs use
2034
# different means to implement remove_user_option and we care only about
2036
self.assertEquals((name,), calls[0][1:])
2038
def test_remove_hook_bazaar(self):
2039
self.assertRemoveHook(self.bazaar_config, 'file')
2041
def test_remove_hook_locations(self):
2042
self.assertRemoveHook(self.locations_config, 'file',
2043
self.locations_config.location)
2045
def test_remove_hook_branch(self):
2046
self.assertRemoveHook(self.branch_config, 'file')
2048
def assertLoadHook(self, name, conf_class, *conf_args):
2052
config.OldConfigHooks.install_named_hook('load', hook, None)
2054
config.OldConfigHooks.uninstall_named_hook, 'load', None)
2055
self.assertLength(0, calls)
2057
conf = conf_class(*conf_args)
2058
# Access an option to trigger a load
2059
conf.get_user_option(name)
2060
self.assertLength(1, calls)
2061
# Since we can't assert about conf, we just use the number of calls ;-/
2063
def test_load_hook_bazaar(self):
2064
self.assertLoadHook('file', config.GlobalConfig)
2066
def test_load_hook_locations(self):
2067
self.assertLoadHook('file', config.LocationConfig, self.tree.basedir)
2069
def test_load_hook_branch(self):
2070
self.assertLoadHook('file', config.BranchConfig, self.tree.branch)
2072
def assertSaveHook(self, conf):
2076
config.OldConfigHooks.install_named_hook('save', hook, None)
2078
config.OldConfigHooks.uninstall_named_hook, 'save', None)
2079
self.assertLength(0, calls)
2080
# Setting an option triggers a save
2081
conf.set_user_option('foo', 'bar')
2082
self.assertLength(1, calls)
2083
# Since we can't assert about conf, we just use the number of calls ;-/
2085
def test_save_hook_bazaar(self):
2086
self.assertSaveHook(self.bazaar_config)
2088
def test_save_hook_locations(self):
2089
self.assertSaveHook(self.locations_config)
2091
def test_save_hook_branch(self):
2092
self.assertSaveHook(self.branch_config)
2095
class TestOldConfigHooksForRemote(tests.TestCaseWithTransport):
2096
"""Tests config hooks for remote configs.
2098
No tests for the remove hook as this is not implemented there.
2102
super(TestOldConfigHooksForRemote, self).setUp()
2103
self.transport_server = test_server.SmartTCPServer_for_testing
2104
create_configs_with_file_option(self)
2106
def assertGetHook(self, conf, name, value):
2110
config.OldConfigHooks.install_named_hook('get', hook, None)
2112
config.OldConfigHooks.uninstall_named_hook, 'get', None)
2113
self.assertLength(0, calls)
2114
actual_value = conf.get_option(name)
2115
self.assertEquals(value, actual_value)
2116
self.assertLength(1, calls)
2117
self.assertEquals((conf, name, value), calls[0])
2119
def test_get_hook_remote_branch(self):
2120
remote_branch = branch.Branch.open(self.get_url('tree'))
2121
self.assertGetHook(remote_branch._get_config(), 'file', 'branch')
2123
def test_get_hook_remote_bzrdir(self):
2124
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2125
conf = remote_bzrdir._get_config()
2126
conf.set_option('remotedir', 'file')
2127
self.assertGetHook(conf, 'file', 'remotedir')
2129
def assertSetHook(self, conf, name, value):
2133
config.OldConfigHooks.install_named_hook('set', hook, None)
2135
config.OldConfigHooks.uninstall_named_hook, 'set', None)
2136
self.assertLength(0, calls)
2137
conf.set_option(value, name)
2138
self.assertLength(1, calls)
2139
# We can't assert the conf object below as different configs use
2140
# different means to implement set_user_option and we care only about
2142
self.assertEquals((name, value), calls[0][1:])
2144
def test_set_hook_remote_branch(self):
2145
remote_branch = branch.Branch.open(self.get_url('tree'))
2146
self.addCleanup(remote_branch.lock_write().unlock)
2147
self.assertSetHook(remote_branch._get_config(), 'file', 'remote')
2149
def test_set_hook_remote_bzrdir(self):
2150
remote_branch = branch.Branch.open(self.get_url('tree'))
2151
self.addCleanup(remote_branch.lock_write().unlock)
2152
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2153
self.assertSetHook(remote_bzrdir._get_config(), 'file', 'remotedir')
2155
def assertLoadHook(self, expected_nb_calls, name, conf_class, *conf_args):
2159
config.OldConfigHooks.install_named_hook('load', hook, None)
2161
config.OldConfigHooks.uninstall_named_hook, 'load', None)
2162
self.assertLength(0, calls)
2164
conf = conf_class(*conf_args)
2165
# Access an option to trigger a load
2166
conf.get_option(name)
2167
self.assertLength(expected_nb_calls, calls)
2168
# Since we can't assert about conf, we just use the number of calls ;-/
2170
def test_load_hook_remote_branch(self):
2171
remote_branch = branch.Branch.open(self.get_url('tree'))
2172
self.assertLoadHook(1, 'file', remote.RemoteBranchConfig, remote_branch)
2174
def test_load_hook_remote_bzrdir(self):
2175
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2176
# The config file doesn't exist, set an option to force its creation
2177
conf = remote_bzrdir._get_config()
2178
conf.set_option('remotedir', 'file')
2179
# We get one call for the server and one call for the client, this is
2180
# caused by the differences in implementations betwen
2181
# SmartServerBzrDirRequestConfigFile (in smart/bzrdir.py) and
2182
# SmartServerBranchGetConfigFile (in smart/branch.py)
2183
self.assertLoadHook(2 ,'file', remote.RemoteBzrDirConfig, remote_bzrdir)
2185
def assertSaveHook(self, conf):
2189
config.OldConfigHooks.install_named_hook('save', hook, None)
2191
config.OldConfigHooks.uninstall_named_hook, 'save', None)
2192
self.assertLength(0, calls)
2193
# Setting an option triggers a save
2194
conf.set_option('foo', 'bar')
2195
self.assertLength(1, calls)
2196
# Since we can't assert about conf, we just use the number of calls ;-/
2198
def test_save_hook_remote_branch(self):
2199
remote_branch = branch.Branch.open(self.get_url('tree'))
2200
self.addCleanup(remote_branch.lock_write().unlock)
2201
self.assertSaveHook(remote_branch._get_config())
2203
def test_save_hook_remote_bzrdir(self):
2204
remote_branch = branch.Branch.open(self.get_url('tree'))
2205
self.addCleanup(remote_branch.lock_write().unlock)
2206
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2207
self.assertSaveHook(remote_bzrdir._get_config())
2210
class TestOption(tests.TestCase):
2212
def test_default_value(self):
2213
opt = config.Option('foo', default='bar')
2214
self.assertEquals('bar', opt.get_default())
2217
class TestOptionRegistry(tests.TestCase):
2220
super(TestOptionRegistry, self).setUp()
2221
# Always start with an empty registry
2222
self.overrideAttr(config, 'option_registry', registry.Registry())
2223
self.registry = config.option_registry
2225
def test_register(self):
2226
opt = config.Option('foo')
2227
self.registry.register('foo', opt)
2228
self.assertIs(opt, self.registry.get('foo'))
2230
lazy_option = config.Option('lazy_foo')
2232
def test_register_lazy(self):
2233
self.registry.register_lazy('foo', self.__module__,
2234
'TestOptionRegistry.lazy_option')
2235
self.assertIs(self.lazy_option, self.registry.get('foo'))
2237
def test_registered_help(self):
2238
opt = config.Option('foo')
2239
self.registry.register('foo', opt, help='A simple option')
2240
self.assertEquals('A simple option', self.registry.get_help('foo'))
2243
class TestRegisteredOptions(tests.TestCase):
2244
"""All registered options should verify some constraints."""
2246
scenarios = [(key, {'option_name': key, 'option': option}) for key, option
2247
in config.option_registry.iteritems()]
2250
super(TestRegisteredOptions, self).setUp()
2251
self.registry = config.option_registry
2253
def test_proper_name(self):
2254
# An option should be registered under its own name, this can't be
2255
# checked at registration time for the lazy ones.
2256
self.assertEquals(self.option_name, self.option.name)
2258
def test_help_is_set(self):
2259
option_help = self.registry.get_help(self.option_name)
2260
self.assertNotEquals(None, option_help)
2261
# Come on, think about the user, he really wants to know whst the
2263
self.assertNotEquals('', option_help)
1849
2266
class TestSection(tests.TestCase):
1851
2268
# FIXME: Parametrize so that all sections produced by Stores run these
1969
2385
self.assertRaises(AssertionError, store._load_from_string, 'bar=baz')
2388
class TestIniFileStoreContent(tests.TestCaseWithTransport):
2389
"""Simulate loading a config store without content of various encodings.
2391
All files produced by bzr are in utf8 content.
2393
Users may modify them manually and end up with a file that can't be
2394
loaded. We need to issue proper error messages in this case.
2397
invalid_utf8_char = '\xff'
2399
def test_load_utf8(self):
2400
"""Ensure we can load an utf8-encoded file."""
2401
t = self.get_transport()
2402
# From http://pad.lv/799212
2403
unicode_user = u'b\N{Euro Sign}ar'
2404
unicode_content = u'user=%s' % (unicode_user,)
2405
utf8_content = unicode_content.encode('utf8')
2406
# Store the raw content in the config file
2407
t.put_bytes('foo.conf', utf8_content)
2408
store = config.IniFileStore(t, 'foo.conf')
2410
stack = config.Stack([store.get_sections], store)
2411
self.assertEquals(unicode_user, stack.get('user'))
2413
def test_load_non_ascii(self):
2414
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2415
t = self.get_transport()
2416
t.put_bytes('foo.conf', 'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2417
store = config.IniFileStore(t, 'foo.conf')
2418
self.assertRaises(errors.ConfigContentError, store.load)
2420
def test_load_erroneous_content(self):
2421
"""Ensure we display a proper error on content that can't be parsed."""
2422
t = self.get_transport()
2423
t.put_bytes('foo.conf', '[open_section\n')
2424
store = config.IniFileStore(t, 'foo.conf')
2425
self.assertRaises(errors.ParseConfigError, store.load)
2428
class TestIniConfigContent(tests.TestCaseWithTransport):
2429
"""Simulate loading a IniBasedConfig without content of various encodings.
2431
All files produced by bzr are in utf8 content.
2433
Users may modify them manually and end up with a file that can't be
2434
loaded. We need to issue proper error messages in this case.
2437
invalid_utf8_char = '\xff'
2439
def test_load_utf8(self):
2440
"""Ensure we can load an utf8-encoded file."""
2441
# From http://pad.lv/799212
2442
unicode_user = u'b\N{Euro Sign}ar'
2443
unicode_content = u'user=%s' % (unicode_user,)
2444
utf8_content = unicode_content.encode('utf8')
2445
# Store the raw content in the config file
2446
with open('foo.conf', 'wb') as f:
2447
f.write(utf8_content)
2448
conf = config.IniBasedConfig(file_name='foo.conf')
2449
self.assertEquals(unicode_user, conf.get_user_option('user'))
2451
def test_load_badly_encoded_content(self):
2452
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2453
with open('foo.conf', 'wb') as f:
2454
f.write('user=foo\n#%s\n' % (self.invalid_utf8_char,))
2455
conf = config.IniBasedConfig(file_name='foo.conf')
2456
self.assertRaises(errors.ConfigContentError, conf._get_parser)
2458
def test_load_erroneous_content(self):
2459
"""Ensure we display a proper error on content that can't be parsed."""
2460
with open('foo.conf', 'wb') as f:
2461
f.write('[open_section\n')
2462
conf = config.IniBasedConfig(file_name='foo.conf')
2463
self.assertRaises(errors.ParseConfigError, conf._get_parser)
1972
2466
class TestMutableStore(TestStore):
1974
scenarios = [(key, {'store_id': key, 'get_store': builder})
1975
for key, builder in test_store_builder_registry.iteritems()]
2468
scenarios = [(key, {'store_id': key, 'get_store': builder}) for key, builder
2469
in config.test_store_builder_registry.iteritems()]
1977
2471
def setUp(self):
1978
2472
super(TestMutableStore, self).setUp()
1979
2473
self.transport = self.get_transport()
1980
self.branch = self.make_branch('branch')
1982
2475
def has_store(self, store):
1983
2476
store_basename = urlutils.relative_url(self.transport.external_url(),
2103
2630
class TestLockableIniFileStore(TestStore):
2105
2632
def test_create_store_in_created_dir(self):
2633
self.assertPathDoesNotExist('dir')
2106
2634
t = self.get_transport('dir/subdir')
2107
2635
store = config.LockableIniFileStore(t, 'foo.conf')
2108
2636
store.get_mutable_section(None).set('foo', 'bar')
2111
# FIXME: We should adapt the tests in TestLockableConfig about concurrent
2112
# writes. Since this requires a clearer rewrite, I'll just rely on using
2113
# the same code in LockableIniFileStore (copied from LockableConfig, but
2114
# trivial enough, the main difference is that we add @needs_write_lock on
2115
# save() instead of set_user_option() and remove_user_option()). The intent
2116
# is to ensure that we always get a valid content for the store even when
2117
# concurrent accesses occur, read/write, write/write. It may be worth
2118
# looking into removing the lock dir when it;s not needed anymore and look
2119
# at possible fallouts for concurrent lockers -- vila 20110-04-06
2638
self.assertPathExists('dir/subdir')
2641
class TestConcurrentStoreUpdates(TestStore):
2642
"""Test that Stores properly handle conccurent updates.
2644
New Store implementation may fail some of these tests but until such
2645
implementations exist it's hard to properly filter them from the scenarios
2646
applied here. If you encounter such a case, contact the bzr devs.
2649
scenarios = [(key, {'get_stack': builder}) for key, builder
2650
in config.test_stack_builder_registry.iteritems()]
2653
super(TestConcurrentStoreUpdates, self).setUp()
2654
self._content = 'one=1\ntwo=2\n'
2655
self.stack = self.get_stack(self)
2656
if not isinstance(self.stack, config._CompatibleStack):
2657
raise tests.TestNotApplicable(
2658
'%s is not meant to be compatible with the old config design'
2660
self.stack.store._load_from_string(self._content)
2662
self.stack.store.save()
2664
def test_simple_read_access(self):
2665
self.assertEquals('1', self.stack.get('one'))
2667
def test_simple_write_access(self):
2668
self.stack.set('one', 'one')
2669
self.assertEquals('one', self.stack.get('one'))
2671
def test_listen_to_the_last_speaker(self):
2673
c2 = self.get_stack(self)
2674
c1.set('one', 'ONE')
2675
c2.set('two', 'TWO')
2676
self.assertEquals('ONE', c1.get('one'))
2677
self.assertEquals('TWO', c2.get('two'))
2678
# The second update respect the first one
2679
self.assertEquals('ONE', c2.get('one'))
2681
def test_last_speaker_wins(self):
2682
# If the same config is not shared, the same variable modified twice
2683
# can only see a single result.
2685
c2 = self.get_stack(self)
2688
self.assertEquals('c2', c2.get('one'))
2689
# The first modification is still available until another refresh
2691
self.assertEquals('c1', c1.get('one'))
2692
c1.set('two', 'done')
2693
self.assertEquals('c2', c1.get('one'))
2695
def test_writes_are_serialized(self):
2697
c2 = self.get_stack(self)
2699
# We spawn a thread that will pause *during* the config saving.
2700
before_writing = threading.Event()
2701
after_writing = threading.Event()
2702
writing_done = threading.Event()
2703
c1_save_without_locking_orig = c1.store.save_without_locking
2704
def c1_save_without_locking():
2705
before_writing.set()
2706
c1_save_without_locking_orig()
2707
# The lock is held. We wait for the main thread to decide when to
2709
after_writing.wait()
2710
c1.store.save_without_locking = c1_save_without_locking
2714
t1 = threading.Thread(target=c1_set)
2715
# Collect the thread after the test
2716
self.addCleanup(t1.join)
2717
# Be ready to unblock the thread if the test goes wrong
2718
self.addCleanup(after_writing.set)
2720
before_writing.wait()
2721
self.assertRaises(errors.LockContention,
2722
c2.set, 'one', 'c2')
2723
self.assertEquals('c1', c1.get('one'))
2724
# Let the lock be released
2728
self.assertEquals('c2', c2.get('one'))
2730
def test_read_while_writing(self):
2732
# We spawn a thread that will pause *during* the write
2733
ready_to_write = threading.Event()
2734
do_writing = threading.Event()
2735
writing_done = threading.Event()
2736
# We override the _save implementation so we know the store is locked
2737
c1_save_without_locking_orig = c1.store.save_without_locking
2738
def c1_save_without_locking():
2739
ready_to_write.set()
2740
# The lock is held. We wait for the main thread to decide when to
2743
c1_save_without_locking_orig()
2745
c1.store.save_without_locking = c1_save_without_locking
2748
t1 = threading.Thread(target=c1_set)
2749
# Collect the thread after the test
2750
self.addCleanup(t1.join)
2751
# Be ready to unblock the thread if the test goes wrong
2752
self.addCleanup(do_writing.set)
2754
# Ensure the thread is ready to write
2755
ready_to_write.wait()
2756
self.assertEquals('c1', c1.get('one'))
2757
# If we read during the write, we get the old value
2758
c2 = self.get_stack(self)
2759
self.assertEquals('1', c2.get('one'))
2760
# Let the writing occur and ensure it occurred
2763
# Now we get the updated value
2764
c3 = self.get_stack(self)
2765
self.assertEquals('c1', c3.get('one'))
2767
# FIXME: It may be worth looking into removing the lock dir when it's not
2768
# needed anymore and look at possible fallouts for concurrent lockers. This
2769
# will matter if/when we use config files outside of bazaar directories
2770
# (.bazaar or .bzr) -- vila 20110-04-11
2122
2773
class TestSectionMatcher(TestStore):
added
removed
bzrlib/tests/test_config.py
