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
37
registry as _mod_registry,
44
from ..sixish import (
47
from ..transport import remote as transport_remote
55
def lockable_config_scenarios():
58
{'config_class': config.GlobalConfig,
60
'config_section': 'DEFAULT'}),
62
{'config_class': config.LocationConfig,
64
'config_section': '.'}),]
67
load_tests = scenarios.load_tests_apply_scenarios
69
# Register helpers to build stores
70
config.test_store_builder_registry.register(
71
'configobj', lambda test: config.TransportIniFileStore(
72
test.get_transport(), 'configobj.conf'))
73
config.test_store_builder_registry.register(
74
'breezy', lambda test: config.GlobalStore())
75
config.test_store_builder_registry.register(
76
'location', lambda test: config.LocationStore())
79
def build_backing_branch(test, relpath,
80
transport_class=None, server_class=None):
81
"""Test helper to create a backing branch only once.
83
Some tests needs multiple stores/stacks to check concurrent update
84
behaviours. As such, they need to build different branch *objects* even if
85
they share the branch on disk.
87
:param relpath: The relative path to the branch. (Note that the helper
88
should always specify the same relpath).
90
:param transport_class: The Transport class the test needs to use.
92
:param server_class: The server associated with the ``transport_class``
95
Either both or neither of ``transport_class`` and ``server_class`` should
98
if transport_class is not None and server_class is not None:
99
test.transport_class = transport_class
100
test.transport_server = server_class
101
elif not (transport_class is None and server_class is None):
102
raise AssertionError('Specify both ``transport_class`` and '
103
'``server_class`` or neither of them')
104
if getattr(test, 'backing_branch', None) is None:
105
# First call, let's build the branch on disk
106
test.backing_branch = test.make_branch(relpath)
109
def build_branch_store(test):
110
build_backing_branch(test, 'branch')
111
b = branch.Branch.open('branch')
112
return config.BranchStore(b)
113
config.test_store_builder_registry.register('branch', build_branch_store)
116
def build_control_store(test):
117
build_backing_branch(test, 'branch')
118
b = controldir.ControlDir.open('branch')
119
return config.ControlStore(b)
120
config.test_store_builder_registry.register('control', build_control_store)
123
def build_remote_branch_store(test):
124
# There is only one permutation (but we won't be able to handle more with
125
# this design anyway)
127
server_class) = transport_remote.get_test_permutations()[0]
128
build_backing_branch(test, 'branch', transport_class, server_class)
129
b = branch.Branch.open(test.get_url('branch'))
130
return config.BranchStore(b)
131
config.test_store_builder_registry.register('remote_branch',
132
build_remote_branch_store)
135
config.test_stack_builder_registry.register(
136
'breezy', lambda test: config.GlobalStack())
137
config.test_stack_builder_registry.register(
138
'location', lambda test: config.LocationStack('.'))
141
def build_branch_stack(test):
142
build_backing_branch(test, 'branch')
143
b = branch.Branch.open('branch')
144
return config.BranchStack(b)
145
config.test_stack_builder_registry.register('branch', build_branch_stack)
148
def build_branch_only_stack(test):
149
# There is only one permutation (but we won't be able to handle more with
150
# this design anyway)
152
server_class) = transport_remote.get_test_permutations()[0]
153
build_backing_branch(test, 'branch', transport_class, server_class)
154
b = branch.Branch.open(test.get_url('branch'))
155
return config.BranchOnlyStack(b)
156
config.test_stack_builder_registry.register('branch_only',
157
build_branch_only_stack)
159
def build_remote_control_stack(test):
160
# There is only one permutation (but we won't be able to handle more with
161
# this design anyway)
163
server_class) = transport_remote.get_test_permutations()[0]
164
# We need only a bzrdir for this, not a full branch, but it's not worth
165
# creating a dedicated helper to create only the bzrdir
166
build_backing_branch(test, 'branch', transport_class, server_class)
167
b = branch.Branch.open(test.get_url('branch'))
168
return config.RemoteControlStack(b.controldir)
169
config.test_stack_builder_registry.register('remote_control',
170
build_remote_control_stack)
173
sample_long_alias="log -r-15..-1 --line"
174
sample_config_text = u"""
176
email=Erik B\u00e5gfors <erik@bagfors.nu>
178
change_editor=vimdiff -of @new_path @old_path
179
gpg_signing_key=DD4D5088
181
validate_signatures_in_log=true
183
user_global_option=something
184
bzr.mergetool.sometool=sometool {base} {this} {other} -o {result}
185
bzr.mergetool.funkytool=funkytool "arg with spaces" {this_temp}
186
bzr.mergetool.newtool='"newtool with spaces" {this_temp}'
187
bzr.default_mergetool=sometool
190
ll=""" + sample_long_alias + "\n"
193
sample_always_signatures = """
195
check_signatures=ignore
196
create_signatures=always
199
sample_ignore_signatures = """
201
check_signatures=require
202
create_signatures=never
205
sample_maybe_signatures = """
207
check_signatures=ignore
208
create_signatures=when-required
211
sample_branches_text = """
212
[http://www.example.com]
214
email=Robert Collins <robertc@example.org>
215
normal_option = normal
216
appendpath_option = append
217
appendpath_option:policy = appendpath
218
norecurse_option = norecurse
219
norecurse_option:policy = norecurse
220
[http://www.example.com/ignoreparent]
221
# different project: ignore parent dir config
223
[http://www.example.com/norecurse]
224
# configuration items that only apply to this dir
226
normal_option = norecurse
227
[http://www.example.com/dir]
228
appendpath_option = normal
230
check_signatures=require
231
# test trailing / matching with no children
233
check_signatures=check-available
234
gpg_signing_key=default
235
user_local_option=local
236
# test trailing / matching
238
#subdirs will match but not the parent
240
check_signatures=ignore
241
post_commit=breezy.tests.test_config.post_commit
242
#testing explicit beats globs
246
def create_configs(test):
247
"""Create configuration files for a given test.
249
This requires creating a tree (and populate the ``test.tree`` attribute)
250
and its associated branch and will populate the following attributes:
252
- branch_config: A BranchConfig for the associated branch.
254
- locations_config : A LocationConfig for the associated branch
256
- breezy_config: A GlobalConfig.
258
The tree and branch are created in a 'tree' subdirectory so the tests can
259
still use the test directory to stay outside of the branch.
261
tree = test.make_branch_and_tree('tree')
263
test.branch_config = config.BranchConfig(tree.branch)
264
test.locations_config = config.LocationConfig(tree.basedir)
265
test.breezy_config = config.GlobalConfig()
268
def create_configs_with_file_option(test):
269
"""Create configuration files with a ``file`` option set in each.
271
This builds on ``create_configs`` and add one ``file`` option in each
272
configuration with a value which allows identifying the configuration file.
275
test.breezy_config.set_user_option('file', 'breezy')
276
test.locations_config.set_user_option('file', 'locations')
277
test.branch_config.set_user_option('file', 'branch')
280
class TestOptionsMixin:
282
def assertOptions(self, expected, conf):
283
# We don't care about the parser (as it will make tests hard to write
284
# and error-prone anyway)
285
self.assertThat([opt[:4] for opt in conf._get_options()],
286
matchers.Equals(expected))
289
class InstrumentedConfigObj(object):
290
"""A config obj look-enough-alike to record calls made to it."""
292
def __contains__(self, thing):
293
self._calls.append(('__contains__', thing))
296
def __getitem__(self, key):
297
self._calls.append(('__getitem__', key))
300
def __init__(self, input, encoding=None):
301
self._calls = [('__init__', input, encoding)]
303
def __setitem__(self, key, value):
304
self._calls.append(('__setitem__', key, value))
306
def __delitem__(self, key):
307
self._calls.append(('__delitem__', key))
310
self._calls.append(('keys',))
314
self._calls.append(('reload',))
316
def write(self, arg):
317
self._calls.append(('write',))
319
def as_bool(self, value):
320
self._calls.append(('as_bool', value))
323
def get_value(self, section, name):
324
self._calls.append(('get_value', section, name))
328
class FakeBranch(object):
330
def __init__(self, base=None):
332
self.base = "http://example.com/branches/demo"
335
self._transport = self.control_files = \
336
FakeControlFilesAndTransport()
338
def _get_config(self):
339
return config.TransportConfig(self._transport, 'branch.conf')
341
def lock_write(self):
348
class FakeControlFilesAndTransport(object):
352
self._transport = self
354
def get(self, filename):
357
return BytesIO(self.files[filename])
359
raise errors.NoSuchFile(filename)
361
def get_bytes(self, filename):
364
return self.files[filename]
366
raise errors.NoSuchFile(filename)
368
def put(self, filename, fileobj):
369
self.files[filename] = fileobj.read()
371
def put_file(self, filename, fileobj):
372
return self.put(filename, fileobj)
375
class InstrumentedConfig(config.Config):
376
"""An instrumented config that supplies stubs for template methods."""
379
super(InstrumentedConfig, self).__init__()
381
self._signatures = config.CHECK_NEVER
383
def _get_user_id(self):
384
self._calls.append('_get_user_id')
385
return "Robert Collins <robert.collins@example.org>"
387
def _get_signature_checking(self):
388
self._calls.append('_get_signature_checking')
389
return self._signatures
391
def _get_change_editor(self):
392
self._calls.append('_get_change_editor')
393
return 'vimdiff -fo @new_path @old_path'
396
bool_config = """[DEFAULT]
405
class TestConfigObj(tests.TestCase):
407
def test_get_bool(self):
408
co = config.ConfigObj(BytesIO(bool_config))
409
self.assertIs(co.get_bool('DEFAULT', 'active'), True)
410
self.assertIs(co.get_bool('DEFAULT', 'inactive'), False)
411
self.assertIs(co.get_bool('UPPERCASE', 'active'), True)
412
self.assertIs(co.get_bool('UPPERCASE', 'nonactive'), False)
414
def test_hash_sign_in_value(self):
416
Before 4.5.0, ConfigObj did not quote # signs in values, so they'd be
417
treated as comments when read in again. (#86838)
419
co = config.ConfigObj()
420
co['test'] = 'foo#bar'
422
co.write(outfile=outfile)
423
lines = outfile.getvalue().splitlines()
424
self.assertEqual(lines, ['test = "foo#bar"'])
425
co2 = config.ConfigObj(lines)
426
self.assertEqual(co2['test'], 'foo#bar')
428
def test_triple_quotes(self):
429
# Bug #710410: if the value string has triple quotes
430
# then ConfigObj versions up to 4.7.2 will quote them wrong
431
# and won't able to read them back
432
triple_quotes_value = '''spam
433
""" that's my spam """
435
co = config.ConfigObj()
436
co['test'] = triple_quotes_value
437
# While writing this test another bug in ConfigObj has been found:
438
# method co.write() without arguments produces list of lines
439
# one option per line, and multiline values are not split
440
# across multiple lines,
441
# and that breaks the parsing these lines back by ConfigObj.
442
# This issue only affects test, but it's better to avoid
443
# `co.write()` construct at all.
444
# [bialix 20110222] bug report sent to ConfigObj's author
446
co.write(outfile=outfile)
447
output = outfile.getvalue()
448
# now we're trying to read it back
449
co2 = config.ConfigObj(BytesIO(output))
450
self.assertEqual(triple_quotes_value, co2['test'])
453
erroneous_config = """[section] # line 1
456
whocares=notme # line 4
460
class TestConfigObjErrors(tests.TestCase):
462
def test_duplicate_section_name_error_line(self):
464
co = configobj.ConfigObj(BytesIO(erroneous_config),
466
except config.configobj.DuplicateError as e:
467
self.assertEqual(3, e.line_number)
469
self.fail('Error in config file not detected')
472
class TestConfig(tests.TestCase):
474
def test_constructs(self):
477
def test_user_email(self):
478
my_config = InstrumentedConfig()
479
self.assertEqual('robert.collins@example.org', my_config.user_email())
480
self.assertEqual(['_get_user_id'], my_config._calls)
482
def test_username(self):
483
my_config = InstrumentedConfig()
484
self.assertEqual('Robert Collins <robert.collins@example.org>',
485
my_config.username())
486
self.assertEqual(['_get_user_id'], my_config._calls)
488
def test_get_user_option_default(self):
489
my_config = config.Config()
490
self.assertEqual(None, my_config.get_user_option('no_option'))
492
def test_validate_signatures_in_log_default(self):
493
my_config = config.Config()
494
self.assertEqual(False, my_config.validate_signatures_in_log())
496
def test_get_change_editor(self):
497
my_config = InstrumentedConfig()
498
change_editor = my_config.get_change_editor('old_tree', 'new_tree')
499
self.assertEqual(['_get_change_editor'], my_config._calls)
500
self.assertIs(diff.DiffFromTool, change_editor.__class__)
501
self.assertEqual(['vimdiff', '-fo', '@new_path', '@old_path'],
502
change_editor.command_template)
505
class TestConfigPath(tests.TestCase):
508
super(TestConfigPath, self).setUp()
509
self.overrideEnv('HOME', '/home/bogus')
510
self.overrideEnv('XDG_CACHE_HOME', '')
511
if sys.platform == 'win32':
514
r'C:\Documents and Settings\bogus\Application Data')
516
'C:/Documents and Settings/bogus/Application Data/breezy'
518
self.brz_home = '/home/bogus/.config/breezy'
520
def test_config_dir(self):
521
self.assertEqual(config.config_dir(), self.brz_home)
523
def test_config_dir_is_unicode(self):
524
self.assertIsInstance(config.config_dir(), unicode)
526
def test_config_filename(self):
527
self.assertEqual(config.config_filename(),
528
self.brz_home + '/breezy.conf')
530
def test_locations_config_filename(self):
531
self.assertEqual(config.locations_config_filename(),
532
self.brz_home + '/locations.conf')
534
def test_authentication_config_filename(self):
535
self.assertEqual(config.authentication_config_filename(),
536
self.brz_home + '/authentication.conf')
538
def test_xdg_cache_dir(self):
539
self.assertEqual(config.xdg_cache_dir(),
540
'/home/bogus/.cache')
543
class TestConfigPathFallback(tests.TestCaseInTempDir):
546
super(TestConfigPathFallback, self).setUp()
547
self.overrideEnv('HOME', self.test_dir)
548
self.overrideEnv('XDG_CACHE_HOME', '')
549
self.bzr_home = os.path.join(self.test_dir, '.bazaar')
550
os.mkdir(self.bzr_home)
552
def test_config_dir(self):
553
self.assertEqual(config.config_dir(), self.bzr_home)
555
def test_config_dir_is_unicode(self):
556
self.assertIsInstance(config.config_dir(), unicode)
558
def test_config_filename(self):
559
self.assertEqual(config.config_filename(),
560
self.bzr_home + '/bazaar.conf')
562
def test_locations_config_filename(self):
563
self.assertEqual(config.locations_config_filename(),
564
self.bzr_home + '/locations.conf')
566
def test_authentication_config_filename(self):
567
self.assertEqual(config.authentication_config_filename(),
568
self.bzr_home + '/authentication.conf')
570
def test_xdg_cache_dir(self):
571
self.assertEqual(config.xdg_cache_dir(),
572
os.path.join(self.test_dir, '.cache'))
575
class TestXDGConfigDir(tests.TestCaseInTempDir):
576
# must be in temp dir because config tests for the existence of the bazaar
577
# subdirectory of $XDG_CONFIG_HOME
580
if sys.platform == 'win32':
581
raise tests.TestNotApplicable(
582
'XDG config dir not used on this platform')
583
super(TestXDGConfigDir, self).setUp()
584
self.overrideEnv('HOME', self.test_home_dir)
585
# BRZ_HOME overrides everything we want to test so unset it.
586
self.overrideEnv('BRZ_HOME', None)
588
def test_xdg_config_dir_exists(self):
589
"""When ~/.config/bazaar exists, use it as the config dir."""
590
newdir = osutils.pathjoin(self.test_home_dir, '.config', 'bazaar')
592
self.assertEqual(config.config_dir(), newdir)
594
def test_xdg_config_home(self):
595
"""When XDG_CONFIG_HOME is set, use it."""
596
xdgconfigdir = osutils.pathjoin(self.test_home_dir, 'xdgconfig')
597
self.overrideEnv('XDG_CONFIG_HOME', xdgconfigdir)
598
newdir = osutils.pathjoin(xdgconfigdir, 'bazaar')
600
self.assertEqual(config.config_dir(), newdir)
603
class TestIniConfig(tests.TestCaseInTempDir):
605
def make_config_parser(self, s):
606
conf = config.IniBasedConfig.from_string(s)
607
return conf, conf._get_parser()
610
class TestIniConfigBuilding(TestIniConfig):
612
def test_contructs(self):
613
config.IniBasedConfig()
615
def test_from_fp(self):
616
my_config = config.IniBasedConfig.from_string(sample_config_text)
617
self.assertIsInstance(my_config._get_parser(), configobj.ConfigObj)
619
def test_cached(self):
620
my_config = config.IniBasedConfig.from_string(sample_config_text)
621
parser = my_config._get_parser()
622
self.assertTrue(my_config._get_parser() is parser)
624
def _dummy_chown(self, path, uid, gid):
625
self.path, self.uid, self.gid = path, uid, gid
627
def test_ini_config_ownership(self):
628
"""Ensure that chown is happening during _write_config_file"""
629
self.requireFeature(features.chown_feature)
630
self.overrideAttr(os, 'chown', self._dummy_chown)
631
self.path = self.uid = self.gid = None
632
conf = config.IniBasedConfig(file_name='./foo.conf')
633
conf._write_config_file()
634
self.assertEqual(self.path, './foo.conf')
635
self.assertTrue(isinstance(self.uid, int))
636
self.assertTrue(isinstance(self.gid, int))
639
class TestIniConfigSaving(tests.TestCaseInTempDir):
641
def test_cant_save_without_a_file_name(self):
642
conf = config.IniBasedConfig()
643
self.assertRaises(AssertionError, conf._write_config_file)
645
def test_saved_with_content(self):
646
content = 'foo = bar\n'
647
config.IniBasedConfig.from_string(content, file_name='./test.conf',
649
self.assertFileEqual(content, 'test.conf')
652
class TestIniConfigOptionExpansion(tests.TestCase):
653
"""Test option expansion from the IniConfig level.
655
What we really want here is to test the Config level, but the class being
656
abstract as far as storing values is concerned, this can't be done
659
# FIXME: This should be rewritten when all configs share a storage
660
# implementation -- vila 2011-02-18
662
def get_config(self, string=None):
665
c = config.IniBasedConfig.from_string(string)
668
def assertExpansion(self, expected, conf, string, env=None):
669
self.assertEqual(expected, conf.expand_options(string, env))
671
def test_no_expansion(self):
672
c = self.get_config('')
673
self.assertExpansion('foo', c, 'foo')
675
def test_env_adding_options(self):
676
c = self.get_config('')
677
self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})
679
def test_env_overriding_options(self):
680
c = self.get_config('foo=baz')
681
self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})
683
def test_simple_ref(self):
684
c = self.get_config('foo=xxx')
685
self.assertExpansion('xxx', c, '{foo}')
687
def test_unknown_ref(self):
688
c = self.get_config('')
689
self.assertRaises(config.ExpandingUnknownOption,
690
c.expand_options, '{foo}')
692
def test_indirect_ref(self):
693
c = self.get_config('''
697
self.assertExpansion('xxx', c, '{bar}')
699
def test_embedded_ref(self):
700
c = self.get_config('''
704
self.assertExpansion('xxx', c, '{{bar}}')
706
def test_simple_loop(self):
707
c = self.get_config('foo={foo}')
708
self.assertRaises(config.OptionExpansionLoop, c.expand_options,
711
def test_indirect_loop(self):
712
c = self.get_config('''
716
e = self.assertRaises(config.OptionExpansionLoop,
717
c.expand_options, '{foo}')
718
self.assertEqual('foo->bar->baz', e.refs)
719
self.assertEqual('{foo}', e.string)
722
conf = self.get_config('''
726
list={foo},{bar},{baz}
728
self.assertEqual(['start', 'middle', 'end'],
729
conf.get_user_option('list', expand=True))
731
def test_cascading_list(self):
732
conf = self.get_config('''
738
self.assertEqual(['start', 'middle', 'end'],
739
conf.get_user_option('list', expand=True))
741
def test_pathological_hidden_list(self):
742
conf = self.get_config('''
748
hidden={start}{middle}{end}
750
# Nope, it's either a string or a list, and the list wins as soon as a
751
# ',' appears, so the string concatenation never occur.
752
self.assertEqual(['{foo', '}', '{', 'bar}'],
753
conf.get_user_option('hidden', expand=True))
756
class TestLocationConfigOptionExpansion(tests.TestCaseInTempDir):
758
def get_config(self, location, string=None):
761
# Since we don't save the config we won't strictly require to inherit
762
# from TestCaseInTempDir, but an error occurs so quickly...
763
c = config.LocationConfig.from_string(string, location)
766
def test_dont_cross_unrelated_section(self):
767
c = self.get_config('/another/branch/path','''
772
[/another/branch/path]
775
self.assertRaises(config.ExpandingUnknownOption,
776
c.get_user_option, 'bar', expand=True)
778
def test_cross_related_sections(self):
779
c = self.get_config('/project/branch/path','''
783
[/project/branch/path]
786
self.assertEqual('quux', c.get_user_option('bar', expand=True))
789
class TestIniBaseConfigOnDisk(tests.TestCaseInTempDir):
791
def test_cannot_reload_without_name(self):
792
conf = config.IniBasedConfig.from_string(sample_config_text)
793
self.assertRaises(AssertionError, conf.reload)
795
def test_reload_see_new_value(self):
796
c1 = config.IniBasedConfig.from_string('editor=vim\n',
797
file_name='./test/conf')
798
c1._write_config_file()
799
c2 = config.IniBasedConfig.from_string('editor=emacs\n',
800
file_name='./test/conf')
801
c2._write_config_file()
802
self.assertEqual('vim', c1.get_user_option('editor'))
803
self.assertEqual('emacs', c2.get_user_option('editor'))
804
# Make sure we get the Right value
806
self.assertEqual('emacs', c1.get_user_option('editor'))
809
class TestLockableConfig(tests.TestCaseInTempDir):
811
scenarios = lockable_config_scenarios()
816
config_section = None
819
super(TestLockableConfig, self).setUp()
820
self._content = '[%s]\none=1\ntwo=2\n' % (self.config_section,)
821
self.config = self.create_config(self._content)
823
def get_existing_config(self):
824
return self.config_class(*self.config_args)
826
def create_config(self, content):
827
kwargs = dict(save=True)
828
c = self.config_class.from_string(content, *self.config_args, **kwargs)
831
def test_simple_read_access(self):
832
self.assertEqual('1', self.config.get_user_option('one'))
834
def test_simple_write_access(self):
835
self.config.set_user_option('one', 'one')
836
self.assertEqual('one', self.config.get_user_option('one'))
838
def test_listen_to_the_last_speaker(self):
840
c2 = self.get_existing_config()
841
c1.set_user_option('one', 'ONE')
842
c2.set_user_option('two', 'TWO')
843
self.assertEqual('ONE', c1.get_user_option('one'))
844
self.assertEqual('TWO', c2.get_user_option('two'))
845
# The second update respect the first one
846
self.assertEqual('ONE', c2.get_user_option('one'))
848
def test_last_speaker_wins(self):
849
# If the same config is not shared, the same variable modified twice
850
# can only see a single result.
852
c2 = self.get_existing_config()
853
c1.set_user_option('one', 'c1')
854
c2.set_user_option('one', 'c2')
855
self.assertEqual('c2', c2._get_user_option('one'))
856
# The first modification is still available until another refresh
858
self.assertEqual('c1', c1._get_user_option('one'))
859
c1.set_user_option('two', 'done')
860
self.assertEqual('c2', c1._get_user_option('one'))
862
def test_writes_are_serialized(self):
864
c2 = self.get_existing_config()
866
# We spawn a thread that will pause *during* the write
867
before_writing = threading.Event()
868
after_writing = threading.Event()
869
writing_done = threading.Event()
870
c1_orig = c1._write_config_file
871
def c1_write_config_file():
874
# The lock is held. We wait for the main thread to decide when to
877
c1._write_config_file = c1_write_config_file
879
c1.set_user_option('one', 'c1')
881
t1 = threading.Thread(target=c1_set_option)
882
# Collect the thread after the test
883
self.addCleanup(t1.join)
884
# Be ready to unblock the thread if the test goes wrong
885
self.addCleanup(after_writing.set)
887
before_writing.wait()
888
self.assertTrue(c1._lock.is_held)
889
self.assertRaises(errors.LockContention,
890
c2.set_user_option, 'one', 'c2')
891
self.assertEqual('c1', c1.get_user_option('one'))
892
# Let the lock be released
895
c2.set_user_option('one', 'c2')
896
self.assertEqual('c2', c2.get_user_option('one'))
898
def test_read_while_writing(self):
900
# We spawn a thread that will pause *during* the write
901
ready_to_write = threading.Event()
902
do_writing = threading.Event()
903
writing_done = threading.Event()
904
c1_orig = c1._write_config_file
905
def c1_write_config_file():
907
# The lock is held. We wait for the main thread to decide when to
912
c1._write_config_file = c1_write_config_file
914
c1.set_user_option('one', 'c1')
915
t1 = threading.Thread(target=c1_set_option)
916
# Collect the thread after the test
917
self.addCleanup(t1.join)
918
# Be ready to unblock the thread if the test goes wrong
919
self.addCleanup(do_writing.set)
921
# Ensure the thread is ready to write
922
ready_to_write.wait()
923
self.assertTrue(c1._lock.is_held)
924
self.assertEqual('c1', c1.get_user_option('one'))
925
# If we read during the write, we get the old value
926
c2 = self.get_existing_config()
927
self.assertEqual('1', c2.get_user_option('one'))
928
# Let the writing occur and ensure it occurred
931
# Now we get the updated value
932
c3 = self.get_existing_config()
933
self.assertEqual('c1', c3.get_user_option('one'))
936
class TestGetUserOptionAs(TestIniConfig):
938
def test_get_user_option_as_bool(self):
939
conf, parser = self.make_config_parser("""
942
an_invalid_bool = maybe
943
a_list = hmm, who knows ? # This is interpreted as a list !
945
get_bool = conf.get_user_option_as_bool
946
self.assertEqual(True, get_bool('a_true_bool'))
947
self.assertEqual(False, get_bool('a_false_bool'))
950
warnings.append(args[0] % args[1:])
951
self.overrideAttr(trace, 'warning', warning)
952
msg = 'Value "%s" is not a boolean for "%s"'
953
self.assertIs(None, get_bool('an_invalid_bool'))
954
self.assertEqual(msg % ('maybe', 'an_invalid_bool'), warnings[0])
956
self.assertIs(None, get_bool('not_defined_in_this_config'))
957
self.assertEqual([], warnings)
959
def test_get_user_option_as_list(self):
960
conf, parser = self.make_config_parser("""
965
get_list = conf.get_user_option_as_list
966
self.assertEqual(['a', 'b', 'c'], get_list('a_list'))
967
self.assertEqual(['1'], get_list('length_1'))
968
self.assertEqual('x', conf.get_user_option('one_item'))
969
# automatically cast to list
970
self.assertEqual(['x'], get_list('one_item'))
973
class TestSupressWarning(TestIniConfig):
975
def make_warnings_config(self, s):
976
conf, parser = self.make_config_parser(s)
977
return conf.suppress_warning
979
def test_suppress_warning_unknown(self):
980
suppress_warning = self.make_warnings_config('')
981
self.assertEqual(False, suppress_warning('unknown_warning'))
983
def test_suppress_warning_known(self):
984
suppress_warning = self.make_warnings_config('suppress_warnings=a,b')
985
self.assertEqual(False, suppress_warning('c'))
986
self.assertEqual(True, suppress_warning('a'))
987
self.assertEqual(True, suppress_warning('b'))
990
class TestGetConfig(tests.TestCaseInTempDir):
992
def test_constructs(self):
993
config.GlobalConfig()
995
def test_calls_read_filenames(self):
996
# replace the class that is constructed, to check its parameters
997
oldparserclass = config.ConfigObj
998
config.ConfigObj = InstrumentedConfigObj
999
my_config = config.GlobalConfig()
1001
parser = my_config._get_parser()
1003
config.ConfigObj = oldparserclass
1004
self.assertIsInstance(parser, InstrumentedConfigObj)
1005
self.assertEqual(parser._calls, [('__init__', config.config_filename(),
1009
class TestBranchConfig(tests.TestCaseWithTransport):
1011
def test_constructs_valid(self):
1012
branch = FakeBranch()
1013
my_config = config.BranchConfig(branch)
1014
self.assertIsNot(None, my_config)
1016
def test_constructs_error(self):
1017
self.assertRaises(TypeError, config.BranchConfig)
1019
def test_get_location_config(self):
1020
branch = FakeBranch()
1021
my_config = config.BranchConfig(branch)
1022
location_config = my_config._get_location_config()
1023
self.assertEqual(branch.base, location_config.location)
1024
self.assertIs(location_config, my_config._get_location_config())
1026
def test_get_config(self):
1027
"""The Branch.get_config method works properly"""
1028
b = controldir.ControlDir.create_standalone_workingtree('.').branch
1029
my_config = b.get_config()
1030
self.assertIs(my_config.get_user_option('wacky'), None)
1031
my_config.set_user_option('wacky', 'unlikely')
1032
self.assertEqual(my_config.get_user_option('wacky'), 'unlikely')
1034
# Ensure we get the same thing if we start again
1035
b2 = branch.Branch.open('.')
1036
my_config2 = b2.get_config()
1037
self.assertEqual(my_config2.get_user_option('wacky'), 'unlikely')
1039
def test_has_explicit_nickname(self):
1040
b = self.make_branch('.')
1041
self.assertFalse(b.get_config().has_explicit_nickname())
1043
self.assertTrue(b.get_config().has_explicit_nickname())
1045
def test_config_url(self):
1046
"""The Branch.get_config will use section that uses a local url"""
1047
branch = self.make_branch('branch')
1048
self.assertEqual('branch', branch.nick)
1050
local_url = urlutils.local_path_to_url('branch')
1051
conf = config.LocationConfig.from_string(
1052
'[%s]\nnickname = foobar' % (local_url,),
1053
local_url, save=True)
1054
self.assertIsNot(None, conf)
1055
self.assertEqual('foobar', branch.nick)
1057
def test_config_local_path(self):
1058
"""The Branch.get_config will use a local system path"""
1059
branch = self.make_branch('branch')
1060
self.assertEqual('branch', branch.nick)
1062
local_path = osutils.getcwd().encode('utf8')
1063
config.LocationConfig.from_string(
1064
'[%s/branch]\nnickname = barry' % (local_path,),
1065
'branch', save=True)
1066
# Now the branch will find its nick via the location config
1067
self.assertEqual('barry', branch.nick)
1069
def test_config_creates_local(self):
1070
"""Creating a new entry in config uses a local path."""
1071
branch = self.make_branch('branch', format='knit')
1072
branch.set_push_location('http://foobar')
1073
local_path = osutils.getcwd().encode('utf8')
1074
# Surprisingly ConfigObj doesn't create a trailing newline
1075
self.check_file_contents(config.locations_config_filename(),
1077
'push_location = http://foobar\n'
1078
'push_location:policy = norecurse\n'
1081
def test_autonick_urlencoded(self):
1082
b = self.make_branch('!repo')
1083
self.assertEqual('!repo', b.get_config().get_nickname())
1085
def test_autonick_uses_branch_name(self):
1086
b = self.make_branch('foo', name='bar')
1087
self.assertEqual('bar', b.get_config().get_nickname())
1089
def test_warn_if_masked(self):
1092
warnings.append(args[0] % args[1:])
1093
self.overrideAttr(trace, 'warning', warning)
1095
def set_option(store, warn_masked=True):
1097
conf.set_user_option('example_option', repr(store), store=store,
1098
warn_masked=warn_masked)
1099
def assertWarning(warning):
1101
self.assertEqual(0, len(warnings))
1103
self.assertEqual(1, len(warnings))
1104
self.assertEqual(warning, warnings[0])
1105
branch = self.make_branch('.')
1106
conf = branch.get_config()
1107
set_option(config.STORE_GLOBAL)
1109
set_option(config.STORE_BRANCH)
1111
set_option(config.STORE_GLOBAL)
1112
assertWarning('Value "4" is masked by "3" from branch.conf')
1113
set_option(config.STORE_GLOBAL, warn_masked=False)
1115
set_option(config.STORE_LOCATION)
1117
set_option(config.STORE_BRANCH)
1118
assertWarning('Value "3" is masked by "0" from locations.conf')
1119
set_option(config.STORE_BRANCH, warn_masked=False)
1123
class TestGlobalConfigItems(tests.TestCaseInTempDir):
1125
def _get_empty_config(self):
1126
my_config = config.GlobalConfig()
1129
def _get_sample_config(self):
1130
my_config = config.GlobalConfig.from_string(sample_config_text)
1133
def test_user_id(self):
1134
my_config = config.GlobalConfig.from_string(sample_config_text)
1135
self.assertEqual(u"Erik B\u00e5gfors <erik@bagfors.nu>",
1136
my_config._get_user_id())
1138
def test_absent_user_id(self):
1139
my_config = config.GlobalConfig()
1140
self.assertEqual(None, my_config._get_user_id())
1142
def test_get_user_option_default(self):
1143
my_config = self._get_empty_config()
1144
self.assertEqual(None, my_config.get_user_option('no_option'))
1146
def test_get_user_option_global(self):
1147
my_config = self._get_sample_config()
1148
self.assertEqual("something",
1149
my_config.get_user_option('user_global_option'))
1151
def test_configured_validate_signatures_in_log(self):
1152
my_config = self._get_sample_config()
1153
self.assertEqual(True, my_config.validate_signatures_in_log())
1155
def test_get_alias(self):
1156
my_config = self._get_sample_config()
1157
self.assertEqual('help', my_config.get_alias('h'))
1159
def test_get_aliases(self):
1160
my_config = self._get_sample_config()
1161
aliases = my_config.get_aliases()
1162
self.assertEqual(2, len(aliases))
1163
sorted_keys = sorted(aliases)
1164
self.assertEqual('help', aliases[sorted_keys[0]])
1165
self.assertEqual(sample_long_alias, aliases[sorted_keys[1]])
1167
def test_get_no_alias(self):
1168
my_config = self._get_sample_config()
1169
self.assertEqual(None, my_config.get_alias('foo'))
1171
def test_get_long_alias(self):
1172
my_config = self._get_sample_config()
1173
self.assertEqual(sample_long_alias, my_config.get_alias('ll'))
1175
def test_get_change_editor(self):
1176
my_config = self._get_sample_config()
1177
change_editor = my_config.get_change_editor('old', 'new')
1178
self.assertIs(diff.DiffFromTool, change_editor.__class__)
1179
self.assertEqual('vimdiff -of @new_path @old_path',
1180
' '.join(change_editor.command_template))
1182
def test_get_no_change_editor(self):
1183
my_config = self._get_empty_config()
1184
change_editor = my_config.get_change_editor('old', 'new')
1185
self.assertIs(None, change_editor)
1187
def test_get_merge_tools(self):
1188
conf = self._get_sample_config()
1189
tools = conf.get_merge_tools()
1190
self.log(repr(tools))
1192
{u'funkytool' : u'funkytool "arg with spaces" {this_temp}',
1193
u'sometool' : u'sometool {base} {this} {other} -o {result}',
1194
u'newtool' : u'"newtool with spaces" {this_temp}'},
1197
def test_get_merge_tools_empty(self):
1198
conf = self._get_empty_config()
1199
tools = conf.get_merge_tools()
1200
self.assertEqual({}, tools)
1202
def test_find_merge_tool(self):
1203
conf = self._get_sample_config()
1204
cmdline = conf.find_merge_tool('sometool')
1205
self.assertEqual('sometool {base} {this} {other} -o {result}', cmdline)
1207
def test_find_merge_tool_not_found(self):
1208
conf = self._get_sample_config()
1209
cmdline = conf.find_merge_tool('DOES NOT EXIST')
1210
self.assertIs(cmdline, None)
1212
def test_find_merge_tool_known(self):
1213
conf = self._get_empty_config()
1214
cmdline = conf.find_merge_tool('kdiff3')
1215
self.assertEqual('kdiff3 {base} {this} {other} -o {result}', cmdline)
1217
def test_find_merge_tool_override_known(self):
1218
conf = self._get_empty_config()
1219
conf.set_user_option('bzr.mergetool.kdiff3', 'kdiff3 blah')
1220
cmdline = conf.find_merge_tool('kdiff3')
1221
self.assertEqual('kdiff3 blah', cmdline)
1224
class TestGlobalConfigSavingOptions(tests.TestCaseInTempDir):
1226
def test_empty(self):
1227
my_config = config.GlobalConfig()
1228
self.assertEqual(0, len(my_config.get_aliases()))
1230
def test_set_alias(self):
1231
my_config = config.GlobalConfig()
1232
alias_value = 'commit --strict'
1233
my_config.set_alias('commit', alias_value)
1234
new_config = config.GlobalConfig()
1235
self.assertEqual(alias_value, new_config.get_alias('commit'))
1237
def test_remove_alias(self):
1238
my_config = config.GlobalConfig()
1239
my_config.set_alias('commit', 'commit --strict')
1240
# Now remove the alias again.
1241
my_config.unset_alias('commit')
1242
new_config = config.GlobalConfig()
1243
self.assertIs(None, new_config.get_alias('commit'))
1246
class TestLocationConfig(tests.TestCaseInTempDir, TestOptionsMixin):
1248
def test_constructs_valid(self):
1249
config.LocationConfig('http://example.com')
1251
def test_constructs_error(self):
1252
self.assertRaises(TypeError, config.LocationConfig)
1254
def test_branch_calls_read_filenames(self):
1255
# This is testing the correct file names are provided.
1256
# TODO: consolidate with the test for GlobalConfigs filename checks.
1258
# replace the class that is constructed, to check its parameters
1259
oldparserclass = config.ConfigObj
1260
config.ConfigObj = InstrumentedConfigObj
1262
my_config = config.LocationConfig('http://www.example.com')
1263
parser = my_config._get_parser()
1265
config.ConfigObj = oldparserclass
1266
self.assertIsInstance(parser, InstrumentedConfigObj)
1267
self.assertEqual(parser._calls,
1268
[('__init__', config.locations_config_filename(),
1271
def test_get_global_config(self):
1272
my_config = config.BranchConfig(FakeBranch('http://example.com'))
1273
global_config = my_config._get_global_config()
1274
self.assertIsInstance(global_config, config.GlobalConfig)
1275
self.assertIs(global_config, my_config._get_global_config())
1277
def assertLocationMatching(self, expected):
1278
self.assertEqual(expected,
1279
list(self.my_location_config._get_matching_sections()))
1281
def test__get_matching_sections_no_match(self):
1282
self.get_branch_config('/')
1283
self.assertLocationMatching([])
1285
def test__get_matching_sections_exact(self):
1286
self.get_branch_config('http://www.example.com')
1287
self.assertLocationMatching([('http://www.example.com', '')])
1289
def test__get_matching_sections_suffix_does_not(self):
1290
self.get_branch_config('http://www.example.com-com')
1291
self.assertLocationMatching([])
1293
def test__get_matching_sections_subdir_recursive(self):
1294
self.get_branch_config('http://www.example.com/com')
1295
self.assertLocationMatching([('http://www.example.com', 'com')])
1297
def test__get_matching_sections_ignoreparent(self):
1298
self.get_branch_config('http://www.example.com/ignoreparent')
1299
self.assertLocationMatching([('http://www.example.com/ignoreparent',
1302
def test__get_matching_sections_ignoreparent_subdir(self):
1303
self.get_branch_config(
1304
'http://www.example.com/ignoreparent/childbranch')
1305
self.assertLocationMatching([('http://www.example.com/ignoreparent',
1308
def test__get_matching_sections_subdir_trailing_slash(self):
1309
self.get_branch_config('/b')
1310
self.assertLocationMatching([('/b/', '')])
1312
def test__get_matching_sections_subdir_child(self):
1313
self.get_branch_config('/a/foo')
1314
self.assertLocationMatching([('/a/*', ''), ('/a/', 'foo')])
1316
def test__get_matching_sections_subdir_child_child(self):
1317
self.get_branch_config('/a/foo/bar')
1318
self.assertLocationMatching([('/a/*', 'bar'), ('/a/', 'foo/bar')])
1320
def test__get_matching_sections_trailing_slash_with_children(self):
1321
self.get_branch_config('/a/')
1322
self.assertLocationMatching([('/a/', '')])
1324
def test__get_matching_sections_explicit_over_glob(self):
1325
# XXX: 2006-09-08 jamesh
1326
# This test only passes because ord('c') > ord('*'). If there
1327
# was a config section for '/a/?', it would get precedence
1329
self.get_branch_config('/a/c')
1330
self.assertLocationMatching([('/a/c', ''), ('/a/*', ''), ('/a/', 'c')])
1332
def test__get_option_policy_normal(self):
1333
self.get_branch_config('http://www.example.com')
1335
self.my_location_config._get_config_policy(
1336
'http://www.example.com', 'normal_option'),
1339
def test__get_option_policy_norecurse(self):
1340
self.get_branch_config('http://www.example.com')
1342
self.my_location_config._get_option_policy(
1343
'http://www.example.com', 'norecurse_option'),
1344
config.POLICY_NORECURSE)
1345
# Test old recurse=False setting:
1347
self.my_location_config._get_option_policy(
1348
'http://www.example.com/norecurse', 'normal_option'),
1349
config.POLICY_NORECURSE)
1351
def test__get_option_policy_normal(self):
1352
self.get_branch_config('http://www.example.com')
1354
self.my_location_config._get_option_policy(
1355
'http://www.example.com', 'appendpath_option'),
1356
config.POLICY_APPENDPATH)
1358
def test__get_options_with_policy(self):
1359
self.get_branch_config('/dir/subdir',
1360
location_config="""\
1362
other_url = /other-dir
1363
other_url:policy = appendpath
1365
other_url = /other-subdir
1368
[(u'other_url', u'/other-subdir', u'/dir/subdir', 'locations'),
1369
(u'other_url', u'/other-dir', u'/dir', 'locations'),
1370
(u'other_url:policy', u'appendpath', u'/dir', 'locations')],
1371
self.my_location_config)
1373
def test_location_without_username(self):
1374
self.get_branch_config('http://www.example.com/ignoreparent')
1375
self.assertEqual(u'Erik B\u00e5gfors <erik@bagfors.nu>',
1376
self.my_config.username())
1378
def test_location_not_listed(self):
1379
"""Test that the global username is used when no location matches"""
1380
self.get_branch_config('/home/robertc/sources')
1381
self.assertEqual(u'Erik B\u00e5gfors <erik@bagfors.nu>',
1382
self.my_config.username())
1384
def test_overriding_location(self):
1385
self.get_branch_config('http://www.example.com/foo')
1386
self.assertEqual('Robert Collins <robertc@example.org>',
1387
self.my_config.username())
1389
def test_get_user_option_global(self):
1390
self.get_branch_config('/a')
1391
self.assertEqual('something',
1392
self.my_config.get_user_option('user_global_option'))
1394
def test_get_user_option_local(self):
1395
self.get_branch_config('/a')
1396
self.assertEqual('local',
1397
self.my_config.get_user_option('user_local_option'))
1399
def test_get_user_option_appendpath(self):
1400
# returned as is for the base path:
1401
self.get_branch_config('http://www.example.com')
1402
self.assertEqual('append',
1403
self.my_config.get_user_option('appendpath_option'))
1404
# Extra path components get appended:
1405
self.get_branch_config('http://www.example.com/a/b/c')
1406
self.assertEqual('append/a/b/c',
1407
self.my_config.get_user_option('appendpath_option'))
1408
# Overriden for http://www.example.com/dir, where it is a
1410
self.get_branch_config('http://www.example.com/dir/a/b/c')
1411
self.assertEqual('normal',
1412
self.my_config.get_user_option('appendpath_option'))
1414
def test_get_user_option_norecurse(self):
1415
self.get_branch_config('http://www.example.com')
1416
self.assertEqual('norecurse',
1417
self.my_config.get_user_option('norecurse_option'))
1418
self.get_branch_config('http://www.example.com/dir')
1419
self.assertEqual(None,
1420
self.my_config.get_user_option('norecurse_option'))
1421
# http://www.example.com/norecurse is a recurse=False section
1422
# that redefines normal_option. Subdirectories do not pick up
1423
# this redefinition.
1424
self.get_branch_config('http://www.example.com/norecurse')
1425
self.assertEqual('norecurse',
1426
self.my_config.get_user_option('normal_option'))
1427
self.get_branch_config('http://www.example.com/norecurse/subdir')
1428
self.assertEqual('normal',
1429
self.my_config.get_user_option('normal_option'))
1431
def test_set_user_option_norecurse(self):
1432
self.get_branch_config('http://www.example.com')
1433
self.my_config.set_user_option('foo', 'bar',
1434
store=config.STORE_LOCATION_NORECURSE)
1436
self.my_location_config._get_option_policy(
1437
'http://www.example.com', 'foo'),
1438
config.POLICY_NORECURSE)
1440
def test_set_user_option_appendpath(self):
1441
self.get_branch_config('http://www.example.com')
1442
self.my_config.set_user_option('foo', 'bar',
1443
store=config.STORE_LOCATION_APPENDPATH)
1445
self.my_location_config._get_option_policy(
1446
'http://www.example.com', 'foo'),
1447
config.POLICY_APPENDPATH)
1449
def test_set_user_option_change_policy(self):
1450
self.get_branch_config('http://www.example.com')
1451
self.my_config.set_user_option('norecurse_option', 'normal',
1452
store=config.STORE_LOCATION)
1454
self.my_location_config._get_option_policy(
1455
'http://www.example.com', 'norecurse_option'),
1458
def get_branch_config(self, location, global_config=None,
1459
location_config=None):
1460
my_branch = FakeBranch(location)
1461
if global_config is None:
1462
global_config = sample_config_text
1463
if location_config is None:
1464
location_config = sample_branches_text
1466
config.GlobalConfig.from_string(global_config, save=True)
1467
config.LocationConfig.from_string(location_config, my_branch.base,
1469
my_config = config.BranchConfig(my_branch)
1470
self.my_config = my_config
1471
self.my_location_config = my_config._get_location_config()
1473
def test_set_user_setting_sets_and_saves2(self):
1474
self.get_branch_config('/a/c')
1475
self.assertIs(self.my_config.get_user_option('foo'), None)
1476
self.my_config.set_user_option('foo', 'bar')
1478
self.my_config.branch.control_files.files['branch.conf'].strip(),
1480
self.assertEqual(self.my_config.get_user_option('foo'), 'bar')
1481
self.my_config.set_user_option('foo', 'baz',
1482
store=config.STORE_LOCATION)
1483
self.assertEqual(self.my_config.get_user_option('foo'), 'baz')
1484
self.my_config.set_user_option('foo', 'qux')
1485
self.assertEqual(self.my_config.get_user_option('foo'), 'baz')
1487
def test_get_bzr_remote_path(self):
1488
my_config = config.LocationConfig('/a/c')
1489
self.assertEqual('bzr', my_config.get_bzr_remote_path())
1490
my_config.set_user_option('bzr_remote_path', '/path-bzr')
1491
self.assertEqual('/path-bzr', my_config.get_bzr_remote_path())
1492
self.overrideEnv('BZR_REMOTE_PATH', '/environ-bzr')
1493
self.assertEqual('/environ-bzr', my_config.get_bzr_remote_path())
1496
precedence_global = 'option = global'
1497
precedence_branch = 'option = branch'
1498
precedence_location = """
1502
[http://example.com/specific]
1506
class TestBranchConfigItems(tests.TestCaseInTempDir):
1508
def get_branch_config(self, global_config=None, location=None,
1509
location_config=None, branch_data_config=None):
1510
my_branch = FakeBranch(location)
1511
if global_config is not None:
1512
config.GlobalConfig.from_string(global_config, save=True)
1513
if location_config is not None:
1514
config.LocationConfig.from_string(location_config, my_branch.base,
1516
my_config = config.BranchConfig(my_branch)
1517
if branch_data_config is not None:
1518
my_config.branch.control_files.files['branch.conf'] = \
1522
def test_user_id(self):
1523
branch = FakeBranch()
1524
my_config = config.BranchConfig(branch)
1525
self.assertIsNot(None, my_config.username())
1526
my_config.branch.control_files.files['email'] = "John"
1527
my_config.set_user_option('email',
1528
"Robert Collins <robertc@example.org>")
1529
self.assertEqual("Robert Collins <robertc@example.org>",
1530
my_config.username())
1532
def test_BRZ_EMAIL_OVERRIDES(self):
1533
self.overrideEnv('BRZ_EMAIL', "Robert Collins <robertc@example.org>")
1534
branch = FakeBranch()
1535
my_config = config.BranchConfig(branch)
1536
self.assertEqual("Robert Collins <robertc@example.org>",
1537
my_config.username())
1539
def test_get_user_option_global(self):
1540
my_config = self.get_branch_config(global_config=sample_config_text)
1541
self.assertEqual('something',
1542
my_config.get_user_option('user_global_option'))
1544
def test_config_precedence(self):
1545
# FIXME: eager test, luckily no persitent config file makes it fail
1547
my_config = self.get_branch_config(global_config=precedence_global)
1548
self.assertEqual(my_config.get_user_option('option'), 'global')
1549
my_config = self.get_branch_config(global_config=precedence_global,
1550
branch_data_config=precedence_branch)
1551
self.assertEqual(my_config.get_user_option('option'), 'branch')
1552
my_config = self.get_branch_config(
1553
global_config=precedence_global,
1554
branch_data_config=precedence_branch,
1555
location_config=precedence_location)
1556
self.assertEqual(my_config.get_user_option('option'), 'recurse')
1557
my_config = self.get_branch_config(
1558
global_config=precedence_global,
1559
branch_data_config=precedence_branch,
1560
location_config=precedence_location,
1561
location='http://example.com/specific')
1562
self.assertEqual(my_config.get_user_option('option'), 'exact')
1565
class TestMailAddressExtraction(tests.TestCase):
1567
def test_extract_email_address(self):
1568
self.assertEqual('jane@test.com',
1569
config.extract_email_address('Jane <jane@test.com>'))
1570
self.assertRaises(config.NoEmailInUsername,
1571
config.extract_email_address, 'Jane Tester')
1573
def test_parse_username(self):
1574
self.assertEqual(('', 'jdoe@example.com'),
1575
config.parse_username('jdoe@example.com'))
1576
self.assertEqual(('', 'jdoe@example.com'),
1577
config.parse_username('<jdoe@example.com>'))
1578
self.assertEqual(('John Doe', 'jdoe@example.com'),
1579
config.parse_username('John Doe <jdoe@example.com>'))
1580
self.assertEqual(('John Doe', ''),
1581
config.parse_username('John Doe'))
1582
self.assertEqual(('John Doe', 'jdoe@example.com'),
1583
config.parse_username('John Doe jdoe@example.com'))
1585
class TestTreeConfig(tests.TestCaseWithTransport):
1587
def test_get_value(self):
1588
"""Test that retreiving a value from a section is possible"""
1589
branch = self.make_branch('.')
1590
tree_config = config.TreeConfig(branch)
1591
tree_config.set_option('value', 'key', 'SECTION')
1592
tree_config.set_option('value2', 'key2')
1593
tree_config.set_option('value3-top', 'key3')
1594
tree_config.set_option('value3-section', 'key3', 'SECTION')
1595
value = tree_config.get_option('key', 'SECTION')
1596
self.assertEqual(value, 'value')
1597
value = tree_config.get_option('key2')
1598
self.assertEqual(value, 'value2')
1599
self.assertEqual(tree_config.get_option('non-existant'), None)
1600
value = tree_config.get_option('non-existant', 'SECTION')
1601
self.assertEqual(value, None)
1602
value = tree_config.get_option('non-existant', default='default')
1603
self.assertEqual(value, 'default')
1604
self.assertEqual(tree_config.get_option('key2', 'NOSECTION'), None)
1605
value = tree_config.get_option('key2', 'NOSECTION', default='default')
1606
self.assertEqual(value, 'default')
1607
value = tree_config.get_option('key3')
1608
self.assertEqual(value, 'value3-top')
1609
value = tree_config.get_option('key3', 'SECTION')
1610
self.assertEqual(value, 'value3-section')
1613
class TestTransportConfig(tests.TestCaseWithTransport):
1615
def test_load_utf8(self):
1616
"""Ensure we can load an utf8-encoded file."""
1617
t = self.get_transport()
1618
unicode_user = u'b\N{Euro Sign}ar'
1619
unicode_content = u'user=%s' % (unicode_user,)
1620
utf8_content = unicode_content.encode('utf8')
1621
# Store the raw content in the config file
1622
t.put_bytes('foo.conf', utf8_content)
1623
conf = config.TransportConfig(t, 'foo.conf')
1624
self.assertEqual(unicode_user, conf.get_option('user'))
1626
def test_load_non_ascii(self):
1627
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
1628
t = self.get_transport()
1629
t.put_bytes('foo.conf', 'user=foo\n#\xff\n')
1630
conf = config.TransportConfig(t, 'foo.conf')
1631
self.assertRaises(config.ConfigContentError, conf._get_configobj)
1633
def test_load_erroneous_content(self):
1634
"""Ensure we display a proper error on content that can't be parsed."""
1635
t = self.get_transport()
1636
t.put_bytes('foo.conf', '[open_section\n')
1637
conf = config.TransportConfig(t, 'foo.conf')
1638
self.assertRaises(config.ParseConfigError, conf._get_configobj)
1640
def test_load_permission_denied(self):
1641
"""Ensure we get an empty config file if the file is inaccessible."""
1644
warnings.append(args[0] % args[1:])
1645
self.overrideAttr(trace, 'warning', warning)
1647
class DenyingTransport(object):
1649
def __init__(self, base):
1652
def get_bytes(self, relpath):
1653
raise errors.PermissionDenied(relpath, "")
1655
cfg = config.TransportConfig(
1656
DenyingTransport("nonexisting://"), 'control.conf')
1657
self.assertIs(None, cfg.get_option('non-existant', 'SECTION'))
1660
[u'Permission denied while trying to open configuration file '
1661
u'nonexisting:///control.conf.'])
1663
def test_get_value(self):
1664
"""Test that retreiving a value from a section is possible"""
1665
bzrdir_config = config.TransportConfig(self.get_transport('.'),
1667
bzrdir_config.set_option('value', 'key', 'SECTION')
1668
bzrdir_config.set_option('value2', 'key2')
1669
bzrdir_config.set_option('value3-top', 'key3')
1670
bzrdir_config.set_option('value3-section', 'key3', 'SECTION')
1671
value = bzrdir_config.get_option('key', 'SECTION')
1672
self.assertEqual(value, 'value')
1673
value = bzrdir_config.get_option('key2')
1674
self.assertEqual(value, 'value2')
1675
self.assertEqual(bzrdir_config.get_option('non-existant'), None)
1676
value = bzrdir_config.get_option('non-existant', 'SECTION')
1677
self.assertEqual(value, None)
1678
value = bzrdir_config.get_option('non-existant', default='default')
1679
self.assertEqual(value, 'default')
1680
self.assertEqual(bzrdir_config.get_option('key2', 'NOSECTION'), None)
1681
value = bzrdir_config.get_option('key2', 'NOSECTION',
1683
self.assertEqual(value, 'default')
1684
value = bzrdir_config.get_option('key3')
1685
self.assertEqual(value, 'value3-top')
1686
value = bzrdir_config.get_option('key3', 'SECTION')
1687
self.assertEqual(value, 'value3-section')
1689
def test_set_unset_default_stack_on(self):
1690
my_dir = self.make_controldir('.')
1691
bzrdir_config = config.BzrDirConfig(my_dir)
1692
self.assertIs(None, bzrdir_config.get_default_stack_on())
1693
bzrdir_config.set_default_stack_on('Foo')
1694
self.assertEqual('Foo', bzrdir_config._config.get_option(
1695
'default_stack_on'))
1696
self.assertEqual('Foo', bzrdir_config.get_default_stack_on())
1697
bzrdir_config.set_default_stack_on(None)
1698
self.assertIs(None, bzrdir_config.get_default_stack_on())
1701
class TestOldConfigHooks(tests.TestCaseWithTransport):
1704
super(TestOldConfigHooks, self).setUp()
1705
create_configs_with_file_option(self)
1707
def assertGetHook(self, conf, name, value):
1711
config.OldConfigHooks.install_named_hook('get', hook, None)
1713
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1714
self.assertLength(0, calls)
1715
actual_value = conf.get_user_option(name)
1716
self.assertEqual(value, actual_value)
1717
self.assertLength(1, calls)
1718
self.assertEqual((conf, name, value), calls[0])
1720
def test_get_hook_breezy(self):
1721
self.assertGetHook(self.breezy_config, 'file', 'breezy')
1723
def test_get_hook_locations(self):
1724
self.assertGetHook(self.locations_config, 'file', 'locations')
1726
def test_get_hook_branch(self):
1727
# Since locations masks branch, we define a different option
1728
self.branch_config.set_user_option('file2', 'branch')
1729
self.assertGetHook(self.branch_config, 'file2', 'branch')
1731
def assertSetHook(self, conf, name, value):
1735
config.OldConfigHooks.install_named_hook('set', hook, None)
1737
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1738
self.assertLength(0, calls)
1739
conf.set_user_option(name, value)
1740
self.assertLength(1, calls)
1741
# We can't assert the conf object below as different configs use
1742
# different means to implement set_user_option and we care only about
1744
self.assertEqual((name, value), calls[0][1:])
1746
def test_set_hook_breezy(self):
1747
self.assertSetHook(self.breezy_config, 'foo', 'breezy')
1749
def test_set_hook_locations(self):
1750
self.assertSetHook(self.locations_config, 'foo', 'locations')
1752
def test_set_hook_branch(self):
1753
self.assertSetHook(self.branch_config, 'foo', 'branch')
1755
def assertRemoveHook(self, conf, name, section_name=None):
1759
config.OldConfigHooks.install_named_hook('remove', hook, None)
1761
config.OldConfigHooks.uninstall_named_hook, 'remove', None)
1762
self.assertLength(0, calls)
1763
conf.remove_user_option(name, section_name)
1764
self.assertLength(1, calls)
1765
# We can't assert the conf object below as different configs use
1766
# different means to implement remove_user_option and we care only about
1768
self.assertEqual((name,), calls[0][1:])
1770
def test_remove_hook_breezy(self):
1771
self.assertRemoveHook(self.breezy_config, 'file')
1773
def test_remove_hook_locations(self):
1774
self.assertRemoveHook(self.locations_config, 'file',
1775
self.locations_config.location)
1777
def test_remove_hook_branch(self):
1778
self.assertRemoveHook(self.branch_config, 'file')
1780
def assertLoadHook(self, name, conf_class, *conf_args):
1784
config.OldConfigHooks.install_named_hook('load', hook, None)
1786
config.OldConfigHooks.uninstall_named_hook, 'load', None)
1787
self.assertLength(0, calls)
1789
conf = conf_class(*conf_args)
1790
# Access an option to trigger a load
1791
conf.get_user_option(name)
1792
self.assertLength(1, calls)
1793
# Since we can't assert about conf, we just use the number of calls ;-/
1795
def test_load_hook_breezy(self):
1796
self.assertLoadHook('file', config.GlobalConfig)
1798
def test_load_hook_locations(self):
1799
self.assertLoadHook('file', config.LocationConfig, self.tree.basedir)
1801
def test_load_hook_branch(self):
1802
self.assertLoadHook('file', config.BranchConfig, self.tree.branch)
1804
def assertSaveHook(self, conf):
1808
config.OldConfigHooks.install_named_hook('save', hook, None)
1810
config.OldConfigHooks.uninstall_named_hook, 'save', None)
1811
self.assertLength(0, calls)
1812
# Setting an option triggers a save
1813
conf.set_user_option('foo', 'bar')
1814
self.assertLength(1, calls)
1815
# Since we can't assert about conf, we just use the number of calls ;-/
1817
def test_save_hook_breezy(self):
1818
self.assertSaveHook(self.breezy_config)
1820
def test_save_hook_locations(self):
1821
self.assertSaveHook(self.locations_config)
1823
def test_save_hook_branch(self):
1824
self.assertSaveHook(self.branch_config)
1827
class TestOldConfigHooksForRemote(tests.TestCaseWithTransport):
1828
"""Tests config hooks for remote configs.
1830
No tests for the remove hook as this is not implemented there.
1834
super(TestOldConfigHooksForRemote, self).setUp()
1835
self.transport_server = test_server.SmartTCPServer_for_testing
1836
create_configs_with_file_option(self)
1838
def assertGetHook(self, conf, name, value):
1842
config.OldConfigHooks.install_named_hook('get', hook, None)
1844
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1845
self.assertLength(0, calls)
1846
actual_value = conf.get_option(name)
1847
self.assertEqual(value, actual_value)
1848
self.assertLength(1, calls)
1849
self.assertEqual((conf, name, value), calls[0])
1851
def test_get_hook_remote_branch(self):
1852
remote_branch = branch.Branch.open(self.get_url('tree'))
1853
self.assertGetHook(remote_branch._get_config(), 'file', 'branch')
1855
def test_get_hook_remote_bzrdir(self):
1856
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1857
conf = remote_bzrdir._get_config()
1858
conf.set_option('remotedir', 'file')
1859
self.assertGetHook(conf, 'file', 'remotedir')
1861
def assertSetHook(self, conf, name, value):
1865
config.OldConfigHooks.install_named_hook('set', hook, None)
1867
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1868
self.assertLength(0, calls)
1869
conf.set_option(value, name)
1870
self.assertLength(1, calls)
1871
# We can't assert the conf object below as different configs use
1872
# different means to implement set_user_option and we care only about
1874
self.assertEqual((name, value), calls[0][1:])
1876
def test_set_hook_remote_branch(self):
1877
remote_branch = branch.Branch.open(self.get_url('tree'))
1878
self.addCleanup(remote_branch.lock_write().unlock)
1879
self.assertSetHook(remote_branch._get_config(), 'file', 'remote')
1881
def test_set_hook_remote_bzrdir(self):
1882
remote_branch = branch.Branch.open(self.get_url('tree'))
1883
self.addCleanup(remote_branch.lock_write().unlock)
1884
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1885
self.assertSetHook(remote_bzrdir._get_config(), 'file', 'remotedir')
1887
def assertLoadHook(self, expected_nb_calls, name, conf_class, *conf_args):
1891
config.OldConfigHooks.install_named_hook('load', hook, None)
1893
config.OldConfigHooks.uninstall_named_hook, 'load', None)
1894
self.assertLength(0, calls)
1896
conf = conf_class(*conf_args)
1897
# Access an option to trigger a load
1898
conf.get_option(name)
1899
self.assertLength(expected_nb_calls, calls)
1900
# Since we can't assert about conf, we just use the number of calls ;-/
1902
def test_load_hook_remote_branch(self):
1903
remote_branch = branch.Branch.open(self.get_url('tree'))
1904
self.assertLoadHook(1, 'file', remote.RemoteBranchConfig, remote_branch)
1906
def test_load_hook_remote_bzrdir(self):
1907
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1908
# The config file doesn't exist, set an option to force its creation
1909
conf = remote_bzrdir._get_config()
1910
conf.set_option('remotedir', 'file')
1911
# We get one call for the server and one call for the client, this is
1912
# caused by the differences in implementations betwen
1913
# SmartServerBzrDirRequestConfigFile (in smart/bzrdir.py) and
1914
# SmartServerBranchGetConfigFile (in smart/branch.py)
1915
self.assertLoadHook(2 ,'file', remote.RemoteBzrDirConfig, remote_bzrdir)
1917
def assertSaveHook(self, conf):
1921
config.OldConfigHooks.install_named_hook('save', hook, None)
1923
config.OldConfigHooks.uninstall_named_hook, 'save', None)
1924
self.assertLength(0, calls)
1925
# Setting an option triggers a save
1926
conf.set_option('foo', 'bar')
1927
self.assertLength(1, calls)
1928
# Since we can't assert about conf, we just use the number of calls ;-/
1930
def test_save_hook_remote_branch(self):
1931
remote_branch = branch.Branch.open(self.get_url('tree'))
1932
self.addCleanup(remote_branch.lock_write().unlock)
1933
self.assertSaveHook(remote_branch._get_config())
1935
def test_save_hook_remote_bzrdir(self):
1936
remote_branch = branch.Branch.open(self.get_url('tree'))
1937
self.addCleanup(remote_branch.lock_write().unlock)
1938
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1939
self.assertSaveHook(remote_bzrdir._get_config())
1942
class TestOptionNames(tests.TestCase):
1944
def is_valid(self, name):
1945
return config._option_ref_re.match('{%s}' % name) is not None
1947
def test_valid_names(self):
1948
self.assertTrue(self.is_valid('foo'))
1949
self.assertTrue(self.is_valid('foo.bar'))
1950
self.assertTrue(self.is_valid('f1'))
1951
self.assertTrue(self.is_valid('_'))
1952
self.assertTrue(self.is_valid('__bar__'))
1953
self.assertTrue(self.is_valid('a_'))
1954
self.assertTrue(self.is_valid('a1'))
1955
# Don't break bzr-svn for no good reason
1956
self.assertTrue(self.is_valid('guessed-layout'))
1958
def test_invalid_names(self):
1959
self.assertFalse(self.is_valid(' foo'))
1960
self.assertFalse(self.is_valid('foo '))
1961
self.assertFalse(self.is_valid('1'))
1962
self.assertFalse(self.is_valid('1,2'))
1963
self.assertFalse(self.is_valid('foo$'))
1964
self.assertFalse(self.is_valid('!foo'))
1965
self.assertFalse(self.is_valid('foo.'))
1966
self.assertFalse(self.is_valid('foo..bar'))
1967
self.assertFalse(self.is_valid('{}'))
1968
self.assertFalse(self.is_valid('{a}'))
1969
self.assertFalse(self.is_valid('a\n'))
1970
self.assertFalse(self.is_valid('-'))
1971
self.assertFalse(self.is_valid('-a'))
1972
self.assertFalse(self.is_valid('a-'))
1973
self.assertFalse(self.is_valid('a--a'))
1975
def assertSingleGroup(self, reference):
1976
# the regexp is used with split and as such should match the reference
1977
# *only*, if more groups needs to be defined, (?:...) should be used.
1978
m = config._option_ref_re.match('{a}')
1979
self.assertLength(1, m.groups())
1981
def test_valid_references(self):
1982
self.assertSingleGroup('{a}')
1983
self.assertSingleGroup('{{a}}')
1986
class TestOption(tests.TestCase):
1988
def test_default_value(self):
1989
opt = config.Option('foo', default='bar')
1990
self.assertEqual('bar', opt.get_default())
1992
def test_callable_default_value(self):
1993
def bar_as_unicode():
1995
opt = config.Option('foo', default=bar_as_unicode)
1996
self.assertEqual('bar', opt.get_default())
1998
def test_default_value_from_env(self):
1999
opt = config.Option('foo', default='bar', default_from_env=['FOO'])
2000
self.overrideEnv('FOO', 'quux')
2001
# Env variable provides a default taking over the option one
2002
self.assertEqual('quux', opt.get_default())
2004
def test_first_default_value_from_env_wins(self):
2005
opt = config.Option('foo', default='bar',
2006
default_from_env=['NO_VALUE', 'FOO', 'BAZ'])
2007
self.overrideEnv('FOO', 'foo')
2008
self.overrideEnv('BAZ', 'baz')
2009
# The first env var set wins
2010
self.assertEqual('foo', opt.get_default())
2012
def test_not_supported_list_default_value(self):
2013
self.assertRaises(AssertionError, config.Option, 'foo', default=[1])
2015
def test_not_supported_object_default_value(self):
2016
self.assertRaises(AssertionError, config.Option, 'foo',
2019
def test_not_supported_callable_default_value_not_unicode(self):
2020
def bar_not_unicode():
2022
opt = config.Option('foo', default=bar_not_unicode)
2023
self.assertRaises(AssertionError, opt.get_default)
2025
def test_get_help_topic(self):
2026
opt = config.Option('foo')
2027
self.assertEqual('foo', opt.get_help_topic())
2030
class TestOptionConverter(tests.TestCase):
2032
def assertConverted(self, expected, opt, value):
2033
self.assertEqual(expected, opt.convert_from_unicode(None, value))
2035
def assertCallsWarning(self, opt, value):
2039
warnings.append(args[0] % args[1:])
2040
self.overrideAttr(trace, 'warning', warning)
2041
self.assertEqual(None, opt.convert_from_unicode(None, value))
2042
self.assertLength(1, warnings)
2044
'Value "%s" is not valid for "%s"' % (value, opt.name),
2047
def assertCallsError(self, opt, value):
2048
self.assertRaises(config.ConfigOptionValueError,
2049
opt.convert_from_unicode, None, value)
2051
def assertConvertInvalid(self, opt, invalid_value):
2053
self.assertEqual(None, opt.convert_from_unicode(None, invalid_value))
2054
opt.invalid = 'warning'
2055
self.assertCallsWarning(opt, invalid_value)
2056
opt.invalid = 'error'
2057
self.assertCallsError(opt, invalid_value)
2060
class TestOptionWithBooleanConverter(TestOptionConverter):
2062
def get_option(self):
2063
return config.Option('foo', help='A boolean.',
2064
from_unicode=config.bool_from_store)
2066
def test_convert_invalid(self):
2067
opt = self.get_option()
2068
# A string that is not recognized as a boolean
2069
self.assertConvertInvalid(opt, u'invalid-boolean')
2070
# A list of strings is never recognized as a boolean
2071
self.assertConvertInvalid(opt, [u'not', u'a', u'boolean'])
2073
def test_convert_valid(self):
2074
opt = self.get_option()
2075
self.assertConverted(True, opt, u'True')
2076
self.assertConverted(True, opt, u'1')
2077
self.assertConverted(False, opt, u'False')
2080
class TestOptionWithIntegerConverter(TestOptionConverter):
2082
def get_option(self):
2083
return config.Option('foo', help='An integer.',
2084
from_unicode=config.int_from_store)
2086
def test_convert_invalid(self):
2087
opt = self.get_option()
2088
# A string that is not recognized as an integer
2089
self.assertConvertInvalid(opt, u'forty-two')
2090
# A list of strings is never recognized as an integer
2091
self.assertConvertInvalid(opt, [u'a', u'list'])
2093
def test_convert_valid(self):
2094
opt = self.get_option()
2095
self.assertConverted(16, opt, u'16')
2098
class TestOptionWithSIUnitConverter(TestOptionConverter):
2100
def get_option(self):
2101
return config.Option('foo', help='An integer in SI units.',
2102
from_unicode=config.int_SI_from_store)
2104
def test_convert_invalid(self):
2105
opt = self.get_option()
2106
self.assertConvertInvalid(opt, u'not-a-unit')
2107
self.assertConvertInvalid(opt, u'Gb') # Forgot the value
2108
self.assertConvertInvalid(opt, u'1b') # Forgot the unit
2109
self.assertConvertInvalid(opt, u'1GG')
2110
self.assertConvertInvalid(opt, u'1Mbb')
2111
self.assertConvertInvalid(opt, u'1MM')
2113
def test_convert_valid(self):
2114
opt = self.get_option()
2115
self.assertConverted(int(5e3), opt, u'5kb')
2116
self.assertConverted(int(5e6), opt, u'5M')
2117
self.assertConverted(int(5e6), opt, u'5MB')
2118
self.assertConverted(int(5e9), opt, u'5g')
2119
self.assertConverted(int(5e9), opt, u'5gB')
2120
self.assertConverted(100, opt, u'100')
2123
class TestListOption(TestOptionConverter):
2125
def get_option(self):
2126
return config.ListOption('foo', help='A list.')
2128
def test_convert_invalid(self):
2129
opt = self.get_option()
2130
# We don't even try to convert a list into a list, we only expect
2132
self.assertConvertInvalid(opt, [1])
2133
# No string is invalid as all forms can be converted to a list
2135
def test_convert_valid(self):
2136
opt = self.get_option()
2137
# An empty string is an empty list
2138
self.assertConverted([], opt, '') # Using a bare str() just in case
2139
self.assertConverted([], opt, u'')
2141
self.assertConverted([u'True'], opt, u'True')
2143
self.assertConverted([u'42'], opt, u'42')
2145
self.assertConverted([u'bar'], opt, u'bar')
2148
class TestRegistryOption(TestOptionConverter):
2150
def get_option(self, registry):
2151
return config.RegistryOption('foo', registry,
2152
help='A registry option.')
2154
def test_convert_invalid(self):
2155
registry = _mod_registry.Registry()
2156
opt = self.get_option(registry)
2157
self.assertConvertInvalid(opt, [1])
2158
self.assertConvertInvalid(opt, u"notregistered")
2160
def test_convert_valid(self):
2161
registry = _mod_registry.Registry()
2162
registry.register("someval", 1234)
2163
opt = self.get_option(registry)
2164
# Using a bare str() just in case
2165
self.assertConverted(1234, opt, "someval")
2166
self.assertConverted(1234, opt, u'someval')
2167
self.assertConverted(None, opt, None)
2169
def test_help(self):
2170
registry = _mod_registry.Registry()
2171
registry.register("someval", 1234, help="some option")
2172
registry.register("dunno", 1234, help="some other option")
2173
opt = self.get_option(registry)
2175
'A registry option.\n'
2177
'The following values are supported:\n'
2178
' dunno - some other option\n'
2179
' someval - some option\n',
2182
def test_get_help_text(self):
2183
registry = _mod_registry.Registry()
2184
registry.register("someval", 1234, help="some option")
2185
registry.register("dunno", 1234, help="some other option")
2186
opt = self.get_option(registry)
2188
'A registry option.\n'
2190
'The following values are supported:\n'
2191
' dunno - some other option\n'
2192
' someval - some option\n',
2193
opt.get_help_text())
2196
class TestOptionRegistry(tests.TestCase):
2199
super(TestOptionRegistry, self).setUp()
2200
# Always start with an empty registry
2201
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
2202
self.registry = config.option_registry
2204
def test_register(self):
2205
opt = config.Option('foo')
2206
self.registry.register(opt)
2207
self.assertIs(opt, self.registry.get('foo'))
2209
def test_registered_help(self):
2210
opt = config.Option('foo', help='A simple option')
2211
self.registry.register(opt)
2212
self.assertEqual('A simple option', self.registry.get_help('foo'))
2214
def test_dont_register_illegal_name(self):
2215
self.assertRaises(config.IllegalOptionName,
2216
self.registry.register, config.Option(' foo'))
2217
self.assertRaises(config.IllegalOptionName,
2218
self.registry.register, config.Option('bar,'))
2220
lazy_option = config.Option('lazy_foo', help='Lazy help')
2222
def test_register_lazy(self):
2223
self.registry.register_lazy('lazy_foo', self.__module__,
2224
'TestOptionRegistry.lazy_option')
2225
self.assertIs(self.lazy_option, self.registry.get('lazy_foo'))
2227
def test_registered_lazy_help(self):
2228
self.registry.register_lazy('lazy_foo', self.__module__,
2229
'TestOptionRegistry.lazy_option')
2230
self.assertEqual('Lazy help', self.registry.get_help('lazy_foo'))
2232
def test_dont_lazy_register_illegal_name(self):
2233
# This is where the root cause of http://pad.lv/1235099 is better
2234
# understood: 'register_lazy' doc string mentions that key should match
2235
# the option name which indirectly requires that the option name is a
2236
# valid python identifier. We violate that rule here (using a key that
2237
# doesn't match the option name) to test the option name checking.
2238
self.assertRaises(config.IllegalOptionName,
2239
self.registry.register_lazy, ' foo', self.__module__,
2240
'TestOptionRegistry.lazy_option')
2241
self.assertRaises(config.IllegalOptionName,
2242
self.registry.register_lazy, '1,2', self.__module__,
2243
'TestOptionRegistry.lazy_option')
2246
class TestRegisteredOptions(tests.TestCase):
2247
"""All registered options should verify some constraints."""
2249
scenarios = [(key, {'option_name': key, 'option': option}) for key, option
2250
in config.option_registry.iteritems()]
2253
super(TestRegisteredOptions, self).setUp()
2254
self.registry = config.option_registry
2256
def test_proper_name(self):
2257
# An option should be registered under its own name, this can't be
2258
# checked at registration time for the lazy ones.
2259
self.assertEqual(self.option_name, self.option.name)
2261
def test_help_is_set(self):
2262
option_help = self.registry.get_help(self.option_name)
2263
# Come on, think about the user, he really wants to know what the
2265
self.assertIsNot(None, option_help)
2266
self.assertNotEqual('', option_help)
2269
class TestSection(tests.TestCase):
2271
# FIXME: Parametrize so that all sections produced by Stores run these
2272
# tests -- vila 2011-04-01
2274
def test_get_a_value(self):
2275
a_dict = dict(foo='bar')
2276
section = config.Section('myID', a_dict)
2277
self.assertEqual('bar', section.get('foo'))
2279
def test_get_unknown_option(self):
2281
section = config.Section(None, a_dict)
2282
self.assertEqual('out of thin air',
2283
section.get('foo', 'out of thin air'))
2285
def test_options_is_shared(self):
2287
section = config.Section(None, a_dict)
2288
self.assertIs(a_dict, section.options)
2291
class TestMutableSection(tests.TestCase):
2293
scenarios = [('mutable',
2295
lambda opts: config.MutableSection('myID', opts)},),
2299
a_dict = dict(foo='bar')
2300
section = self.get_section(a_dict)
2301
section.set('foo', 'new_value')
2302
self.assertEqual('new_value', section.get('foo'))
2303
# The change appears in the shared section
2304
self.assertEqual('new_value', a_dict.get('foo'))
2305
# We keep track of the change
2306
self.assertTrue('foo' in section.orig)
2307
self.assertEqual('bar', section.orig.get('foo'))
2309
def test_set_preserve_original_once(self):
2310
a_dict = dict(foo='bar')
2311
section = self.get_section(a_dict)
2312
section.set('foo', 'first_value')
2313
section.set('foo', 'second_value')
2314
# We keep track of the original value
2315
self.assertTrue('foo' in section.orig)
2316
self.assertEqual('bar', section.orig.get('foo'))
2318
def test_remove(self):
2319
a_dict = dict(foo='bar')
2320
section = self.get_section(a_dict)
2321
section.remove('foo')
2322
# We get None for unknown options via the default value
2323
self.assertEqual(None, section.get('foo'))
2324
# Or we just get the default value
2325
self.assertEqual('unknown', section.get('foo', 'unknown'))
2326
self.assertFalse('foo' in section.options)
2327
# We keep track of the deletion
2328
self.assertTrue('foo' in section.orig)
2329
self.assertEqual('bar', section.orig.get('foo'))
2331
def test_remove_new_option(self):
2333
section = self.get_section(a_dict)
2334
section.set('foo', 'bar')
2335
section.remove('foo')
2336
self.assertFalse('foo' in section.options)
2337
# The option didn't exist initially so it we need to keep track of it
2338
# with a special value
2339
self.assertTrue('foo' in section.orig)
2340
self.assertEqual(config._NewlyCreatedOption, section.orig['foo'])
2343
class TestCommandLineStore(tests.TestCase):
2346
super(TestCommandLineStore, self).setUp()
2347
self.store = config.CommandLineStore()
2348
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
2350
def get_section(self):
2351
"""Get the unique section for the command line overrides."""
2352
sections = list(self.store.get_sections())
2353
self.assertLength(1, sections)
2354
store, section = sections[0]
2355
self.assertEqual(self.store, store)
2358
def test_no_override(self):
2359
self.store._from_cmdline([])
2360
section = self.get_section()
2361
self.assertLength(0, list(section.iter_option_names()))
2363
def test_simple_override(self):
2364
self.store._from_cmdline(['a=b'])
2365
section = self.get_section()
2366
self.assertEqual('b', section.get('a'))
2368
def test_list_override(self):
2369
opt = config.ListOption('l')
2370
config.option_registry.register(opt)
2371
self.store._from_cmdline(['l=1,2,3'])
2372
val = self.get_section().get('l')
2373
self.assertEqual('1,2,3', val)
2374
# Reminder: lists should be registered as such explicitely, otherwise
2375
# the conversion needs to be done afterwards.
2376
self.assertEqual(['1', '2', '3'],
2377
opt.convert_from_unicode(self.store, val))
2379
def test_multiple_overrides(self):
2380
self.store._from_cmdline(['a=b', 'x=y'])
2381
section = self.get_section()
2382
self.assertEqual('b', section.get('a'))
2383
self.assertEqual('y', section.get('x'))
2385
def test_wrong_syntax(self):
2386
self.assertRaises(errors.BzrCommandError,
2387
self.store._from_cmdline, ['a=b', 'c'])
2389
class TestStoreMinimalAPI(tests.TestCaseWithTransport):
2391
scenarios = [(key, {'get_store': builder}) for key, builder
2392
in config.test_store_builder_registry.iteritems()] + [
2393
('cmdline', {'get_store': lambda test: config.CommandLineStore()})]
2396
store = self.get_store(self)
2397
if isinstance(store, config.TransportIniFileStore):
2398
raise tests.TestNotApplicable(
2399
"%s is not a concrete Store implementation"
2400
" so it doesn't need an id" % (store.__class__.__name__,))
2401
self.assertIsNot(None, store.id)
2404
class TestStore(tests.TestCaseWithTransport):
2406
def assertSectionContent(self, expected, store_and_section):
2407
"""Assert that some options have the proper values in a section."""
2408
_, section = store_and_section
2409
expected_name, expected_options = expected
2410
self.assertEqual(expected_name, section.id)
2413
dict([(k, section.get(k)) for k in expected_options.keys()]))
2416
class TestReadonlyStore(TestStore):
2418
scenarios = [(key, {'get_store': builder}) for key, builder
2419
in config.test_store_builder_registry.iteritems()]
2421
def test_building_delays_load(self):
2422
store = self.get_store(self)
2423
self.assertEqual(False, store.is_loaded())
2424
store._load_from_string('')
2425
self.assertEqual(True, store.is_loaded())
2427
def test_get_no_sections_for_empty(self):
2428
store = self.get_store(self)
2429
store._load_from_string('')
2430
self.assertEqual([], list(store.get_sections()))
2432
def test_get_default_section(self):
2433
store = self.get_store(self)
2434
store._load_from_string('foo=bar')
2435
sections = list(store.get_sections())
2436
self.assertLength(1, sections)
2437
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2439
def test_get_named_section(self):
2440
store = self.get_store(self)
2441
store._load_from_string('[baz]\nfoo=bar')
2442
sections = list(store.get_sections())
2443
self.assertLength(1, sections)
2444
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2446
def test_load_from_string_fails_for_non_empty_store(self):
2447
store = self.get_store(self)
2448
store._load_from_string('foo=bar')
2449
self.assertRaises(AssertionError, store._load_from_string, 'bar=baz')
2452
class TestStoreQuoting(TestStore):
2454
scenarios = [(key, {'get_store': builder}) for key, builder
2455
in config.test_store_builder_registry.iteritems()]
2458
super(TestStoreQuoting, self).setUp()
2459
self.store = self.get_store(self)
2460
# We need a loaded store but any content will do
2461
self.store._load_from_string('')
2463
def assertIdempotent(self, s):
2464
"""Assert that quoting an unquoted string is a no-op and vice-versa.
2466
What matters here is that option values, as they appear in a store, can
2467
be safely round-tripped out of the store and back.
2469
:param s: A string, quoted if required.
2471
self.assertEqual(s, self.store.quote(self.store.unquote(s)))
2472
self.assertEqual(s, self.store.unquote(self.store.quote(s)))
2474
def test_empty_string(self):
2475
if isinstance(self.store, config.IniFileStore):
2476
# configobj._quote doesn't handle empty values
2477
self.assertRaises(AssertionError,
2478
self.assertIdempotent, '')
2480
self.assertIdempotent('')
2481
# But quoted empty strings are ok
2482
self.assertIdempotent('""')
2484
def test_embedded_spaces(self):
2485
self.assertIdempotent('" a b c "')
2487
def test_embedded_commas(self):
2488
self.assertIdempotent('" a , b c "')
2490
def test_simple_comma(self):
2491
if isinstance(self.store, config.IniFileStore):
2492
# configobj requires that lists are special-cased
2493
self.assertRaises(AssertionError,
2494
self.assertIdempotent, ',')
2496
self.assertIdempotent(',')
2497
# When a single comma is required, quoting is also required
2498
self.assertIdempotent('","')
2500
def test_list(self):
2501
if isinstance(self.store, config.IniFileStore):
2502
# configobj requires that lists are special-cased
2503
self.assertRaises(AssertionError,
2504
self.assertIdempotent, 'a,b')
2506
self.assertIdempotent('a,b')
2509
class TestDictFromStore(tests.TestCase):
2511
def test_unquote_not_string(self):
2512
conf = config.MemoryStack('x=2\n[a_section]\na=1\n')
2513
value = conf.get('a_section')
2514
# Urgh, despite 'conf' asking for the no-name section, we get the
2515
# content of another section as a dict o_O
2516
self.assertEqual({'a': '1'}, value)
2517
unquoted = conf.store.unquote(value)
2518
# Which cannot be unquoted but shouldn't crash either (the use cases
2519
# are getting the value or displaying it. In the later case, '%s' will
2521
self.assertEqual({'a': '1'}, unquoted)
2522
self.assertEqual("{u'a': u'1'}", '%s' % (unquoted,))
2525
class TestIniFileStoreContent(tests.TestCaseWithTransport):
2526
"""Simulate loading a config store with content of various encodings.
2528
All files produced by bzr are in utf8 content.
2530
Users may modify them manually and end up with a file that can't be
2531
loaded. We need to issue proper error messages in this case.
2534
invalid_utf8_char = '\xff'
2536
def test_load_utf8(self):
2537
"""Ensure we can load an utf8-encoded file."""
2538
t = self.get_transport()
2539
# From http://pad.lv/799212
2540
unicode_user = u'b\N{Euro Sign}ar'
2541
unicode_content = u'user=%s' % (unicode_user,)
2542
utf8_content = unicode_content.encode('utf8')
2543
# Store the raw content in the config file
2544
t.put_bytes('foo.conf', utf8_content)
2545
store = config.TransportIniFileStore(t, 'foo.conf')
2547
stack = config.Stack([store.get_sections], store)
2548
self.assertEqual(unicode_user, stack.get('user'))
2550
def test_load_non_ascii(self):
2551
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2552
t = self.get_transport()
2553
t.put_bytes('foo.conf', 'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2554
store = config.TransportIniFileStore(t, 'foo.conf')
2555
self.assertRaises(config.ConfigContentError, store.load)
2557
def test_load_erroneous_content(self):
2558
"""Ensure we display a proper error on content that can't be parsed."""
2559
t = self.get_transport()
2560
t.put_bytes('foo.conf', '[open_section\n')
2561
store = config.TransportIniFileStore(t, 'foo.conf')
2562
self.assertRaises(config.ParseConfigError, store.load)
2564
def test_load_permission_denied(self):
2565
"""Ensure we get warned when trying to load an inaccessible file."""
2568
warnings.append(args[0] % args[1:])
2569
self.overrideAttr(trace, 'warning', warning)
2571
t = self.get_transport()
2573
def get_bytes(relpath):
2574
raise errors.PermissionDenied(relpath, "")
2575
t.get_bytes = get_bytes
2576
store = config.TransportIniFileStore(t, 'foo.conf')
2577
self.assertRaises(errors.PermissionDenied, store.load)
2580
[u'Permission denied while trying to load configuration store %s.'
2581
% store.external_url()])
2584
class TestIniConfigContent(tests.TestCaseWithTransport):
2585
"""Simulate loading a IniBasedConfig with content of various encodings.
2587
All files produced by bzr are in utf8 content.
2589
Users may modify them manually and end up with a file that can't be
2590
loaded. We need to issue proper error messages in this case.
2593
invalid_utf8_char = '\xff'
2595
def test_load_utf8(self):
2596
"""Ensure we can load an utf8-encoded file."""
2597
# From http://pad.lv/799212
2598
unicode_user = u'b\N{Euro Sign}ar'
2599
unicode_content = u'user=%s' % (unicode_user,)
2600
utf8_content = unicode_content.encode('utf8')
2601
# Store the raw content in the config file
2602
with open('foo.conf', 'wb') as f:
2603
f.write(utf8_content)
2604
conf = config.IniBasedConfig(file_name='foo.conf')
2605
self.assertEqual(unicode_user, conf.get_user_option('user'))
2607
def test_load_badly_encoded_content(self):
2608
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2609
with open('foo.conf', 'wb') as f:
2610
f.write('user=foo\n#%s\n' % (self.invalid_utf8_char,))
2611
conf = config.IniBasedConfig(file_name='foo.conf')
2612
self.assertRaises(config.ConfigContentError, conf._get_parser)
2614
def test_load_erroneous_content(self):
2615
"""Ensure we display a proper error on content that can't be parsed."""
2616
with open('foo.conf', 'wb') as f:
2617
f.write('[open_section\n')
2618
conf = config.IniBasedConfig(file_name='foo.conf')
2619
self.assertRaises(config.ParseConfigError, conf._get_parser)
2622
class TestMutableStore(TestStore):
2624
scenarios = [(key, {'store_id': key, 'get_store': builder}) for key, builder
2625
in config.test_store_builder_registry.iteritems()]
2628
super(TestMutableStore, self).setUp()
2629
self.transport = self.get_transport()
2631
def has_store(self, store):
2632
store_basename = urlutils.relative_url(self.transport.external_url(),
2633
store.external_url())
2634
return self.transport.has(store_basename)
2636
def test_save_empty_creates_no_file(self):
2637
# FIXME: There should be a better way than relying on the test
2638
# parametrization to identify branch.conf -- vila 2011-0526
2639
if self.store_id in ('branch', 'remote_branch'):
2640
raise tests.TestNotApplicable(
2641
'branch.conf is *always* created when a branch is initialized')
2642
store = self.get_store(self)
2644
self.assertEqual(False, self.has_store(store))
2646
def test_mutable_section_shared(self):
2647
store = self.get_store(self)
2648
store._load_from_string('foo=bar\n')
2649
# FIXME: There should be a better way than relying on the test
2650
# parametrization to identify branch.conf -- vila 2011-0526
2651
if self.store_id in ('branch', 'remote_branch'):
2652
# branch stores requires write locked branches
2653
self.addCleanup(store.branch.lock_write().unlock)
2654
section1 = store.get_mutable_section(None)
2655
section2 = store.get_mutable_section(None)
2656
# If we get different sections, different callers won't share the
2658
self.assertIs(section1, section2)
2660
def test_save_emptied_succeeds(self):
2661
store = self.get_store(self)
2662
store._load_from_string('foo=bar\n')
2663
# FIXME: There should be a better way than relying on the test
2664
# parametrization to identify branch.conf -- vila 2011-0526
2665
if self.store_id in ('branch', 'remote_branch'):
2666
# branch stores requires write locked branches
2667
self.addCleanup(store.branch.lock_write().unlock)
2668
section = store.get_mutable_section(None)
2669
section.remove('foo')
2671
self.assertEqual(True, self.has_store(store))
2672
modified_store = self.get_store(self)
2673
sections = list(modified_store.get_sections())
2674
self.assertLength(0, sections)
2676
def test_save_with_content_succeeds(self):
2677
# FIXME: There should be a better way than relying on the test
2678
# parametrization to identify branch.conf -- vila 2011-0526
2679
if self.store_id in ('branch', 'remote_branch'):
2680
raise tests.TestNotApplicable(
2681
'branch.conf is *always* created when a branch is initialized')
2682
store = self.get_store(self)
2683
store._load_from_string('foo=bar\n')
2684
self.assertEqual(False, self.has_store(store))
2686
self.assertEqual(True, self.has_store(store))
2687
modified_store = self.get_store(self)
2688
sections = list(modified_store.get_sections())
2689
self.assertLength(1, sections)
2690
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2692
def test_set_option_in_empty_store(self):
2693
store = self.get_store(self)
2694
# FIXME: There should be a better way than relying on the test
2695
# parametrization to identify branch.conf -- vila 2011-0526
2696
if self.store_id in ('branch', 'remote_branch'):
2697
# branch stores requires write locked branches
2698
self.addCleanup(store.branch.lock_write().unlock)
2699
section = store.get_mutable_section(None)
2700
section.set('foo', 'bar')
2702
modified_store = self.get_store(self)
2703
sections = list(modified_store.get_sections())
2704
self.assertLength(1, sections)
2705
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2707
def test_set_option_in_default_section(self):
2708
store = self.get_store(self)
2709
store._load_from_string('')
2710
# FIXME: There should be a better way than relying on the test
2711
# parametrization to identify branch.conf -- vila 2011-0526
2712
if self.store_id in ('branch', 'remote_branch'):
2713
# branch stores requires write locked branches
2714
self.addCleanup(store.branch.lock_write().unlock)
2715
section = store.get_mutable_section(None)
2716
section.set('foo', 'bar')
2718
modified_store = self.get_store(self)
2719
sections = list(modified_store.get_sections())
2720
self.assertLength(1, sections)
2721
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2723
def test_set_option_in_named_section(self):
2724
store = self.get_store(self)
2725
store._load_from_string('')
2726
# FIXME: There should be a better way than relying on the test
2727
# parametrization to identify branch.conf -- vila 2011-0526
2728
if self.store_id in ('branch', 'remote_branch'):
2729
# branch stores requires write locked branches
2730
self.addCleanup(store.branch.lock_write().unlock)
2731
section = store.get_mutable_section('baz')
2732
section.set('foo', 'bar')
2734
modified_store = self.get_store(self)
2735
sections = list(modified_store.get_sections())
2736
self.assertLength(1, sections)
2737
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2739
def test_load_hook(self):
2740
# First, we need to ensure that the store exists
2741
store = self.get_store(self)
2742
# FIXME: There should be a better way than relying on the test
2743
# parametrization to identify branch.conf -- vila 2011-0526
2744
if self.store_id in ('branch', 'remote_branch'):
2745
# branch stores requires write locked branches
2746
self.addCleanup(store.branch.lock_write().unlock)
2747
section = store.get_mutable_section('baz')
2748
section.set('foo', 'bar')
2750
# Now we can try to load it
2751
store = self.get_store(self)
2755
config.ConfigHooks.install_named_hook('load', hook, None)
2756
self.assertLength(0, calls)
2758
self.assertLength(1, calls)
2759
self.assertEqual((store,), calls[0])
2761
def test_save_hook(self):
2765
config.ConfigHooks.install_named_hook('save', hook, None)
2766
self.assertLength(0, calls)
2767
store = self.get_store(self)
2768
# FIXME: There should be a better way than relying on the test
2769
# parametrization to identify branch.conf -- vila 2011-0526
2770
if self.store_id in ('branch', 'remote_branch'):
2771
# branch stores requires write locked branches
2772
self.addCleanup(store.branch.lock_write().unlock)
2773
section = store.get_mutable_section('baz')
2774
section.set('foo', 'bar')
2776
self.assertLength(1, calls)
2777
self.assertEqual((store,), calls[0])
2779
def test_set_mark_dirty(self):
2780
stack = config.MemoryStack('')
2781
self.assertLength(0, stack.store.dirty_sections)
2782
stack.set('foo', 'baz')
2783
self.assertLength(1, stack.store.dirty_sections)
2784
self.assertTrue(stack.store._need_saving())
2786
def test_remove_mark_dirty(self):
2787
stack = config.MemoryStack('foo=bar')
2788
self.assertLength(0, stack.store.dirty_sections)
2790
self.assertLength(1, stack.store.dirty_sections)
2791
self.assertTrue(stack.store._need_saving())
2794
class TestStoreSaveChanges(tests.TestCaseWithTransport):
2795
"""Tests that config changes are kept in memory and saved on-demand."""
2798
super(TestStoreSaveChanges, self).setUp()
2799
self.transport = self.get_transport()
2800
# Most of the tests involve two stores pointing to the same persistent
2801
# storage to observe the effects of concurrent changes
2802
self.st1 = config.TransportIniFileStore(self.transport, 'foo.conf')
2803
self.st2 = config.TransportIniFileStore(self.transport, 'foo.conf')
2806
self.warnings.append(args[0] % args[1:])
2807
self.overrideAttr(trace, 'warning', warning)
2809
def has_store(self, store):
2810
store_basename = urlutils.relative_url(self.transport.external_url(),
2811
store.external_url())
2812
return self.transport.has(store_basename)
2814
def get_stack(self, store):
2815
# Any stack will do as long as it uses the right store, just a single
2816
# no-name section is enough
2817
return config.Stack([store.get_sections], store)
2819
def test_no_changes_no_save(self):
2820
s = self.get_stack(self.st1)
2821
s.store.save_changes()
2822
self.assertEqual(False, self.has_store(self.st1))
2824
def test_unrelated_concurrent_update(self):
2825
s1 = self.get_stack(self.st1)
2826
s2 = self.get_stack(self.st2)
2827
s1.set('foo', 'bar')
2828
s2.set('baz', 'quux')
2830
# Changes don't propagate magically
2831
self.assertEqual(None, s1.get('baz'))
2832
s2.store.save_changes()
2833
self.assertEqual('quux', s2.get('baz'))
2834
# Changes are acquired when saving
2835
self.assertEqual('bar', s2.get('foo'))
2836
# Since there is no overlap, no warnings are emitted
2837
self.assertLength(0, self.warnings)
2839
def test_concurrent_update_modified(self):
2840
s1 = self.get_stack(self.st1)
2841
s2 = self.get_stack(self.st2)
2842
s1.set('foo', 'bar')
2843
s2.set('foo', 'baz')
2846
s2.store.save_changes()
2847
self.assertEqual('baz', s2.get('foo'))
2848
# But the user get a warning
2849
self.assertLength(1, self.warnings)
2850
warning = self.warnings[0]
2851
self.assertStartsWith(warning, 'Option foo in section None')
2852
self.assertEndsWith(warning, 'was changed from <CREATED> to bar.'
2853
' The baz value will be saved.')
2855
def test_concurrent_deletion(self):
2856
self.st1._load_from_string('foo=bar')
2858
s1 = self.get_stack(self.st1)
2859
s2 = self.get_stack(self.st2)
2862
s1.store.save_changes()
2864
self.assertLength(0, self.warnings)
2865
s2.store.save_changes()
2867
self.assertLength(1, self.warnings)
2868
warning = self.warnings[0]
2869
self.assertStartsWith(warning, 'Option foo in section None')
2870
self.assertEndsWith(warning, 'was changed from bar to <CREATED>.'
2871
' The <DELETED> value will be saved.')
2874
class TestQuotingIniFileStore(tests.TestCaseWithTransport):
2876
def get_store(self):
2877
return config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2879
def test_get_quoted_string(self):
2880
store = self.get_store()
2881
store._load_from_string('foo= " abc "')
2882
stack = config.Stack([store.get_sections])
2883
self.assertEqual(' abc ', stack.get('foo'))
2885
def test_set_quoted_string(self):
2886
store = self.get_store()
2887
stack = config.Stack([store.get_sections], store)
2888
stack.set('foo', ' a b c ')
2890
self.assertFileEqual('foo = " a b c "' + os.linesep, 'foo.conf')
2893
class TestTransportIniFileStore(TestStore):
2895
def test_loading_unknown_file_fails(self):
2896
store = config.TransportIniFileStore(self.get_transport(),
2898
self.assertRaises(errors.NoSuchFile, store.load)
2900
def test_invalid_content(self):
2901
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2902
self.assertEqual(False, store.is_loaded())
2903
exc = self.assertRaises(
2904
config.ParseConfigError, store._load_from_string,
2905
'this is invalid !')
2906
self.assertEndsWith(exc.filename, 'foo.conf')
2907
# And the load failed
2908
self.assertEqual(False, store.is_loaded())
2910
def test_get_embedded_sections(self):
2911
# A more complicated example (which also shows that section names and
2912
# option names share the same name space...)
2913
# FIXME: This should be fixed by forbidding dicts as values ?
2914
# -- vila 2011-04-05
2915
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2916
store._load_from_string('''
2920
foo_in_DEFAULT=foo_DEFAULT
2928
sections = list(store.get_sections())
2929
self.assertLength(4, sections)
2930
# The default section has no name.
2931
# List values are provided as strings and need to be explicitly
2932
# converted by specifying from_unicode=list_from_store at option
2934
self.assertSectionContent((None, {'foo': 'bar', 'l': u'1,2'}),
2936
self.assertSectionContent(
2937
('DEFAULT', {'foo_in_DEFAULT': 'foo_DEFAULT'}), sections[1])
2938
self.assertSectionContent(
2939
('bar', {'foo_in_bar': 'barbar'}), sections[2])
2940
# sub sections are provided as embedded dicts.
2941
self.assertSectionContent(
2942
('baz', {'foo_in_baz': 'barbaz', 'qux': {'foo_in_qux': 'quux'}}),
2946
class TestLockableIniFileStore(TestStore):
2948
def test_create_store_in_created_dir(self):
2949
self.assertPathDoesNotExist('dir')
2950
t = self.get_transport('dir/subdir')
2951
store = config.LockableIniFileStore(t, 'foo.conf')
2952
store.get_mutable_section(None).set('foo', 'bar')
2954
self.assertPathExists('dir/subdir')
2957
class TestConcurrentStoreUpdates(TestStore):
2958
"""Test that Stores properly handle conccurent updates.
2960
New Store implementation may fail some of these tests but until such
2961
implementations exist it's hard to properly filter them from the scenarios
2962
applied here. If you encounter such a case, contact the bzr devs.
2965
scenarios = [(key, {'get_stack': builder}) for key, builder
2966
in config.test_stack_builder_registry.iteritems()]
2969
super(TestConcurrentStoreUpdates, self).setUp()
2970
self.stack = self.get_stack(self)
2971
if not isinstance(self.stack, config._CompatibleStack):
2972
raise tests.TestNotApplicable(
2973
'%s is not meant to be compatible with the old config design'
2975
self.stack.set('one', '1')
2976
self.stack.set('two', '2')
2978
self.stack.store.save()
2980
def test_simple_read_access(self):
2981
self.assertEqual('1', self.stack.get('one'))
2983
def test_simple_write_access(self):
2984
self.stack.set('one', 'one')
2985
self.assertEqual('one', self.stack.get('one'))
2987
def test_listen_to_the_last_speaker(self):
2989
c2 = self.get_stack(self)
2990
c1.set('one', 'ONE')
2991
c2.set('two', 'TWO')
2992
self.assertEqual('ONE', c1.get('one'))
2993
self.assertEqual('TWO', c2.get('two'))
2994
# The second update respect the first one
2995
self.assertEqual('ONE', c2.get('one'))
2997
def test_last_speaker_wins(self):
2998
# If the same config is not shared, the same variable modified twice
2999
# can only see a single result.
3001
c2 = self.get_stack(self)
3004
self.assertEqual('c2', c2.get('one'))
3005
# The first modification is still available until another refresh
3007
self.assertEqual('c1', c1.get('one'))
3008
c1.set('two', 'done')
3009
self.assertEqual('c2', c1.get('one'))
3011
def test_writes_are_serialized(self):
3013
c2 = self.get_stack(self)
3015
# We spawn a thread that will pause *during* the config saving.
3016
before_writing = threading.Event()
3017
after_writing = threading.Event()
3018
writing_done = threading.Event()
3019
c1_save_without_locking_orig = c1.store.save_without_locking
3020
def c1_save_without_locking():
3021
before_writing.set()
3022
c1_save_without_locking_orig()
3023
# The lock is held. We wait for the main thread to decide when to
3025
after_writing.wait()
3026
c1.store.save_without_locking = c1_save_without_locking
3030
t1 = threading.Thread(target=c1_set)
3031
# Collect the thread after the test
3032
self.addCleanup(t1.join)
3033
# Be ready to unblock the thread if the test goes wrong
3034
self.addCleanup(after_writing.set)
3036
before_writing.wait()
3037
self.assertRaises(errors.LockContention,
3038
c2.set, 'one', 'c2')
3039
self.assertEqual('c1', c1.get('one'))
3040
# Let the lock be released
3044
self.assertEqual('c2', c2.get('one'))
3046
def test_read_while_writing(self):
3048
# We spawn a thread that will pause *during* the write
3049
ready_to_write = threading.Event()
3050
do_writing = threading.Event()
3051
writing_done = threading.Event()
3052
# We override the _save implementation so we know the store is locked
3053
c1_save_without_locking_orig = c1.store.save_without_locking
3054
def c1_save_without_locking():
3055
ready_to_write.set()
3056
# The lock is held. We wait for the main thread to decide when to
3059
c1_save_without_locking_orig()
3061
c1.store.save_without_locking = c1_save_without_locking
3064
t1 = threading.Thread(target=c1_set)
3065
# Collect the thread after the test
3066
self.addCleanup(t1.join)
3067
# Be ready to unblock the thread if the test goes wrong
3068
self.addCleanup(do_writing.set)
3070
# Ensure the thread is ready to write
3071
ready_to_write.wait()
3072
self.assertEqual('c1', c1.get('one'))
3073
# If we read during the write, we get the old value
3074
c2 = self.get_stack(self)
3075
self.assertEqual('1', c2.get('one'))
3076
# Let the writing occur and ensure it occurred
3079
# Now we get the updated value
3080
c3 = self.get_stack(self)
3081
self.assertEqual('c1', c3.get('one'))
3083
# FIXME: It may be worth looking into removing the lock dir when it's not
3084
# needed anymore and look at possible fallouts for concurrent lockers. This
3085
# will matter if/when we use config files outside of breezy directories
3086
# (.config/breezy or .bzr) -- vila 20110-04-111
3089
class TestSectionMatcher(TestStore):
3091
scenarios = [('location', {'matcher': config.LocationMatcher}),
3092
('id', {'matcher': config.NameMatcher}),]
3095
super(TestSectionMatcher, self).setUp()
3096
# Any simple store is good enough
3097
self.get_store = config.test_store_builder_registry.get('configobj')
3099
def test_no_matches_for_empty_stores(self):
3100
store = self.get_store(self)
3101
store._load_from_string('')
3102
matcher = self.matcher(store, '/bar')
3103
self.assertEqual([], list(matcher.get_sections()))
3105
def test_build_doesnt_load_store(self):
3106
store = self.get_store(self)
3107
self.matcher(store, '/bar')
3108
self.assertFalse(store.is_loaded())
3111
class TestLocationSection(tests.TestCase):
3113
def get_section(self, options, extra_path):
3114
section = config.Section('foo', options)
3115
return config.LocationSection(section, extra_path)
3117
def test_simple_option(self):
3118
section = self.get_section({'foo': 'bar'}, '')
3119
self.assertEqual('bar', section.get('foo'))
3121
def test_option_with_extra_path(self):
3122
section = self.get_section({'foo': 'bar', 'foo:policy': 'appendpath'},
3124
self.assertEqual('bar/baz', section.get('foo'))
3126
def test_invalid_policy(self):
3127
section = self.get_section({'foo': 'bar', 'foo:policy': 'die'},
3129
# invalid policies are ignored
3130
self.assertEqual('bar', section.get('foo'))
3133
class TestLocationMatcher(TestStore):
3136
super(TestLocationMatcher, self).setUp()
3137
# Any simple store is good enough
3138
self.get_store = config.test_store_builder_registry.get('configobj')
3140
def test_unrelated_section_excluded(self):
3141
store = self.get_store(self)
3142
store._load_from_string('''
3150
section=/foo/bar/baz
3154
self.assertEqual(['/foo', '/foo/baz', '/foo/bar', '/foo/bar/baz',
3156
[section.id for _, section in store.get_sections()])
3157
matcher = config.LocationMatcher(store, '/foo/bar/quux')
3158
sections = [section for _, section in matcher.get_sections()]
3159
self.assertEqual(['/foo/bar', '/foo'],
3160
[section.id for section in sections])
3161
self.assertEqual(['quux', 'bar/quux'],
3162
[section.extra_path for section in sections])
3164
def test_more_specific_sections_first(self):
3165
store = self.get_store(self)
3166
store._load_from_string('''
3172
self.assertEqual(['/foo', '/foo/bar'],
3173
[section.id for _, section in store.get_sections()])
3174
matcher = config.LocationMatcher(store, '/foo/bar/baz')
3175
sections = [section for _, section in matcher.get_sections()]
3176
self.assertEqual(['/foo/bar', '/foo'],
3177
[section.id for section in sections])
3178
self.assertEqual(['baz', 'bar/baz'],
3179
[section.extra_path for section in sections])
3181
def test_appendpath_in_no_name_section(self):
3182
# It's a bit weird to allow appendpath in a no-name section, but
3183
# someone may found a use for it
3184
store = self.get_store(self)
3185
store._load_from_string('''
3187
foo:policy = appendpath
3189
matcher = config.LocationMatcher(store, 'dir/subdir')
3190
sections = list(matcher.get_sections())
3191
self.assertLength(1, sections)
3192
self.assertEqual('bar/dir/subdir', sections[0][1].get('foo'))
3194
def test_file_urls_are_normalized(self):
3195
store = self.get_store(self)
3196
if sys.platform == 'win32':
3197
expected_url = 'file:///C:/dir/subdir'
3198
expected_location = 'C:/dir/subdir'
3200
expected_url = 'file:///dir/subdir'
3201
expected_location = '/dir/subdir'
3202
matcher = config.LocationMatcher(store, expected_url)
3203
self.assertEqual(expected_location, matcher.location)
3205
def test_branch_name_colo(self):
3206
store = self.get_store(self)
3207
store._load_from_string(dedent("""\
3209
push_location=my{branchname}
3211
matcher = config.LocationMatcher(store, 'file:///,branch=example%3c')
3212
self.assertEqual('example<', matcher.branch_name)
3213
((_, section),) = matcher.get_sections()
3214
self.assertEqual('example<', section.locals['branchname'])
3216
def test_branch_name_basename(self):
3217
store = self.get_store(self)
3218
store._load_from_string(dedent("""\
3220
push_location=my{branchname}
3222
matcher = config.LocationMatcher(store, 'file:///parent/example%3c')
3223
self.assertEqual('example<', matcher.branch_name)
3224
((_, section),) = matcher.get_sections()
3225
self.assertEqual('example<', section.locals['branchname'])
3228
class TestStartingPathMatcher(TestStore):
3231
super(TestStartingPathMatcher, self).setUp()
3232
# Any simple store is good enough
3233
self.store = config.IniFileStore()
3235
def assertSectionIDs(self, expected, location, content):
3236
self.store._load_from_string(content)
3237
matcher = config.StartingPathMatcher(self.store, location)
3238
sections = list(matcher.get_sections())
3239
self.assertLength(len(expected), sections)
3240
self.assertEqual(expected, [section.id for _, section in sections])
3243
def test_empty(self):
3244
self.assertSectionIDs([], self.get_url(), '')
3246
def test_url_vs_local_paths(self):
3247
# The matcher location is an url and the section names are local paths
3248
self.assertSectionIDs(['/foo/bar', '/foo'],
3249
'file:///foo/bar/baz', '''\
3254
def test_local_path_vs_url(self):
3255
# The matcher location is a local path and the section names are urls
3256
self.assertSectionIDs(['file:///foo/bar', 'file:///foo'],
3257
'/foo/bar/baz', '''\
3263
def test_no_name_section_included_when_present(self):
3264
# Note that other tests will cover the case where the no-name section
3265
# is empty and as such, not included.
3266
sections = self.assertSectionIDs(['/foo/bar', '/foo', None],
3267
'/foo/bar/baz', '''\
3268
option = defined so the no-name section exists
3272
self.assertEqual(['baz', 'bar/baz', '/foo/bar/baz'],
3273
[s.locals['relpath'] for _, s in sections])
3275
def test_order_reversed(self):
3276
self.assertSectionIDs(['/foo/bar', '/foo'], '/foo/bar/baz', '''\
3281
def test_unrelated_section_excluded(self):
3282
self.assertSectionIDs(['/foo/bar', '/foo'], '/foo/bar/baz', '''\
3288
def test_glob_included(self):
3289
sections = self.assertSectionIDs(['/foo/*/baz', '/foo/b*', '/foo'],
3290
'/foo/bar/baz', '''\
3296
# Note that 'baz' as a relpath for /foo/b* is not fully correct, but
3297
# nothing really is... as far using {relpath} to append it to something
3298
# else, this seems good enough though.
3299
self.assertEqual(['', 'baz', 'bar/baz'],
3300
[s.locals['relpath'] for _, s in sections])
3302
def test_respect_order(self):
3303
self.assertSectionIDs(['/foo', '/foo/b*', '/foo/*/baz'],
3304
'/foo/bar/baz', '''\
3312
class TestNameMatcher(TestStore):
3315
super(TestNameMatcher, self).setUp()
3316
self.matcher = config.NameMatcher
3317
# Any simple store is good enough
3318
self.get_store = config.test_store_builder_registry.get('configobj')
3320
def get_matching_sections(self, name):
3321
store = self.get_store(self)
3322
store._load_from_string('''
3330
matcher = self.matcher(store, name)
3331
return list(matcher.get_sections())
3333
def test_matching(self):
3334
sections = self.get_matching_sections('foo')
3335
self.assertLength(1, sections)
3336
self.assertSectionContent(('foo', {'option': 'foo'}), sections[0])
3338
def test_not_matching(self):
3339
sections = self.get_matching_sections('baz')
3340
self.assertLength(0, sections)
3343
class TestBaseStackGet(tests.TestCase):
3346
super(TestBaseStackGet, self).setUp()
3347
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3349
def test_get_first_definition(self):
3350
store1 = config.IniFileStore()
3351
store1._load_from_string('foo=bar')
3352
store2 = config.IniFileStore()
3353
store2._load_from_string('foo=baz')
3354
conf = config.Stack([store1.get_sections, store2.get_sections])
3355
self.assertEqual('bar', conf.get('foo'))
3357
def test_get_with_registered_default_value(self):
3358
config.option_registry.register(config.Option('foo', default='bar'))
3359
conf_stack = config.Stack([])
3360
self.assertEqual('bar', conf_stack.get('foo'))
3362
def test_get_without_registered_default_value(self):
3363
config.option_registry.register(config.Option('foo'))
3364
conf_stack = config.Stack([])
3365
self.assertEqual(None, conf_stack.get('foo'))
3367
def test_get_without_default_value_for_not_registered(self):
3368
conf_stack = config.Stack([])
3369
self.assertEqual(None, conf_stack.get('foo'))
3371
def test_get_for_empty_section_callable(self):
3372
conf_stack = config.Stack([lambda : []])
3373
self.assertEqual(None, conf_stack.get('foo'))
3375
def test_get_for_broken_callable(self):
3376
# Trying to use and invalid callable raises an exception on first use
3377
conf_stack = config.Stack([object])
3378
self.assertRaises(TypeError, conf_stack.get, 'foo')
3381
class TestStackWithSimpleStore(tests.TestCase):
3384
super(TestStackWithSimpleStore, self).setUp()
3385
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3386
self.registry = config.option_registry
3388
def get_conf(self, content=None):
3389
return config.MemoryStack(content)
3391
def test_override_value_from_env(self):
3392
self.overrideEnv('FOO', None)
3393
self.registry.register(
3394
config.Option('foo', default='bar', override_from_env=['FOO']))
3395
self.overrideEnv('FOO', 'quux')
3396
# Env variable provides a default taking over the option one
3397
conf = self.get_conf('foo=store')
3398
self.assertEqual('quux', conf.get('foo'))
3400
def test_first_override_value_from_env_wins(self):
3401
self.overrideEnv('NO_VALUE', None)
3402
self.overrideEnv('FOO', None)
3403
self.overrideEnv('BAZ', None)
3404
self.registry.register(
3405
config.Option('foo', default='bar',
3406
override_from_env=['NO_VALUE', 'FOO', 'BAZ']))
3407
self.overrideEnv('FOO', 'foo')
3408
self.overrideEnv('BAZ', 'baz')
3409
# The first env var set wins
3410
conf = self.get_conf('foo=store')
3411
self.assertEqual('foo', conf.get('foo'))
3414
class TestMemoryStack(tests.TestCase):
3417
conf = config.MemoryStack('foo=bar')
3418
self.assertEqual('bar', conf.get('foo'))
3421
conf = config.MemoryStack('foo=bar')
3422
conf.set('foo', 'baz')
3423
self.assertEqual('baz', conf.get('foo'))
3425
def test_no_content(self):
3426
conf = config.MemoryStack()
3427
# No content means no loading
3428
self.assertFalse(conf.store.is_loaded())
3429
self.assertRaises(NotImplementedError, conf.get, 'foo')
3430
# But a content can still be provided
3431
conf.store._load_from_string('foo=bar')
3432
self.assertEqual('bar', conf.get('foo'))
3435
class TestStackIterSections(tests.TestCase):
3437
def test_empty_stack(self):
3438
conf = config.Stack([])
3439
sections = list(conf.iter_sections())
3440
self.assertLength(0, sections)
3442
def test_empty_store(self):
3443
store = config.IniFileStore()
3444
store._load_from_string('')
3445
conf = config.Stack([store.get_sections])
3446
sections = list(conf.iter_sections())
3447
self.assertLength(0, sections)
3449
def test_simple_store(self):
3450
store = config.IniFileStore()
3451
store._load_from_string('foo=bar')
3452
conf = config.Stack([store.get_sections])
3453
tuples = list(conf.iter_sections())
3454
self.assertLength(1, tuples)
3455
(found_store, found_section) = tuples[0]
3456
self.assertIs(store, found_store)
3458
def test_two_stores(self):
3459
store1 = config.IniFileStore()
3460
store1._load_from_string('foo=bar')
3461
store2 = config.IniFileStore()
3462
store2._load_from_string('bar=qux')
3463
conf = config.Stack([store1.get_sections, store2.get_sections])
3464
tuples = list(conf.iter_sections())
3465
self.assertLength(2, tuples)
3466
self.assertIs(store1, tuples[0][0])
3467
self.assertIs(store2, tuples[1][0])
3470
class TestStackWithTransport(tests.TestCaseWithTransport):
3472
scenarios = [(key, {'get_stack': builder}) for key, builder
3473
in config.test_stack_builder_registry.iteritems()]
3476
class TestConcreteStacks(TestStackWithTransport):
3478
def test_build_stack(self):
3479
# Just a smoke test to help debug builders
3480
self.get_stack(self)
3483
class TestStackGet(TestStackWithTransport):
3486
super(TestStackGet, self).setUp()
3487
self.conf = self.get_stack(self)
3489
def test_get_for_empty_stack(self):
3490
self.assertEqual(None, self.conf.get('foo'))
3492
def test_get_hook(self):
3493
self.conf.set('foo', 'bar')
3497
config.ConfigHooks.install_named_hook('get', hook, None)
3498
self.assertLength(0, calls)
3499
value = self.conf.get('foo')
3500
self.assertEqual('bar', value)
3501
self.assertLength(1, calls)
3502
self.assertEqual((self.conf, 'foo', 'bar'), calls[0])
3505
class TestStackGetWithConverter(tests.TestCase):
3508
super(TestStackGetWithConverter, self).setUp()
3509
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3510
self.registry = config.option_registry
3512
def get_conf(self, content=None):
3513
return config.MemoryStack(content)
3515
def register_bool_option(self, name, default=None, default_from_env=None):
3516
b = config.Option(name, help='A boolean.',
3517
default=default, default_from_env=default_from_env,
3518
from_unicode=config.bool_from_store)
3519
self.registry.register(b)
3521
def test_get_default_bool_None(self):
3522
self.register_bool_option('foo')
3523
conf = self.get_conf('')
3524
self.assertEqual(None, conf.get('foo'))
3526
def test_get_default_bool_True(self):
3527
self.register_bool_option('foo', u'True')
3528
conf = self.get_conf('')
3529
self.assertEqual(True, conf.get('foo'))
3531
def test_get_default_bool_False(self):
3532
self.register_bool_option('foo', False)
3533
conf = self.get_conf('')
3534
self.assertEqual(False, conf.get('foo'))
3536
def test_get_default_bool_False_as_string(self):
3537
self.register_bool_option('foo', u'False')
3538
conf = self.get_conf('')
3539
self.assertEqual(False, conf.get('foo'))
3541
def test_get_default_bool_from_env_converted(self):
3542
self.register_bool_option('foo', u'True', default_from_env=['FOO'])
3543
self.overrideEnv('FOO', 'False')
3544
conf = self.get_conf('')
3545
self.assertEqual(False, conf.get('foo'))
3547
def test_get_default_bool_when_conversion_fails(self):
3548
self.register_bool_option('foo', default='True')
3549
conf = self.get_conf('foo=invalid boolean')
3550
self.assertEqual(True, conf.get('foo'))
3552
def register_integer_option(self, name,
3553
default=None, default_from_env=None):
3554
i = config.Option(name, help='An integer.',
3555
default=default, default_from_env=default_from_env,
3556
from_unicode=config.int_from_store)
3557
self.registry.register(i)
3559
def test_get_default_integer_None(self):
3560
self.register_integer_option('foo')
3561
conf = self.get_conf('')
3562
self.assertEqual(None, conf.get('foo'))
3564
def test_get_default_integer(self):
3565
self.register_integer_option('foo', 42)
3566
conf = self.get_conf('')
3567
self.assertEqual(42, conf.get('foo'))
3569
def test_get_default_integer_as_string(self):
3570
self.register_integer_option('foo', u'42')
3571
conf = self.get_conf('')
3572
self.assertEqual(42, conf.get('foo'))
3574
def test_get_default_integer_from_env(self):
3575
self.register_integer_option('foo', default_from_env=['FOO'])
3576
self.overrideEnv('FOO', '18')
3577
conf = self.get_conf('')
3578
self.assertEqual(18, conf.get('foo'))
3580
def test_get_default_integer_when_conversion_fails(self):
3581
self.register_integer_option('foo', default='12')
3582
conf = self.get_conf('foo=invalid integer')
3583
self.assertEqual(12, conf.get('foo'))
3585
def register_list_option(self, name, default=None, default_from_env=None):
3586
l = config.ListOption(name, help='A list.', default=default,
3587
default_from_env=default_from_env)
3588
self.registry.register(l)
3590
def test_get_default_list_None(self):
3591
self.register_list_option('foo')
3592
conf = self.get_conf('')
3593
self.assertEqual(None, conf.get('foo'))
3595
def test_get_default_list_empty(self):
3596
self.register_list_option('foo', '')
3597
conf = self.get_conf('')
3598
self.assertEqual([], conf.get('foo'))
3600
def test_get_default_list_from_env(self):
3601
self.register_list_option('foo', default_from_env=['FOO'])
3602
self.overrideEnv('FOO', '')
3603
conf = self.get_conf('')
3604
self.assertEqual([], conf.get('foo'))
3606
def test_get_with_list_converter_no_item(self):
3607
self.register_list_option('foo', None)
3608
conf = self.get_conf('foo=,')
3609
self.assertEqual([], conf.get('foo'))
3611
def test_get_with_list_converter_many_items(self):
3612
self.register_list_option('foo', None)
3613
conf = self.get_conf('foo=m,o,r,e')
3614
self.assertEqual(['m', 'o', 'r', 'e'], conf.get('foo'))
3616
def test_get_with_list_converter_embedded_spaces_many_items(self):
3617
self.register_list_option('foo', None)
3618
conf = self.get_conf('foo=" bar", "baz "')
3619
self.assertEqual([' bar', 'baz '], conf.get('foo'))
3621
def test_get_with_list_converter_stripped_spaces_many_items(self):
3622
self.register_list_option('foo', None)
3623
conf = self.get_conf('foo= bar , baz ')
3624
self.assertEqual(['bar', 'baz'], conf.get('foo'))
3627
class TestIterOptionRefs(tests.TestCase):
3628
"""iter_option_refs is a bit unusual, document some cases."""
3630
def assertRefs(self, expected, string):
3631
self.assertEqual(expected, list(config.iter_option_refs(string)))
3633
def test_empty(self):
3634
self.assertRefs([(False, '')], '')
3636
def test_no_refs(self):
3637
self.assertRefs([(False, 'foo bar')], 'foo bar')
3639
def test_single_ref(self):
3640
self.assertRefs([(False, ''), (True, '{foo}'), (False, '')], '{foo}')
3642
def test_broken_ref(self):
3643
self.assertRefs([(False, '{foo')], '{foo')
3645
def test_embedded_ref(self):
3646
self.assertRefs([(False, '{'), (True, '{foo}'), (False, '}')],
3649
def test_two_refs(self):
3650
self.assertRefs([(False, ''), (True, '{foo}'),
3651
(False, ''), (True, '{bar}'),
3655
def test_newline_in_refs_are_not_matched(self):
3656
self.assertRefs([(False, '{\nxx}{xx\n}{{\n}}')], '{\nxx}{xx\n}{{\n}}')
3659
class TestStackExpandOptions(tests.TestCaseWithTransport):
3662
super(TestStackExpandOptions, self).setUp()
3663
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3664
self.registry = config.option_registry
3665
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
3666
self.conf = config.Stack([store.get_sections], store)
3668
def assertExpansion(self, expected, string, env=None):
3669
self.assertEqual(expected, self.conf.expand_options(string, env))
3671
def test_no_expansion(self):
3672
self.assertExpansion('foo', 'foo')
3674
def test_expand_default_value(self):
3675
self.conf.store._load_from_string('bar=baz')
3676
self.registry.register(config.Option('foo', default=u'{bar}'))
3677
self.assertEqual('baz', self.conf.get('foo', expand=True))
3679
def test_expand_default_from_env(self):
3680
self.conf.store._load_from_string('bar=baz')
3681
self.registry.register(config.Option('foo', default_from_env=['FOO']))
3682
self.overrideEnv('FOO', '{bar}')
3683
self.assertEqual('baz', self.conf.get('foo', expand=True))
3685
def test_expand_default_on_failed_conversion(self):
3686
self.conf.store._load_from_string('baz=bogus\nbar=42\nfoo={baz}')
3687
self.registry.register(
3688
config.Option('foo', default=u'{bar}',
3689
from_unicode=config.int_from_store))
3690
self.assertEqual(42, self.conf.get('foo', expand=True))
3692
def test_env_adding_options(self):
3693
self.assertExpansion('bar', '{foo}', {'foo': 'bar'})
3695
def test_env_overriding_options(self):
3696
self.conf.store._load_from_string('foo=baz')
3697
self.assertExpansion('bar', '{foo}', {'foo': 'bar'})
3699
def test_simple_ref(self):
3700
self.conf.store._load_from_string('foo=xxx')
3701
self.assertExpansion('xxx', '{foo}')
3703
def test_unknown_ref(self):
3704
self.assertRaises(config.ExpandingUnknownOption,
3705
self.conf.expand_options, '{foo}')
3707
def test_illegal_def_is_ignored(self):
3708
self.assertExpansion('{1,2}', '{1,2}')
3709
self.assertExpansion('{ }', '{ }')
3710
self.assertExpansion('${Foo,f}', '${Foo,f}')
3712
def test_indirect_ref(self):
3713
self.conf.store._load_from_string('''
3717
self.assertExpansion('xxx', '{bar}')
3719
def test_embedded_ref(self):
3720
self.conf.store._load_from_string('''
3724
self.assertExpansion('xxx', '{{bar}}')
3726
def test_simple_loop(self):
3727
self.conf.store._load_from_string('foo={foo}')
3728
self.assertRaises(config.OptionExpansionLoop,
3729
self.conf.expand_options, '{foo}')
3731
def test_indirect_loop(self):
3732
self.conf.store._load_from_string('''
3736
e = self.assertRaises(config.OptionExpansionLoop,
3737
self.conf.expand_options, '{foo}')
3738
self.assertEqual('foo->bar->baz', e.refs)
3739
self.assertEqual('{foo}', e.string)
3741
def test_list(self):
3742
self.conf.store._load_from_string('''
3746
list={foo},{bar},{baz}
3748
self.registry.register(
3749
config.ListOption('list'))
3750
self.assertEqual(['start', 'middle', 'end'],
3751
self.conf.get('list', expand=True))
3753
def test_cascading_list(self):
3754
self.conf.store._load_from_string('''
3760
self.registry.register(config.ListOption('list'))
3761
# Register an intermediate option as a list to ensure no conversion
3762
# happen while expanding. Conversion should only occur for the original
3763
# option ('list' here).
3764
self.registry.register(config.ListOption('baz'))
3765
self.assertEqual(['start', 'middle', 'end'],
3766
self.conf.get('list', expand=True))
3768
def test_pathologically_hidden_list(self):
3769
self.conf.store._load_from_string('''
3775
hidden={start}{middle}{end}
3777
# What matters is what the registration says, the conversion happens
3778
# only after all expansions have been performed
3779
self.registry.register(config.ListOption('hidden'))
3780
self.assertEqual(['bin', 'go'],
3781
self.conf.get('hidden', expand=True))
3784
class TestStackCrossSectionsExpand(tests.TestCaseWithTransport):
3787
super(TestStackCrossSectionsExpand, self).setUp()
3789
def get_config(self, location, string):
3792
# Since we don't save the config we won't strictly require to inherit
3793
# from TestCaseInTempDir, but an error occurs so quickly...
3794
c = config.LocationStack(location)
3795
c.store._load_from_string(string)
3798
def test_dont_cross_unrelated_section(self):
3799
c = self.get_config('/another/branch/path','''
3804
[/another/branch/path]
3807
self.assertRaises(config.ExpandingUnknownOption,
3808
c.get, 'bar', expand=True)
3810
def test_cross_related_sections(self):
3811
c = self.get_config('/project/branch/path','''
3815
[/project/branch/path]
3818
self.assertEqual('quux', c.get('bar', expand=True))
3821
class TestStackCrossStoresExpand(tests.TestCaseWithTransport):
3823
def test_cross_global_locations(self):
3824
l_store = config.LocationStore()
3825
l_store._load_from_string('''
3831
g_store = config.GlobalStore()
3832
g_store._load_from_string('''
3838
stack = config.LocationStack('/branch')
3839
self.assertEqual('glob-bar', stack.get('lbar', expand=True))
3840
self.assertEqual('loc-foo', stack.get('gfoo', expand=True))
3843
class TestStackExpandSectionLocals(tests.TestCaseWithTransport):
3845
def test_expand_locals_empty(self):
3846
l_store = config.LocationStore()
3847
l_store._load_from_string('''
3848
[/home/user/project]
3853
stack = config.LocationStack('/home/user/project/')
3854
self.assertEqual('', stack.get('base', expand=True))
3855
self.assertEqual('', stack.get('rel', expand=True))
3857
def test_expand_basename_locally(self):
3858
l_store = config.LocationStore()
3859
l_store._load_from_string('''
3860
[/home/user/project]
3864
stack = config.LocationStack('/home/user/project/branch')
3865
self.assertEqual('branch', stack.get('bfoo', expand=True))
3867
def test_expand_basename_locally_longer_path(self):
3868
l_store = config.LocationStore()
3869
l_store._load_from_string('''
3874
stack = config.LocationStack('/home/user/project/dir/branch')
3875
self.assertEqual('branch', stack.get('bfoo', expand=True))
3877
def test_expand_relpath_locally(self):
3878
l_store = config.LocationStore()
3879
l_store._load_from_string('''
3880
[/home/user/project]
3881
lfoo = loc-foo/{relpath}
3884
stack = config.LocationStack('/home/user/project/branch')
3885
self.assertEqual('loc-foo/branch', stack.get('lfoo', expand=True))
3887
def test_expand_relpath_unknonw_in_global(self):
3888
g_store = config.GlobalStore()
3889
g_store._load_from_string('''
3894
stack = config.LocationStack('/home/user/project/branch')
3895
self.assertRaises(config.ExpandingUnknownOption,
3896
stack.get, 'gfoo', expand=True)
3898
def test_expand_local_option_locally(self):
3899
l_store = config.LocationStore()
3900
l_store._load_from_string('''
3901
[/home/user/project]
3902
lfoo = loc-foo/{relpath}
3906
g_store = config.GlobalStore()
3907
g_store._load_from_string('''
3913
stack = config.LocationStack('/home/user/project/branch')
3914
self.assertEqual('glob-bar', stack.get('lbar', expand=True))
3915
self.assertEqual('loc-foo/branch', stack.get('gfoo', expand=True))
3917
def test_locals_dont_leak(self):
3918
"""Make sure we chose the right local in presence of several sections.
3920
l_store = config.LocationStore()
3921
l_store._load_from_string('''
3923
lfoo = loc-foo/{relpath}
3924
[/home/user/project]
3925
lfoo = loc-foo/{relpath}
3928
stack = config.LocationStack('/home/user/project/branch')
3929
self.assertEqual('loc-foo/branch', stack.get('lfoo', expand=True))
3930
stack = config.LocationStack('/home/user/bar/baz')
3931
self.assertEqual('loc-foo/bar/baz', stack.get('lfoo', expand=True))
3935
class TestStackSet(TestStackWithTransport):
3937
def test_simple_set(self):
3938
conf = self.get_stack(self)
3939
self.assertEqual(None, conf.get('foo'))
3940
conf.set('foo', 'baz')
3941
# Did we get it back ?
3942
self.assertEqual('baz', conf.get('foo'))
3944
def test_set_creates_a_new_section(self):
3945
conf = self.get_stack(self)
3946
conf.set('foo', 'baz')
3947
self.assertEqual, 'baz', conf.get('foo')
3949
def test_set_hook(self):
3953
config.ConfigHooks.install_named_hook('set', hook, None)
3954
self.assertLength(0, calls)
3955
conf = self.get_stack(self)
3956
conf.set('foo', 'bar')
3957
self.assertLength(1, calls)
3958
self.assertEqual((conf, 'foo', 'bar'), calls[0])
3961
class TestStackRemove(TestStackWithTransport):
3963
def test_remove_existing(self):
3964
conf = self.get_stack(self)
3965
conf.set('foo', 'bar')
3966
self.assertEqual('bar', conf.get('foo'))
3968
# Did we get it back ?
3969
self.assertEqual(None, conf.get('foo'))
3971
def test_remove_unknown(self):
3972
conf = self.get_stack(self)
3973
self.assertRaises(KeyError, conf.remove, 'I_do_not_exist')
3975
def test_remove_hook(self):
3979
config.ConfigHooks.install_named_hook('remove', hook, None)
3980
self.assertLength(0, calls)
3981
conf = self.get_stack(self)
3982
conf.set('foo', 'bar')
3984
self.assertLength(1, calls)
3985
self.assertEqual((conf, 'foo'), calls[0])
3988
class TestConfigGetOptions(tests.TestCaseWithTransport, TestOptionsMixin):
3991
super(TestConfigGetOptions, self).setUp()
3992
create_configs(self)
3994
def test_no_variable(self):
3995
# Using branch should query branch, locations and breezy
3996
self.assertOptions([], self.branch_config)
3998
def test_option_in_breezy(self):
3999
self.breezy_config.set_user_option('file', 'breezy')
4000
self.assertOptions([('file', 'breezy', 'DEFAULT', 'breezy')],
4003
def test_option_in_locations(self):
4004
self.locations_config.set_user_option('file', 'locations')
4006
[('file', 'locations', self.tree.basedir, 'locations')],
4007
self.locations_config)
4009
def test_option_in_branch(self):
4010
self.branch_config.set_user_option('file', 'branch')
4011
self.assertOptions([('file', 'branch', 'DEFAULT', 'branch')],
4014
def test_option_in_breezy_and_branch(self):
4015
self.breezy_config.set_user_option('file', 'breezy')
4016
self.branch_config.set_user_option('file', 'branch')
4017
self.assertOptions([('file', 'branch', 'DEFAULT', 'branch'),
4018
('file', 'breezy', 'DEFAULT', 'breezy'),],
4021
def test_option_in_branch_and_locations(self):
4022
# Hmm, locations override branch :-/
4023
self.locations_config.set_user_option('file', 'locations')
4024
self.branch_config.set_user_option('file', 'branch')
4026
[('file', 'locations', self.tree.basedir, 'locations'),
4027
('file', 'branch', 'DEFAULT', 'branch'),],
4030
def test_option_in_breezy_locations_and_branch(self):
4031
self.breezy_config.set_user_option('file', 'breezy')
4032
self.locations_config.set_user_option('file', 'locations')
4033
self.branch_config.set_user_option('file', 'branch')
4035
[('file', 'locations', self.tree.basedir, 'locations'),
4036
('file', 'branch', 'DEFAULT', 'branch'),
4037
('file', 'breezy', 'DEFAULT', 'breezy'),],
4041
class TestConfigRemoveOption(tests.TestCaseWithTransport, TestOptionsMixin):
4044
super(TestConfigRemoveOption, self).setUp()
4045
create_configs_with_file_option(self)
4047
def test_remove_in_locations(self):
4048
self.locations_config.remove_user_option('file', self.tree.basedir)
4050
[('file', 'branch', 'DEFAULT', 'branch'),
4051
('file', 'breezy', 'DEFAULT', 'breezy'),],
4054
def test_remove_in_branch(self):
4055
self.branch_config.remove_user_option('file')
4057
[('file', 'locations', self.tree.basedir, 'locations'),
4058
('file', 'breezy', 'DEFAULT', 'breezy'),],
4061
def test_remove_in_breezy(self):
4062
self.breezy_config.remove_user_option('file')
4064
[('file', 'locations', self.tree.basedir, 'locations'),
4065
('file', 'branch', 'DEFAULT', 'branch'),],
4069
class TestConfigGetSections(tests.TestCaseWithTransport):
4072
super(TestConfigGetSections, self).setUp()
4073
create_configs(self)
4075
def assertSectionNames(self, expected, conf, name=None):
4076
"""Check which sections are returned for a given config.
4078
If fallback configurations exist their sections can be included.
4080
:param expected: A list of section names.
4082
:param conf: The configuration that will be queried.
4084
:param name: An optional section name that will be passed to
4087
sections = list(conf._get_sections(name))
4088
self.assertLength(len(expected), sections)
4089
self.assertEqual(expected, [n for n, _, _ in sections])
4091
def test_breezy_default_section(self):
4092
self.assertSectionNames(['DEFAULT'], self.breezy_config)
4094
def test_locations_default_section(self):
4095
# No sections are defined in an empty file
4096
self.assertSectionNames([], self.locations_config)
4098
def test_locations_named_section(self):
4099
self.locations_config.set_user_option('file', 'locations')
4100
self.assertSectionNames([self.tree.basedir], self.locations_config)
4102
def test_locations_matching_sections(self):
4103
loc_config = self.locations_config
4104
loc_config.set_user_option('file', 'locations')
4105
# We need to cheat a bit here to create an option in sections above and
4106
# below the 'location' one.
4107
parser = loc_config._get_parser()
4108
# locations.cong deals with '/' ignoring native os.sep
4109
location_names = self.tree.basedir.split('/')
4110
parent = '/'.join(location_names[:-1])
4111
child = '/'.join(location_names + ['child'])
4113
parser[parent]['file'] = 'parent'
4115
parser[child]['file'] = 'child'
4116
self.assertSectionNames([self.tree.basedir, parent], loc_config)
4118
def test_branch_data_default_section(self):
4119
self.assertSectionNames([None],
4120
self.branch_config._get_branch_data_config())
4122
def test_branch_default_sections(self):
4123
# No sections are defined in an empty locations file
4124
self.assertSectionNames([None, 'DEFAULT'],
4126
# Unless we define an option
4127
self.branch_config._get_location_config().set_user_option(
4128
'file', 'locations')
4129
self.assertSectionNames([self.tree.basedir, None, 'DEFAULT'],
4132
def test_breezy_named_section(self):
4133
# We need to cheat as the API doesn't give direct access to sections
4134
# other than DEFAULT.
4135
self.breezy_config.set_alias('breezy', 'bzr')
4136
self.assertSectionNames(['ALIASES'], self.breezy_config, 'ALIASES')
4139
class TestSharedStores(tests.TestCaseInTempDir):
4141
def test_breezy_conf_shared(self):
4142
g1 = config.GlobalStack()
4143
g2 = config.GlobalStack()
4144
# The two stacks share the same store
4145
self.assertIs(g1.store, g2.store)
4148
class TestAuthenticationConfigFilePermissions(tests.TestCaseInTempDir):
4149
"""Test warning for permissions of authentication.conf."""
4152
super(TestAuthenticationConfigFilePermissions, self).setUp()
4153
self.path = osutils.pathjoin(self.test_dir, 'authentication.conf')
4154
with open(self.path, 'w') as f:
4155
f.write(b"""[broken]
4158
port=port # Error: Not an int
4160
self.overrideAttr(config, 'authentication_config_filename',
4162
osutils.chmod_if_possible(self.path, 0o755)
4164
def test_check_warning(self):
4165
conf = config.AuthenticationConfig()
4166
self.assertEqual(conf._filename, self.path)
4167
self.assertContainsRe(self.get_log(),
4168
'Saved passwords may be accessible by other users.')
4170
def test_check_suppressed_warning(self):
4171
global_config = config.GlobalConfig()
4172
global_config.set_user_option('suppress_warnings',
4173
'insecure_permissions')
4174
conf = config.AuthenticationConfig()
4175
self.assertEqual(conf._filename, self.path)
4176
self.assertNotContainsRe(self.get_log(),
4177
'Saved passwords may be accessible by other users.')
4180
class TestAuthenticationConfigFile(tests.TestCase):
4181
"""Test the authentication.conf file matching"""
4183
def _got_user_passwd(self, expected_user, expected_password,
4184
config, *args, **kwargs):
4185
credentials = config.get_credentials(*args, **kwargs)
4186
if credentials is None:
4190
user = credentials['user']
4191
password = credentials['password']
4192
self.assertEqual(expected_user, user)
4193
self.assertEqual(expected_password, password)
4195
def test_empty_config(self):
4196
conf = config.AuthenticationConfig(_file=BytesIO())
4197
self.assertEqual({}, conf._get_config())
4198
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
4200
def test_non_utf8_config(self):
4201
conf = config.AuthenticationConfig(_file=BytesIO(b'foo = bar\xff'))
4202
self.assertRaises(config.ConfigContentError, conf._get_config)
4204
def test_missing_auth_section_header(self):
4205
conf = config.AuthenticationConfig(_file=BytesIO(b'foo = bar'))
4206
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4208
def test_auth_section_header_not_closed(self):
4209
conf = config.AuthenticationConfig(_file=BytesIO(b'[DEF'))
4210
self.assertRaises(config.ParseConfigError, conf._get_config)
4212
def test_auth_value_not_boolean(self):
4213
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4217
verify_certificates=askme # Error: Not a boolean
4219
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4221
def test_auth_value_not_int(self):
4222
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4226
port=port # Error: Not an int
4228
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4230
def test_unknown_password_encoding(self):
4231
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4235
password_encoding=unknown
4237
self.assertRaises(ValueError, conf.get_password,
4238
'ftp', 'foo.net', 'joe')
4240
def test_credentials_for_scheme_host(self):
4241
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4242
# Identity on foo.net
4247
password=secret-pass
4250
self._got_user_passwd('joe', 'secret-pass', conf, 'ftp', 'foo.net')
4252
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
4254
self._got_user_passwd(None, None, conf, 'ftp', 'bar.net')
4256
def test_credentials_for_host_port(self):
4257
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4258
# Identity on foo.net
4264
password=secret-pass
4267
self._got_user_passwd('joe', 'secret-pass',
4268
conf, 'ftp', 'foo.net', port=10021)
4270
self._got_user_passwd(None, None, conf, 'ftp', 'foo.net')
4272
def test_for_matching_host(self):
4273
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4274
# Identity on foo.net
4280
[sourceforge domain]
4287
self._got_user_passwd('georges', 'bendover',
4288
conf, 'bzr', 'foo.bzr.sf.net')
4290
self._got_user_passwd(None, None,
4291
conf, 'bzr', 'bbzr.sf.net')
4293
def test_for_matching_host_None(self):
4294
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4295
# Identity on foo.net
4305
self._got_user_passwd('joe', 'joepass',
4306
conf, 'bzr', 'quux.net')
4307
# no host but different scheme
4308
self._got_user_passwd('georges', 'bendover',
4309
conf, 'ftp', 'quux.net')
4311
def test_credentials_for_path(self):
4312
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4327
self._got_user_passwd(None, None,
4328
conf, 'http', host='bar.org', path='/dir3')
4330
self._got_user_passwd('georges', 'bendover',
4331
conf, 'http', host='bar.org', path='/dir2')
4333
self._got_user_passwd('jim', 'jimpass',
4334
conf, 'http', host='bar.org',path='/dir1/subdir')
4336
def test_credentials_for_user(self):
4337
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4345
self._got_user_passwd('jim', 'jimpass',
4346
conf, 'http', 'bar.org')
4348
self._got_user_passwd('jim', 'jimpass',
4349
conf, 'http', 'bar.org', user='jim')
4350
# Don't get a different user if one is specified
4351
self._got_user_passwd(None, None,
4352
conf, 'http', 'bar.org', user='georges')
4354
def test_credentials_for_user_without_password(self):
4355
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4361
# Get user but no password
4362
self._got_user_passwd('jim', None,
4363
conf, 'http', 'bar.org')
4365
def test_verify_certificates(self):
4366
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4372
verify_certificates=False
4379
credentials = conf.get_credentials('https', 'bar.org')
4380
self.assertEqual(False, credentials.get('verify_certificates'))
4381
credentials = conf.get_credentials('https', 'foo.net')
4382
self.assertEqual(True, credentials.get('verify_certificates'))
4385
class TestAuthenticationStorage(tests.TestCaseInTempDir):
4387
def test_set_credentials(self):
4388
conf = config.AuthenticationConfig()
4389
conf.set_credentials('name', 'host', 'user', 'scheme', 'password',
4390
99, path='/foo', verify_certificates=False, realm='realm')
4391
credentials = conf.get_credentials(host='host', scheme='scheme',
4392
port=99, path='/foo',
4394
CREDENTIALS = {'name': 'name', 'user': 'user', 'password': 'password',
4395
'verify_certificates': False, 'scheme': 'scheme',
4396
'host': 'host', 'port': 99, 'path': '/foo',
4398
self.assertEqual(CREDENTIALS, credentials)
4399
credentials_from_disk = config.AuthenticationConfig().get_credentials(
4400
host='host', scheme='scheme', port=99, path='/foo', realm='realm')
4401
self.assertEqual(CREDENTIALS, credentials_from_disk)
4403
def test_reset_credentials_different_name(self):
4404
conf = config.AuthenticationConfig()
4405
conf.set_credentials('name', 'host', 'user', 'scheme', 'password'),
4406
conf.set_credentials('name2', 'host', 'user2', 'scheme', 'password'),
4407
self.assertIs(None, conf._get_config().get('name'))
4408
credentials = conf.get_credentials(host='host', scheme='scheme')
4409
CREDENTIALS = {'name': 'name2', 'user': 'user2', 'password':
4410
'password', 'verify_certificates': True,
4411
'scheme': 'scheme', 'host': 'host', 'port': None,
4412
'path': None, 'realm': None}
4413
self.assertEqual(CREDENTIALS, credentials)
4416
class TestAuthenticationConfig(tests.TestCaseInTempDir):
4417
"""Test AuthenticationConfig behaviour"""
4419
def _check_default_password_prompt(self, expected_prompt_format, scheme,
4420
host=None, port=None, realm=None,
4424
user, password = 'jim', 'precious'
4425
expected_prompt = expected_prompt_format % {
4426
'scheme': scheme, 'host': host, 'port': port,
4427
'user': user, 'realm': realm}
4429
ui.ui_factory = tests.TestUIFactory(stdin=password + '\n')
4430
# We use an empty conf so that the user is always prompted
4431
conf = config.AuthenticationConfig()
4432
self.assertEqual(password,
4433
conf.get_password(scheme, host, user, port=port,
4434
realm=realm, path=path))
4435
self.assertEqual(expected_prompt, ui.ui_factory.stderr.getvalue())
4436
self.assertEqual('', ui.ui_factory.stdout.getvalue())
4438
def _check_default_username_prompt(self, expected_prompt_format, scheme,
4439
host=None, port=None, realm=None,
4444
expected_prompt = expected_prompt_format % {
4445
'scheme': scheme, 'host': host, 'port': port,
4447
ui.ui_factory = tests.TestUIFactory(stdin=username+ '\n')
4448
# We use an empty conf so that the user is always prompted
4449
conf = config.AuthenticationConfig()
4450
self.assertEqual(username, conf.get_user(scheme, host, port=port,
4451
realm=realm, path=path, ask=True))
4452
self.assertEqual(expected_prompt, ui.ui_factory.stderr.getvalue())
4453
self.assertEqual('', ui.ui_factory.stdout.getvalue())
4455
def test_username_defaults_prompts(self):
4456
# HTTP prompts can't be tested here, see test_http.py
4457
self._check_default_username_prompt(u'FTP %(host)s username: ', 'ftp')
4458
self._check_default_username_prompt(
4459
u'FTP %(host)s:%(port)d username: ', 'ftp', port=10020)
4460
self._check_default_username_prompt(
4461
u'SSH %(host)s:%(port)d username: ', 'ssh', port=12345)
4463
def test_username_default_no_prompt(self):
4464
conf = config.AuthenticationConfig()
4465
self.assertEqual(None,
4466
conf.get_user('ftp', 'example.com'))
4467
self.assertEqual("explicitdefault",
4468
conf.get_user('ftp', 'example.com', default="explicitdefault"))
4470
def test_password_default_prompts(self):
4471
# HTTP prompts can't be tested here, see test_http.py
4472
self._check_default_password_prompt(
4473
u'FTP %(user)s@%(host)s password: ', 'ftp')
4474
self._check_default_password_prompt(
4475
u'FTP %(user)s@%(host)s:%(port)d password: ', 'ftp', port=10020)
4476
self._check_default_password_prompt(
4477
u'SSH %(user)s@%(host)s:%(port)d password: ', 'ssh', port=12345)
4478
# SMTP port handling is a bit special (it's handled if embedded in the
4480
# FIXME: should we: forbid that, extend it to other schemes, leave
4481
# things as they are that's fine thank you ?
4482
self._check_default_password_prompt(
4483
u'SMTP %(user)s@%(host)s password: ', 'smtp')
4484
self._check_default_password_prompt(
4485
u'SMTP %(user)s@%(host)s password: ', 'smtp', host='bar.org:10025')
4486
self._check_default_password_prompt(
4487
u'SMTP %(user)s@%(host)s:%(port)d password: ', 'smtp', port=10025)
4489
def test_ssh_password_emits_warning(self):
4490
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4497
entered_password = 'typed-by-hand'
4498
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n')
4500
# Since the password defined in the authentication config is ignored,
4501
# the user is prompted
4502
self.assertEqual(entered_password,
4503
conf.get_password('ssh', 'bar.org', user='jim'))
4504
self.assertContainsRe(
4506
'password ignored in section \[ssh with password\]')
4508
def test_ssh_without_password_doesnt_emit_warning(self):
4509
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4515
entered_password = 'typed-by-hand'
4516
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n')
4518
# Since the password defined in the authentication config is ignored,
4519
# the user is prompted
4520
self.assertEqual(entered_password,
4521
conf.get_password('ssh', 'bar.org', user='jim'))
4522
# No warning shoud be emitted since there is no password. We are only
4524
self.assertNotContainsRe(
4526
'password ignored in section \[ssh with password\]')
4528
def test_uses_fallback_stores(self):
4529
self.overrideAttr(config, 'credential_store_registry',
4530
config.CredentialStoreRegistry())
4531
store = StubCredentialStore()
4532
store.add_credentials("http", "example.com", "joe", "secret")
4533
config.credential_store_registry.register("stub", store, fallback=True)
4534
conf = config.AuthenticationConfig(_file=BytesIO())
4535
creds = conf.get_credentials("http", "example.com")
4536
self.assertEqual("joe", creds["user"])
4537
self.assertEqual("secret", creds["password"])
4540
class StubCredentialStore(config.CredentialStore):
4546
def add_credentials(self, scheme, host, user, password=None):
4547
self._username[(scheme, host)] = user
4548
self._password[(scheme, host)] = password
4550
def get_credentials(self, scheme, host, port=None, user=None,
4551
path=None, realm=None):
4552
key = (scheme, host)
4553
if not key in self._username:
4555
return { "scheme": scheme, "host": host, "port": port,
4556
"user": self._username[key], "password": self._password[key]}
4559
class CountingCredentialStore(config.CredentialStore):
4564
def get_credentials(self, scheme, host, port=None, user=None,
4565
path=None, realm=None):
4570
class TestCredentialStoreRegistry(tests.TestCase):
4572
def _get_cs_registry(self):
4573
return config.credential_store_registry
4575
def test_default_credential_store(self):
4576
r = self._get_cs_registry()
4577
default = r.get_credential_store(None)
4578
self.assertIsInstance(default, config.PlainTextCredentialStore)
4580
def test_unknown_credential_store(self):
4581
r = self._get_cs_registry()
4582
# It's hard to imagine someone creating a credential store named
4583
# 'unknown' so we use that as an never registered key.
4584
self.assertRaises(KeyError, r.get_credential_store, 'unknown')
4586
def test_fallback_none_registered(self):
4587
r = config.CredentialStoreRegistry()
4588
self.assertEqual(None,
4589
r.get_fallback_credentials("http", "example.com"))
4591
def test_register(self):
4592
r = config.CredentialStoreRegistry()
4593
r.register("stub", StubCredentialStore(), fallback=False)
4594
r.register("another", StubCredentialStore(), fallback=True)
4595
self.assertEqual(["another", "stub"], r.keys())
4597
def test_register_lazy(self):
4598
r = config.CredentialStoreRegistry()
4599
r.register_lazy("stub", "breezy.tests.test_config",
4600
"StubCredentialStore", fallback=False)
4601
self.assertEqual(["stub"], r.keys())
4602
self.assertIsInstance(r.get_credential_store("stub"),
4603
StubCredentialStore)
4605
def test_is_fallback(self):
4606
r = config.CredentialStoreRegistry()
4607
r.register("stub1", None, fallback=False)
4608
r.register("stub2", None, fallback=True)
4609
self.assertEqual(False, r.is_fallback("stub1"))
4610
self.assertEqual(True, r.is_fallback("stub2"))
4612
def test_no_fallback(self):
4613
r = config.CredentialStoreRegistry()
4614
store = CountingCredentialStore()
4615
r.register("count", store, fallback=False)
4616
self.assertEqual(None,
4617
r.get_fallback_credentials("http", "example.com"))
4618
self.assertEqual(0, store._calls)
4620
def test_fallback_credentials(self):
4621
r = config.CredentialStoreRegistry()
4622
store = StubCredentialStore()
4623
store.add_credentials("http", "example.com",
4624
"somebody", "geheim")
4625
r.register("stub", store, fallback=True)
4626
creds = r.get_fallback_credentials("http", "example.com")
4627
self.assertEqual("somebody", creds["user"])
4628
self.assertEqual("geheim", creds["password"])
4630
def test_fallback_first_wins(self):
4631
r = config.CredentialStoreRegistry()
4632
stub1 = StubCredentialStore()
4633
stub1.add_credentials("http", "example.com",
4634
"somebody", "stub1")
4635
r.register("stub1", stub1, fallback=True)
4636
stub2 = StubCredentialStore()
4637
stub2.add_credentials("http", "example.com",
4638
"somebody", "stub2")
4639
r.register("stub2", stub1, fallback=True)
4640
creds = r.get_fallback_credentials("http", "example.com")
4641
self.assertEqual("somebody", creds["user"])
4642
self.assertEqual("stub1", creds["password"])
4645
class TestPlainTextCredentialStore(tests.TestCase):
4647
def test_decode_password(self):
4648
r = config.credential_store_registry
4649
plain_text = r.get_credential_store()
4650
decoded = plain_text.decode_password(dict(password='secret'))
4651
self.assertEqual('secret', decoded)
4654
class TestBase64CredentialStore(tests.TestCase):
4656
def test_decode_password(self):
4657
r = config.credential_store_registry
4658
plain_text = r.get_credential_store('base64')
4659
decoded = plain_text.decode_password(dict(password='c2VjcmV0'))
4660
self.assertEqual('secret', decoded)
4663
# FIXME: Once we have a way to declare authentication to all test servers, we
4664
# can implement generic tests.
4665
# test_user_password_in_url
4666
# test_user_in_url_password_from_config
4667
# test_user_in_url_password_prompted
4668
# test_user_in_config
4669
# test_user_getpass.getuser
4670
# test_user_prompted ?
4671
class TestAuthenticationRing(tests.TestCaseWithTransport):
4675
class TestAutoUserId(tests.TestCase):
4676
"""Test inferring an automatic user name."""
4678
def test_auto_user_id(self):
4679
"""Automatic inference of user name.
4681
This is a bit hard to test in an isolated way, because it depends on
4682
system functions that go direct to /etc or perhaps somewhere else.
4683
But it's reasonable to say that on Unix, with an /etc/mailname, we ought
4684
to be able to choose a user name with no configuration.
4686
if sys.platform == 'win32':
4687
raise tests.TestSkipped(
4688
"User name inference not implemented on win32")
4689
realname, address = config._auto_user_id()
4690
if os.path.exists('/etc/mailname'):
4691
self.assertIsNot(None, realname)
4692
self.assertIsNot(None, address)
4694
self.assertEqual((None, None), (realname, address))
4697
class TestDefaultMailDomain(tests.TestCaseInTempDir):
4698
"""Test retrieving default domain from mailname file"""
4700
def test_default_mail_domain_simple(self):
4701
f = file('simple', 'w')
4703
f.write("domainname.com\n")
4706
r = config._get_default_mail_domain('simple')
4707
self.assertEqual('domainname.com', r)
4709
def test_default_mail_domain_no_eol(self):
4710
f = file('no_eol', 'w')
4712
f.write("domainname.com")
4715
r = config._get_default_mail_domain('no_eol')
4716
self.assertEqual('domainname.com', r)
4718
def test_default_mail_domain_multiple_lines(self):
4719
f = file('multiple_lines', 'w')
4721
f.write("domainname.com\nsome other text\n")
4724
r = config._get_default_mail_domain('multiple_lines')
4725
self.assertEqual('domainname.com', r)
4728
class EmailOptionTests(tests.TestCase):
4730
def test_default_email_uses_BRZ_EMAIL(self):
4731
conf = config.MemoryStack('email=jelmer@debian.org')
4732
# BRZ_EMAIL takes precedence over EMAIL
4733
self.overrideEnv('BRZ_EMAIL', 'jelmer@samba.org')
4734
self.overrideEnv('EMAIL', 'jelmer@apache.org')
4735
self.assertEqual('jelmer@samba.org', conf.get('email'))
4737
def test_default_email_uses_EMAIL(self):
4738
conf = config.MemoryStack('')
4739
self.overrideEnv('BRZ_EMAIL', None)
4740
self.overrideEnv('EMAIL', 'jelmer@apache.org')
4741
self.assertEqual('jelmer@apache.org', conf.get('email'))
4743
def test_BRZ_EMAIL_overrides(self):
4744
conf = config.MemoryStack('email=jelmer@debian.org')
4745
self.overrideEnv('BRZ_EMAIL', 'jelmer@apache.org')
4746
self.assertEqual('jelmer@apache.org', conf.get('email'))
4747
self.overrideEnv('BRZ_EMAIL', None)
4748
self.overrideEnv('EMAIL', 'jelmer@samba.org')
4749
self.assertEqual('jelmer@debian.org', conf.get('email'))
4752
class MailClientOptionTests(tests.TestCase):
4754
def test_default(self):
4755
conf = config.MemoryStack('')
4756
client = conf.get('mail_client')
4757
self.assertIs(client, mail_client.DefaultMail)
4759
def test_evolution(self):
4760
conf = config.MemoryStack('mail_client=evolution')
4761
client = conf.get('mail_client')
4762
self.assertIs(client, mail_client.Evolution)
4764
def test_kmail(self):
4765
conf = config.MemoryStack('mail_client=kmail')
4766
client = conf.get('mail_client')
4767
self.assertIs(client, mail_client.KMail)
4769
def test_mutt(self):
4770
conf = config.MemoryStack('mail_client=mutt')
4771
client = conf.get('mail_client')
4772
self.assertIs(client, mail_client.Mutt)
4774
def test_thunderbird(self):
4775
conf = config.MemoryStack('mail_client=thunderbird')
4776
client = conf.get('mail_client')
4777
self.assertIs(client, mail_client.Thunderbird)
4779
def test_explicit_default(self):
4780
conf = config.MemoryStack('mail_client=default')
4781
client = conf.get('mail_client')
4782
self.assertIs(client, mail_client.DefaultMail)
4784
def test_editor(self):
4785
conf = config.MemoryStack('mail_client=editor')
4786
client = conf.get('mail_client')
4787
self.assertIs(client, mail_client.Editor)
4789
def test_mapi(self):
4790
conf = config.MemoryStack('mail_client=mapi')
4791
client = conf.get('mail_client')
4792
self.assertIs(client, mail_client.MAPIClient)
4794
def test_xdg_email(self):
4795
conf = config.MemoryStack('mail_client=xdg-email')
4796
client = conf.get('mail_client')
4797
self.assertIs(client, mail_client.XDGEmail)
4799
def test_unknown(self):
4800
conf = config.MemoryStack('mail_client=firebird')
4801
self.assertRaises(config.ConfigOptionValueError, conf.get,
added
removed
breezy/tests/test_config.py
