1
# Copyright (C) 2005-2014, 2016 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for finding and reading the bzr config file[s]."""
19
from textwrap import dedent
25
from testtools import matchers
38
registry as _mod_registry,
45
from ..sixish import (
49
from ..transport import remote as transport_remote
57
def lockable_config_scenarios():
60
{'config_class': config.GlobalConfig,
62
'config_section': 'DEFAULT'}),
64
{'config_class': config.LocationConfig,
66
'config_section': '.'}), ]
69
load_tests = scenarios.load_tests_apply_scenarios
71
# Register helpers to build stores
72
config.test_store_builder_registry.register(
73
'configobj', lambda test: config.TransportIniFileStore(
74
test.get_transport(), 'configobj.conf'))
75
config.test_store_builder_registry.register(
76
'breezy', lambda test: config.GlobalStore())
77
config.test_store_builder_registry.register(
78
'location', lambda test: config.LocationStore())
81
def build_backing_branch(test, relpath,
82
transport_class=None, server_class=None):
83
"""Test helper to create a backing branch only once.
85
Some tests needs multiple stores/stacks to check concurrent update
86
behaviours. As such, they need to build different branch *objects* even if
87
they share the branch on disk.
89
:param relpath: The relative path to the branch. (Note that the helper
90
should always specify the same relpath).
92
:param transport_class: The Transport class the test needs to use.
94
:param server_class: The server associated with the ``transport_class``
97
Either both or neither of ``transport_class`` and ``server_class`` should
100
if transport_class is not None and server_class is not None:
101
test.transport_class = transport_class
102
test.transport_server = server_class
103
elif not (transport_class is None and server_class is None):
104
raise AssertionError('Specify both ``transport_class`` and '
105
'``server_class`` or neither of them')
106
if getattr(test, 'backing_branch', None) is None:
107
# First call, let's build the branch on disk
108
test.backing_branch = test.make_branch(relpath)
111
def build_branch_store(test):
112
build_backing_branch(test, 'branch')
113
b = branch.Branch.open('branch')
114
return config.BranchStore(b)
117
config.test_store_builder_registry.register('branch', build_branch_store)
120
def build_control_store(test):
121
build_backing_branch(test, 'branch')
122
b = controldir.ControlDir.open('branch')
123
return config.ControlStore(b)
126
config.test_store_builder_registry.register('control', build_control_store)
129
def build_remote_branch_store(test):
130
# There is only one permutation (but we won't be able to handle more with
131
# this design anyway)
133
server_class) = transport_remote.get_test_permutations()[0]
134
build_backing_branch(test, 'branch', transport_class, server_class)
135
b = branch.Branch.open(test.get_url('branch'))
136
return config.BranchStore(b)
139
config.test_store_builder_registry.register('remote_branch',
140
build_remote_branch_store)
143
config.test_stack_builder_registry.register(
144
'breezy', lambda test: config.GlobalStack())
145
config.test_stack_builder_registry.register(
146
'location', lambda test: config.LocationStack('.'))
149
def build_branch_stack(test):
150
build_backing_branch(test, 'branch')
151
b = branch.Branch.open('branch')
152
return config.BranchStack(b)
155
config.test_stack_builder_registry.register('branch', build_branch_stack)
158
def build_branch_only_stack(test):
159
# There is only one permutation (but we won't be able to handle more with
160
# this design anyway)
162
server_class) = transport_remote.get_test_permutations()[0]
163
build_backing_branch(test, 'branch', transport_class, server_class)
164
b = branch.Branch.open(test.get_url('branch'))
165
return config.BranchOnlyStack(b)
168
config.test_stack_builder_registry.register('branch_only',
169
build_branch_only_stack)
172
def build_remote_control_stack(test):
173
# There is only one permutation (but we won't be able to handle more with
174
# this design anyway)
176
server_class) = transport_remote.get_test_permutations()[0]
177
# We need only a bzrdir for this, not a full branch, but it's not worth
178
# creating a dedicated helper to create only the bzrdir
179
build_backing_branch(test, 'branch', transport_class, server_class)
180
b = branch.Branch.open(test.get_url('branch'))
181
return config.RemoteControlStack(b.controldir)
184
config.test_stack_builder_registry.register('remote_control',
185
build_remote_control_stack)
188
sample_long_alias = "log -r-15..-1 --line"
189
sample_config_text = u"""
191
email=Erik B\u00e5gfors <erik@bagfors.nu>
193
change_editor=vimdiff -of @new_path @old_path
194
gpg_signing_key=DD4D5088
196
validate_signatures_in_log=true
198
user_global_option=something
199
bzr.mergetool.sometool=sometool {base} {this} {other} -o {result}
200
bzr.mergetool.funkytool=funkytool "arg with spaces" {this_temp}
201
bzr.mergetool.newtool='"newtool with spaces" {this_temp}'
202
bzr.default_mergetool=sometool
205
ll=""".encode('utf-8') + sample_long_alias.encode('utf-8') + b"\n"
208
sample_always_signatures = b"""
210
check_signatures=ignore
211
create_signatures=always
214
sample_ignore_signatures = b"""
216
check_signatures=require
217
create_signatures=never
220
sample_maybe_signatures = b"""
222
check_signatures=ignore
223
create_signatures=when-required
226
sample_branches_text = b"""
227
[http://www.example.com]
229
email=Robert Collins <robertc@example.org>
230
normal_option = normal
231
appendpath_option = append
232
appendpath_option:policy = appendpath
233
norecurse_option = norecurse
234
norecurse_option:policy = norecurse
235
[http://www.example.com/ignoreparent]
236
# different project: ignore parent dir config
238
[http://www.example.com/norecurse]
239
# configuration items that only apply to this dir
241
normal_option = norecurse
242
[http://www.example.com/dir]
243
appendpath_option = normal
245
check_signatures=require
246
# test trailing / matching with no children
248
check_signatures=check-available
249
gpg_signing_key=default
250
user_local_option=local
251
# test trailing / matching
253
#subdirs will match but not the parent
255
check_signatures=ignore
256
post_commit=breezy.tests.test_config.post_commit
257
#testing explicit beats globs
261
def create_configs(test):
262
"""Create configuration files for a given test.
264
This requires creating a tree (and populate the ``test.tree`` attribute)
265
and its associated branch and will populate the following attributes:
267
- branch_config: A BranchConfig for the associated branch.
269
- locations_config : A LocationConfig for the associated branch
271
- breezy_config: A GlobalConfig.
273
The tree and branch are created in a 'tree' subdirectory so the tests can
274
still use the test directory to stay outside of the branch.
276
tree = test.make_branch_and_tree('tree')
278
test.branch_config = config.BranchConfig(tree.branch)
279
test.locations_config = config.LocationConfig(tree.basedir)
280
test.breezy_config = config.GlobalConfig()
283
def create_configs_with_file_option(test):
284
"""Create configuration files with a ``file`` option set in each.
286
This builds on ``create_configs`` and add one ``file`` option in each
287
configuration with a value which allows identifying the configuration file.
290
test.breezy_config.set_user_option('file', 'breezy')
291
test.locations_config.set_user_option('file', 'locations')
292
test.branch_config.set_user_option('file', 'branch')
295
class TestOptionsMixin:
297
def assertOptions(self, expected, conf):
298
# We don't care about the parser (as it will make tests hard to write
299
# and error-prone anyway)
300
self.assertThat([opt[:4] for opt in conf._get_options()],
301
matchers.Equals(expected))
304
class InstrumentedConfigObj(object):
305
"""A config obj look-enough-alike to record calls made to it."""
307
def __contains__(self, thing):
308
self._calls.append(('__contains__', thing))
311
def __getitem__(self, key):
312
self._calls.append(('__getitem__', key))
315
def __init__(self, input, encoding=None):
316
self._calls = [('__init__', input, encoding)]
318
def __setitem__(self, key, value):
319
self._calls.append(('__setitem__', key, value))
321
def __delitem__(self, key):
322
self._calls.append(('__delitem__', key))
325
self._calls.append(('keys',))
329
self._calls.append(('reload',))
331
def write(self, arg):
332
self._calls.append(('write',))
334
def as_bool(self, value):
335
self._calls.append(('as_bool', value))
338
def get_value(self, section, name):
339
self._calls.append(('get_value', section, name))
343
class FakeBranch(object):
345
def __init__(self, base=None):
347
self.base = "http://example.com/branches/demo"
350
self._transport = self.control_files = \
351
FakeControlFilesAndTransport()
353
def _get_config(self):
354
return config.TransportConfig(self._transport, 'branch.conf')
356
def lock_write(self):
357
return lock.LogicalLockResult(self.unlock)
363
class FakeControlFilesAndTransport(object):
367
self._transport = self
369
def get(self, filename):
372
return BytesIO(self.files[filename])
374
raise errors.NoSuchFile(filename)
376
def get_bytes(self, filename):
379
return self.files[filename]
381
raise errors.NoSuchFile(filename)
383
def put(self, filename, fileobj):
384
self.files[filename] = fileobj.read()
386
def put_file(self, filename, fileobj):
387
return self.put(filename, fileobj)
390
class InstrumentedConfig(config.Config):
391
"""An instrumented config that supplies stubs for template methods."""
394
super(InstrumentedConfig, self).__init__()
396
self._signatures = config.CHECK_NEVER
398
def _get_user_id(self):
399
self._calls.append('_get_user_id')
400
return "Robert Collins <robert.collins@example.org>"
402
def _get_signature_checking(self):
403
self._calls.append('_get_signature_checking')
404
return self._signatures
406
def _get_change_editor(self):
407
self._calls.append('_get_change_editor')
408
return 'vimdiff -fo @new_path @old_path'
411
bool_config = b"""[DEFAULT]
420
class TestConfigObj(tests.TestCase):
422
def test_get_bool(self):
423
co = config.ConfigObj(BytesIO(bool_config))
424
self.assertIs(co.get_bool('DEFAULT', 'active'), True)
425
self.assertIs(co.get_bool('DEFAULT', 'inactive'), False)
426
self.assertIs(co.get_bool('UPPERCASE', 'active'), True)
427
self.assertIs(co.get_bool('UPPERCASE', 'nonactive'), False)
429
def test_hash_sign_in_value(self):
431
Before 4.5.0, ConfigObj did not quote # signs in values, so they'd be
432
treated as comments when read in again. (#86838)
434
co = config.ConfigObj()
435
co['test'] = 'foo#bar'
437
co.write(outfile=outfile)
438
lines = outfile.getvalue().splitlines()
439
self.assertEqual(lines, [b'test = "foo#bar"'])
440
co2 = config.ConfigObj(lines)
441
self.assertEqual(co2['test'], 'foo#bar')
443
def test_triple_quotes(self):
444
# Bug #710410: if the value string has triple quotes
445
# then ConfigObj versions up to 4.7.2 will quote them wrong
446
# and won't able to read them back
447
triple_quotes_value = '''spam
448
""" that's my spam """
450
co = config.ConfigObj()
451
co['test'] = triple_quotes_value
452
# While writing this test another bug in ConfigObj has been found:
453
# method co.write() without arguments produces list of lines
454
# one option per line, and multiline values are not split
455
# across multiple lines,
456
# and that breaks the parsing these lines back by ConfigObj.
457
# This issue only affects test, but it's better to avoid
458
# `co.write()` construct at all.
459
# [bialix 20110222] bug report sent to ConfigObj's author
461
co.write(outfile=outfile)
462
output = outfile.getvalue()
463
# now we're trying to read it back
464
co2 = config.ConfigObj(BytesIO(output))
465
self.assertEqual(triple_quotes_value, co2['test'])
468
erroneous_config = b"""[section] # line 1
471
whocares=notme # line 4
475
class TestConfigObjErrors(tests.TestCase):
477
def test_duplicate_section_name_error_line(self):
479
co = configobj.ConfigObj(BytesIO(erroneous_config),
481
except config.configobj.DuplicateError as e:
482
self.assertEqual(3, e.line_number)
484
self.fail('Error in config file not detected')
487
class TestConfig(tests.TestCase):
489
def test_constructs(self):
492
def test_user_email(self):
493
my_config = InstrumentedConfig()
494
self.assertEqual('robert.collins@example.org', my_config.user_email())
495
self.assertEqual(['_get_user_id'], my_config._calls)
497
def test_username(self):
498
my_config = InstrumentedConfig()
499
self.assertEqual('Robert Collins <robert.collins@example.org>',
500
my_config.username())
501
self.assertEqual(['_get_user_id'], my_config._calls)
503
def test_get_user_option_default(self):
504
my_config = config.Config()
505
self.assertEqual(None, my_config.get_user_option('no_option'))
507
def test_validate_signatures_in_log_default(self):
508
my_config = config.Config()
509
self.assertEqual(False, my_config.validate_signatures_in_log())
511
def test_get_change_editor(self):
512
my_config = InstrumentedConfig()
513
change_editor = my_config.get_change_editor('old_tree', 'new_tree')
514
self.assertEqual(['_get_change_editor'], my_config._calls)
515
self.assertIs(diff.DiffFromTool, change_editor.__class__)
516
self.assertEqual(['vimdiff', '-fo', '@new_path', '@old_path'],
517
change_editor.command_template)
520
class TestConfigPath(tests.TestCase):
523
super(TestConfigPath, self).setUp()
524
self.overrideEnv('HOME', '/home/bogus')
525
self.overrideEnv('XDG_CACHE_HOME', '')
526
if sys.platform == 'win32':
529
r'C:\Documents and Settings\bogus\Application Data')
531
'C:/Documents and Settings/bogus/Application Data/breezy'
533
self.brz_home = '/home/bogus/.config/breezy'
535
def test_config_dir(self):
536
self.assertEqual(config.config_dir(), self.brz_home)
538
def test_config_dir_is_unicode(self):
539
self.assertIsInstance(config.config_dir(), text_type)
541
def test_config_filename(self):
542
self.assertEqual(config.config_filename(),
543
self.brz_home + '/breezy.conf')
545
def test_locations_config_filename(self):
546
self.assertEqual(config.locations_config_filename(),
547
self.brz_home + '/locations.conf')
549
def test_authentication_config_filename(self):
550
self.assertEqual(config.authentication_config_filename(),
551
self.brz_home + '/authentication.conf')
553
def test_xdg_cache_dir(self):
554
self.assertEqual(config.xdg_cache_dir(),
555
'/home/bogus/.cache')
558
class TestConfigPathFallback(tests.TestCaseInTempDir):
561
super(TestConfigPathFallback, self).setUp()
562
self.overrideEnv('HOME', self.test_dir)
563
self.overrideEnv('XDG_CACHE_HOME', '')
564
self.bzr_home = os.path.join(self.test_dir, '.bazaar')
565
os.mkdir(self.bzr_home)
567
def test_config_dir(self):
568
self.assertEqual(config.config_dir(), self.bzr_home)
570
def test_config_dir_is_unicode(self):
571
self.assertIsInstance(config.config_dir(), text_type)
573
def test_config_filename(self):
574
self.assertEqual(config.config_filename(),
575
self.bzr_home + '/bazaar.conf')
577
def test_locations_config_filename(self):
578
self.assertEqual(config.locations_config_filename(),
579
self.bzr_home + '/locations.conf')
581
def test_authentication_config_filename(self):
582
self.assertEqual(config.authentication_config_filename(),
583
self.bzr_home + '/authentication.conf')
585
def test_xdg_cache_dir(self):
586
self.assertEqual(config.xdg_cache_dir(),
587
os.path.join(self.test_dir, '.cache'))
590
class TestXDGConfigDir(tests.TestCaseInTempDir):
591
# must be in temp dir because config tests for the existence of the bazaar
592
# subdirectory of $XDG_CONFIG_HOME
595
if sys.platform == 'win32':
596
raise tests.TestNotApplicable(
597
'XDG config dir not used on this platform')
598
super(TestXDGConfigDir, self).setUp()
599
self.overrideEnv('HOME', self.test_home_dir)
600
# BRZ_HOME overrides everything we want to test so unset it.
601
self.overrideEnv('BRZ_HOME', None)
603
def test_xdg_config_dir_exists(self):
604
"""When ~/.config/bazaar exists, use it as the config dir."""
605
newdir = osutils.pathjoin(self.test_home_dir, '.config', 'bazaar')
607
self.assertEqual(config.config_dir(), newdir)
609
def test_xdg_config_home(self):
610
"""When XDG_CONFIG_HOME is set, use it."""
611
xdgconfigdir = osutils.pathjoin(self.test_home_dir, 'xdgconfig')
612
self.overrideEnv('XDG_CONFIG_HOME', xdgconfigdir)
613
newdir = osutils.pathjoin(xdgconfigdir, 'bazaar')
615
self.assertEqual(config.config_dir(), newdir)
618
class TestIniConfig(tests.TestCaseInTempDir):
620
def make_config_parser(self, s):
621
conf = config.IniBasedConfig.from_string(s)
622
return conf, conf._get_parser()
625
class TestIniConfigBuilding(TestIniConfig):
627
def test_contructs(self):
628
config.IniBasedConfig()
630
def test_from_fp(self):
631
my_config = config.IniBasedConfig.from_string(sample_config_text)
632
self.assertIsInstance(my_config._get_parser(), configobj.ConfigObj)
634
def test_cached(self):
635
my_config = config.IniBasedConfig.from_string(sample_config_text)
636
parser = my_config._get_parser()
637
self.assertTrue(my_config._get_parser() is parser)
639
def _dummy_chown(self, path, uid, gid):
640
self.path, self.uid, self.gid = path, uid, gid
642
def test_ini_config_ownership(self):
643
"""Ensure that chown is happening during _write_config_file"""
644
self.requireFeature(features.chown_feature)
645
self.overrideAttr(os, 'chown', self._dummy_chown)
646
self.path = self.uid = self.gid = None
647
conf = config.IniBasedConfig(file_name='./foo.conf')
648
conf._write_config_file()
649
self.assertEqual(self.path, './foo.conf')
650
self.assertTrue(isinstance(self.uid, int))
651
self.assertTrue(isinstance(self.gid, int))
654
class TestIniConfigSaving(tests.TestCaseInTempDir):
656
def test_cant_save_without_a_file_name(self):
657
conf = config.IniBasedConfig()
658
self.assertRaises(AssertionError, conf._write_config_file)
660
def test_saved_with_content(self):
661
content = b'foo = bar\n'
662
config.IniBasedConfig.from_string(content, file_name='./test.conf',
664
self.assertFileEqual(content, 'test.conf')
667
class TestIniConfigOptionExpansion(tests.TestCase):
668
"""Test option expansion from the IniConfig level.
670
What we really want here is to test the Config level, but the class being
671
abstract as far as storing values is concerned, this can't be done
674
# FIXME: This should be rewritten when all configs share a storage
675
# implementation -- vila 2011-02-18
677
def get_config(self, string=None):
680
c = config.IniBasedConfig.from_string(string)
683
def assertExpansion(self, expected, conf, string, env=None):
684
self.assertEqual(expected, conf.expand_options(string, env))
686
def test_no_expansion(self):
687
c = self.get_config('')
688
self.assertExpansion('foo', c, 'foo')
690
def test_env_adding_options(self):
691
c = self.get_config('')
692
self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})
694
def test_env_overriding_options(self):
695
c = self.get_config('foo=baz')
696
self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})
698
def test_simple_ref(self):
699
c = self.get_config('foo=xxx')
700
self.assertExpansion('xxx', c, '{foo}')
702
def test_unknown_ref(self):
703
c = self.get_config('')
704
self.assertRaises(config.ExpandingUnknownOption,
705
c.expand_options, '{foo}')
707
def test_indirect_ref(self):
708
c = self.get_config('''
712
self.assertExpansion('xxx', c, '{bar}')
714
def test_embedded_ref(self):
715
c = self.get_config('''
719
self.assertExpansion('xxx', c, '{{bar}}')
721
def test_simple_loop(self):
722
c = self.get_config('foo={foo}')
723
self.assertRaises(config.OptionExpansionLoop, c.expand_options,
726
def test_indirect_loop(self):
727
c = self.get_config('''
731
e = self.assertRaises(config.OptionExpansionLoop,
732
c.expand_options, '{foo}')
733
self.assertEqual('foo->bar->baz', e.refs)
734
self.assertEqual('{foo}', e.string)
737
conf = self.get_config('''
741
list={foo},{bar},{baz}
743
self.assertEqual(['start', 'middle', 'end'],
744
conf.get_user_option('list', expand=True))
746
def test_cascading_list(self):
747
conf = self.get_config('''
753
self.assertEqual(['start', 'middle', 'end'],
754
conf.get_user_option('list', expand=True))
756
def test_pathological_hidden_list(self):
757
conf = self.get_config('''
763
hidden={start}{middle}{end}
765
# Nope, it's either a string or a list, and the list wins as soon as a
766
# ',' appears, so the string concatenation never occur.
767
self.assertEqual(['{foo', '}', '{', 'bar}'],
768
conf.get_user_option('hidden', expand=True))
771
class TestLocationConfigOptionExpansion(tests.TestCaseInTempDir):
773
def get_config(self, location, string=None):
776
# Since we don't save the config we won't strictly require to inherit
777
# from TestCaseInTempDir, but an error occurs so quickly...
778
c = config.LocationConfig.from_string(string, location)
781
def test_dont_cross_unrelated_section(self):
782
c = self.get_config('/another/branch/path', '''
787
[/another/branch/path]
790
self.assertRaises(config.ExpandingUnknownOption,
791
c.get_user_option, 'bar', expand=True)
793
def test_cross_related_sections(self):
794
c = self.get_config('/project/branch/path', '''
798
[/project/branch/path]
801
self.assertEqual('quux', c.get_user_option('bar', expand=True))
804
class TestIniBaseConfigOnDisk(tests.TestCaseInTempDir):
806
def test_cannot_reload_without_name(self):
807
conf = config.IniBasedConfig.from_string(sample_config_text)
808
self.assertRaises(AssertionError, conf.reload)
810
def test_reload_see_new_value(self):
811
c1 = config.IniBasedConfig.from_string('editor=vim\n',
812
file_name='./test/conf')
813
c1._write_config_file()
814
c2 = config.IniBasedConfig.from_string('editor=emacs\n',
815
file_name='./test/conf')
816
c2._write_config_file()
817
self.assertEqual('vim', c1.get_user_option('editor'))
818
self.assertEqual('emacs', c2.get_user_option('editor'))
819
# Make sure we get the Right value
821
self.assertEqual('emacs', c1.get_user_option('editor'))
824
class TestLockableConfig(tests.TestCaseInTempDir):
826
scenarios = lockable_config_scenarios()
831
config_section = None
834
super(TestLockableConfig, self).setUp()
835
self._content = '[%s]\none=1\ntwo=2\n' % (self.config_section,)
836
self.config = self.create_config(self._content)
838
def get_existing_config(self):
839
return self.config_class(*self.config_args)
841
def create_config(self, content):
842
kwargs = dict(save=True)
843
c = self.config_class.from_string(content, *self.config_args, **kwargs)
846
def test_simple_read_access(self):
847
self.assertEqual('1', self.config.get_user_option('one'))
849
def test_simple_write_access(self):
850
self.config.set_user_option('one', 'one')
851
self.assertEqual('one', self.config.get_user_option('one'))
853
def test_listen_to_the_last_speaker(self):
855
c2 = self.get_existing_config()
856
c1.set_user_option('one', 'ONE')
857
c2.set_user_option('two', 'TWO')
858
self.assertEqual('ONE', c1.get_user_option('one'))
859
self.assertEqual('TWO', c2.get_user_option('two'))
860
# The second update respect the first one
861
self.assertEqual('ONE', c2.get_user_option('one'))
863
def test_last_speaker_wins(self):
864
# If the same config is not shared, the same variable modified twice
865
# can only see a single result.
867
c2 = self.get_existing_config()
868
c1.set_user_option('one', 'c1')
869
c2.set_user_option('one', 'c2')
870
self.assertEqual('c2', c2._get_user_option('one'))
871
# The first modification is still available until another refresh
873
self.assertEqual('c1', c1._get_user_option('one'))
874
c1.set_user_option('two', 'done')
875
self.assertEqual('c2', c1._get_user_option('one'))
877
def test_writes_are_serialized(self):
879
c2 = self.get_existing_config()
881
# We spawn a thread that will pause *during* the write
882
before_writing = threading.Event()
883
after_writing = threading.Event()
884
writing_done = threading.Event()
885
c1_orig = c1._write_config_file
887
def c1_write_config_file():
890
# The lock is held. We wait for the main thread to decide when to
893
c1._write_config_file = c1_write_config_file
896
c1.set_user_option('one', 'c1')
898
t1 = threading.Thread(target=c1_set_option)
899
# Collect the thread after the test
900
self.addCleanup(t1.join)
901
# Be ready to unblock the thread if the test goes wrong
902
self.addCleanup(after_writing.set)
904
before_writing.wait()
905
self.assertTrue(c1._lock.is_held)
906
self.assertRaises(errors.LockContention,
907
c2.set_user_option, 'one', 'c2')
908
self.assertEqual('c1', c1.get_user_option('one'))
909
# Let the lock be released
912
c2.set_user_option('one', 'c2')
913
self.assertEqual('c2', c2.get_user_option('one'))
915
def test_read_while_writing(self):
917
# We spawn a thread that will pause *during* the write
918
ready_to_write = threading.Event()
919
do_writing = threading.Event()
920
writing_done = threading.Event()
921
c1_orig = c1._write_config_file
923
def c1_write_config_file():
925
# The lock is held. We wait for the main thread to decide when to
930
c1._write_config_file = c1_write_config_file
933
c1.set_user_option('one', 'c1')
934
t1 = threading.Thread(target=c1_set_option)
935
# Collect the thread after the test
936
self.addCleanup(t1.join)
937
# Be ready to unblock the thread if the test goes wrong
938
self.addCleanup(do_writing.set)
940
# Ensure the thread is ready to write
941
ready_to_write.wait()
942
self.assertTrue(c1._lock.is_held)
943
self.assertEqual('c1', c1.get_user_option('one'))
944
# If we read during the write, we get the old value
945
c2 = self.get_existing_config()
946
self.assertEqual('1', c2.get_user_option('one'))
947
# Let the writing occur and ensure it occurred
950
# Now we get the updated value
951
c3 = self.get_existing_config()
952
self.assertEqual('c1', c3.get_user_option('one'))
955
class TestGetUserOptionAs(TestIniConfig):
957
def test_get_user_option_as_bool(self):
958
conf, parser = self.make_config_parser("""
961
an_invalid_bool = maybe
962
a_list = hmm, who knows ? # This is interpreted as a list !
964
get_bool = conf.get_user_option_as_bool
965
self.assertEqual(True, get_bool('a_true_bool'))
966
self.assertEqual(False, get_bool('a_false_bool'))
970
warnings.append(args[0] % args[1:])
971
self.overrideAttr(trace, 'warning', warning)
972
msg = 'Value "%s" is not a boolean for "%s"'
973
self.assertIs(None, get_bool('an_invalid_bool'))
974
self.assertEqual(msg % ('maybe', 'an_invalid_bool'), warnings[0])
976
self.assertIs(None, get_bool('not_defined_in_this_config'))
977
self.assertEqual([], warnings)
979
def test_get_user_option_as_list(self):
980
conf, parser = self.make_config_parser("""
985
get_list = conf.get_user_option_as_list
986
self.assertEqual(['a', 'b', 'c'], get_list('a_list'))
987
self.assertEqual(['1'], get_list('length_1'))
988
self.assertEqual('x', conf.get_user_option('one_item'))
989
# automatically cast to list
990
self.assertEqual(['x'], get_list('one_item'))
993
class TestSupressWarning(TestIniConfig):
995
def make_warnings_config(self, s):
996
conf, parser = self.make_config_parser(s)
997
return conf.suppress_warning
999
def test_suppress_warning_unknown(self):
1000
suppress_warning = self.make_warnings_config('')
1001
self.assertEqual(False, suppress_warning('unknown_warning'))
1003
def test_suppress_warning_known(self):
1004
suppress_warning = self.make_warnings_config('suppress_warnings=a,b')
1005
self.assertEqual(False, suppress_warning('c'))
1006
self.assertEqual(True, suppress_warning('a'))
1007
self.assertEqual(True, suppress_warning('b'))
1010
class TestGetConfig(tests.TestCaseInTempDir):
1012
def test_constructs(self):
1013
config.GlobalConfig()
1015
def test_calls_read_filenames(self):
1016
# replace the class that is constructed, to check its parameters
1017
oldparserclass = config.ConfigObj
1018
config.ConfigObj = InstrumentedConfigObj
1019
my_config = config.GlobalConfig()
1021
parser = my_config._get_parser()
1023
config.ConfigObj = oldparserclass
1024
self.assertIsInstance(parser, InstrumentedConfigObj)
1025
self.assertEqual(parser._calls, [('__init__', config.config_filename(),
1029
class TestBranchConfig(tests.TestCaseWithTransport):
1031
def test_constructs_valid(self):
1032
branch = FakeBranch()
1033
my_config = config.BranchConfig(branch)
1034
self.assertIsNot(None, my_config)
1036
def test_constructs_error(self):
1037
self.assertRaises(TypeError, config.BranchConfig)
1039
def test_get_location_config(self):
1040
branch = FakeBranch()
1041
my_config = config.BranchConfig(branch)
1042
location_config = my_config._get_location_config()
1043
self.assertEqual(branch.base, location_config.location)
1044
self.assertIs(location_config, my_config._get_location_config())
1046
def test_get_config(self):
1047
"""The Branch.get_config method works properly"""
1048
b = controldir.ControlDir.create_standalone_workingtree('.').branch
1049
my_config = b.get_config()
1050
self.assertIs(my_config.get_user_option('wacky'), None)
1051
my_config.set_user_option('wacky', 'unlikely')
1052
self.assertEqual(my_config.get_user_option('wacky'), 'unlikely')
1054
# Ensure we get the same thing if we start again
1055
b2 = branch.Branch.open('.')
1056
my_config2 = b2.get_config()
1057
self.assertEqual(my_config2.get_user_option('wacky'), 'unlikely')
1059
def test_has_explicit_nickname(self):
1060
b = self.make_branch('.')
1061
self.assertFalse(b.get_config().has_explicit_nickname())
1063
self.assertTrue(b.get_config().has_explicit_nickname())
1065
def test_config_url(self):
1066
"""The Branch.get_config will use section that uses a local url"""
1067
branch = self.make_branch('branch')
1068
self.assertEqual('branch', branch.nick)
1070
local_url = urlutils.local_path_to_url('branch')
1071
conf = config.LocationConfig.from_string(
1072
'[%s]\nnickname = foobar' % (local_url,),
1073
local_url, save=True)
1074
self.assertIsNot(None, conf)
1075
self.assertEqual('foobar', branch.nick)
1077
def test_config_local_path(self):
1078
"""The Branch.get_config will use a local system path"""
1079
branch = self.make_branch('branch')
1080
self.assertEqual('branch', branch.nick)
1082
local_path = osutils.getcwd().encode('utf8')
1083
config.LocationConfig.from_string(
1084
b'[%s/branch]\nnickname = barry' % (local_path,),
1085
'branch', save=True)
1086
# Now the branch will find its nick via the location config
1087
self.assertEqual('barry', branch.nick)
1089
def test_config_creates_local(self):
1090
"""Creating a new entry in config uses a local path."""
1091
branch = self.make_branch('branch', format='knit')
1092
branch.set_push_location('http://foobar')
1093
local_path = osutils.getcwd().encode('utf8')
1094
# Surprisingly ConfigObj doesn't create a trailing newline
1095
self.check_file_contents(config.locations_config_filename(),
1097
b'push_location = http://foobar\n'
1098
b'push_location:policy = norecurse\n'
1101
def test_autonick_urlencoded(self):
1102
b = self.make_branch('!repo')
1103
self.assertEqual('!repo', b.get_config().get_nickname())
1105
def test_autonick_uses_branch_name(self):
1106
b = self.make_branch('foo', name='bar')
1107
self.assertEqual('bar', b.get_config().get_nickname())
1109
def test_warn_if_masked(self):
1113
warnings.append(args[0] % args[1:])
1114
self.overrideAttr(trace, 'warning', warning)
1116
def set_option(store, warn_masked=True):
1118
conf.set_user_option('example_option', repr(store), store=store,
1119
warn_masked=warn_masked)
1121
def assertWarning(warning):
1123
self.assertEqual(0, len(warnings))
1125
self.assertEqual(1, len(warnings))
1126
self.assertEqual(warning, warnings[0])
1127
branch = self.make_branch('.')
1128
conf = branch.get_config()
1129
set_option(config.STORE_GLOBAL)
1131
set_option(config.STORE_BRANCH)
1133
set_option(config.STORE_GLOBAL)
1134
assertWarning('Value "4" is masked by "3" from branch.conf')
1135
set_option(config.STORE_GLOBAL, warn_masked=False)
1137
set_option(config.STORE_LOCATION)
1139
set_option(config.STORE_BRANCH)
1140
assertWarning('Value "3" is masked by "0" from locations.conf')
1141
set_option(config.STORE_BRANCH, warn_masked=False)
1145
class TestGlobalConfigItems(tests.TestCaseInTempDir):
1147
def _get_empty_config(self):
1148
my_config = config.GlobalConfig()
1151
def _get_sample_config(self):
1152
my_config = config.GlobalConfig.from_string(sample_config_text)
1155
def test_user_id(self):
1156
my_config = config.GlobalConfig.from_string(sample_config_text)
1157
self.assertEqual(u"Erik B\u00e5gfors <erik@bagfors.nu>",
1158
my_config._get_user_id())
1160
def test_absent_user_id(self):
1161
my_config = config.GlobalConfig()
1162
self.assertEqual(None, my_config._get_user_id())
1164
def test_get_user_option_default(self):
1165
my_config = self._get_empty_config()
1166
self.assertEqual(None, my_config.get_user_option('no_option'))
1168
def test_get_user_option_global(self):
1169
my_config = self._get_sample_config()
1170
self.assertEqual("something",
1171
my_config.get_user_option('user_global_option'))
1173
def test_configured_validate_signatures_in_log(self):
1174
my_config = self._get_sample_config()
1175
self.assertEqual(True, my_config.validate_signatures_in_log())
1177
def test_get_alias(self):
1178
my_config = self._get_sample_config()
1179
self.assertEqual('help', my_config.get_alias('h'))
1181
def test_get_aliases(self):
1182
my_config = self._get_sample_config()
1183
aliases = my_config.get_aliases()
1184
self.assertEqual(2, len(aliases))
1185
sorted_keys = sorted(aliases)
1186
self.assertEqual('help', aliases[sorted_keys[0]])
1187
self.assertEqual(sample_long_alias, aliases[sorted_keys[1]])
1189
def test_get_no_alias(self):
1190
my_config = self._get_sample_config()
1191
self.assertEqual(None, my_config.get_alias('foo'))
1193
def test_get_long_alias(self):
1194
my_config = self._get_sample_config()
1195
self.assertEqual(sample_long_alias, my_config.get_alias('ll'))
1197
def test_get_change_editor(self):
1198
my_config = self._get_sample_config()
1199
change_editor = my_config.get_change_editor('old', 'new')
1200
self.assertIs(diff.DiffFromTool, change_editor.__class__)
1201
self.assertEqual('vimdiff -of @new_path @old_path',
1202
' '.join(change_editor.command_template))
1204
def test_get_no_change_editor(self):
1205
my_config = self._get_empty_config()
1206
change_editor = my_config.get_change_editor('old', 'new')
1207
self.assertIs(None, change_editor)
1209
def test_get_merge_tools(self):
1210
conf = self._get_sample_config()
1211
tools = conf.get_merge_tools()
1212
self.log(repr(tools))
1214
{u'funkytool': u'funkytool "arg with spaces" {this_temp}',
1215
u'sometool': u'sometool {base} {this} {other} -o {result}',
1216
u'newtool': u'"newtool with spaces" {this_temp}'},
1219
def test_get_merge_tools_empty(self):
1220
conf = self._get_empty_config()
1221
tools = conf.get_merge_tools()
1222
self.assertEqual({}, tools)
1224
def test_find_merge_tool(self):
1225
conf = self._get_sample_config()
1226
cmdline = conf.find_merge_tool('sometool')
1227
self.assertEqual('sometool {base} {this} {other} -o {result}', cmdline)
1229
def test_find_merge_tool_not_found(self):
1230
conf = self._get_sample_config()
1231
cmdline = conf.find_merge_tool('DOES NOT EXIST')
1232
self.assertIs(cmdline, None)
1234
def test_find_merge_tool_known(self):
1235
conf = self._get_empty_config()
1236
cmdline = conf.find_merge_tool('kdiff3')
1237
self.assertEqual('kdiff3 {base} {this} {other} -o {result}', cmdline)
1239
def test_find_merge_tool_override_known(self):
1240
conf = self._get_empty_config()
1241
conf.set_user_option('bzr.mergetool.kdiff3', 'kdiff3 blah')
1242
cmdline = conf.find_merge_tool('kdiff3')
1243
self.assertEqual('kdiff3 blah', cmdline)
1246
class TestGlobalConfigSavingOptions(tests.TestCaseInTempDir):
1248
def test_empty(self):
1249
my_config = config.GlobalConfig()
1250
self.assertEqual(0, len(my_config.get_aliases()))
1252
def test_set_alias(self):
1253
my_config = config.GlobalConfig()
1254
alias_value = 'commit --strict'
1255
my_config.set_alias('commit', alias_value)
1256
new_config = config.GlobalConfig()
1257
self.assertEqual(alias_value, new_config.get_alias('commit'))
1259
def test_remove_alias(self):
1260
my_config = config.GlobalConfig()
1261
my_config.set_alias('commit', 'commit --strict')
1262
# Now remove the alias again.
1263
my_config.unset_alias('commit')
1264
new_config = config.GlobalConfig()
1265
self.assertIs(None, new_config.get_alias('commit'))
1268
class TestLocationConfig(tests.TestCaseInTempDir, TestOptionsMixin):
1270
def test_constructs_valid(self):
1271
config.LocationConfig('http://example.com')
1273
def test_constructs_error(self):
1274
self.assertRaises(TypeError, config.LocationConfig)
1276
def test_branch_calls_read_filenames(self):
1277
# This is testing the correct file names are provided.
1278
# TODO: consolidate with the test for GlobalConfigs filename checks.
1280
# replace the class that is constructed, to check its parameters
1281
oldparserclass = config.ConfigObj
1282
config.ConfigObj = InstrumentedConfigObj
1284
my_config = config.LocationConfig('http://www.example.com')
1285
parser = my_config._get_parser()
1287
config.ConfigObj = oldparserclass
1288
self.assertIsInstance(parser, InstrumentedConfigObj)
1289
self.assertEqual(parser._calls,
1290
[('__init__', config.locations_config_filename(),
1293
def test_get_global_config(self):
1294
my_config = config.BranchConfig(FakeBranch('http://example.com'))
1295
global_config = my_config._get_global_config()
1296
self.assertIsInstance(global_config, config.GlobalConfig)
1297
self.assertIs(global_config, my_config._get_global_config())
1299
def assertLocationMatching(self, expected):
1300
self.assertEqual(expected,
1301
list(self.my_location_config._get_matching_sections()))
1303
def test__get_matching_sections_no_match(self):
1304
self.get_branch_config('/')
1305
self.assertLocationMatching([])
1307
def test__get_matching_sections_exact(self):
1308
self.get_branch_config('http://www.example.com')
1309
self.assertLocationMatching([('http://www.example.com', '')])
1311
def test__get_matching_sections_suffix_does_not(self):
1312
self.get_branch_config('http://www.example.com-com')
1313
self.assertLocationMatching([])
1315
def test__get_matching_sections_subdir_recursive(self):
1316
self.get_branch_config('http://www.example.com/com')
1317
self.assertLocationMatching([('http://www.example.com', 'com')])
1319
def test__get_matching_sections_ignoreparent(self):
1320
self.get_branch_config('http://www.example.com/ignoreparent')
1321
self.assertLocationMatching([('http://www.example.com/ignoreparent',
1324
def test__get_matching_sections_ignoreparent_subdir(self):
1325
self.get_branch_config(
1326
'http://www.example.com/ignoreparent/childbranch')
1327
self.assertLocationMatching([('http://www.example.com/ignoreparent',
1330
def test__get_matching_sections_subdir_trailing_slash(self):
1331
self.get_branch_config('/b')
1332
self.assertLocationMatching([('/b/', '')])
1334
def test__get_matching_sections_subdir_child(self):
1335
self.get_branch_config('/a/foo')
1336
self.assertLocationMatching([('/a/*', ''), ('/a/', 'foo')])
1338
def test__get_matching_sections_subdir_child_child(self):
1339
self.get_branch_config('/a/foo/bar')
1340
self.assertLocationMatching([('/a/*', 'bar'), ('/a/', 'foo/bar')])
1342
def test__get_matching_sections_trailing_slash_with_children(self):
1343
self.get_branch_config('/a/')
1344
self.assertLocationMatching([('/a/', '')])
1346
def test__get_matching_sections_explicit_over_glob(self):
1347
# XXX: 2006-09-08 jamesh
1348
# This test only passes because ord('c') > ord('*'). If there
1349
# was a config section for '/a/?', it would get precedence
1351
self.get_branch_config('/a/c')
1352
self.assertLocationMatching([('/a/c', ''), ('/a/*', ''), ('/a/', 'c')])
1354
def test__get_option_policy_normal(self):
1355
self.get_branch_config('http://www.example.com')
1357
self.my_location_config._get_config_policy(
1358
'http://www.example.com', 'normal_option'),
1361
def test__get_option_policy_norecurse(self):
1362
self.get_branch_config('http://www.example.com')
1364
self.my_location_config._get_option_policy(
1365
'http://www.example.com', 'norecurse_option'),
1366
config.POLICY_NORECURSE)
1367
# Test old recurse=False setting:
1369
self.my_location_config._get_option_policy(
1370
'http://www.example.com/norecurse', 'normal_option'),
1371
config.POLICY_NORECURSE)
1373
def test__get_option_policy_normal(self):
1374
self.get_branch_config('http://www.example.com')
1376
self.my_location_config._get_option_policy(
1377
'http://www.example.com', 'appendpath_option'),
1378
config.POLICY_APPENDPATH)
1380
def test__get_options_with_policy(self):
1381
self.get_branch_config('/dir/subdir',
1382
location_config="""\
1384
other_url = /other-dir
1385
other_url:policy = appendpath
1387
other_url = /other-subdir
1390
[(u'other_url', u'/other-subdir', u'/dir/subdir', 'locations'),
1391
(u'other_url', u'/other-dir', u'/dir', 'locations'),
1392
(u'other_url:policy', u'appendpath', u'/dir', 'locations')],
1393
self.my_location_config)
1395
def test_location_without_username(self):
1396
self.get_branch_config('http://www.example.com/ignoreparent')
1397
self.assertEqual(u'Erik B\u00e5gfors <erik@bagfors.nu>',
1398
self.my_config.username())
1400
def test_location_not_listed(self):
1401
"""Test that the global username is used when no location matches"""
1402
self.get_branch_config('/home/robertc/sources')
1403
self.assertEqual(u'Erik B\u00e5gfors <erik@bagfors.nu>',
1404
self.my_config.username())
1406
def test_overriding_location(self):
1407
self.get_branch_config('http://www.example.com/foo')
1408
self.assertEqual('Robert Collins <robertc@example.org>',
1409
self.my_config.username())
1411
def test_get_user_option_global(self):
1412
self.get_branch_config('/a')
1413
self.assertEqual('something',
1414
self.my_config.get_user_option('user_global_option'))
1416
def test_get_user_option_local(self):
1417
self.get_branch_config('/a')
1418
self.assertEqual('local',
1419
self.my_config.get_user_option('user_local_option'))
1421
def test_get_user_option_appendpath(self):
1422
# returned as is for the base path:
1423
self.get_branch_config('http://www.example.com')
1424
self.assertEqual('append',
1425
self.my_config.get_user_option('appendpath_option'))
1426
# Extra path components get appended:
1427
self.get_branch_config('http://www.example.com/a/b/c')
1428
self.assertEqual('append/a/b/c',
1429
self.my_config.get_user_option('appendpath_option'))
1430
# Overriden for http://www.example.com/dir, where it is a
1432
self.get_branch_config('http://www.example.com/dir/a/b/c')
1433
self.assertEqual('normal',
1434
self.my_config.get_user_option('appendpath_option'))
1436
def test_get_user_option_norecurse(self):
1437
self.get_branch_config('http://www.example.com')
1438
self.assertEqual('norecurse',
1439
self.my_config.get_user_option('norecurse_option'))
1440
self.get_branch_config('http://www.example.com/dir')
1441
self.assertEqual(None,
1442
self.my_config.get_user_option('norecurse_option'))
1443
# http://www.example.com/norecurse is a recurse=False section
1444
# that redefines normal_option. Subdirectories do not pick up
1445
# this redefinition.
1446
self.get_branch_config('http://www.example.com/norecurse')
1447
self.assertEqual('norecurse',
1448
self.my_config.get_user_option('normal_option'))
1449
self.get_branch_config('http://www.example.com/norecurse/subdir')
1450
self.assertEqual('normal',
1451
self.my_config.get_user_option('normal_option'))
1453
def test_set_user_option_norecurse(self):
1454
self.get_branch_config('http://www.example.com')
1455
self.my_config.set_user_option('foo', 'bar',
1456
store=config.STORE_LOCATION_NORECURSE)
1458
self.my_location_config._get_option_policy(
1459
'http://www.example.com', 'foo'),
1460
config.POLICY_NORECURSE)
1462
def test_set_user_option_appendpath(self):
1463
self.get_branch_config('http://www.example.com')
1464
self.my_config.set_user_option('foo', 'bar',
1465
store=config.STORE_LOCATION_APPENDPATH)
1467
self.my_location_config._get_option_policy(
1468
'http://www.example.com', 'foo'),
1469
config.POLICY_APPENDPATH)
1471
def test_set_user_option_change_policy(self):
1472
self.get_branch_config('http://www.example.com')
1473
self.my_config.set_user_option('norecurse_option', 'normal',
1474
store=config.STORE_LOCATION)
1476
self.my_location_config._get_option_policy(
1477
'http://www.example.com', 'norecurse_option'),
1480
def get_branch_config(self, location, global_config=None,
1481
location_config=None):
1482
my_branch = FakeBranch(location)
1483
if global_config is None:
1484
global_config = sample_config_text
1485
if location_config is None:
1486
location_config = sample_branches_text
1488
config.GlobalConfig.from_string(global_config, save=True)
1489
config.LocationConfig.from_string(location_config, my_branch.base,
1491
my_config = config.BranchConfig(my_branch)
1492
self.my_config = my_config
1493
self.my_location_config = my_config._get_location_config()
1495
def test_set_user_setting_sets_and_saves2(self):
1496
self.get_branch_config('/a/c')
1497
self.assertIs(self.my_config.get_user_option('foo'), None)
1498
self.my_config.set_user_option('foo', 'bar')
1500
self.my_config.branch.control_files.files['branch.conf'].strip(),
1502
self.assertEqual(self.my_config.get_user_option('foo'), 'bar')
1503
self.my_config.set_user_option('foo', 'baz',
1504
store=config.STORE_LOCATION)
1505
self.assertEqual(self.my_config.get_user_option('foo'), 'baz')
1506
self.my_config.set_user_option('foo', 'qux')
1507
self.assertEqual(self.my_config.get_user_option('foo'), 'baz')
1509
def test_get_bzr_remote_path(self):
1510
my_config = config.LocationConfig('/a/c')
1511
self.assertEqual('bzr', my_config.get_bzr_remote_path())
1512
my_config.set_user_option('bzr_remote_path', '/path-bzr')
1513
self.assertEqual('/path-bzr', my_config.get_bzr_remote_path())
1514
self.overrideEnv('BZR_REMOTE_PATH', '/environ-bzr')
1515
self.assertEqual('/environ-bzr', my_config.get_bzr_remote_path())
1518
precedence_global = b'option = global'
1519
precedence_branch = b'option = branch'
1520
precedence_location = b"""
1524
[http://example.com/specific]
1529
class TestBranchConfigItems(tests.TestCaseInTempDir):
1531
def get_branch_config(self, global_config=None, location=None,
1532
location_config=None, branch_data_config=None):
1533
my_branch = FakeBranch(location)
1534
if global_config is not None:
1535
config.GlobalConfig.from_string(global_config, save=True)
1536
if location_config is not None:
1537
config.LocationConfig.from_string(location_config, my_branch.base,
1539
my_config = config.BranchConfig(my_branch)
1540
if branch_data_config is not None:
1541
my_config.branch.control_files.files['branch.conf'] = \
1545
def test_user_id(self):
1546
branch = FakeBranch()
1547
my_config = config.BranchConfig(branch)
1548
self.assertIsNot(None, my_config.username())
1549
my_config.branch.control_files.files['email'] = "John"
1550
my_config.set_user_option('email',
1551
"Robert Collins <robertc@example.org>")
1552
self.assertEqual("Robert Collins <robertc@example.org>",
1553
my_config.username())
1555
def test_BRZ_EMAIL_OVERRIDES(self):
1556
self.overrideEnv('BRZ_EMAIL', "Robert Collins <robertc@example.org>")
1557
branch = FakeBranch()
1558
my_config = config.BranchConfig(branch)
1559
self.assertEqual("Robert Collins <robertc@example.org>",
1560
my_config.username())
1562
def test_get_user_option_global(self):
1563
my_config = self.get_branch_config(global_config=sample_config_text)
1564
self.assertEqual('something',
1565
my_config.get_user_option('user_global_option'))
1567
def test_config_precedence(self):
1568
# FIXME: eager test, luckily no persitent config file makes it fail
1570
my_config = self.get_branch_config(global_config=precedence_global)
1571
self.assertEqual(my_config.get_user_option('option'), 'global')
1572
my_config = self.get_branch_config(global_config=precedence_global,
1573
branch_data_config=precedence_branch)
1574
self.assertEqual(my_config.get_user_option('option'), 'branch')
1575
my_config = self.get_branch_config(
1576
global_config=precedence_global,
1577
branch_data_config=precedence_branch,
1578
location_config=precedence_location)
1579
self.assertEqual(my_config.get_user_option('option'), 'recurse')
1580
my_config = self.get_branch_config(
1581
global_config=precedence_global,
1582
branch_data_config=precedence_branch,
1583
location_config=precedence_location,
1584
location='http://example.com/specific')
1585
self.assertEqual(my_config.get_user_option('option'), 'exact')
1588
class TestMailAddressExtraction(tests.TestCase):
1590
def test_extract_email_address(self):
1591
self.assertEqual('jane@test.com',
1592
config.extract_email_address('Jane <jane@test.com>'))
1593
self.assertRaises(config.NoEmailInUsername,
1594
config.extract_email_address, 'Jane Tester')
1596
def test_parse_username(self):
1597
self.assertEqual(('', 'jdoe@example.com'),
1598
config.parse_username('jdoe@example.com'))
1599
self.assertEqual(('', 'jdoe@example.com'),
1600
config.parse_username('<jdoe@example.com>'))
1601
self.assertEqual(('John Doe', 'jdoe@example.com'),
1602
config.parse_username('John Doe <jdoe@example.com>'))
1603
self.assertEqual(('John Doe', ''),
1604
config.parse_username('John Doe'))
1605
self.assertEqual(('John Doe', 'jdoe@example.com'),
1606
config.parse_username('John Doe jdoe@example.com'))
1609
class TestTreeConfig(tests.TestCaseWithTransport):
1611
def test_get_value(self):
1612
"""Test that retreiving a value from a section is possible"""
1613
branch = self.make_branch('.')
1614
tree_config = config.TreeConfig(branch)
1615
tree_config.set_option('value', 'key', 'SECTION')
1616
tree_config.set_option('value2', 'key2')
1617
tree_config.set_option('value3-top', 'key3')
1618
tree_config.set_option('value3-section', 'key3', 'SECTION')
1619
value = tree_config.get_option('key', 'SECTION')
1620
self.assertEqual(value, 'value')
1621
value = tree_config.get_option('key2')
1622
self.assertEqual(value, 'value2')
1623
self.assertEqual(tree_config.get_option('non-existant'), None)
1624
value = tree_config.get_option('non-existant', 'SECTION')
1625
self.assertEqual(value, None)
1626
value = tree_config.get_option('non-existant', default='default')
1627
self.assertEqual(value, 'default')
1628
self.assertEqual(tree_config.get_option('key2', 'NOSECTION'), None)
1629
value = tree_config.get_option('key2', 'NOSECTION', default='default')
1630
self.assertEqual(value, 'default')
1631
value = tree_config.get_option('key3')
1632
self.assertEqual(value, 'value3-top')
1633
value = tree_config.get_option('key3', 'SECTION')
1634
self.assertEqual(value, 'value3-section')
1637
class TestTransportConfig(tests.TestCaseWithTransport):
1639
def test_load_utf8(self):
1640
"""Ensure we can load an utf8-encoded file."""
1641
t = self.get_transport()
1642
unicode_user = u'b\N{Euro Sign}ar'
1643
unicode_content = u'user=%s' % (unicode_user,)
1644
utf8_content = unicode_content.encode('utf8')
1645
# Store the raw content in the config file
1646
t.put_bytes('foo.conf', utf8_content)
1647
conf = config.TransportConfig(t, 'foo.conf')
1648
self.assertEqual(unicode_user, conf.get_option('user'))
1650
def test_load_non_ascii(self):
1651
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
1652
t = self.get_transport()
1653
t.put_bytes('foo.conf', b'user=foo\n#\xff\n')
1654
conf = config.TransportConfig(t, 'foo.conf')
1655
self.assertRaises(config.ConfigContentError, conf._get_configobj)
1657
def test_load_erroneous_content(self):
1658
"""Ensure we display a proper error on content that can't be parsed."""
1659
t = self.get_transport()
1660
t.put_bytes('foo.conf', b'[open_section\n')
1661
conf = config.TransportConfig(t, 'foo.conf')
1662
self.assertRaises(config.ParseConfigError, conf._get_configobj)
1664
def test_load_permission_denied(self):
1665
"""Ensure we get an empty config file if the file is inaccessible."""
1669
warnings.append(args[0] % args[1:])
1670
self.overrideAttr(trace, 'warning', warning)
1672
class DenyingTransport(object):
1674
def __init__(self, base):
1677
def get_bytes(self, relpath):
1678
raise errors.PermissionDenied(relpath, "")
1680
cfg = config.TransportConfig(
1681
DenyingTransport("nonexisting://"), 'control.conf')
1682
self.assertIs(None, cfg.get_option('non-existant', 'SECTION'))
1685
[u'Permission denied while trying to open configuration file '
1686
u'nonexisting:///control.conf.'])
1688
def test_get_value(self):
1689
"""Test that retreiving a value from a section is possible"""
1690
bzrdir_config = config.TransportConfig(self.get_transport('.'),
1692
bzrdir_config.set_option('value', 'key', 'SECTION')
1693
bzrdir_config.set_option('value2', 'key2')
1694
bzrdir_config.set_option('value3-top', 'key3')
1695
bzrdir_config.set_option('value3-section', 'key3', 'SECTION')
1696
value = bzrdir_config.get_option('key', 'SECTION')
1697
self.assertEqual(value, 'value')
1698
value = bzrdir_config.get_option('key2')
1699
self.assertEqual(value, 'value2')
1700
self.assertEqual(bzrdir_config.get_option('non-existant'), None)
1701
value = bzrdir_config.get_option('non-existant', 'SECTION')
1702
self.assertEqual(value, None)
1703
value = bzrdir_config.get_option('non-existant', default='default')
1704
self.assertEqual(value, 'default')
1705
self.assertEqual(bzrdir_config.get_option('key2', 'NOSECTION'), None)
1706
value = bzrdir_config.get_option('key2', 'NOSECTION',
1708
self.assertEqual(value, 'default')
1709
value = bzrdir_config.get_option('key3')
1710
self.assertEqual(value, 'value3-top')
1711
value = bzrdir_config.get_option('key3', 'SECTION')
1712
self.assertEqual(value, 'value3-section')
1714
def test_set_unset_default_stack_on(self):
1715
my_dir = self.make_controldir('.')
1716
bzrdir_config = config.BzrDirConfig(my_dir)
1717
self.assertIs(None, bzrdir_config.get_default_stack_on())
1718
bzrdir_config.set_default_stack_on('Foo')
1719
self.assertEqual('Foo', bzrdir_config._config.get_option(
1720
'default_stack_on'))
1721
self.assertEqual('Foo', bzrdir_config.get_default_stack_on())
1722
bzrdir_config.set_default_stack_on(None)
1723
self.assertIs(None, bzrdir_config.get_default_stack_on())
1726
class TestOldConfigHooks(tests.TestCaseWithTransport):
1729
super(TestOldConfigHooks, self).setUp()
1730
create_configs_with_file_option(self)
1732
def assertGetHook(self, conf, name, value):
1737
config.OldConfigHooks.install_named_hook('get', hook, None)
1739
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1740
self.assertLength(0, calls)
1741
actual_value = conf.get_user_option(name)
1742
self.assertEqual(value, actual_value)
1743
self.assertLength(1, calls)
1744
self.assertEqual((conf, name, value), calls[0])
1746
def test_get_hook_breezy(self):
1747
self.assertGetHook(self.breezy_config, 'file', 'breezy')
1749
def test_get_hook_locations(self):
1750
self.assertGetHook(self.locations_config, 'file', 'locations')
1752
def test_get_hook_branch(self):
1753
# Since locations masks branch, we define a different option
1754
self.branch_config.set_user_option('file2', 'branch')
1755
self.assertGetHook(self.branch_config, 'file2', 'branch')
1757
def assertSetHook(self, conf, name, value):
1762
config.OldConfigHooks.install_named_hook('set', hook, None)
1764
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1765
self.assertLength(0, calls)
1766
conf.set_user_option(name, value)
1767
self.assertLength(1, calls)
1768
# We can't assert the conf object below as different configs use
1769
# different means to implement set_user_option and we care only about
1771
self.assertEqual((name, value), calls[0][1:])
1773
def test_set_hook_breezy(self):
1774
self.assertSetHook(self.breezy_config, 'foo', 'breezy')
1776
def test_set_hook_locations(self):
1777
self.assertSetHook(self.locations_config, 'foo', 'locations')
1779
def test_set_hook_branch(self):
1780
self.assertSetHook(self.branch_config, 'foo', 'branch')
1782
def assertRemoveHook(self, conf, name, section_name=None):
1787
config.OldConfigHooks.install_named_hook('remove', hook, None)
1789
config.OldConfigHooks.uninstall_named_hook, 'remove', None)
1790
self.assertLength(0, calls)
1791
conf.remove_user_option(name, section_name)
1792
self.assertLength(1, calls)
1793
# We can't assert the conf object below as different configs use
1794
# different means to implement remove_user_option and we care only about
1796
self.assertEqual((name,), calls[0][1:])
1798
def test_remove_hook_breezy(self):
1799
self.assertRemoveHook(self.breezy_config, 'file')
1801
def test_remove_hook_locations(self):
1802
self.assertRemoveHook(self.locations_config, 'file',
1803
self.locations_config.location)
1805
def test_remove_hook_branch(self):
1806
self.assertRemoveHook(self.branch_config, 'file')
1808
def assertLoadHook(self, name, conf_class, *conf_args):
1813
config.OldConfigHooks.install_named_hook('load', hook, None)
1815
config.OldConfigHooks.uninstall_named_hook, 'load', None)
1816
self.assertLength(0, calls)
1818
conf = conf_class(*conf_args)
1819
# Access an option to trigger a load
1820
conf.get_user_option(name)
1821
self.assertLength(1, calls)
1822
# Since we can't assert about conf, we just use the number of calls ;-/
1824
def test_load_hook_breezy(self):
1825
self.assertLoadHook('file', config.GlobalConfig)
1827
def test_load_hook_locations(self):
1828
self.assertLoadHook('file', config.LocationConfig, self.tree.basedir)
1830
def test_load_hook_branch(self):
1831
self.assertLoadHook('file', config.BranchConfig, self.tree.branch)
1833
def assertSaveHook(self, conf):
1838
config.OldConfigHooks.install_named_hook('save', hook, None)
1840
config.OldConfigHooks.uninstall_named_hook, 'save', None)
1841
self.assertLength(0, calls)
1842
# Setting an option triggers a save
1843
conf.set_user_option('foo', 'bar')
1844
self.assertLength(1, calls)
1845
# Since we can't assert about conf, we just use the number of calls ;-/
1847
def test_save_hook_breezy(self):
1848
self.assertSaveHook(self.breezy_config)
1850
def test_save_hook_locations(self):
1851
self.assertSaveHook(self.locations_config)
1853
def test_save_hook_branch(self):
1854
self.assertSaveHook(self.branch_config)
1857
class TestOldConfigHooksForRemote(tests.TestCaseWithTransport):
1858
"""Tests config hooks for remote configs.
1860
No tests for the remove hook as this is not implemented there.
1864
super(TestOldConfigHooksForRemote, self).setUp()
1865
self.transport_server = test_server.SmartTCPServer_for_testing
1866
create_configs_with_file_option(self)
1868
def assertGetHook(self, conf, name, value):
1873
config.OldConfigHooks.install_named_hook('get', hook, None)
1875
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1876
self.assertLength(0, calls)
1877
actual_value = conf.get_option(name)
1878
self.assertEqual(value, actual_value)
1879
self.assertLength(1, calls)
1880
self.assertEqual((conf, name, value), calls[0])
1882
def test_get_hook_remote_branch(self):
1883
remote_branch = branch.Branch.open(self.get_url('tree'))
1884
self.assertGetHook(remote_branch._get_config(), 'file', 'branch')
1886
def test_get_hook_remote_bzrdir(self):
1887
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1888
conf = remote_bzrdir._get_config()
1889
conf.set_option('remotedir', 'file')
1890
self.assertGetHook(conf, 'file', 'remotedir')
1892
def assertSetHook(self, conf, name, value):
1897
config.OldConfigHooks.install_named_hook('set', hook, None)
1899
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1900
self.assertLength(0, calls)
1901
conf.set_option(value, name)
1902
self.assertLength(1, calls)
1903
# We can't assert the conf object below as different configs use
1904
# different means to implement set_user_option and we care only about
1906
self.assertEqual((name, value), calls[0][1:])
1908
def test_set_hook_remote_branch(self):
1909
remote_branch = branch.Branch.open(self.get_url('tree'))
1910
self.addCleanup(remote_branch.lock_write().unlock)
1911
self.assertSetHook(remote_branch._get_config(), 'file', 'remote')
1913
def test_set_hook_remote_bzrdir(self):
1914
remote_branch = branch.Branch.open(self.get_url('tree'))
1915
self.addCleanup(remote_branch.lock_write().unlock)
1916
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1917
self.assertSetHook(remote_bzrdir._get_config(), 'file', 'remotedir')
1919
def assertLoadHook(self, expected_nb_calls, name, conf_class, *conf_args):
1924
config.OldConfigHooks.install_named_hook('load', hook, None)
1926
config.OldConfigHooks.uninstall_named_hook, 'load', None)
1927
self.assertLength(0, calls)
1929
conf = conf_class(*conf_args)
1930
# Access an option to trigger a load
1931
conf.get_option(name)
1932
self.assertLength(expected_nb_calls, calls)
1933
# Since we can't assert about conf, we just use the number of calls ;-/
1935
def test_load_hook_remote_branch(self):
1936
remote_branch = branch.Branch.open(self.get_url('tree'))
1937
self.assertLoadHook(
1938
1, 'file', remote.RemoteBranchConfig, remote_branch)
1940
def test_load_hook_remote_bzrdir(self):
1941
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1942
# The config file doesn't exist, set an option to force its creation
1943
conf = remote_bzrdir._get_config()
1944
conf.set_option('remotedir', 'file')
1945
# We get one call for the server and one call for the client, this is
1946
# caused by the differences in implementations betwen
1947
# SmartServerBzrDirRequestConfigFile (in smart/bzrdir.py) and
1948
# SmartServerBranchGetConfigFile (in smart/branch.py)
1949
self.assertLoadHook(
1950
2, 'file', remote.RemoteBzrDirConfig, remote_bzrdir)
1952
def assertSaveHook(self, conf):
1957
config.OldConfigHooks.install_named_hook('save', hook, None)
1959
config.OldConfigHooks.uninstall_named_hook, 'save', None)
1960
self.assertLength(0, calls)
1961
# Setting an option triggers a save
1962
conf.set_option('foo', 'bar')
1963
self.assertLength(1, calls)
1964
# Since we can't assert about conf, we just use the number of calls ;-/
1966
def test_save_hook_remote_branch(self):
1967
remote_branch = branch.Branch.open(self.get_url('tree'))
1968
self.addCleanup(remote_branch.lock_write().unlock)
1969
self.assertSaveHook(remote_branch._get_config())
1971
def test_save_hook_remote_bzrdir(self):
1972
remote_branch = branch.Branch.open(self.get_url('tree'))
1973
self.addCleanup(remote_branch.lock_write().unlock)
1974
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1975
self.assertSaveHook(remote_bzrdir._get_config())
1978
class TestOptionNames(tests.TestCase):
1980
def is_valid(self, name):
1981
return config._option_ref_re.match('{%s}' % name) is not None
1983
def test_valid_names(self):
1984
self.assertTrue(self.is_valid('foo'))
1985
self.assertTrue(self.is_valid('foo.bar'))
1986
self.assertTrue(self.is_valid('f1'))
1987
self.assertTrue(self.is_valid('_'))
1988
self.assertTrue(self.is_valid('__bar__'))
1989
self.assertTrue(self.is_valid('a_'))
1990
self.assertTrue(self.is_valid('a1'))
1991
# Don't break bzr-svn for no good reason
1992
self.assertTrue(self.is_valid('guessed-layout'))
1994
def test_invalid_names(self):
1995
self.assertFalse(self.is_valid(' foo'))
1996
self.assertFalse(self.is_valid('foo '))
1997
self.assertFalse(self.is_valid('1'))
1998
self.assertFalse(self.is_valid('1,2'))
1999
self.assertFalse(self.is_valid('foo$'))
2000
self.assertFalse(self.is_valid('!foo'))
2001
self.assertFalse(self.is_valid('foo.'))
2002
self.assertFalse(self.is_valid('foo..bar'))
2003
self.assertFalse(self.is_valid('{}'))
2004
self.assertFalse(self.is_valid('{a}'))
2005
self.assertFalse(self.is_valid('a\n'))
2006
self.assertFalse(self.is_valid('-'))
2007
self.assertFalse(self.is_valid('-a'))
2008
self.assertFalse(self.is_valid('a-'))
2009
self.assertFalse(self.is_valid('a--a'))
2011
def assertSingleGroup(self, reference):
2012
# the regexp is used with split and as such should match the reference
2013
# *only*, if more groups needs to be defined, (?:...) should be used.
2014
m = config._option_ref_re.match('{a}')
2015
self.assertLength(1, m.groups())
2017
def test_valid_references(self):
2018
self.assertSingleGroup('{a}')
2019
self.assertSingleGroup('{{a}}')
2022
class TestOption(tests.TestCase):
2024
def test_default_value(self):
2025
opt = config.Option('foo', default='bar')
2026
self.assertEqual('bar', opt.get_default())
2028
def test_callable_default_value(self):
2029
def bar_as_unicode():
2031
opt = config.Option('foo', default=bar_as_unicode)
2032
self.assertEqual('bar', opt.get_default())
2034
def test_default_value_from_env(self):
2035
opt = config.Option('foo', default='bar', default_from_env=['FOO'])
2036
self.overrideEnv('FOO', 'quux')
2037
# Env variable provides a default taking over the option one
2038
self.assertEqual('quux', opt.get_default())
2040
def test_first_default_value_from_env_wins(self):
2041
opt = config.Option('foo', default='bar',
2042
default_from_env=['NO_VALUE', 'FOO', 'BAZ'])
2043
self.overrideEnv('FOO', 'foo')
2044
self.overrideEnv('BAZ', 'baz')
2045
# The first env var set wins
2046
self.assertEqual('foo', opt.get_default())
2048
def test_not_supported_list_default_value(self):
2049
self.assertRaises(AssertionError, config.Option, 'foo', default=[1])
2051
def test_not_supported_object_default_value(self):
2052
self.assertRaises(AssertionError, config.Option, 'foo',
2055
def test_not_supported_callable_default_value_not_unicode(self):
2056
def bar_not_unicode():
2058
opt = config.Option('foo', default=bar_not_unicode)
2059
self.assertRaises(AssertionError, opt.get_default)
2061
def test_get_help_topic(self):
2062
opt = config.Option('foo')
2063
self.assertEqual('foo', opt.get_help_topic())
2066
class TestOptionConverter(tests.TestCase):
2068
def assertConverted(self, expected, opt, value):
2069
self.assertEqual(expected, opt.convert_from_unicode(None, value))
2071
def assertCallsWarning(self, opt, value):
2075
warnings.append(args[0] % args[1:])
2076
self.overrideAttr(trace, 'warning', warning)
2077
self.assertEqual(None, opt.convert_from_unicode(None, value))
2078
self.assertLength(1, warnings)
2080
'Value "%s" is not valid for "%s"' % (value, opt.name),
2083
def assertCallsError(self, opt, value):
2084
self.assertRaises(config.ConfigOptionValueError,
2085
opt.convert_from_unicode, None, value)
2087
def assertConvertInvalid(self, opt, invalid_value):
2089
self.assertEqual(None, opt.convert_from_unicode(None, invalid_value))
2090
opt.invalid = 'warning'
2091
self.assertCallsWarning(opt, invalid_value)
2092
opt.invalid = 'error'
2093
self.assertCallsError(opt, invalid_value)
2096
class TestOptionWithBooleanConverter(TestOptionConverter):
2098
def get_option(self):
2099
return config.Option('foo', help='A boolean.',
2100
from_unicode=config.bool_from_store)
2102
def test_convert_invalid(self):
2103
opt = self.get_option()
2104
# A string that is not recognized as a boolean
2105
self.assertConvertInvalid(opt, u'invalid-boolean')
2106
# A list of strings is never recognized as a boolean
2107
self.assertConvertInvalid(opt, [u'not', u'a', u'boolean'])
2109
def test_convert_valid(self):
2110
opt = self.get_option()
2111
self.assertConverted(True, opt, u'True')
2112
self.assertConverted(True, opt, u'1')
2113
self.assertConverted(False, opt, u'False')
2116
class TestOptionWithIntegerConverter(TestOptionConverter):
2118
def get_option(self):
2119
return config.Option('foo', help='An integer.',
2120
from_unicode=config.int_from_store)
2122
def test_convert_invalid(self):
2123
opt = self.get_option()
2124
# A string that is not recognized as an integer
2125
self.assertConvertInvalid(opt, u'forty-two')
2126
# A list of strings is never recognized as an integer
2127
self.assertConvertInvalid(opt, [u'a', u'list'])
2129
def test_convert_valid(self):
2130
opt = self.get_option()
2131
self.assertConverted(16, opt, u'16')
2134
class TestOptionWithSIUnitConverter(TestOptionConverter):
2136
def get_option(self):
2137
return config.Option('foo', help='An integer in SI units.',
2138
from_unicode=config.int_SI_from_store)
2140
def test_convert_invalid(self):
2141
opt = self.get_option()
2142
self.assertConvertInvalid(opt, u'not-a-unit')
2143
self.assertConvertInvalid(opt, u'Gb') # Forgot the value
2144
self.assertConvertInvalid(opt, u'1b') # Forgot the unit
2145
self.assertConvertInvalid(opt, u'1GG')
2146
self.assertConvertInvalid(opt, u'1Mbb')
2147
self.assertConvertInvalid(opt, u'1MM')
2149
def test_convert_valid(self):
2150
opt = self.get_option()
2151
self.assertConverted(int(5e3), opt, u'5kb')
2152
self.assertConverted(int(5e6), opt, u'5M')
2153
self.assertConverted(int(5e6), opt, u'5MB')
2154
self.assertConverted(int(5e9), opt, u'5g')
2155
self.assertConverted(int(5e9), opt, u'5gB')
2156
self.assertConverted(100, opt, u'100')
2159
class TestListOption(TestOptionConverter):
2161
def get_option(self):
2162
return config.ListOption('foo', help='A list.')
2164
def test_convert_invalid(self):
2165
opt = self.get_option()
2166
# We don't even try to convert a list into a list, we only expect
2168
self.assertConvertInvalid(opt, [1])
2169
# No string is invalid as all forms can be converted to a list
2171
def test_convert_valid(self):
2172
opt = self.get_option()
2173
# An empty string is an empty list
2174
self.assertConverted([], opt, '') # Using a bare str() just in case
2175
self.assertConverted([], opt, u'')
2177
self.assertConverted([u'True'], opt, u'True')
2179
self.assertConverted([u'42'], opt, u'42')
2181
self.assertConverted([u'bar'], opt, u'bar')
2184
class TestRegistryOption(TestOptionConverter):
2186
def get_option(self, registry):
2187
return config.RegistryOption('foo', registry,
2188
help='A registry option.')
2190
def test_convert_invalid(self):
2191
registry = _mod_registry.Registry()
2192
opt = self.get_option(registry)
2193
self.assertConvertInvalid(opt, [1])
2194
self.assertConvertInvalid(opt, u"notregistered")
2196
def test_convert_valid(self):
2197
registry = _mod_registry.Registry()
2198
registry.register("someval", 1234)
2199
opt = self.get_option(registry)
2200
# Using a bare str() just in case
2201
self.assertConverted(1234, opt, "someval")
2202
self.assertConverted(1234, opt, u'someval')
2203
self.assertConverted(None, opt, None)
2205
def test_help(self):
2206
registry = _mod_registry.Registry()
2207
registry.register("someval", 1234, help="some option")
2208
registry.register("dunno", 1234, help="some other option")
2209
opt = self.get_option(registry)
2211
'A registry option.\n'
2213
'The following values are supported:\n'
2214
' dunno - some other option\n'
2215
' someval - some option\n',
2218
def test_get_help_text(self):
2219
registry = _mod_registry.Registry()
2220
registry.register("someval", 1234, help="some option")
2221
registry.register("dunno", 1234, help="some other option")
2222
opt = self.get_option(registry)
2224
'A registry option.\n'
2226
'The following values are supported:\n'
2227
' dunno - some other option\n'
2228
' someval - some option\n',
2229
opt.get_help_text())
2232
class TestOptionRegistry(tests.TestCase):
2235
super(TestOptionRegistry, self).setUp()
2236
# Always start with an empty registry
2237
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
2238
self.registry = config.option_registry
2240
def test_register(self):
2241
opt = config.Option('foo')
2242
self.registry.register(opt)
2243
self.assertIs(opt, self.registry.get('foo'))
2245
def test_registered_help(self):
2246
opt = config.Option('foo', help='A simple option')
2247
self.registry.register(opt)
2248
self.assertEqual('A simple option', self.registry.get_help('foo'))
2250
def test_dont_register_illegal_name(self):
2251
self.assertRaises(config.IllegalOptionName,
2252
self.registry.register, config.Option(' foo'))
2253
self.assertRaises(config.IllegalOptionName,
2254
self.registry.register, config.Option('bar,'))
2256
lazy_option = config.Option('lazy_foo', help='Lazy help')
2258
def test_register_lazy(self):
2259
self.registry.register_lazy('lazy_foo', self.__module__,
2260
'TestOptionRegistry.lazy_option')
2261
self.assertIs(self.lazy_option, self.registry.get('lazy_foo'))
2263
def test_registered_lazy_help(self):
2264
self.registry.register_lazy('lazy_foo', self.__module__,
2265
'TestOptionRegistry.lazy_option')
2266
self.assertEqual('Lazy help', self.registry.get_help('lazy_foo'))
2268
def test_dont_lazy_register_illegal_name(self):
2269
# This is where the root cause of http://pad.lv/1235099 is better
2270
# understood: 'register_lazy' doc string mentions that key should match
2271
# the option name which indirectly requires that the option name is a
2272
# valid python identifier. We violate that rule here (using a key that
2273
# doesn't match the option name) to test the option name checking.
2274
self.assertRaises(config.IllegalOptionName,
2275
self.registry.register_lazy, ' foo', self.__module__,
2276
'TestOptionRegistry.lazy_option')
2277
self.assertRaises(config.IllegalOptionName,
2278
self.registry.register_lazy, '1,2', self.__module__,
2279
'TestOptionRegistry.lazy_option')
2282
class TestRegisteredOptions(tests.TestCase):
2283
"""All registered options should verify some constraints."""
2285
scenarios = [(key, {'option_name': key, 'option': option}) for key, option
2286
in config.option_registry.iteritems()]
2289
super(TestRegisteredOptions, self).setUp()
2290
self.registry = config.option_registry
2292
def test_proper_name(self):
2293
# An option should be registered under its own name, this can't be
2294
# checked at registration time for the lazy ones.
2295
self.assertEqual(self.option_name, self.option.name)
2297
def test_help_is_set(self):
2298
option_help = self.registry.get_help(self.option_name)
2299
# Come on, think about the user, he really wants to know what the
2301
self.assertIsNot(None, option_help)
2302
self.assertNotEqual('', option_help)
2305
class TestSection(tests.TestCase):
2307
# FIXME: Parametrize so that all sections produced by Stores run these
2308
# tests -- vila 2011-04-01
2310
def test_get_a_value(self):
2311
a_dict = dict(foo='bar')
2312
section = config.Section('myID', a_dict)
2313
self.assertEqual('bar', section.get('foo'))
2315
def test_get_unknown_option(self):
2317
section = config.Section(None, a_dict)
2318
self.assertEqual('out of thin air',
2319
section.get('foo', 'out of thin air'))
2321
def test_options_is_shared(self):
2323
section = config.Section(None, a_dict)
2324
self.assertIs(a_dict, section.options)
2327
class TestMutableSection(tests.TestCase):
2329
scenarios = [('mutable',
2331
lambda opts: config.MutableSection('myID', opts)},),
2335
a_dict = dict(foo='bar')
2336
section = self.get_section(a_dict)
2337
section.set('foo', 'new_value')
2338
self.assertEqual('new_value', section.get('foo'))
2339
# The change appears in the shared section
2340
self.assertEqual('new_value', a_dict.get('foo'))
2341
# We keep track of the change
2342
self.assertTrue('foo' in section.orig)
2343
self.assertEqual('bar', section.orig.get('foo'))
2345
def test_set_preserve_original_once(self):
2346
a_dict = dict(foo='bar')
2347
section = self.get_section(a_dict)
2348
section.set('foo', 'first_value')
2349
section.set('foo', 'second_value')
2350
# We keep track of the original value
2351
self.assertTrue('foo' in section.orig)
2352
self.assertEqual('bar', section.orig.get('foo'))
2354
def test_remove(self):
2355
a_dict = dict(foo='bar')
2356
section = self.get_section(a_dict)
2357
section.remove('foo')
2358
# We get None for unknown options via the default value
2359
self.assertEqual(None, section.get('foo'))
2360
# Or we just get the default value
2361
self.assertEqual('unknown', section.get('foo', 'unknown'))
2362
self.assertFalse('foo' in section.options)
2363
# We keep track of the deletion
2364
self.assertTrue('foo' in section.orig)
2365
self.assertEqual('bar', section.orig.get('foo'))
2367
def test_remove_new_option(self):
2369
section = self.get_section(a_dict)
2370
section.set('foo', 'bar')
2371
section.remove('foo')
2372
self.assertFalse('foo' in section.options)
2373
# The option didn't exist initially so it we need to keep track of it
2374
# with a special value
2375
self.assertTrue('foo' in section.orig)
2376
self.assertEqual(config._NewlyCreatedOption, section.orig['foo'])
2379
class TestCommandLineStore(tests.TestCase):
2382
super(TestCommandLineStore, self).setUp()
2383
self.store = config.CommandLineStore()
2384
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
2386
def get_section(self):
2387
"""Get the unique section for the command line overrides."""
2388
sections = list(self.store.get_sections())
2389
self.assertLength(1, sections)
2390
store, section = sections[0]
2391
self.assertEqual(self.store, store)
2394
def test_no_override(self):
2395
self.store._from_cmdline([])
2396
section = self.get_section()
2397
self.assertLength(0, list(section.iter_option_names()))
2399
def test_simple_override(self):
2400
self.store._from_cmdline(['a=b'])
2401
section = self.get_section()
2402
self.assertEqual('b', section.get('a'))
2404
def test_list_override(self):
2405
opt = config.ListOption('l')
2406
config.option_registry.register(opt)
2407
self.store._from_cmdline(['l=1,2,3'])
2408
val = self.get_section().get('l')
2409
self.assertEqual('1,2,3', val)
2410
# Reminder: lists should be registered as such explicitely, otherwise
2411
# the conversion needs to be done afterwards.
2412
self.assertEqual(['1', '2', '3'],
2413
opt.convert_from_unicode(self.store, val))
2415
def test_multiple_overrides(self):
2416
self.store._from_cmdline(['a=b', 'x=y'])
2417
section = self.get_section()
2418
self.assertEqual('b', section.get('a'))
2419
self.assertEqual('y', section.get('x'))
2421
def test_wrong_syntax(self):
2422
self.assertRaises(errors.BzrCommandError,
2423
self.store._from_cmdline, ['a=b', 'c'])
2426
class TestStoreMinimalAPI(tests.TestCaseWithTransport):
2428
scenarios = [(key, {'get_store': builder}) for key, builder
2429
in config.test_store_builder_registry.iteritems()] + [
2430
('cmdline', {'get_store': lambda test: config.CommandLineStore()})]
2433
store = self.get_store(self)
2434
if isinstance(store, config.TransportIniFileStore):
2435
raise tests.TestNotApplicable(
2436
"%s is not a concrete Store implementation"
2437
" so it doesn't need an id" % (store.__class__.__name__,))
2438
self.assertIsNot(None, store.id)
2441
class TestStore(tests.TestCaseWithTransport):
2443
def assertSectionContent(self, expected, store_and_section):
2444
"""Assert that some options have the proper values in a section."""
2445
_, section = store_and_section
2446
expected_name, expected_options = expected
2447
self.assertEqual(expected_name, section.id)
2450
dict([(k, section.get(k)) for k in expected_options.keys()]))
2453
class TestReadonlyStore(TestStore):
2455
scenarios = [(key, {'get_store': builder}) for key, builder
2456
in config.test_store_builder_registry.iteritems()]
2458
def test_building_delays_load(self):
2459
store = self.get_store(self)
2460
self.assertEqual(False, store.is_loaded())
2461
store._load_from_string(b'')
2462
self.assertEqual(True, store.is_loaded())
2464
def test_get_no_sections_for_empty(self):
2465
store = self.get_store(self)
2466
store._load_from_string(b'')
2467
self.assertEqual([], list(store.get_sections()))
2469
def test_get_default_section(self):
2470
store = self.get_store(self)
2471
store._load_from_string(b'foo=bar')
2472
sections = list(store.get_sections())
2473
self.assertLength(1, sections)
2474
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2476
def test_get_named_section(self):
2477
store = self.get_store(self)
2478
store._load_from_string(b'[baz]\nfoo=bar')
2479
sections = list(store.get_sections())
2480
self.assertLength(1, sections)
2481
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2483
def test_load_from_string_fails_for_non_empty_store(self):
2484
store = self.get_store(self)
2485
store._load_from_string(b'foo=bar')
2486
self.assertRaises(AssertionError, store._load_from_string, b'bar=baz')
2489
class TestStoreQuoting(TestStore):
2491
scenarios = [(key, {'get_store': builder}) for key, builder
2492
in config.test_store_builder_registry.iteritems()]
2495
super(TestStoreQuoting, self).setUp()
2496
self.store = self.get_store(self)
2497
# We need a loaded store but any content will do
2498
self.store._load_from_string(b'')
2500
def assertIdempotent(self, s):
2501
"""Assert that quoting an unquoted string is a no-op and vice-versa.
2503
What matters here is that option values, as they appear in a store, can
2504
be safely round-tripped out of the store and back.
2506
:param s: A string, quoted if required.
2508
self.assertEqual(s, self.store.quote(self.store.unquote(s)))
2509
self.assertEqual(s, self.store.unquote(self.store.quote(s)))
2511
def test_empty_string(self):
2512
if isinstance(self.store, config.IniFileStore):
2513
# configobj._quote doesn't handle empty values
2514
self.assertRaises(AssertionError,
2515
self.assertIdempotent, '')
2517
self.assertIdempotent('')
2518
# But quoted empty strings are ok
2519
self.assertIdempotent('""')
2521
def test_embedded_spaces(self):
2522
self.assertIdempotent('" a b c "')
2524
def test_embedded_commas(self):
2525
self.assertIdempotent('" a , b c "')
2527
def test_simple_comma(self):
2528
if isinstance(self.store, config.IniFileStore):
2529
# configobj requires that lists are special-cased
2530
self.assertRaises(AssertionError,
2531
self.assertIdempotent, ',')
2533
self.assertIdempotent(',')
2534
# When a single comma is required, quoting is also required
2535
self.assertIdempotent('","')
2537
def test_list(self):
2538
if isinstance(self.store, config.IniFileStore):
2539
# configobj requires that lists are special-cased
2540
self.assertRaises(AssertionError,
2541
self.assertIdempotent, 'a,b')
2543
self.assertIdempotent('a,b')
2546
class TestDictFromStore(tests.TestCase):
2548
def test_unquote_not_string(self):
2549
conf = config.MemoryStack(b'x=2\n[a_section]\na=1\n')
2550
value = conf.get('a_section')
2551
# Urgh, despite 'conf' asking for the no-name section, we get the
2552
# content of another section as a dict o_O
2553
self.assertEqual({'a': '1'}, value)
2554
unquoted = conf.store.unquote(value)
2555
# Which cannot be unquoted but shouldn't crash either (the use cases
2556
# are getting the value or displaying it. In the later case, '%s' will
2558
self.assertEqual({'a': '1'}, unquoted)
2559
self.assertIn('%s' % (unquoted,), ("{u'a': u'1'}", "{'a': '1'}"))
2562
class TestIniFileStoreContent(tests.TestCaseWithTransport):
2563
"""Simulate loading a config store with content of various encodings.
2565
All files produced by bzr are in utf8 content.
2567
Users may modify them manually and end up with a file that can't be
2568
loaded. We need to issue proper error messages in this case.
2571
invalid_utf8_char = b'\xff'
2573
def test_load_utf8(self):
2574
"""Ensure we can load an utf8-encoded file."""
2575
t = self.get_transport()
2576
# From http://pad.lv/799212
2577
unicode_user = u'b\N{Euro Sign}ar'
2578
unicode_content = u'user=%s' % (unicode_user,)
2579
utf8_content = unicode_content.encode('utf8')
2580
# Store the raw content in the config file
2581
t.put_bytes('foo.conf', utf8_content)
2582
store = config.TransportIniFileStore(t, 'foo.conf')
2584
stack = config.Stack([store.get_sections], store)
2585
self.assertEqual(unicode_user, stack.get('user'))
2587
def test_load_non_ascii(self):
2588
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2589
t = self.get_transport()
2590
t.put_bytes('foo.conf', b'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2591
store = config.TransportIniFileStore(t, 'foo.conf')
2592
self.assertRaises(config.ConfigContentError, store.load)
2594
def test_load_erroneous_content(self):
2595
"""Ensure we display a proper error on content that can't be parsed."""
2596
t = self.get_transport()
2597
t.put_bytes('foo.conf', b'[open_section\n')
2598
store = config.TransportIniFileStore(t, 'foo.conf')
2599
self.assertRaises(config.ParseConfigError, store.load)
2601
def test_load_permission_denied(self):
2602
"""Ensure we get warned when trying to load an inaccessible file."""
2606
warnings.append(args[0] % args[1:])
2607
self.overrideAttr(trace, 'warning', warning)
2609
t = self.get_transport()
2611
def get_bytes(relpath):
2612
raise errors.PermissionDenied(relpath, "")
2613
t.get_bytes = get_bytes
2614
store = config.TransportIniFileStore(t, 'foo.conf')
2615
self.assertRaises(errors.PermissionDenied, store.load)
2618
[u'Permission denied while trying to load configuration store %s.'
2619
% store.external_url()])
2622
class TestIniConfigContent(tests.TestCaseWithTransport):
2623
"""Simulate loading a IniBasedConfig with content of various encodings.
2625
All files produced by bzr are in utf8 content.
2627
Users may modify them manually and end up with a file that can't be
2628
loaded. We need to issue proper error messages in this case.
2631
invalid_utf8_char = b'\xff'
2633
def test_load_utf8(self):
2634
"""Ensure we can load an utf8-encoded file."""
2635
# From http://pad.lv/799212
2636
unicode_user = u'b\N{Euro Sign}ar'
2637
unicode_content = u'user=%s' % (unicode_user,)
2638
utf8_content = unicode_content.encode('utf8')
2639
# Store the raw content in the config file
2640
with open('foo.conf', 'wb') as f:
2641
f.write(utf8_content)
2642
conf = config.IniBasedConfig(file_name='foo.conf')
2643
self.assertEqual(unicode_user, conf.get_user_option('user'))
2645
def test_load_badly_encoded_content(self):
2646
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2647
with open('foo.conf', 'wb') as f:
2648
f.write(b'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2649
conf = config.IniBasedConfig(file_name='foo.conf')
2650
self.assertRaises(config.ConfigContentError, conf._get_parser)
2652
def test_load_erroneous_content(self):
2653
"""Ensure we display a proper error on content that can't be parsed."""
2654
with open('foo.conf', 'wb') as f:
2655
f.write(b'[open_section\n')
2656
conf = config.IniBasedConfig(file_name='foo.conf')
2657
self.assertRaises(config.ParseConfigError, conf._get_parser)
2660
class TestMutableStore(TestStore):
2662
scenarios = [(key, {'store_id': key, 'get_store': builder}) for key, builder
2663
in config.test_store_builder_registry.iteritems()]
2666
super(TestMutableStore, self).setUp()
2667
self.transport = self.get_transport()
2669
def has_store(self, store):
2670
store_basename = urlutils.relative_url(self.transport.external_url(),
2671
store.external_url())
2672
return self.transport.has(store_basename)
2674
def test_save_empty_creates_no_file(self):
2675
# FIXME: There should be a better way than relying on the test
2676
# parametrization to identify branch.conf -- vila 2011-0526
2677
if self.store_id in ('branch', 'remote_branch'):
2678
raise tests.TestNotApplicable(
2679
'branch.conf is *always* created when a branch is initialized')
2680
store = self.get_store(self)
2682
self.assertEqual(False, self.has_store(store))
2684
def test_mutable_section_shared(self):
2685
store = self.get_store(self)
2686
store._load_from_string(b'foo=bar\n')
2687
# FIXME: There should be a better way than relying on the test
2688
# parametrization to identify branch.conf -- vila 2011-0526
2689
if self.store_id in ('branch', 'remote_branch'):
2690
# branch stores requires write locked branches
2691
self.addCleanup(store.branch.lock_write().unlock)
2692
section1 = store.get_mutable_section(None)
2693
section2 = store.get_mutable_section(None)
2694
# If we get different sections, different callers won't share the
2696
self.assertIs(section1, section2)
2698
def test_save_emptied_succeeds(self):
2699
store = self.get_store(self)
2700
store._load_from_string(b'foo=bar\n')
2701
# FIXME: There should be a better way than relying on the test
2702
# parametrization to identify branch.conf -- vila 2011-0526
2703
if self.store_id in ('branch', 'remote_branch'):
2704
# branch stores requires write locked branches
2705
self.addCleanup(store.branch.lock_write().unlock)
2706
section = store.get_mutable_section(None)
2707
section.remove('foo')
2709
self.assertEqual(True, self.has_store(store))
2710
modified_store = self.get_store(self)
2711
sections = list(modified_store.get_sections())
2712
self.assertLength(0, sections)
2714
def test_save_with_content_succeeds(self):
2715
# FIXME: There should be a better way than relying on the test
2716
# parametrization to identify branch.conf -- vila 2011-0526
2717
if self.store_id in ('branch', 'remote_branch'):
2718
raise tests.TestNotApplicable(
2719
'branch.conf is *always* created when a branch is initialized')
2720
store = self.get_store(self)
2721
store._load_from_string(b'foo=bar\n')
2722
self.assertEqual(False, self.has_store(store))
2724
self.assertEqual(True, self.has_store(store))
2725
modified_store = self.get_store(self)
2726
sections = list(modified_store.get_sections())
2727
self.assertLength(1, sections)
2728
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2730
def test_set_option_in_empty_store(self):
2731
store = self.get_store(self)
2732
# FIXME: There should be a better way than relying on the test
2733
# parametrization to identify branch.conf -- vila 2011-0526
2734
if self.store_id in ('branch', 'remote_branch'):
2735
# branch stores requires write locked branches
2736
self.addCleanup(store.branch.lock_write().unlock)
2737
section = store.get_mutable_section(None)
2738
section.set('foo', 'bar')
2740
modified_store = self.get_store(self)
2741
sections = list(modified_store.get_sections())
2742
self.assertLength(1, sections)
2743
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2745
def test_set_option_in_default_section(self):
2746
store = self.get_store(self)
2747
store._load_from_string(b'')
2748
# FIXME: There should be a better way than relying on the test
2749
# parametrization to identify branch.conf -- vila 2011-0526
2750
if self.store_id in ('branch', 'remote_branch'):
2751
# branch stores requires write locked branches
2752
self.addCleanup(store.branch.lock_write().unlock)
2753
section = store.get_mutable_section(None)
2754
section.set('foo', 'bar')
2756
modified_store = self.get_store(self)
2757
sections = list(modified_store.get_sections())
2758
self.assertLength(1, sections)
2759
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2761
def test_set_option_in_named_section(self):
2762
store = self.get_store(self)
2763
store._load_from_string(b'')
2764
# FIXME: There should be a better way than relying on the test
2765
# parametrization to identify branch.conf -- vila 2011-0526
2766
if self.store_id in ('branch', 'remote_branch'):
2767
# branch stores requires write locked branches
2768
self.addCleanup(store.branch.lock_write().unlock)
2769
section = store.get_mutable_section('baz')
2770
section.set('foo', 'bar')
2772
modified_store = self.get_store(self)
2773
sections = list(modified_store.get_sections())
2774
self.assertLength(1, sections)
2775
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2777
def test_load_hook(self):
2778
# First, we need to ensure that the store exists
2779
store = self.get_store(self)
2780
# FIXME: There should be a better way than relying on the test
2781
# parametrization to identify branch.conf -- vila 2011-0526
2782
if self.store_id in ('branch', 'remote_branch'):
2783
# branch stores requires write locked branches
2784
self.addCleanup(store.branch.lock_write().unlock)
2785
section = store.get_mutable_section('baz')
2786
section.set('foo', 'bar')
2788
# Now we can try to load it
2789
store = self.get_store(self)
2794
config.ConfigHooks.install_named_hook('load', hook, None)
2795
self.assertLength(0, calls)
2797
self.assertLength(1, calls)
2798
self.assertEqual((store,), calls[0])
2800
def test_save_hook(self):
2805
config.ConfigHooks.install_named_hook('save', hook, None)
2806
self.assertLength(0, calls)
2807
store = self.get_store(self)
2808
# FIXME: There should be a better way than relying on the test
2809
# parametrization to identify branch.conf -- vila 2011-0526
2810
if self.store_id in ('branch', 'remote_branch'):
2811
# branch stores requires write locked branches
2812
self.addCleanup(store.branch.lock_write().unlock)
2813
section = store.get_mutable_section('baz')
2814
section.set('foo', 'bar')
2816
self.assertLength(1, calls)
2817
self.assertEqual((store,), calls[0])
2819
def test_set_mark_dirty(self):
2820
stack = config.MemoryStack(b'')
2821
self.assertLength(0, stack.store.dirty_sections)
2822
stack.set('foo', 'baz')
2823
self.assertLength(1, stack.store.dirty_sections)
2824
self.assertTrue(stack.store._need_saving())
2826
def test_remove_mark_dirty(self):
2827
stack = config.MemoryStack(b'foo=bar')
2828
self.assertLength(0, stack.store.dirty_sections)
2830
self.assertLength(1, stack.store.dirty_sections)
2831
self.assertTrue(stack.store._need_saving())
2834
class TestStoreSaveChanges(tests.TestCaseWithTransport):
2835
"""Tests that config changes are kept in memory and saved on-demand."""
2838
super(TestStoreSaveChanges, self).setUp()
2839
self.transport = self.get_transport()
2840
# Most of the tests involve two stores pointing to the same persistent
2841
# storage to observe the effects of concurrent changes
2842
self.st1 = config.TransportIniFileStore(self.transport, 'foo.conf')
2843
self.st2 = config.TransportIniFileStore(self.transport, 'foo.conf')
2847
self.warnings.append(args[0] % args[1:])
2848
self.overrideAttr(trace, 'warning', warning)
2850
def has_store(self, store):
2851
store_basename = urlutils.relative_url(self.transport.external_url(),
2852
store.external_url())
2853
return self.transport.has(store_basename)
2855
def get_stack(self, store):
2856
# Any stack will do as long as it uses the right store, just a single
2857
# no-name section is enough
2858
return config.Stack([store.get_sections], store)
2860
def test_no_changes_no_save(self):
2861
s = self.get_stack(self.st1)
2862
s.store.save_changes()
2863
self.assertEqual(False, self.has_store(self.st1))
2865
def test_unrelated_concurrent_update(self):
2866
s1 = self.get_stack(self.st1)
2867
s2 = self.get_stack(self.st2)
2868
s1.set('foo', 'bar')
2869
s2.set('baz', 'quux')
2871
# Changes don't propagate magically
2872
self.assertEqual(None, s1.get('baz'))
2873
s2.store.save_changes()
2874
self.assertEqual('quux', s2.get('baz'))
2875
# Changes are acquired when saving
2876
self.assertEqual('bar', s2.get('foo'))
2877
# Since there is no overlap, no warnings are emitted
2878
self.assertLength(0, self.warnings)
2880
def test_concurrent_update_modified(self):
2881
s1 = self.get_stack(self.st1)
2882
s2 = self.get_stack(self.st2)
2883
s1.set('foo', 'bar')
2884
s2.set('foo', 'baz')
2887
s2.store.save_changes()
2888
self.assertEqual('baz', s2.get('foo'))
2889
# But the user get a warning
2890
self.assertLength(1, self.warnings)
2891
warning = self.warnings[0]
2892
self.assertStartsWith(warning, 'Option foo in section None')
2893
self.assertEndsWith(warning, 'was changed from <CREATED> to bar.'
2894
' The baz value will be saved.')
2896
def test_concurrent_deletion(self):
2897
self.st1._load_from_string(b'foo=bar')
2899
s1 = self.get_stack(self.st1)
2900
s2 = self.get_stack(self.st2)
2903
s1.store.save_changes()
2905
self.assertLength(0, self.warnings)
2906
s2.store.save_changes()
2908
self.assertLength(1, self.warnings)
2909
warning = self.warnings[0]
2910
self.assertStartsWith(warning, 'Option foo in section None')
2911
self.assertEndsWith(warning, 'was changed from bar to <CREATED>.'
2912
' The <DELETED> value will be saved.')
2915
class TestQuotingIniFileStore(tests.TestCaseWithTransport):
2917
def get_store(self):
2918
return config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2920
def test_get_quoted_string(self):
2921
store = self.get_store()
2922
store._load_from_string(b'foo= " abc "')
2923
stack = config.Stack([store.get_sections])
2924
self.assertEqual(' abc ', stack.get('foo'))
2926
def test_set_quoted_string(self):
2927
store = self.get_store()
2928
stack = config.Stack([store.get_sections], store)
2929
stack.set('foo', ' a b c ')
2931
self.assertFileEqual(b'foo = " a b c "' +
2932
os.linesep.encode('ascii'), 'foo.conf')
2935
class TestTransportIniFileStore(TestStore):
2937
def test_loading_unknown_file_fails(self):
2938
store = config.TransportIniFileStore(self.get_transport(),
2940
self.assertRaises(errors.NoSuchFile, store.load)
2942
def test_invalid_content(self):
2943
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2944
self.assertEqual(False, store.is_loaded())
2945
exc = self.assertRaises(
2946
config.ParseConfigError, store._load_from_string,
2947
b'this is invalid !')
2948
self.assertEndsWith(exc.filename, 'foo.conf')
2949
# And the load failed
2950
self.assertEqual(False, store.is_loaded())
2952
def test_get_embedded_sections(self):
2953
# A more complicated example (which also shows that section names and
2954
# option names share the same name space...)
2955
# FIXME: This should be fixed by forbidding dicts as values ?
2956
# -- vila 2011-04-05
2957
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2958
store._load_from_string(b'''
2962
foo_in_DEFAULT=foo_DEFAULT
2970
sections = list(store.get_sections())
2971
self.assertLength(4, sections)
2972
# The default section has no name.
2973
# List values are provided as strings and need to be explicitly
2974
# converted by specifying from_unicode=list_from_store at option
2976
self.assertSectionContent((None, {'foo': 'bar', 'l': u'1,2'}),
2978
self.assertSectionContent(
2979
('DEFAULT', {'foo_in_DEFAULT': 'foo_DEFAULT'}), sections[1])
2980
self.assertSectionContent(
2981
('bar', {'foo_in_bar': 'barbar'}), sections[2])
2982
# sub sections are provided as embedded dicts.
2983
self.assertSectionContent(
2984
('baz', {'foo_in_baz': 'barbaz', 'qux': {'foo_in_qux': 'quux'}}),
2988
class TestLockableIniFileStore(TestStore):
2990
def test_create_store_in_created_dir(self):
2991
self.assertPathDoesNotExist('dir')
2992
t = self.get_transport('dir/subdir')
2993
store = config.LockableIniFileStore(t, 'foo.conf')
2994
store.get_mutable_section(None).set('foo', 'bar')
2996
self.assertPathExists('dir/subdir')
2999
class TestConcurrentStoreUpdates(TestStore):
3000
"""Test that Stores properly handle conccurent updates.
3002
New Store implementation may fail some of these tests but until such
3003
implementations exist it's hard to properly filter them from the scenarios
3004
applied here. If you encounter such a case, contact the bzr devs.
3007
scenarios = [(key, {'get_stack': builder}) for key, builder
3008
in config.test_stack_builder_registry.iteritems()]
3011
super(TestConcurrentStoreUpdates, self).setUp()
3012
self.stack = self.get_stack(self)
3013
if not isinstance(self.stack, config._CompatibleStack):
3014
raise tests.TestNotApplicable(
3015
'%s is not meant to be compatible with the old config design'
3017
self.stack.set('one', '1')
3018
self.stack.set('two', '2')
3020
self.stack.store.save()
3022
def test_simple_read_access(self):
3023
self.assertEqual('1', self.stack.get('one'))
3025
def test_simple_write_access(self):
3026
self.stack.set('one', 'one')
3027
self.assertEqual('one', self.stack.get('one'))
3029
def test_listen_to_the_last_speaker(self):
3031
c2 = self.get_stack(self)
3032
c1.set('one', 'ONE')
3033
c2.set('two', 'TWO')
3034
self.assertEqual('ONE', c1.get('one'))
3035
self.assertEqual('TWO', c2.get('two'))
3036
# The second update respect the first one
3037
self.assertEqual('ONE', c2.get('one'))
3039
def test_last_speaker_wins(self):
3040
# If the same config is not shared, the same variable modified twice
3041
# can only see a single result.
3043
c2 = self.get_stack(self)
3046
self.assertEqual('c2', c2.get('one'))
3047
# The first modification is still available until another refresh
3049
self.assertEqual('c1', c1.get('one'))
3050
c1.set('two', 'done')
3051
self.assertEqual('c2', c1.get('one'))
3053
def test_writes_are_serialized(self):
3055
c2 = self.get_stack(self)
3057
# We spawn a thread that will pause *during* the config saving.
3058
before_writing = threading.Event()
3059
after_writing = threading.Event()
3060
writing_done = threading.Event()
3061
c1_save_without_locking_orig = c1.store.save_without_locking
3063
def c1_save_without_locking():
3064
before_writing.set()
3065
c1_save_without_locking_orig()
3066
# The lock is held. We wait for the main thread to decide when to
3068
after_writing.wait()
3069
c1.store.save_without_locking = c1_save_without_locking
3074
t1 = threading.Thread(target=c1_set)
3075
# Collect the thread after the test
3076
self.addCleanup(t1.join)
3077
# Be ready to unblock the thread if the test goes wrong
3078
self.addCleanup(after_writing.set)
3080
before_writing.wait()
3081
self.assertRaises(errors.LockContention,
3082
c2.set, 'one', 'c2')
3083
self.assertEqual('c1', c1.get('one'))
3084
# Let the lock be released
3088
self.assertEqual('c2', c2.get('one'))
3090
def test_read_while_writing(self):
3092
# We spawn a thread that will pause *during* the write
3093
ready_to_write = threading.Event()
3094
do_writing = threading.Event()
3095
writing_done = threading.Event()
3096
# We override the _save implementation so we know the store is locked
3097
c1_save_without_locking_orig = c1.store.save_without_locking
3099
def c1_save_without_locking():
3100
ready_to_write.set()
3101
# The lock is held. We wait for the main thread to decide when to
3104
c1_save_without_locking_orig()
3106
c1.store.save_without_locking = c1_save_without_locking
3110
t1 = threading.Thread(target=c1_set)
3111
# Collect the thread after the test
3112
self.addCleanup(t1.join)
3113
# Be ready to unblock the thread if the test goes wrong
3114
self.addCleanup(do_writing.set)
3116
# Ensure the thread is ready to write
3117
ready_to_write.wait()
3118
self.assertEqual('c1', c1.get('one'))
3119
# If we read during the write, we get the old value
3120
c2 = self.get_stack(self)
3121
self.assertEqual('1', c2.get('one'))
3122
# Let the writing occur and ensure it occurred
3125
# Now we get the updated value
3126
c3 = self.get_stack(self)
3127
self.assertEqual('c1', c3.get('one'))
3129
# FIXME: It may be worth looking into removing the lock dir when it's not
3130
# needed anymore and look at possible fallouts for concurrent lockers. This
3131
# will matter if/when we use config files outside of breezy directories
3132
# (.config/breezy or .bzr) -- vila 20110-04-111
3135
class TestSectionMatcher(TestStore):
3137
scenarios = [('location', {'matcher': config.LocationMatcher}),
3138
('id', {'matcher': config.NameMatcher}), ]
3141
super(TestSectionMatcher, self).setUp()
3142
# Any simple store is good enough
3143
self.get_store = config.test_store_builder_registry.get('configobj')
3145
def test_no_matches_for_empty_stores(self):
3146
store = self.get_store(self)
3147
store._load_from_string(b'')
3148
matcher = self.matcher(store, '/bar')
3149
self.assertEqual([], list(matcher.get_sections()))
3151
def test_build_doesnt_load_store(self):
3152
store = self.get_store(self)
3153
self.matcher(store, '/bar')
3154
self.assertFalse(store.is_loaded())
3157
class TestLocationSection(tests.TestCase):
3159
def get_section(self, options, extra_path):
3160
section = config.Section('foo', options)
3161
return config.LocationSection(section, extra_path)
3163
def test_simple_option(self):
3164
section = self.get_section({'foo': 'bar'}, '')
3165
self.assertEqual('bar', section.get('foo'))
3167
def test_option_with_extra_path(self):
3168
section = self.get_section({'foo': 'bar', 'foo:policy': 'appendpath'},
3170
self.assertEqual('bar/baz', section.get('foo'))
3172
def test_invalid_policy(self):
3173
section = self.get_section({'foo': 'bar', 'foo:policy': 'die'},
3175
# invalid policies are ignored
3176
self.assertEqual('bar', section.get('foo'))
3179
class TestLocationMatcher(TestStore):
3182
super(TestLocationMatcher, self).setUp()
3183
# Any simple store is good enough
3184
self.get_store = config.test_store_builder_registry.get('configobj')
3186
def test_unrelated_section_excluded(self):
3187
store = self.get_store(self)
3188
store._load_from_string(b'''
3196
section=/foo/bar/baz
3200
self.assertEqual(['/foo', '/foo/baz', '/foo/bar', '/foo/bar/baz',
3202
[section.id for _, section in store.get_sections()])
3203
matcher = config.LocationMatcher(store, '/foo/bar/quux')
3204
sections = [section for _, section in matcher.get_sections()]
3205
self.assertEqual(['/foo/bar', '/foo'],
3206
[section.id for section in sections])
3207
self.assertEqual(['quux', 'bar/quux'],
3208
[section.extra_path for section in sections])
3210
def test_more_specific_sections_first(self):
3211
store = self.get_store(self)
3212
store._load_from_string(b'''
3218
self.assertEqual(['/foo', '/foo/bar'],
3219
[section.id for _, section in store.get_sections()])
3220
matcher = config.LocationMatcher(store, '/foo/bar/baz')
3221
sections = [section for _, section in matcher.get_sections()]
3222
self.assertEqual(['/foo/bar', '/foo'],
3223
[section.id for section in sections])
3224
self.assertEqual(['baz', 'bar/baz'],
3225
[section.extra_path for section in sections])
3227
def test_appendpath_in_no_name_section(self):
3228
# It's a bit weird to allow appendpath in a no-name section, but
3229
# someone may found a use for it
3230
store = self.get_store(self)
3231
store._load_from_string(b'''
3233
foo:policy = appendpath
3235
matcher = config.LocationMatcher(store, 'dir/subdir')
3236
sections = list(matcher.get_sections())
3237
self.assertLength(1, sections)
3238
self.assertEqual('bar/dir/subdir', sections[0][1].get('foo'))
3240
def test_file_urls_are_normalized(self):
3241
store = self.get_store(self)
3242
if sys.platform == 'win32':
3243
expected_url = 'file:///C:/dir/subdir'
3244
expected_location = 'C:/dir/subdir'
3246
expected_url = 'file:///dir/subdir'
3247
expected_location = '/dir/subdir'
3248
matcher = config.LocationMatcher(store, expected_url)
3249
self.assertEqual(expected_location, matcher.location)
3251
def test_branch_name_colo(self):
3252
store = self.get_store(self)
3253
store._load_from_string(dedent("""\
3255
push_location=my{branchname}
3256
""").encode('ascii'))
3257
matcher = config.LocationMatcher(store, 'file:///,branch=example%3c')
3258
self.assertEqual('example<', matcher.branch_name)
3259
((_, section),) = matcher.get_sections()
3260
self.assertEqual('example<', section.locals['branchname'])
3262
def test_branch_name_basename(self):
3263
store = self.get_store(self)
3264
store._load_from_string(dedent("""\
3266
push_location=my{branchname}
3267
""").encode('ascii'))
3268
matcher = config.LocationMatcher(store, 'file:///parent/example%3c')
3269
self.assertEqual('example<', matcher.branch_name)
3270
((_, section),) = matcher.get_sections()
3271
self.assertEqual('example<', section.locals['branchname'])
3274
class TestStartingPathMatcher(TestStore):
3277
super(TestStartingPathMatcher, self).setUp()
3278
# Any simple store is good enough
3279
self.store = config.IniFileStore()
3281
def assertSectionIDs(self, expected, location, content):
3282
self.store._load_from_string(content)
3283
matcher = config.StartingPathMatcher(self.store, location)
3284
sections = list(matcher.get_sections())
3285
self.assertLength(len(expected), sections)
3286
self.assertEqual(expected, [section.id for _, section in sections])
3289
def test_empty(self):
3290
self.assertSectionIDs([], self.get_url(), b'')
3292
def test_url_vs_local_paths(self):
3293
# The matcher location is an url and the section names are local paths
3294
self.assertSectionIDs(['/foo/bar', '/foo'],
3295
'file:///foo/bar/baz', b'''\
3300
def test_local_path_vs_url(self):
3301
# The matcher location is a local path and the section names are urls
3302
self.assertSectionIDs(['file:///foo/bar', 'file:///foo'],
3303
'/foo/bar/baz', b'''\
3308
def test_no_name_section_included_when_present(self):
3309
# Note that other tests will cover the case where the no-name section
3310
# is empty and as such, not included.
3311
sections = self.assertSectionIDs(['/foo/bar', '/foo', None],
3312
'/foo/bar/baz', b'''\
3313
option = defined so the no-name section exists
3317
self.assertEqual(['baz', 'bar/baz', '/foo/bar/baz'],
3318
[s.locals['relpath'] for _, s in sections])
3320
def test_order_reversed(self):
3321
self.assertSectionIDs(['/foo/bar', '/foo'], '/foo/bar/baz', b'''\
3326
def test_unrelated_section_excluded(self):
3327
self.assertSectionIDs(['/foo/bar', '/foo'], '/foo/bar/baz', b'''\
3333
def test_glob_included(self):
3334
sections = self.assertSectionIDs(['/foo/*/baz', '/foo/b*', '/foo'],
3335
'/foo/bar/baz', b'''\
3341
# Note that 'baz' as a relpath for /foo/b* is not fully correct, but
3342
# nothing really is... as far using {relpath} to append it to something
3343
# else, this seems good enough though.
3344
self.assertEqual(['', 'baz', 'bar/baz'],
3345
[s.locals['relpath'] for _, s in sections])
3347
def test_respect_order(self):
3348
self.assertSectionIDs(['/foo', '/foo/b*', '/foo/*/baz'],
3349
'/foo/bar/baz', b'''\
3357
class TestNameMatcher(TestStore):
3360
super(TestNameMatcher, self).setUp()
3361
self.matcher = config.NameMatcher
3362
# Any simple store is good enough
3363
self.get_store = config.test_store_builder_registry.get('configobj')
3365
def get_matching_sections(self, name):
3366
store = self.get_store(self)
3367
store._load_from_string(b'''
3375
matcher = self.matcher(store, name)
3376
return list(matcher.get_sections())
3378
def test_matching(self):
3379
sections = self.get_matching_sections('foo')
3380
self.assertLength(1, sections)
3381
self.assertSectionContent(('foo', {'option': 'foo'}), sections[0])
3383
def test_not_matching(self):
3384
sections = self.get_matching_sections('baz')
3385
self.assertLength(0, sections)
3388
class TestBaseStackGet(tests.TestCase):
3391
super(TestBaseStackGet, self).setUp()
3392
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3394
def test_get_first_definition(self):
3395
store1 = config.IniFileStore()
3396
store1._load_from_string(b'foo=bar')
3397
store2 = config.IniFileStore()
3398
store2._load_from_string(b'foo=baz')
3399
conf = config.Stack([store1.get_sections, store2.get_sections])
3400
self.assertEqual('bar', conf.get('foo'))
3402
def test_get_with_registered_default_value(self):
3403
config.option_registry.register(config.Option('foo', default='bar'))
3404
conf_stack = config.Stack([])
3405
self.assertEqual('bar', conf_stack.get('foo'))
3407
def test_get_without_registered_default_value(self):
3408
config.option_registry.register(config.Option('foo'))
3409
conf_stack = config.Stack([])
3410
self.assertEqual(None, conf_stack.get('foo'))
3412
def test_get_without_default_value_for_not_registered(self):
3413
conf_stack = config.Stack([])
3414
self.assertEqual(None, conf_stack.get('foo'))
3416
def test_get_for_empty_section_callable(self):
3417
conf_stack = config.Stack([lambda: []])
3418
self.assertEqual(None, conf_stack.get('foo'))
3420
def test_get_for_broken_callable(self):
3421
# Trying to use and invalid callable raises an exception on first use
3422
conf_stack = config.Stack([object])
3423
self.assertRaises(TypeError, conf_stack.get, 'foo')
3426
class TestStackWithSimpleStore(tests.TestCase):
3429
super(TestStackWithSimpleStore, self).setUp()
3430
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3431
self.registry = config.option_registry
3433
def get_conf(self, content=None):
3434
return config.MemoryStack(content)
3436
def test_override_value_from_env(self):
3437
self.overrideEnv('FOO', None)
3438
self.registry.register(
3439
config.Option('foo', default='bar', override_from_env=['FOO']))
3440
self.overrideEnv('FOO', 'quux')
3441
# Env variable provides a default taking over the option one
3442
conf = self.get_conf(b'foo=store')
3443
self.assertEqual('quux', conf.get('foo'))
3445
def test_first_override_value_from_env_wins(self):
3446
self.overrideEnv('NO_VALUE', None)
3447
self.overrideEnv('FOO', None)
3448
self.overrideEnv('BAZ', None)
3449
self.registry.register(
3450
config.Option('foo', default='bar',
3451
override_from_env=['NO_VALUE', 'FOO', 'BAZ']))
3452
self.overrideEnv('FOO', 'foo')
3453
self.overrideEnv('BAZ', 'baz')
3454
# The first env var set wins
3455
conf = self.get_conf(b'foo=store')
3456
self.assertEqual('foo', conf.get('foo'))
3459
class TestMemoryStack(tests.TestCase):
3462
conf = config.MemoryStack(b'foo=bar')
3463
self.assertEqual('bar', conf.get('foo'))
3466
conf = config.MemoryStack(b'foo=bar')
3467
conf.set('foo', 'baz')
3468
self.assertEqual('baz', conf.get('foo'))
3470
def test_no_content(self):
3471
conf = config.MemoryStack()
3472
# No content means no loading
3473
self.assertFalse(conf.store.is_loaded())
3474
self.assertRaises(NotImplementedError, conf.get, 'foo')
3475
# But a content can still be provided
3476
conf.store._load_from_string(b'foo=bar')
3477
self.assertEqual('bar', conf.get('foo'))
3480
class TestStackIterSections(tests.TestCase):
3482
def test_empty_stack(self):
3483
conf = config.Stack([])
3484
sections = list(conf.iter_sections())
3485
self.assertLength(0, sections)
3487
def test_empty_store(self):
3488
store = config.IniFileStore()
3489
store._load_from_string(b'')
3490
conf = config.Stack([store.get_sections])
3491
sections = list(conf.iter_sections())
3492
self.assertLength(0, sections)
3494
def test_simple_store(self):
3495
store = config.IniFileStore()
3496
store._load_from_string(b'foo=bar')
3497
conf = config.Stack([store.get_sections])
3498
tuples = list(conf.iter_sections())
3499
self.assertLength(1, tuples)
3500
(found_store, found_section) = tuples[0]
3501
self.assertIs(store, found_store)
3503
def test_two_stores(self):
3504
store1 = config.IniFileStore()
3505
store1._load_from_string(b'foo=bar')
3506
store2 = config.IniFileStore()
3507
store2._load_from_string(b'bar=qux')
3508
conf = config.Stack([store1.get_sections, store2.get_sections])
3509
tuples = list(conf.iter_sections())
3510
self.assertLength(2, tuples)
3511
self.assertIs(store1, tuples[0][0])
3512
self.assertIs(store2, tuples[1][0])
3515
class TestStackWithTransport(tests.TestCaseWithTransport):
3517
scenarios = [(key, {'get_stack': builder}) for key, builder
3518
in config.test_stack_builder_registry.iteritems()]
3521
class TestConcreteStacks(TestStackWithTransport):
3523
def test_build_stack(self):
3524
# Just a smoke test to help debug builders
3525
self.get_stack(self)
3528
class TestStackGet(TestStackWithTransport):
3531
super(TestStackGet, self).setUp()
3532
self.conf = self.get_stack(self)
3534
def test_get_for_empty_stack(self):
3535
self.assertEqual(None, self.conf.get('foo'))
3537
def test_get_hook(self):
3538
self.conf.set('foo', 'bar')
3543
config.ConfigHooks.install_named_hook('get', hook, None)
3544
self.assertLength(0, calls)
3545
value = self.conf.get('foo')
3546
self.assertEqual('bar', value)
3547
self.assertLength(1, calls)
3548
self.assertEqual((self.conf, 'foo', 'bar'), calls[0])
3551
class TestStackGetWithConverter(tests.TestCase):
3554
super(TestStackGetWithConverter, self).setUp()
3555
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3556
self.registry = config.option_registry
3558
def get_conf(self, content=None):
3559
return config.MemoryStack(content)
3561
def register_bool_option(self, name, default=None, default_from_env=None):
3562
b = config.Option(name, help='A boolean.',
3563
default=default, default_from_env=default_from_env,
3564
from_unicode=config.bool_from_store)
3565
self.registry.register(b)
3567
def test_get_default_bool_None(self):
3568
self.register_bool_option('foo')
3569
conf = self.get_conf(b'')
3570
self.assertEqual(None, conf.get('foo'))
3572
def test_get_default_bool_True(self):
3573
self.register_bool_option('foo', u'True')
3574
conf = self.get_conf(b'')
3575
self.assertEqual(True, conf.get('foo'))
3577
def test_get_default_bool_False(self):
3578
self.register_bool_option('foo', False)
3579
conf = self.get_conf(b'')
3580
self.assertEqual(False, conf.get('foo'))
3582
def test_get_default_bool_False_as_string(self):
3583
self.register_bool_option('foo', u'False')
3584
conf = self.get_conf(b'')
3585
self.assertEqual(False, conf.get('foo'))
3587
def test_get_default_bool_from_env_converted(self):
3588
self.register_bool_option('foo', u'True', default_from_env=['FOO'])
3589
self.overrideEnv('FOO', 'False')
3590
conf = self.get_conf(b'')
3591
self.assertEqual(False, conf.get('foo'))
3593
def test_get_default_bool_when_conversion_fails(self):
3594
self.register_bool_option('foo', default='True')
3595
conf = self.get_conf(b'foo=invalid boolean')
3596
self.assertEqual(True, conf.get('foo'))
3598
def register_integer_option(self, name,
3599
default=None, default_from_env=None):
3600
i = config.Option(name, help='An integer.',
3601
default=default, default_from_env=default_from_env,
3602
from_unicode=config.int_from_store)
3603
self.registry.register(i)
3605
def test_get_default_integer_None(self):
3606
self.register_integer_option('foo')
3607
conf = self.get_conf(b'')
3608
self.assertEqual(None, conf.get('foo'))
3610
def test_get_default_integer(self):
3611
self.register_integer_option('foo', 42)
3612
conf = self.get_conf(b'')
3613
self.assertEqual(42, conf.get('foo'))
3615
def test_get_default_integer_as_string(self):
3616
self.register_integer_option('foo', u'42')
3617
conf = self.get_conf(b'')
3618
self.assertEqual(42, conf.get('foo'))
3620
def test_get_default_integer_from_env(self):
3621
self.register_integer_option('foo', default_from_env=['FOO'])
3622
self.overrideEnv('FOO', '18')
3623
conf = self.get_conf(b'')
3624
self.assertEqual(18, conf.get('foo'))
3626
def test_get_default_integer_when_conversion_fails(self):
3627
self.register_integer_option('foo', default='12')
3628
conf = self.get_conf(b'foo=invalid integer')
3629
self.assertEqual(12, conf.get('foo'))
3631
def register_list_option(self, name, default=None, default_from_env=None):
3632
l = config.ListOption(name, help='A list.', default=default,
3633
default_from_env=default_from_env)
3634
self.registry.register(l)
3636
def test_get_default_list_None(self):
3637
self.register_list_option('foo')
3638
conf = self.get_conf(b'')
3639
self.assertEqual(None, conf.get('foo'))
3641
def test_get_default_list_empty(self):
3642
self.register_list_option('foo', '')
3643
conf = self.get_conf(b'')
3644
self.assertEqual([], conf.get('foo'))
3646
def test_get_default_list_from_env(self):
3647
self.register_list_option('foo', default_from_env=['FOO'])
3648
self.overrideEnv('FOO', '')
3649
conf = self.get_conf(b'')
3650
self.assertEqual([], conf.get('foo'))
3652
def test_get_with_list_converter_no_item(self):
3653
self.register_list_option('foo', None)
3654
conf = self.get_conf(b'foo=,')
3655
self.assertEqual([], conf.get('foo'))
3657
def test_get_with_list_converter_many_items(self):
3658
self.register_list_option('foo', None)
3659
conf = self.get_conf(b'foo=m,o,r,e')
3660
self.assertEqual(['m', 'o', 'r', 'e'], conf.get('foo'))
3662
def test_get_with_list_converter_embedded_spaces_many_items(self):
3663
self.register_list_option('foo', None)
3664
conf = self.get_conf(b'foo=" bar", "baz "')
3665
self.assertEqual([' bar', 'baz '], conf.get('foo'))
3667
def test_get_with_list_converter_stripped_spaces_many_items(self):
3668
self.register_list_option('foo', None)
3669
conf = self.get_conf(b'foo= bar , baz ')
3670
self.assertEqual(['bar', 'baz'], conf.get('foo'))
3673
class TestIterOptionRefs(tests.TestCase):
3674
"""iter_option_refs is a bit unusual, document some cases."""
3676
def assertRefs(self, expected, string):
3677
self.assertEqual(expected, list(config.iter_option_refs(string)))
3679
def test_empty(self):
3680
self.assertRefs([(False, '')], '')
3682
def test_no_refs(self):
3683
self.assertRefs([(False, 'foo bar')], 'foo bar')
3685
def test_single_ref(self):
3686
self.assertRefs([(False, ''), (True, '{foo}'), (False, '')], '{foo}')
3688
def test_broken_ref(self):
3689
self.assertRefs([(False, '{foo')], '{foo')
3691
def test_embedded_ref(self):
3692
self.assertRefs([(False, '{'), (True, '{foo}'), (False, '}')],
3695
def test_two_refs(self):
3696
self.assertRefs([(False, ''), (True, '{foo}'),
3697
(False, ''), (True, '{bar}'),
3701
def test_newline_in_refs_are_not_matched(self):
3702
self.assertRefs([(False, '{\nxx}{xx\n}{{\n}}')], '{\nxx}{xx\n}{{\n}}')
3705
class TestStackExpandOptions(tests.TestCaseWithTransport):
3708
super(TestStackExpandOptions, self).setUp()
3709
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3710
self.registry = config.option_registry
3711
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
3712
self.conf = config.Stack([store.get_sections], store)
3714
def assertExpansion(self, expected, string, env=None):
3715
self.assertEqual(expected, self.conf.expand_options(string, env))
3717
def test_no_expansion(self):
3718
self.assertExpansion('foo', 'foo')
3720
def test_expand_default_value(self):
3721
self.conf.store._load_from_string(b'bar=baz')
3722
self.registry.register(config.Option('foo', default=u'{bar}'))
3723
self.assertEqual('baz', self.conf.get('foo', expand=True))
3725
def test_expand_default_from_env(self):
3726
self.conf.store._load_from_string(b'bar=baz')
3727
self.registry.register(config.Option('foo', default_from_env=['FOO']))
3728
self.overrideEnv('FOO', '{bar}')
3729
self.assertEqual('baz', self.conf.get('foo', expand=True))
3731
def test_expand_default_on_failed_conversion(self):
3732
self.conf.store._load_from_string(b'baz=bogus\nbar=42\nfoo={baz}')
3733
self.registry.register(
3734
config.Option('foo', default=u'{bar}',
3735
from_unicode=config.int_from_store))
3736
self.assertEqual(42, self.conf.get('foo', expand=True))
3738
def test_env_adding_options(self):
3739
self.assertExpansion('bar', '{foo}', {'foo': 'bar'})
3741
def test_env_overriding_options(self):
3742
self.conf.store._load_from_string(b'foo=baz')
3743
self.assertExpansion('bar', '{foo}', {'foo': 'bar'})
3745
def test_simple_ref(self):
3746
self.conf.store._load_from_string(b'foo=xxx')
3747
self.assertExpansion('xxx', '{foo}')
3749
def test_unknown_ref(self):
3750
self.assertRaises(config.ExpandingUnknownOption,
3751
self.conf.expand_options, '{foo}')
3753
def test_illegal_def_is_ignored(self):
3754
self.assertExpansion('{1,2}', '{1,2}')
3755
self.assertExpansion('{ }', '{ }')
3756
self.assertExpansion('${Foo,f}', '${Foo,f}')
3758
def test_indirect_ref(self):
3759
self.conf.store._load_from_string(b'''
3763
self.assertExpansion('xxx', '{bar}')
3765
def test_embedded_ref(self):
3766
self.conf.store._load_from_string(b'''
3770
self.assertExpansion('xxx', '{{bar}}')
3772
def test_simple_loop(self):
3773
self.conf.store._load_from_string(b'foo={foo}')
3774
self.assertRaises(config.OptionExpansionLoop,
3775
self.conf.expand_options, '{foo}')
3777
def test_indirect_loop(self):
3778
self.conf.store._load_from_string(b'''
3782
e = self.assertRaises(config.OptionExpansionLoop,
3783
self.conf.expand_options, '{foo}')
3784
self.assertEqual('foo->bar->baz', e.refs)
3785
self.assertEqual('{foo}', e.string)
3787
def test_list(self):
3788
self.conf.store._load_from_string(b'''
3792
list={foo},{bar},{baz}
3794
self.registry.register(
3795
config.ListOption('list'))
3796
self.assertEqual(['start', 'middle', 'end'],
3797
self.conf.get('list', expand=True))
3799
def test_cascading_list(self):
3800
self.conf.store._load_from_string(b'''
3806
self.registry.register(config.ListOption('list'))
3807
# Register an intermediate option as a list to ensure no conversion
3808
# happen while expanding. Conversion should only occur for the original
3809
# option ('list' here).
3810
self.registry.register(config.ListOption('baz'))
3811
self.assertEqual(['start', 'middle', 'end'],
3812
self.conf.get('list', expand=True))
3814
def test_pathologically_hidden_list(self):
3815
self.conf.store._load_from_string(b'''
3821
hidden={start}{middle}{end}
3823
# What matters is what the registration says, the conversion happens
3824
# only after all expansions have been performed
3825
self.registry.register(config.ListOption('hidden'))
3826
self.assertEqual(['bin', 'go'],
3827
self.conf.get('hidden', expand=True))
3830
class TestStackCrossSectionsExpand(tests.TestCaseWithTransport):
3833
super(TestStackCrossSectionsExpand, self).setUp()
3835
def get_config(self, location, string):
3838
# Since we don't save the config we won't strictly require to inherit
3839
# from TestCaseInTempDir, but an error occurs so quickly...
3840
c = config.LocationStack(location)
3841
c.store._load_from_string(string)
3844
def test_dont_cross_unrelated_section(self):
3845
c = self.get_config('/another/branch/path', b'''
3850
[/another/branch/path]
3853
self.assertRaises(config.ExpandingUnknownOption,
3854
c.get, 'bar', expand=True)
3856
def test_cross_related_sections(self):
3857
c = self.get_config('/project/branch/path', b'''
3861
[/project/branch/path]
3864
self.assertEqual('quux', c.get('bar', expand=True))
3867
class TestStackCrossStoresExpand(tests.TestCaseWithTransport):
3869
def test_cross_global_locations(self):
3870
l_store = config.LocationStore()
3871
l_store._load_from_string(b'''
3877
g_store = config.GlobalStore()
3878
g_store._load_from_string(b'''
3884
stack = config.LocationStack('/branch')
3885
self.assertEqual('glob-bar', stack.get('lbar', expand=True))
3886
self.assertEqual('loc-foo', stack.get('gfoo', expand=True))
3889
class TestStackExpandSectionLocals(tests.TestCaseWithTransport):
3891
def test_expand_locals_empty(self):
3892
l_store = config.LocationStore()
3893
l_store._load_from_string(b'''
3894
[/home/user/project]
3899
stack = config.LocationStack('/home/user/project/')
3900
self.assertEqual('', stack.get('base', expand=True))
3901
self.assertEqual('', stack.get('rel', expand=True))
3903
def test_expand_basename_locally(self):
3904
l_store = config.LocationStore()
3905
l_store._load_from_string(b'''
3906
[/home/user/project]
3910
stack = config.LocationStack('/home/user/project/branch')
3911
self.assertEqual('branch', stack.get('bfoo', expand=True))
3913
def test_expand_basename_locally_longer_path(self):
3914
l_store = config.LocationStore()
3915
l_store._load_from_string(b'''
3920
stack = config.LocationStack('/home/user/project/dir/branch')
3921
self.assertEqual('branch', stack.get('bfoo', expand=True))
3923
def test_expand_relpath_locally(self):
3924
l_store = config.LocationStore()
3925
l_store._load_from_string(b'''
3926
[/home/user/project]
3927
lfoo = loc-foo/{relpath}
3930
stack = config.LocationStack('/home/user/project/branch')
3931
self.assertEqual('loc-foo/branch', stack.get('lfoo', expand=True))
3933
def test_expand_relpath_unknonw_in_global(self):
3934
g_store = config.GlobalStore()
3935
g_store._load_from_string(b'''
3940
stack = config.LocationStack('/home/user/project/branch')
3941
self.assertRaises(config.ExpandingUnknownOption,
3942
stack.get, 'gfoo', expand=True)
3944
def test_expand_local_option_locally(self):
3945
l_store = config.LocationStore()
3946
l_store._load_from_string(b'''
3947
[/home/user/project]
3948
lfoo = loc-foo/{relpath}
3952
g_store = config.GlobalStore()
3953
g_store._load_from_string(b'''
3959
stack = config.LocationStack('/home/user/project/branch')
3960
self.assertEqual('glob-bar', stack.get('lbar', expand=True))
3961
self.assertEqual('loc-foo/branch', stack.get('gfoo', expand=True))
3963
def test_locals_dont_leak(self):
3964
"""Make sure we chose the right local in presence of several sections.
3966
l_store = config.LocationStore()
3967
l_store._load_from_string(b'''
3969
lfoo = loc-foo/{relpath}
3970
[/home/user/project]
3971
lfoo = loc-foo/{relpath}
3974
stack = config.LocationStack('/home/user/project/branch')
3975
self.assertEqual('loc-foo/branch', stack.get('lfoo', expand=True))
3976
stack = config.LocationStack('/home/user/bar/baz')
3977
self.assertEqual('loc-foo/bar/baz', stack.get('lfoo', expand=True))
3980
class TestStackSet(TestStackWithTransport):
3982
def test_simple_set(self):
3983
conf = self.get_stack(self)
3984
self.assertEqual(None, conf.get('foo'))
3985
conf.set('foo', 'baz')
3986
# Did we get it back ?
3987
self.assertEqual('baz', conf.get('foo'))
3989
def test_set_creates_a_new_section(self):
3990
conf = self.get_stack(self)
3991
conf.set('foo', 'baz')
3992
self.assertEqual, 'baz', conf.get('foo')
3994
def test_set_hook(self):
3999
config.ConfigHooks.install_named_hook('set', hook, None)
4000
self.assertLength(0, calls)
4001
conf = self.get_stack(self)
4002
conf.set('foo', 'bar')
4003
self.assertLength(1, calls)
4004
self.assertEqual((conf, 'foo', 'bar'), calls[0])
4007
class TestStackRemove(TestStackWithTransport):
4009
def test_remove_existing(self):
4010
conf = self.get_stack(self)
4011
conf.set('foo', 'bar')
4012
self.assertEqual('bar', conf.get('foo'))
4014
# Did we get it back ?
4015
self.assertEqual(None, conf.get('foo'))
4017
def test_remove_unknown(self):
4018
conf = self.get_stack(self)
4019
self.assertRaises(KeyError, conf.remove, 'I_do_not_exist')
4021
def test_remove_hook(self):
4026
config.ConfigHooks.install_named_hook('remove', hook, None)
4027
self.assertLength(0, calls)
4028
conf = self.get_stack(self)
4029
conf.set('foo', 'bar')
4031
self.assertLength(1, calls)
4032
self.assertEqual((conf, 'foo'), calls[0])
4035
class TestConfigGetOptions(tests.TestCaseWithTransport, TestOptionsMixin):
4038
super(TestConfigGetOptions, self).setUp()
4039
create_configs(self)
4041
def test_no_variable(self):
4042
# Using branch should query branch, locations and breezy
4043
self.assertOptions([], self.branch_config)
4045
def test_option_in_breezy(self):
4046
self.breezy_config.set_user_option('file', 'breezy')
4047
self.assertOptions([('file', 'breezy', 'DEFAULT', 'breezy')],
4050
def test_option_in_locations(self):
4051
self.locations_config.set_user_option('file', 'locations')
4053
[('file', 'locations', self.tree.basedir, 'locations')],
4054
self.locations_config)
4056
def test_option_in_branch(self):
4057
self.branch_config.set_user_option('file', 'branch')
4058
self.assertOptions([('file', 'branch', 'DEFAULT', 'branch')],
4061
def test_option_in_breezy_and_branch(self):
4062
self.breezy_config.set_user_option('file', 'breezy')
4063
self.branch_config.set_user_option('file', 'branch')
4064
self.assertOptions([('file', 'branch', 'DEFAULT', 'branch'),
4065
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4068
def test_option_in_branch_and_locations(self):
4069
# Hmm, locations override branch :-/
4070
self.locations_config.set_user_option('file', 'locations')
4071
self.branch_config.set_user_option('file', 'branch')
4073
[('file', 'locations', self.tree.basedir, 'locations'),
4074
('file', 'branch', 'DEFAULT', 'branch'), ],
4077
def test_option_in_breezy_locations_and_branch(self):
4078
self.breezy_config.set_user_option('file', 'breezy')
4079
self.locations_config.set_user_option('file', 'locations')
4080
self.branch_config.set_user_option('file', 'branch')
4082
[('file', 'locations', self.tree.basedir, 'locations'),
4083
('file', 'branch', 'DEFAULT', 'branch'),
4084
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4088
class TestConfigRemoveOption(tests.TestCaseWithTransport, TestOptionsMixin):
4091
super(TestConfigRemoveOption, self).setUp()
4092
create_configs_with_file_option(self)
4094
def test_remove_in_locations(self):
4095
self.locations_config.remove_user_option('file', self.tree.basedir)
4097
[('file', 'branch', 'DEFAULT', 'branch'),
4098
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4101
def test_remove_in_branch(self):
4102
self.branch_config.remove_user_option('file')
4104
[('file', 'locations', self.tree.basedir, 'locations'),
4105
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4108
def test_remove_in_breezy(self):
4109
self.breezy_config.remove_user_option('file')
4111
[('file', 'locations', self.tree.basedir, 'locations'),
4112
('file', 'branch', 'DEFAULT', 'branch'), ],
4116
class TestConfigGetSections(tests.TestCaseWithTransport):
4119
super(TestConfigGetSections, self).setUp()
4120
create_configs(self)
4122
def assertSectionNames(self, expected, conf, name=None):
4123
"""Check which sections are returned for a given config.
4125
If fallback configurations exist their sections can be included.
4127
:param expected: A list of section names.
4129
:param conf: The configuration that will be queried.
4131
:param name: An optional section name that will be passed to
4134
sections = list(conf._get_sections(name))
4135
self.assertLength(len(expected), sections)
4136
self.assertEqual(expected, [n for n, _, _ in sections])
4138
def test_breezy_default_section(self):
4139
self.assertSectionNames(['DEFAULT'], self.breezy_config)
4141
def test_locations_default_section(self):
4142
# No sections are defined in an empty file
4143
self.assertSectionNames([], self.locations_config)
4145
def test_locations_named_section(self):
4146
self.locations_config.set_user_option('file', 'locations')
4147
self.assertSectionNames([self.tree.basedir], self.locations_config)
4149
def test_locations_matching_sections(self):
4150
loc_config = self.locations_config
4151
loc_config.set_user_option('file', 'locations')
4152
# We need to cheat a bit here to create an option in sections above and
4153
# below the 'location' one.
4154
parser = loc_config._get_parser()
4155
# locations.cong deals with '/' ignoring native os.sep
4156
location_names = self.tree.basedir.split('/')
4157
parent = '/'.join(location_names[:-1])
4158
child = '/'.join(location_names + ['child'])
4160
parser[parent]['file'] = 'parent'
4162
parser[child]['file'] = 'child'
4163
self.assertSectionNames([self.tree.basedir, parent], loc_config)
4165
def test_branch_data_default_section(self):
4166
self.assertSectionNames([None],
4167
self.branch_config._get_branch_data_config())
4169
def test_branch_default_sections(self):
4170
# No sections are defined in an empty locations file
4171
self.assertSectionNames([None, 'DEFAULT'],
4173
# Unless we define an option
4174
self.branch_config._get_location_config().set_user_option(
4175
'file', 'locations')
4176
self.assertSectionNames([self.tree.basedir, None, 'DEFAULT'],
4179
def test_breezy_named_section(self):
4180
# We need to cheat as the API doesn't give direct access to sections
4181
# other than DEFAULT.
4182
self.breezy_config.set_alias('breezy', 'bzr')
4183
self.assertSectionNames(['ALIASES'], self.breezy_config, 'ALIASES')
4186
class TestSharedStores(tests.TestCaseInTempDir):
4188
def test_breezy_conf_shared(self):
4189
g1 = config.GlobalStack()
4190
g2 = config.GlobalStack()
4191
# The two stacks share the same store
4192
self.assertIs(g1.store, g2.store)
4195
class TestAuthenticationConfigFilePermissions(tests.TestCaseInTempDir):
4196
"""Test warning for permissions of authentication.conf."""
4199
super(TestAuthenticationConfigFilePermissions, self).setUp()
4200
self.path = osutils.pathjoin(self.test_dir, 'authentication.conf')
4201
with open(self.path, 'wb') as f:
4202
f.write(b"""[broken]
4205
port=port # Error: Not an int
4207
self.overrideAttr(config, 'authentication_config_filename',
4209
osutils.chmod_if_possible(self.path, 0o755)
4211
def test_check_warning(self):
4212
conf = config.AuthenticationConfig()
4213
self.assertEqual(conf._filename, self.path)
4214
self.assertContainsRe(self.get_log(),
4215
'Saved passwords may be accessible by other users.')
4217
def test_check_suppressed_warning(self):
4218
global_config = config.GlobalConfig()
4219
global_config.set_user_option('suppress_warnings',
4220
'insecure_permissions')
4221
conf = config.AuthenticationConfig()
4222
self.assertEqual(conf._filename, self.path)
4223
self.assertNotContainsRe(self.get_log(),
4224
'Saved passwords may be accessible by other users.')
4227
class TestAuthenticationConfigFile(tests.TestCase):
4228
"""Test the authentication.conf file matching"""
4230
def _got_user_passwd(self, expected_user, expected_password,
4231
config, *args, **kwargs):
4232
credentials = config.get_credentials(*args, **kwargs)
4233
if credentials is None:
4237
user = credentials['user']
4238
password = credentials['password']
4239
self.assertEqual(expected_user, user)
4240
self.assertEqual(expected_password, password)
4242
def test_empty_config(self):
4243
conf = config.AuthenticationConfig(_file=BytesIO())
4244
self.assertEqual({}, conf._get_config())
4245
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
4247
def test_non_utf8_config(self):
4248
conf = config.AuthenticationConfig(_file=BytesIO(b'foo = bar\xff'))
4249
self.assertRaises(config.ConfigContentError, conf._get_config)
4251
def test_missing_auth_section_header(self):
4252
conf = config.AuthenticationConfig(_file=BytesIO(b'foo = bar'))
4253
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4255
def test_auth_section_header_not_closed(self):
4256
conf = config.AuthenticationConfig(_file=BytesIO(b'[DEF'))
4257
self.assertRaises(config.ParseConfigError, conf._get_config)
4259
def test_auth_value_not_boolean(self):
4260
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4264
verify_certificates=askme # Error: Not a boolean
4266
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4268
def test_auth_value_not_int(self):
4269
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4273
port=port # Error: Not an int
4275
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4277
def test_unknown_password_encoding(self):
4278
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4282
password_encoding=unknown
4284
self.assertRaises(ValueError, conf.get_password,
4285
'ftp', 'foo.net', 'joe')
4287
def test_credentials_for_scheme_host(self):
4288
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4289
# Identity on foo.net
4294
password=secret-pass
4297
self._got_user_passwd('joe', 'secret-pass', conf, 'ftp', 'foo.net')
4299
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
4301
self._got_user_passwd(None, None, conf, 'ftp', 'bar.net')
4303
def test_credentials_for_host_port(self):
4304
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4305
# Identity on foo.net
4311
password=secret-pass
4314
self._got_user_passwd('joe', 'secret-pass',
4315
conf, 'ftp', 'foo.net', port=10021)
4317
self._got_user_passwd(None, None, conf, 'ftp', 'foo.net')
4319
def test_for_matching_host(self):
4320
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4321
# Identity on foo.net
4327
[sourceforge domain]
4334
self._got_user_passwd('georges', 'bendover',
4335
conf, 'bzr', 'foo.bzr.sf.net')
4337
self._got_user_passwd(None, None,
4338
conf, 'bzr', 'bbzr.sf.net')
4340
def test_for_matching_host_None(self):
4341
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4342
# Identity on foo.net
4352
self._got_user_passwd('joe', 'joepass',
4353
conf, 'bzr', 'quux.net')
4354
# no host but different scheme
4355
self._got_user_passwd('georges', 'bendover',
4356
conf, 'ftp', 'quux.net')
4358
def test_credentials_for_path(self):
4359
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4374
self._got_user_passwd(None, None,
4375
conf, 'http', host='bar.org', path='/dir3')
4377
self._got_user_passwd('georges', 'bendover',
4378
conf, 'http', host='bar.org', path='/dir2')
4380
self._got_user_passwd('jim', 'jimpass',
4381
conf, 'http', host='bar.org', path='/dir1/subdir')
4383
def test_credentials_for_user(self):
4384
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4392
self._got_user_passwd('jim', 'jimpass',
4393
conf, 'http', 'bar.org')
4395
self._got_user_passwd('jim', 'jimpass',
4396
conf, 'http', 'bar.org', user='jim')
4397
# Don't get a different user if one is specified
4398
self._got_user_passwd(None, None,
4399
conf, 'http', 'bar.org', user='georges')
4401
def test_credentials_for_user_without_password(self):
4402
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4408
# Get user but no password
4409
self._got_user_passwd('jim', None,
4410
conf, 'http', 'bar.org')
4412
def test_verify_certificates(self):
4413
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4419
verify_certificates=False
4426
credentials = conf.get_credentials('https', 'bar.org')
4427
self.assertEqual(False, credentials.get('verify_certificates'))
4428
credentials = conf.get_credentials('https', 'foo.net')
4429
self.assertEqual(True, credentials.get('verify_certificates'))
4432
class TestAuthenticationStorage(tests.TestCaseInTempDir):
4434
def test_set_credentials(self):
4435
conf = config.AuthenticationConfig()
4436
conf.set_credentials('name', 'host', 'user', 'scheme', 'password',
4437
99, path='/foo', verify_certificates=False, realm='realm')
4438
credentials = conf.get_credentials(host='host', scheme='scheme',
4439
port=99, path='/foo',
4441
CREDENTIALS = {'name': 'name', 'user': 'user', 'password': 'password',
4442
'verify_certificates': False, 'scheme': 'scheme',
4443
'host': 'host', 'port': 99, 'path': '/foo',
4445
self.assertEqual(CREDENTIALS, credentials)
4446
credentials_from_disk = config.AuthenticationConfig().get_credentials(
4447
host='host', scheme='scheme', port=99, path='/foo', realm='realm')
4448
self.assertEqual(CREDENTIALS, credentials_from_disk)
4450
def test_reset_credentials_different_name(self):
4451
conf = config.AuthenticationConfig()
4452
conf.set_credentials('name', 'host', 'user', 'scheme', 'password'),
4453
conf.set_credentials('name2', 'host', 'user2', 'scheme', 'password'),
4454
self.assertIs(None, conf._get_config().get('name'))
4455
credentials = conf.get_credentials(host='host', scheme='scheme')
4456
CREDENTIALS = {'name': 'name2', 'user': 'user2', 'password':
4457
'password', 'verify_certificates': True,
4458
'scheme': 'scheme', 'host': 'host', 'port': None,
4459
'path': None, 'realm': None}
4460
self.assertEqual(CREDENTIALS, credentials)
4463
class TestAuthenticationConfig(tests.TestCaseInTempDir):
4464
"""Test AuthenticationConfig behaviour"""
4466
def _check_default_password_prompt(self, expected_prompt_format, scheme,
4467
host=None, port=None, realm=None,
4471
user, password = 'jim', 'precious'
4472
expected_prompt = expected_prompt_format % {
4473
'scheme': scheme, 'host': host, 'port': port,
4474
'user': user, 'realm': realm}
4476
ui.ui_factory = tests.TestUIFactory(stdin=password + '\n')
4477
# We use an empty conf so that the user is always prompted
4478
conf = config.AuthenticationConfig()
4479
self.assertEqual(password,
4480
conf.get_password(scheme, host, user, port=port,
4481
realm=realm, path=path))
4482
self.assertEqual(expected_prompt, ui.ui_factory.stderr.getvalue())
4483
self.assertEqual('', ui.ui_factory.stdout.getvalue())
4485
def _check_default_username_prompt(self, expected_prompt_format, scheme,
4486
host=None, port=None, realm=None,
4491
expected_prompt = expected_prompt_format % {
4492
'scheme': scheme, 'host': host, 'port': port,
4494
ui.ui_factory = tests.TestUIFactory(stdin=username + '\n')
4495
# We use an empty conf so that the user is always prompted
4496
conf = config.AuthenticationConfig()
4497
self.assertEqual(username, conf.get_user(scheme, host, port=port,
4498
realm=realm, path=path, ask=True))
4499
self.assertEqual(expected_prompt, ui.ui_factory.stderr.getvalue())
4500
self.assertEqual('', ui.ui_factory.stdout.getvalue())
4502
def test_username_defaults_prompts(self):
4503
# HTTP prompts can't be tested here, see test_http.py
4504
self._check_default_username_prompt(u'FTP %(host)s username: ', 'ftp')
4505
self._check_default_username_prompt(
4506
u'FTP %(host)s:%(port)d username: ', 'ftp', port=10020)
4507
self._check_default_username_prompt(
4508
u'SSH %(host)s:%(port)d username: ', 'ssh', port=12345)
4510
def test_username_default_no_prompt(self):
4511
conf = config.AuthenticationConfig()
4512
self.assertEqual(None,
4513
conf.get_user('ftp', 'example.com'))
4514
self.assertEqual("explicitdefault",
4515
conf.get_user('ftp', 'example.com', default="explicitdefault"))
4517
def test_password_default_prompts(self):
4518
# HTTP prompts can't be tested here, see test_http.py
4519
self._check_default_password_prompt(
4520
u'FTP %(user)s@%(host)s password: ', 'ftp')
4521
self._check_default_password_prompt(
4522
u'FTP %(user)s@%(host)s:%(port)d password: ', 'ftp', port=10020)
4523
self._check_default_password_prompt(
4524
u'SSH %(user)s@%(host)s:%(port)d password: ', 'ssh', port=12345)
4525
# SMTP port handling is a bit special (it's handled if embedded in the
4527
# FIXME: should we: forbid that, extend it to other schemes, leave
4528
# things as they are that's fine thank you ?
4529
self._check_default_password_prompt(
4530
u'SMTP %(user)s@%(host)s password: ', 'smtp')
4531
self._check_default_password_prompt(
4532
u'SMTP %(user)s@%(host)s password: ', 'smtp', host='bar.org:10025')
4533
self._check_default_password_prompt(
4534
u'SMTP %(user)s@%(host)s:%(port)d password: ', 'smtp', port=10025)
4536
def test_ssh_password_emits_warning(self):
4537
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4544
entered_password = 'typed-by-hand'
4545
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n')
4547
# Since the password defined in the authentication config is ignored,
4548
# the user is prompted
4549
self.assertEqual(entered_password,
4550
conf.get_password('ssh', 'bar.org', user='jim'))
4551
self.assertContainsRe(
4553
'password ignored in section \\[ssh with password\\]')
4555
def test_ssh_without_password_doesnt_emit_warning(self):
4556
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4562
entered_password = 'typed-by-hand'
4563
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n')
4565
# Since the password defined in the authentication config is ignored,
4566
# the user is prompted
4567
self.assertEqual(entered_password,
4568
conf.get_password('ssh', 'bar.org', user='jim'))
4569
# No warning shoud be emitted since there is no password. We are only
4571
self.assertNotContainsRe(
4573
'password ignored in section \\[ssh with password\\]')
4575
def test_uses_fallback_stores(self):
4576
self.overrideAttr(config, 'credential_store_registry',
4577
config.CredentialStoreRegistry())
4578
store = StubCredentialStore()
4579
store.add_credentials("http", "example.com", "joe", "secret")
4580
config.credential_store_registry.register("stub", store, fallback=True)
4581
conf = config.AuthenticationConfig(_file=BytesIO())
4582
creds = conf.get_credentials("http", "example.com")
4583
self.assertEqual("joe", creds["user"])
4584
self.assertEqual("secret", creds["password"])
4587
class StubCredentialStore(config.CredentialStore):
4593
def add_credentials(self, scheme, host, user, password=None):
4594
self._username[(scheme, host)] = user
4595
self._password[(scheme, host)] = password
4597
def get_credentials(self, scheme, host, port=None, user=None,
4598
path=None, realm=None):
4599
key = (scheme, host)
4600
if key not in self._username:
4602
return {"scheme": scheme, "host": host, "port": port,
4603
"user": self._username[key], "password": self._password[key]}
4606
class CountingCredentialStore(config.CredentialStore):
4611
def get_credentials(self, scheme, host, port=None, user=None,
4612
path=None, realm=None):
4617
class TestCredentialStoreRegistry(tests.TestCase):
4619
def _get_cs_registry(self):
4620
return config.credential_store_registry
4622
def test_default_credential_store(self):
4623
r = self._get_cs_registry()
4624
default = r.get_credential_store(None)
4625
self.assertIsInstance(default, config.PlainTextCredentialStore)
4627
def test_unknown_credential_store(self):
4628
r = self._get_cs_registry()
4629
# It's hard to imagine someone creating a credential store named
4630
# 'unknown' so we use that as an never registered key.
4631
self.assertRaises(KeyError, r.get_credential_store, 'unknown')
4633
def test_fallback_none_registered(self):
4634
r = config.CredentialStoreRegistry()
4635
self.assertEqual(None,
4636
r.get_fallback_credentials("http", "example.com"))
4638
def test_register(self):
4639
r = config.CredentialStoreRegistry()
4640
r.register("stub", StubCredentialStore(), fallback=False)
4641
r.register("another", StubCredentialStore(), fallback=True)
4642
self.assertEqual(["another", "stub"], r.keys())
4644
def test_register_lazy(self):
4645
r = config.CredentialStoreRegistry()
4646
r.register_lazy("stub", "breezy.tests.test_config",
4647
"StubCredentialStore", fallback=False)
4648
self.assertEqual(["stub"], r.keys())
4649
self.assertIsInstance(r.get_credential_store("stub"),
4650
StubCredentialStore)
4652
def test_is_fallback(self):
4653
r = config.CredentialStoreRegistry()
4654
r.register("stub1", None, fallback=False)
4655
r.register("stub2", None, fallback=True)
4656
self.assertEqual(False, r.is_fallback("stub1"))
4657
self.assertEqual(True, r.is_fallback("stub2"))
4659
def test_no_fallback(self):
4660
r = config.CredentialStoreRegistry()
4661
store = CountingCredentialStore()
4662
r.register("count", store, fallback=False)
4663
self.assertEqual(None,
4664
r.get_fallback_credentials("http", "example.com"))
4665
self.assertEqual(0, store._calls)
4667
def test_fallback_credentials(self):
4668
r = config.CredentialStoreRegistry()
4669
store = StubCredentialStore()
4670
store.add_credentials("http", "example.com",
4671
"somebody", "geheim")
4672
r.register("stub", store, fallback=True)
4673
creds = r.get_fallback_credentials("http", "example.com")
4674
self.assertEqual("somebody", creds["user"])
4675
self.assertEqual("geheim", creds["password"])
4677
def test_fallback_first_wins(self):
4678
r = config.CredentialStoreRegistry()
4679
stub1 = StubCredentialStore()
4680
stub1.add_credentials("http", "example.com",
4681
"somebody", "stub1")
4682
r.register("stub1", stub1, fallback=True)
4683
stub2 = StubCredentialStore()
4684
stub2.add_credentials("http", "example.com",
4685
"somebody", "stub2")
4686
r.register("stub2", stub1, fallback=True)
4687
creds = r.get_fallback_credentials("http", "example.com")
4688
self.assertEqual("somebody", creds["user"])
4689
self.assertEqual("stub1", creds["password"])
4692
class TestPlainTextCredentialStore(tests.TestCase):
4694
def test_decode_password(self):
4695
r = config.credential_store_registry
4696
plain_text = r.get_credential_store()
4697
decoded = plain_text.decode_password(dict(password='secret'))
4698
self.assertEqual('secret', decoded)
4701
class TestBase64CredentialStore(tests.TestCase):
4703
def test_decode_password(self):
4704
r = config.credential_store_registry
4705
plain_text = r.get_credential_store('base64')
4706
decoded = plain_text.decode_password(dict(password='c2VjcmV0'))
4707
self.assertEqual(b'secret', decoded)
4710
# FIXME: Once we have a way to declare authentication to all test servers, we
4711
# can implement generic tests.
4712
# test_user_password_in_url
4713
# test_user_in_url_password_from_config
4714
# test_user_in_url_password_prompted
4715
# test_user_in_config
4716
# test_user_getpass.getuser
4717
# test_user_prompted ?
4718
class TestAuthenticationRing(tests.TestCaseWithTransport):
4722
class TestAutoUserId(tests.TestCase):
4723
"""Test inferring an automatic user name."""
4725
def test_auto_user_id(self):
4726
"""Automatic inference of user name.
4728
This is a bit hard to test in an isolated way, because it depends on
4729
system functions that go direct to /etc or perhaps somewhere else.
4730
But it's reasonable to say that on Unix, with an /etc/mailname, we ought
4731
to be able to choose a user name with no configuration.
4733
if sys.platform == 'win32':
4734
raise tests.TestSkipped(
4735
"User name inference not implemented on win32")
4736
realname, address = config._auto_user_id()
4737
if os.path.exists('/etc/mailname'):
4738
self.assertIsNot(None, realname)
4739
self.assertIsNot(None, address)
4741
self.assertEqual((None, None), (realname, address))
4744
class TestDefaultMailDomain(tests.TestCaseInTempDir):
4745
"""Test retrieving default domain from mailname file"""
4747
def test_default_mail_domain_simple(self):
4748
with open('simple', 'w') as f:
4749
f.write("domainname.com\n")
4750
r = config._get_default_mail_domain('simple')
4751
self.assertEqual('domainname.com', r)
4753
def test_default_mail_domain_no_eol(self):
4754
with open('no_eol', 'w') as f:
4755
f.write("domainname.com")
4756
r = config._get_default_mail_domain('no_eol')
4757
self.assertEqual('domainname.com', r)
4759
def test_default_mail_domain_multiple_lines(self):
4760
with open('multiple_lines', 'w') as f:
4761
f.write("domainname.com\nsome other text\n")
4762
r = config._get_default_mail_domain('multiple_lines')
4763
self.assertEqual('domainname.com', r)
4766
class EmailOptionTests(tests.TestCase):
4768
def test_default_email_uses_BRZ_EMAIL(self):
4769
conf = config.MemoryStack(b'email=jelmer@debian.org')
4770
# BRZ_EMAIL takes precedence over EMAIL
4771
self.overrideEnv('BRZ_EMAIL', 'jelmer@samba.org')
4772
self.overrideEnv('EMAIL', 'jelmer@apache.org')
4773
self.assertEqual('jelmer@samba.org', conf.get('email'))
4775
def test_default_email_uses_EMAIL(self):
4776
conf = config.MemoryStack(b'')
4777
self.overrideEnv('BRZ_EMAIL', None)
4778
self.overrideEnv('EMAIL', 'jelmer@apache.org')
4779
self.assertEqual('jelmer@apache.org', conf.get('email'))
4781
def test_BRZ_EMAIL_overrides(self):
4782
conf = config.MemoryStack(b'email=jelmer@debian.org')
4783
self.overrideEnv('BRZ_EMAIL', 'jelmer@apache.org')
4784
self.assertEqual('jelmer@apache.org', conf.get('email'))
4785
self.overrideEnv('BRZ_EMAIL', None)
4786
self.overrideEnv('EMAIL', 'jelmer@samba.org')
4787
self.assertEqual('jelmer@debian.org', conf.get('email'))
4790
class MailClientOptionTests(tests.TestCase):
4792
def test_default(self):
4793
conf = config.MemoryStack(b'')
4794
client = conf.get('mail_client')
4795
self.assertIs(client, mail_client.DefaultMail)
4797
def test_evolution(self):
4798
conf = config.MemoryStack(b'mail_client=evolution')
4799
client = conf.get('mail_client')
4800
self.assertIs(client, mail_client.Evolution)
4802
def test_kmail(self):
4803
conf = config.MemoryStack(b'mail_client=kmail')
4804
client = conf.get('mail_client')
4805
self.assertIs(client, mail_client.KMail)
4807
def test_mutt(self):
4808
conf = config.MemoryStack(b'mail_client=mutt')
4809
client = conf.get('mail_client')
4810
self.assertIs(client, mail_client.Mutt)
4812
def test_thunderbird(self):
4813
conf = config.MemoryStack(b'mail_client=thunderbird')
4814
client = conf.get('mail_client')
4815
self.assertIs(client, mail_client.Thunderbird)
4817
def test_explicit_default(self):
4818
conf = config.MemoryStack(b'mail_client=default')
4819
client = conf.get('mail_client')
4820
self.assertIs(client, mail_client.DefaultMail)
4822
def test_editor(self):
4823
conf = config.MemoryStack(b'mail_client=editor')
4824
client = conf.get('mail_client')
4825
self.assertIs(client, mail_client.Editor)
4827
def test_mapi(self):
4828
conf = config.MemoryStack(b'mail_client=mapi')
4829
client = conf.get('mail_client')
4830
self.assertIs(client, mail_client.MAPIClient)
4832
def test_xdg_email(self):
4833
conf = config.MemoryStack(b'mail_client=xdg-email')
4834
client = conf.get('mail_client')
4835
self.assertIs(client, mail_client.XDGEmail)
4837
def test_unknown(self):
4838
conf = config.MemoryStack(b'mail_client=firebird')
4839
self.assertRaises(config.ConfigOptionValueError, conf.get,
added
removed
breezy/tests/test_config.py
