70
64
load_tests = scenarios.load_tests_apply_scenarios
72
# Register helpers to build stores
73
config.test_store_builder_registry.register(
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(
74
79
'configobj', lambda test: config.IniFileStore(test.get_transport(),
76
config.test_store_builder_registry.register(
81
test_store_builder_registry.register(
77
82
'bazaar', lambda test: config.GlobalStore())
78
config.test_store_builder_registry.register(
83
test_store_builder_registry.register(
79
84
'location', lambda test: config.LocationStore())
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(
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(
132
91
'bazaar', lambda test: config.GlobalStack())
133
config.test_stack_builder_registry.register(
92
test_stack_builder_registry.register(
134
93
'location', lambda test: config.LocationStack('.'))
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)
94
test_stack_builder_registry.register(
95
'branch', lambda test: config.BranchStack(test.branch))
156
98
sample_long_alias="log -r-15..-1 --line"
1952
1846
self.assertIs(None, bzrdir_config.get_default_stack_on())
1955
class TestOldConfigHooks(tests.TestCaseWithTransport):
1958
super(TestOldConfigHooks, self).setUp()
1959
create_configs_with_file_option(self)
1961
def assertGetHook(self, conf, name, value):
1965
config.OldConfigHooks.install_named_hook('get', hook, None)
1967
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1968
self.assertLength(0, calls)
1969
actual_value = conf.get_user_option(name)
1970
self.assertEquals(value, actual_value)
1971
self.assertLength(1, calls)
1972
self.assertEquals((conf, name, value), calls[0])
1974
def test_get_hook_bazaar(self):
1975
self.assertGetHook(self.bazaar_config, 'file', 'bazaar')
1977
def test_get_hook_locations(self):
1978
self.assertGetHook(self.locations_config, 'file', 'locations')
1980
def test_get_hook_branch(self):
1981
# Since locations masks branch, we define a different option
1982
self.branch_config.set_user_option('file2', 'branch')
1983
self.assertGetHook(self.branch_config, 'file2', 'branch')
1985
def assertSetHook(self, conf, name, value):
1989
config.OldConfigHooks.install_named_hook('set', hook, None)
1991
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1992
self.assertLength(0, calls)
1993
conf.set_user_option(name, value)
1994
self.assertLength(1, calls)
1995
# We can't assert the conf object below as different configs use
1996
# different means to implement set_user_option and we care only about
1998
self.assertEquals((name, value), calls[0][1:])
2000
def test_set_hook_bazaar(self):
2001
self.assertSetHook(self.bazaar_config, 'foo', 'bazaar')
2003
def test_set_hook_locations(self):
2004
self.assertSetHook(self.locations_config, 'foo', 'locations')
2006
def test_set_hook_branch(self):
2007
self.assertSetHook(self.branch_config, 'foo', 'branch')
2009
def assertRemoveHook(self, conf, name, section_name=None):
2013
config.OldConfigHooks.install_named_hook('remove', hook, None)
2015
config.OldConfigHooks.uninstall_named_hook, 'remove', None)
2016
self.assertLength(0, calls)
2017
conf.remove_user_option(name, section_name)
2018
self.assertLength(1, calls)
2019
# We can't assert the conf object below as different configs use
2020
# different means to implement remove_user_option and we care only about
2022
self.assertEquals((name,), calls[0][1:])
2024
def test_remove_hook_bazaar(self):
2025
self.assertRemoveHook(self.bazaar_config, 'file')
2027
def test_remove_hook_locations(self):
2028
self.assertRemoveHook(self.locations_config, 'file',
2029
self.locations_config.location)
2031
def test_remove_hook_branch(self):
2032
self.assertRemoveHook(self.branch_config, 'file')
2034
def assertLoadHook(self, name, conf_class, *conf_args):
2038
config.OldConfigHooks.install_named_hook('load', hook, None)
2040
config.OldConfigHooks.uninstall_named_hook, 'load', None)
2041
self.assertLength(0, calls)
2043
conf = conf_class(*conf_args)
2044
# Access an option to trigger a load
2045
conf.get_user_option(name)
2046
self.assertLength(1, calls)
2047
# Since we can't assert about conf, we just use the number of calls ;-/
2049
def test_load_hook_bazaar(self):
2050
self.assertLoadHook('file', config.GlobalConfig)
2052
def test_load_hook_locations(self):
2053
self.assertLoadHook('file', config.LocationConfig, self.tree.basedir)
2055
def test_load_hook_branch(self):
2056
self.assertLoadHook('file', config.BranchConfig, self.tree.branch)
2058
def assertSaveHook(self, conf):
2062
config.OldConfigHooks.install_named_hook('save', hook, None)
2064
config.OldConfigHooks.uninstall_named_hook, 'save', None)
2065
self.assertLength(0, calls)
2066
# Setting an option triggers a save
2067
conf.set_user_option('foo', 'bar')
2068
self.assertLength(1, calls)
2069
# Since we can't assert about conf, we just use the number of calls ;-/
2071
def test_save_hook_bazaar(self):
2072
self.assertSaveHook(self.bazaar_config)
2074
def test_save_hook_locations(self):
2075
self.assertSaveHook(self.locations_config)
2077
def test_save_hook_branch(self):
2078
self.assertSaveHook(self.branch_config)
2081
class TestOldConfigHooksForRemote(tests.TestCaseWithTransport):
2082
"""Tests config hooks for remote configs.
2084
No tests for the remove hook as this is not implemented there.
2088
super(TestOldConfigHooksForRemote, self).setUp()
2089
self.transport_server = test_server.SmartTCPServer_for_testing
2090
create_configs_with_file_option(self)
2092
def assertGetHook(self, conf, name, value):
2096
config.OldConfigHooks.install_named_hook('get', hook, None)
2098
config.OldConfigHooks.uninstall_named_hook, 'get', None)
2099
self.assertLength(0, calls)
2100
actual_value = conf.get_option(name)
2101
self.assertEquals(value, actual_value)
2102
self.assertLength(1, calls)
2103
self.assertEquals((conf, name, value), calls[0])
2105
def test_get_hook_remote_branch(self):
2106
remote_branch = branch.Branch.open(self.get_url('tree'))
2107
self.assertGetHook(remote_branch._get_config(), 'file', 'branch')
2109
def test_get_hook_remote_bzrdir(self):
2110
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2111
conf = remote_bzrdir._get_config()
2112
conf.set_option('remotedir', 'file')
2113
self.assertGetHook(conf, 'file', 'remotedir')
2115
def assertSetHook(self, conf, name, value):
2119
config.OldConfigHooks.install_named_hook('set', hook, None)
2121
config.OldConfigHooks.uninstall_named_hook, 'set', None)
2122
self.assertLength(0, calls)
2123
conf.set_option(value, name)
2124
self.assertLength(1, calls)
2125
# We can't assert the conf object below as different configs use
2126
# different means to implement set_user_option and we care only about
2128
self.assertEquals((name, value), calls[0][1:])
2130
def test_set_hook_remote_branch(self):
2131
remote_branch = branch.Branch.open(self.get_url('tree'))
2132
self.addCleanup(remote_branch.lock_write().unlock)
2133
self.assertSetHook(remote_branch._get_config(), 'file', 'remote')
2135
def test_set_hook_remote_bzrdir(self):
2136
remote_branch = branch.Branch.open(self.get_url('tree'))
2137
self.addCleanup(remote_branch.lock_write().unlock)
2138
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2139
self.assertSetHook(remote_bzrdir._get_config(), 'file', 'remotedir')
2141
def assertLoadHook(self, expected_nb_calls, name, conf_class, *conf_args):
2145
config.OldConfigHooks.install_named_hook('load', hook, None)
2147
config.OldConfigHooks.uninstall_named_hook, 'load', None)
2148
self.assertLength(0, calls)
2150
conf = conf_class(*conf_args)
2151
# Access an option to trigger a load
2152
conf.get_option(name)
2153
self.assertLength(expected_nb_calls, calls)
2154
# Since we can't assert about conf, we just use the number of calls ;-/
2156
def test_load_hook_remote_branch(self):
2157
remote_branch = branch.Branch.open(self.get_url('tree'))
2158
self.assertLoadHook(1, 'file', remote.RemoteBranchConfig, remote_branch)
2160
def test_load_hook_remote_bzrdir(self):
2161
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2162
# The config file doesn't exist, set an option to force its creation
2163
conf = remote_bzrdir._get_config()
2164
conf.set_option('remotedir', 'file')
2165
# We get one call for the server and one call for the client, this is
2166
# caused by the differences in implementations betwen
2167
# SmartServerBzrDirRequestConfigFile (in smart/bzrdir.py) and
2168
# SmartServerBranchGetConfigFile (in smart/branch.py)
2169
self.assertLoadHook(2 ,'file', remote.RemoteBzrDirConfig, remote_bzrdir)
2171
def assertSaveHook(self, conf):
2175
config.OldConfigHooks.install_named_hook('save', hook, None)
2177
config.OldConfigHooks.uninstall_named_hook, 'save', None)
2178
self.assertLength(0, calls)
2179
# Setting an option triggers a save
2180
conf.set_option('foo', 'bar')
2181
self.assertLength(1, calls)
2182
# Since we can't assert about conf, we just use the number of calls ;-/
2184
def test_save_hook_remote_branch(self):
2185
remote_branch = branch.Branch.open(self.get_url('tree'))
2186
self.addCleanup(remote_branch.lock_write().unlock)
2187
self.assertSaveHook(remote_branch._get_config())
2189
def test_save_hook_remote_bzrdir(self):
2190
remote_branch = branch.Branch.open(self.get_url('tree'))
2191
self.addCleanup(remote_branch.lock_write().unlock)
2192
remote_bzrdir = bzrdir.BzrDir.open(self.get_url('tree'))
2193
self.assertSaveHook(remote_bzrdir._get_config())
2196
class TestOption(tests.TestCase):
2198
def test_default_value(self):
2199
opt = config.Option('foo', default='bar')
2200
self.assertEquals('bar', opt.get_default())
2203
class TestOptionRegistry(tests.TestCase):
2206
super(TestOptionRegistry, self).setUp()
2207
# Always start with an empty registry
2208
self.overrideAttr(config, 'option_registry', registry.Registry())
2209
self.registry = config.option_registry
2211
def test_register(self):
2212
opt = config.Option('foo')
2213
self.registry.register('foo', opt)
2214
self.assertIs(opt, self.registry.get('foo'))
2216
lazy_option = config.Option('lazy_foo')
2218
def test_register_lazy(self):
2219
self.registry.register_lazy('foo', self.__module__,
2220
'TestOptionRegistry.lazy_option')
2221
self.assertIs(self.lazy_option, self.registry.get('foo'))
2223
def test_registered_help(self):
2224
opt = config.Option('foo')
2225
self.registry.register('foo', opt, help='A simple option')
2226
self.assertEquals('A simple option', self.registry.get_help('foo'))
2229
class TestRegisteredOptions(tests.TestCase):
2230
"""All registered options should verify some constraints."""
2232
scenarios = [(key, {'option_name': key, 'option': option}) for key, option
2233
in config.option_registry.iteritems()]
2236
super(TestRegisteredOptions, self).setUp()
2237
self.registry = config.option_registry
2239
def test_proper_name(self):
2240
# An option should be registered under its own name, this can't be
2241
# checked at registration time for the lazy ones.
2242
self.assertEquals(self.option_name, self.option.name)
2244
def test_help_is_set(self):
2245
option_help = self.registry.get_help(self.option_name)
2246
self.assertNotEquals(None, option_help)
2247
# Come on, think about the user, he really wants to know whst the
2249
self.assertNotEquals('', option_help)
2252
1849
class TestSection(tests.TestCase):
2254
1851
# FIXME: Parametrize so that all sections produced by Stores run these
2371
1969
self.assertRaises(AssertionError, store._load_from_string, 'bar=baz')
2374
class TestIniFileStoreContent(tests.TestCaseWithTransport):
2375
"""Simulate loading a config store without content of various encodings.
2377
All files produced by bzr are in utf8 content.
2379
Users may modify them manually and end up with a file that can't be
2380
loaded. We need to issue proper error messages in this case.
2383
invalid_utf8_char = '\xff'
2385
def test_load_utf8(self):
2386
"""Ensure we can load an utf8-encoded file."""
2387
t = self.get_transport()
2388
# From http://pad.lv/799212
2389
unicode_user = u'b\N{Euro Sign}ar'
2390
unicode_content = u'user=%s' % (unicode_user,)
2391
utf8_content = unicode_content.encode('utf8')
2392
# Store the raw content in the config file
2393
t.put_bytes('foo.conf', utf8_content)
2394
store = config.IniFileStore(t, 'foo.conf')
2396
stack = config.Stack([store.get_sections], store)
2397
self.assertEquals(unicode_user, stack.get('user'))
2399
def test_load_non_ascii(self):
2400
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2401
t = self.get_transport()
2402
t.put_bytes('foo.conf', 'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2403
store = config.IniFileStore(t, 'foo.conf')
2404
self.assertRaises(errors.ConfigContentError, store.load)
2406
def test_load_erroneous_content(self):
2407
"""Ensure we display a proper error on content that can't be parsed."""
2408
t = self.get_transport()
2409
t.put_bytes('foo.conf', '[open_section\n')
2410
store = config.IniFileStore(t, 'foo.conf')
2411
self.assertRaises(errors.ParseConfigError, store.load)
2414
class TestIniConfigContent(tests.TestCaseWithTransport):
2415
"""Simulate loading a IniBasedConfig without content of various encodings.
2417
All files produced by bzr are in utf8 content.
2419
Users may modify them manually and end up with a file that can't be
2420
loaded. We need to issue proper error messages in this case.
2423
invalid_utf8_char = '\xff'
2425
def test_load_utf8(self):
2426
"""Ensure we can load an utf8-encoded file."""
2427
# From http://pad.lv/799212
2428
unicode_user = u'b\N{Euro Sign}ar'
2429
unicode_content = u'user=%s' % (unicode_user,)
2430
utf8_content = unicode_content.encode('utf8')
2431
# Store the raw content in the config file
2432
with open('foo.conf', 'wb') as f:
2433
f.write(utf8_content)
2434
conf = config.IniBasedConfig(file_name='foo.conf')
2435
self.assertEquals(unicode_user, conf.get_user_option('user'))
2437
def test_load_badly_encoded_content(self):
2438
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2439
with open('foo.conf', 'wb') as f:
2440
f.write('user=foo\n#%s\n' % (self.invalid_utf8_char,))
2441
conf = config.IniBasedConfig(file_name='foo.conf')
2442
self.assertRaises(errors.ConfigContentError, conf._get_parser)
2444
def test_load_erroneous_content(self):
2445
"""Ensure we display a proper error on content that can't be parsed."""
2446
with open('foo.conf', 'wb') as f:
2447
f.write('[open_section\n')
2448
conf = config.IniBasedConfig(file_name='foo.conf')
2449
self.assertRaises(errors.ParseConfigError, conf._get_parser)
2452
1972
class TestMutableStore(TestStore):
2454
scenarios = [(key, {'store_id': key, 'get_store': builder}) for key, builder
2455
in config.test_store_builder_registry.iteritems()]
1974
scenarios = [(key, {'store_id': key, 'get_store': builder})
1975
for key, builder in test_store_builder_registry.iteritems()]
2457
1977
def setUp(self):
2458
1978
super(TestMutableStore, self).setUp()
2459
1979
self.transport = self.get_transport()
1980
self.branch = self.make_branch('branch')
2461
1982
def has_store(self, store):
2462
1983
store_basename = urlutils.relative_url(self.transport.external_url(),
2616
2103
class TestLockableIniFileStore(TestStore):
2618
2105
def test_create_store_in_created_dir(self):
2619
self.assertPathDoesNotExist('dir')
2620
2106
t = self.get_transport('dir/subdir')
2621
2107
store = config.LockableIniFileStore(t, 'foo.conf')
2622
2108
store.get_mutable_section(None).set('foo', 'bar')
2624
self.assertPathExists('dir/subdir')
2627
class TestConcurrentStoreUpdates(TestStore):
2628
"""Test that Stores properly handle conccurent updates.
2630
New Store implementation may fail some of these tests but until such
2631
implementations exist it's hard to properly filter them from the scenarios
2632
applied here. If you encounter such a case, contact the bzr devs.
2635
scenarios = [(key, {'get_stack': builder}) for key, builder
2636
in config.test_stack_builder_registry.iteritems()]
2639
super(TestConcurrentStoreUpdates, self).setUp()
2640
self._content = 'one=1\ntwo=2\n'
2641
self.stack = self.get_stack(self)
2642
if not isinstance(self.stack, config._CompatibleStack):
2643
raise tests.TestNotApplicable(
2644
'%s is not meant to be compatible with the old config design'
2646
self.stack.store._load_from_string(self._content)
2648
self.stack.store.save()
2650
def test_simple_read_access(self):
2651
self.assertEquals('1', self.stack.get('one'))
2653
def test_simple_write_access(self):
2654
self.stack.set('one', 'one')
2655
self.assertEquals('one', self.stack.get('one'))
2657
def test_listen_to_the_last_speaker(self):
2659
c2 = self.get_stack(self)
2660
c1.set('one', 'ONE')
2661
c2.set('two', 'TWO')
2662
self.assertEquals('ONE', c1.get('one'))
2663
self.assertEquals('TWO', c2.get('two'))
2664
# The second update respect the first one
2665
self.assertEquals('ONE', c2.get('one'))
2667
def test_last_speaker_wins(self):
2668
# If the same config is not shared, the same variable modified twice
2669
# can only see a single result.
2671
c2 = self.get_stack(self)
2674
self.assertEquals('c2', c2.get('one'))
2675
# The first modification is still available until another refresh
2677
self.assertEquals('c1', c1.get('one'))
2678
c1.set('two', 'done')
2679
self.assertEquals('c2', c1.get('one'))
2681
def test_writes_are_serialized(self):
2683
c2 = self.get_stack(self)
2685
# We spawn a thread that will pause *during* the config saving.
2686
before_writing = threading.Event()
2687
after_writing = threading.Event()
2688
writing_done = threading.Event()
2689
c1_save_without_locking_orig = c1.store.save_without_locking
2690
def c1_save_without_locking():
2691
before_writing.set()
2692
c1_save_without_locking_orig()
2693
# The lock is held. We wait for the main thread to decide when to
2695
after_writing.wait()
2696
c1.store.save_without_locking = c1_save_without_locking
2700
t1 = threading.Thread(target=c1_set)
2701
# Collect the thread after the test
2702
self.addCleanup(t1.join)
2703
# Be ready to unblock the thread if the test goes wrong
2704
self.addCleanup(after_writing.set)
2706
before_writing.wait()
2707
self.assertRaises(errors.LockContention,
2708
c2.set, 'one', 'c2')
2709
self.assertEquals('c1', c1.get('one'))
2710
# Let the lock be released
2714
self.assertEquals('c2', c2.get('one'))
2716
def test_read_while_writing(self):
2718
# We spawn a thread that will pause *during* the write
2719
ready_to_write = threading.Event()
2720
do_writing = threading.Event()
2721
writing_done = threading.Event()
2722
# We override the _save implementation so we know the store is locked
2723
c1_save_without_locking_orig = c1.store.save_without_locking
2724
def c1_save_without_locking():
2725
ready_to_write.set()
2726
# The lock is held. We wait for the main thread to decide when to
2729
c1_save_without_locking_orig()
2731
c1.store.save_without_locking = c1_save_without_locking
2734
t1 = threading.Thread(target=c1_set)
2735
# Collect the thread after the test
2736
self.addCleanup(t1.join)
2737
# Be ready to unblock the thread if the test goes wrong
2738
self.addCleanup(do_writing.set)
2740
# Ensure the thread is ready to write
2741
ready_to_write.wait()
2742
self.assertEquals('c1', c1.get('one'))
2743
# If we read during the write, we get the old value
2744
c2 = self.get_stack(self)
2745
self.assertEquals('1', c2.get('one'))
2746
# Let the writing occur and ensure it occurred
2749
# Now we get the updated value
2750
c3 = self.get_stack(self)
2751
self.assertEquals('c1', c3.get('one'))
2753
# FIXME: It may be worth looking into removing the lock dir when it's not
2754
# needed anymore and look at possible fallouts for concurrent lockers. This
2755
# will matter if/when we use config files outside of bazaar directories
2756
# (.bazaar or .bzr) -- vila 20110-04-11
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
2759
2122
class TestSectionMatcher(TestStore):
added
removed
bzrlib/tests/test_config.py
