1952
1559
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
class TestSection(tests.TestCase):
2254
# FIXME: Parametrize so that all sections produced by Stores run these
2255
# tests -- vila 2011-04-01
2257
def test_get_a_value(self):
2258
a_dict = dict(foo='bar')
2259
section = config.Section('myID', a_dict)
2260
self.assertEquals('bar', section.get('foo'))
2262
def test_get_unknown_option(self):
2264
section = config.Section(None, a_dict)
2265
self.assertEquals('out of thin air',
2266
section.get('foo', 'out of thin air'))
2268
def test_options_is_shared(self):
2270
section = config.Section(None, a_dict)
2271
self.assertIs(a_dict, section.options)
2274
class TestMutableSection(tests.TestCase):
2276
# FIXME: Parametrize so that all sections (including os.environ and the
2277
# ones produced by Stores) run these tests -- vila 2011-04-01
2280
a_dict = dict(foo='bar')
2281
section = config.MutableSection('myID', a_dict)
2282
section.set('foo', 'new_value')
2283
self.assertEquals('new_value', section.get('foo'))
2284
# The change appears in the shared section
2285
self.assertEquals('new_value', a_dict.get('foo'))
2286
# We keep track of the change
2287
self.assertTrue('foo' in section.orig)
2288
self.assertEquals('bar', section.orig.get('foo'))
2290
def test_set_preserve_original_once(self):
2291
a_dict = dict(foo='bar')
2292
section = config.MutableSection('myID', a_dict)
2293
section.set('foo', 'first_value')
2294
section.set('foo', 'second_value')
2295
# We keep track of the original value
2296
self.assertTrue('foo' in section.orig)
2297
self.assertEquals('bar', section.orig.get('foo'))
2299
def test_remove(self):
2300
a_dict = dict(foo='bar')
2301
section = config.MutableSection('myID', a_dict)
2302
section.remove('foo')
2303
# We get None for unknown options via the default value
2304
self.assertEquals(None, section.get('foo'))
2305
# Or we just get the default value
2306
self.assertEquals('unknown', section.get('foo', 'unknown'))
2307
self.assertFalse('foo' in section.options)
2308
# We keep track of the deletion
2309
self.assertTrue('foo' in section.orig)
2310
self.assertEquals('bar', section.orig.get('foo'))
2312
def test_remove_new_option(self):
2314
section = config.MutableSection('myID', a_dict)
2315
section.set('foo', 'bar')
2316
section.remove('foo')
2317
self.assertFalse('foo' in section.options)
2318
# The option didn't exist initially so it we need to keep track of it
2319
# with a special value
2320
self.assertTrue('foo' in section.orig)
2321
self.assertEquals(config._NewlyCreatedOption, section.orig['foo'])
2324
class TestStore(tests.TestCaseWithTransport):
2326
def assertSectionContent(self, expected, section):
2327
"""Assert that some options have the proper values in a section."""
2328
expected_name, expected_options = expected
2329
self.assertEquals(expected_name, section.id)
2332
dict([(k, section.get(k)) for k in expected_options.keys()]))
2335
class TestReadonlyStore(TestStore):
2337
scenarios = [(key, {'get_store': builder}) for key, builder
2338
in config.test_store_builder_registry.iteritems()]
2341
super(TestReadonlyStore, self).setUp()
2343
def test_building_delays_load(self):
2344
store = self.get_store(self)
2345
self.assertEquals(False, store.is_loaded())
2346
store._load_from_string('')
2347
self.assertEquals(True, store.is_loaded())
2349
def test_get_no_sections_for_empty(self):
2350
store = self.get_store(self)
2351
store._load_from_string('')
2352
self.assertEquals([], list(store.get_sections()))
2354
def test_get_default_section(self):
2355
store = self.get_store(self)
2356
store._load_from_string('foo=bar')
2357
sections = list(store.get_sections())
2358
self.assertLength(1, sections)
2359
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2361
def test_get_named_section(self):
2362
store = self.get_store(self)
2363
store._load_from_string('[baz]\nfoo=bar')
2364
sections = list(store.get_sections())
2365
self.assertLength(1, sections)
2366
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2368
def test_load_from_string_fails_for_non_empty_store(self):
2369
store = self.get_store(self)
2370
store._load_from_string('foo=bar')
2371
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
class TestMutableStore(TestStore):
2454
scenarios = [(key, {'store_id': key, 'get_store': builder}) for key, builder
2455
in config.test_store_builder_registry.iteritems()]
2458
super(TestMutableStore, self).setUp()
2459
self.transport = self.get_transport()
2461
def has_store(self, store):
2462
store_basename = urlutils.relative_url(self.transport.external_url(),
2463
store.external_url())
2464
return self.transport.has(store_basename)
2466
def test_save_empty_creates_no_file(self):
2467
# FIXME: There should be a better way than relying on the test
2468
# parametrization to identify branch.conf -- vila 2011-0526
2469
if self.store_id in ('branch', 'remote_branch'):
2470
raise tests.TestNotApplicable(
2471
'branch.conf is *always* created when a branch is initialized')
2472
store = self.get_store(self)
2474
self.assertEquals(False, self.has_store(store))
2476
def test_save_emptied_succeeds(self):
2477
store = self.get_store(self)
2478
store._load_from_string('foo=bar\n')
2479
section = store.get_mutable_section(None)
2480
section.remove('foo')
2482
self.assertEquals(True, self.has_store(store))
2483
modified_store = self.get_store(self)
2484
sections = list(modified_store.get_sections())
2485
self.assertLength(0, sections)
2487
def test_save_with_content_succeeds(self):
2488
# FIXME: There should be a better way than relying on the test
2489
# parametrization to identify branch.conf -- vila 2011-0526
2490
if self.store_id in ('branch', 'remote_branch'):
2491
raise tests.TestNotApplicable(
2492
'branch.conf is *always* created when a branch is initialized')
2493
store = self.get_store(self)
2494
store._load_from_string('foo=bar\n')
2495
self.assertEquals(False, self.has_store(store))
2497
self.assertEquals(True, self.has_store(store))
2498
modified_store = self.get_store(self)
2499
sections = list(modified_store.get_sections())
2500
self.assertLength(1, sections)
2501
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2503
def test_set_option_in_empty_store(self):
2504
store = self.get_store(self)
2505
section = store.get_mutable_section(None)
2506
section.set('foo', 'bar')
2508
modified_store = self.get_store(self)
2509
sections = list(modified_store.get_sections())
2510
self.assertLength(1, sections)
2511
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2513
def test_set_option_in_default_section(self):
2514
store = self.get_store(self)
2515
store._load_from_string('')
2516
section = store.get_mutable_section(None)
2517
section.set('foo', 'bar')
2519
modified_store = self.get_store(self)
2520
sections = list(modified_store.get_sections())
2521
self.assertLength(1, sections)
2522
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2524
def test_set_option_in_named_section(self):
2525
store = self.get_store(self)
2526
store._load_from_string('')
2527
section = store.get_mutable_section('baz')
2528
section.set('foo', 'bar')
2530
modified_store = self.get_store(self)
2531
sections = list(modified_store.get_sections())
2532
self.assertLength(1, sections)
2533
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2535
def test_load_hook(self):
2536
# We first needs to ensure that the store exists
2537
store = self.get_store(self)
2538
section = store.get_mutable_section('baz')
2539
section.set('foo', 'bar')
2541
# Now we can try to load it
2542
store = self.get_store(self)
2546
config.ConfigHooks.install_named_hook('load', hook, None)
2547
self.assertLength(0, calls)
2549
self.assertLength(1, calls)
2550
self.assertEquals((store,), calls[0])
2552
def test_save_hook(self):
2556
config.ConfigHooks.install_named_hook('save', hook, None)
2557
self.assertLength(0, calls)
2558
store = self.get_store(self)
2559
section = store.get_mutable_section('baz')
2560
section.set('foo', 'bar')
2562
self.assertLength(1, calls)
2563
self.assertEquals((store,), calls[0])
2566
class TestIniFileStore(TestStore):
2568
def test_loading_unknown_file_fails(self):
2569
store = config.IniFileStore(self.get_transport(), 'I-do-not-exist')
2570
self.assertRaises(errors.NoSuchFile, store.load)
2572
def test_invalid_content(self):
2573
store = config.IniFileStore(self.get_transport(), 'foo.conf', )
2574
self.assertEquals(False, store.is_loaded())
2575
exc = self.assertRaises(
2576
errors.ParseConfigError, store._load_from_string,
2577
'this is invalid !')
2578
self.assertEndsWith(exc.filename, 'foo.conf')
2579
# And the load failed
2580
self.assertEquals(False, store.is_loaded())
2582
def test_get_embedded_sections(self):
2583
# A more complicated example (which also shows that section names and
2584
# option names share the same name space...)
2585
# FIXME: This should be fixed by forbidding dicts as values ?
2586
# -- vila 2011-04-05
2587
store = config.IniFileStore(self.get_transport(), 'foo.conf', )
2588
store._load_from_string('''
2592
foo_in_DEFAULT=foo_DEFAULT
2600
sections = list(store.get_sections())
2601
self.assertLength(4, sections)
2602
# The default section has no name.
2603
# List values are provided as lists
2604
self.assertSectionContent((None, {'foo': 'bar', 'l': ['1', '2']}),
2606
self.assertSectionContent(
2607
('DEFAULT', {'foo_in_DEFAULT': 'foo_DEFAULT'}), sections[1])
2608
self.assertSectionContent(
2609
('bar', {'foo_in_bar': 'barbar'}), sections[2])
2610
# sub sections are provided as embedded dicts.
2611
self.assertSectionContent(
2612
('baz', {'foo_in_baz': 'barbaz', 'qux': {'foo_in_qux': 'quux'}}),
2616
class TestLockableIniFileStore(TestStore):
2618
def test_create_store_in_created_dir(self):
2619
self.assertPathDoesNotExist('dir')
2620
t = self.get_transport('dir/subdir')
2621
store = config.LockableIniFileStore(t, 'foo.conf')
2622
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
2759
class TestSectionMatcher(TestStore):
2761
scenarios = [('location', {'matcher': config.LocationMatcher})]
2763
def get_store(self, file_name):
2764
return config.IniFileStore(self.get_readonly_transport(), file_name)
2766
def test_no_matches_for_empty_stores(self):
2767
store = self.get_store('foo.conf')
2768
store._load_from_string('')
2769
matcher = self.matcher(store, '/bar')
2770
self.assertEquals([], list(matcher.get_sections()))
2772
def test_build_doesnt_load_store(self):
2773
store = self.get_store('foo.conf')
2774
matcher = self.matcher(store, '/bar')
2775
self.assertFalse(store.is_loaded())
2778
class TestLocationSection(tests.TestCase):
2780
def get_section(self, options, extra_path):
2781
section = config.Section('foo', options)
2782
# We don't care about the length so we use '0'
2783
return config.LocationSection(section, 0, extra_path)
2785
def test_simple_option(self):
2786
section = self.get_section({'foo': 'bar'}, '')
2787
self.assertEquals('bar', section.get('foo'))
2789
def test_option_with_extra_path(self):
2790
section = self.get_section({'foo': 'bar', 'foo:policy': 'appendpath'},
2792
self.assertEquals('bar/baz', section.get('foo'))
2794
def test_invalid_policy(self):
2795
section = self.get_section({'foo': 'bar', 'foo:policy': 'die'},
2797
# invalid policies are ignored
2798
self.assertEquals('bar', section.get('foo'))
2801
class TestLocationMatcher(TestStore):
2803
def get_store(self, file_name):
2804
return config.IniFileStore(self.get_readonly_transport(), file_name)
2806
def test_more_specific_sections_first(self):
2807
store = self.get_store('foo.conf')
2808
store._load_from_string('''
2814
self.assertEquals(['/foo', '/foo/bar'],
2815
[section.id for section in store.get_sections()])
2816
matcher = config.LocationMatcher(store, '/foo/bar/baz')
2817
sections = list(matcher.get_sections())
2818
self.assertEquals([3, 2],
2819
[section.length for section in sections])
2820
self.assertEquals(['/foo/bar', '/foo'],
2821
[section.id for section in sections])
2822
self.assertEquals(['baz', 'bar/baz'],
2823
[section.extra_path for section in sections])
2825
def test_appendpath_in_no_name_section(self):
2826
# It's a bit weird to allow appendpath in a no-name section, but
2827
# someone may found a use for it
2828
store = self.get_store('foo.conf')
2829
store._load_from_string('''
2831
foo:policy = appendpath
2833
matcher = config.LocationMatcher(store, 'dir/subdir')
2834
sections = list(matcher.get_sections())
2835
self.assertLength(1, sections)
2836
self.assertEquals('bar/dir/subdir', sections[0].get('foo'))
2838
def test_file_urls_are_normalized(self):
2839
store = self.get_store('foo.conf')
2840
if sys.platform == 'win32':
2841
expected_url = 'file:///C:/dir/subdir'
2842
expected_location = 'C:/dir/subdir'
2844
expected_url = 'file:///dir/subdir'
2845
expected_location = '/dir/subdir'
2846
matcher = config.LocationMatcher(store, expected_url)
2847
self.assertEquals(expected_location, matcher.location)
2850
class TestStackGet(tests.TestCase):
2852
# FIXME: This should be parametrized for all known Stack or dedicated
2853
# paramerized tests created to avoid bloating -- vila 2011-03-31
2855
def test_single_config_get(self):
2856
conf = dict(foo='bar')
2857
conf_stack = config.Stack([conf])
2858
self.assertEquals('bar', conf_stack.get('foo'))
2860
def test_get_with_registered_default_value(self):
2861
conf_stack = config.Stack([dict()])
2862
opt = config.Option('foo', default='bar')
2863
self.overrideAttr(config, 'option_registry', registry.Registry())
2864
config.option_registry.register('foo', opt)
2865
self.assertEquals('bar', conf_stack.get('foo'))
2867
def test_get_without_registered_default_value(self):
2868
conf_stack = config.Stack([dict()])
2869
opt = config.Option('foo')
2870
self.overrideAttr(config, 'option_registry', registry.Registry())
2871
config.option_registry.register('foo', opt)
2872
self.assertEquals(None, conf_stack.get('foo'))
2874
def test_get_without_default_value_for_not_registered(self):
2875
conf_stack = config.Stack([dict()])
2876
opt = config.Option('foo')
2877
self.overrideAttr(config, 'option_registry', registry.Registry())
2878
self.assertEquals(None, conf_stack.get('foo'))
2880
def test_get_first_definition(self):
2881
conf1 = dict(foo='bar')
2882
conf2 = dict(foo='baz')
2883
conf_stack = config.Stack([conf1, conf2])
2884
self.assertEquals('bar', conf_stack.get('foo'))
2886
def test_get_embedded_definition(self):
2887
conf1 = dict(yy='12')
2888
conf2 = config.Stack([dict(xx='42'), dict(foo='baz')])
2889
conf_stack = config.Stack([conf1, conf2])
2890
self.assertEquals('baz', conf_stack.get('foo'))
2892
def test_get_for_empty_section_callable(self):
2893
conf_stack = config.Stack([lambda : []])
2894
self.assertEquals(None, conf_stack.get('foo'))
2896
def test_get_for_broken_callable(self):
2897
# Trying to use and invalid callable raises an exception on first use
2898
conf_stack = config.Stack([lambda : object()])
2899
self.assertRaises(TypeError, conf_stack.get, 'foo')
2902
class TestStackWithTransport(tests.TestCaseWithTransport):
2904
scenarios = [(key, {'get_stack': builder}) for key, builder
2905
in config.test_stack_builder_registry.iteritems()]
2908
class TestConcreteStacks(TestStackWithTransport):
2910
def test_build_stack(self):
2911
# Just a smoke test to help debug builders
2912
stack = self.get_stack(self)
2915
class TestStackGet(TestStackWithTransport):
2917
def test_get_for_empty_stack(self):
2918
conf = self.get_stack(self)
2919
self.assertEquals(None, conf.get('foo'))
2921
def test_get_hook(self):
2922
conf = self.get_stack(self)
2923
conf.store._load_from_string('foo=bar')
2927
config.ConfigHooks.install_named_hook('get', hook, None)
2928
self.assertLength(0, calls)
2929
value = conf.get('foo')
2930
self.assertEquals('bar', value)
2931
self.assertLength(1, calls)
2932
self.assertEquals((conf, 'foo', 'bar'), calls[0])
2935
class TestStackSet(TestStackWithTransport):
2937
def test_simple_set(self):
2938
conf = self.get_stack(self)
2939
conf.store._load_from_string('foo=bar')
2940
self.assertEquals('bar', conf.get('foo'))
2941
conf.set('foo', 'baz')
2942
# Did we get it back ?
2943
self.assertEquals('baz', conf.get('foo'))
2945
def test_set_creates_a_new_section(self):
2946
conf = self.get_stack(self)
2947
conf.set('foo', 'baz')
2948
self.assertEquals, 'baz', conf.get('foo')
2950
def test_set_hook(self):
2954
config.ConfigHooks.install_named_hook('set', hook, None)
2955
self.assertLength(0, calls)
2956
conf = self.get_stack(self)
2957
conf.set('foo', 'bar')
2958
self.assertLength(1, calls)
2959
self.assertEquals((conf, 'foo', 'bar'), calls[0])
2962
class TestStackRemove(TestStackWithTransport):
2964
def test_remove_existing(self):
2965
conf = self.get_stack(self)
2966
conf.store._load_from_string('foo=bar')
2967
self.assertEquals('bar', conf.get('foo'))
2969
# Did we get it back ?
2970
self.assertEquals(None, conf.get('foo'))
2972
def test_remove_unknown(self):
2973
conf = self.get_stack(self)
2974
self.assertRaises(KeyError, conf.remove, 'I_do_not_exist')
2976
def test_remove_hook(self):
2980
config.ConfigHooks.install_named_hook('remove', hook, None)
2981
self.assertLength(0, calls)
2982
conf = self.get_stack(self)
2983
conf.store._load_from_string('foo=bar')
2985
self.assertLength(1, calls)
2986
self.assertEquals((conf, 'foo'), calls[0])
2989
1562
class TestConfigGetOptions(tests.TestCaseWithTransport, TestOptionsMixin):
2991
1564
def setUp(self):
2992
1565
super(TestConfigGetOptions, self).setUp()
2993
1566
create_configs(self)
1568
# One variable in none of the above
2995
1569
def test_no_variable(self):
2996
1570
# Using branch should query branch, locations and bazaar
2997
1571
self.assertOptions([], self.branch_config)
added
removed
bzrlib/tests/test_config.py
