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
39
registry as _mod_registry,
46
from ..sixish import (
50
from ..transport import remote as transport_remote
58
def lockable_config_scenarios():
61
{'config_class': config.GlobalConfig,
63
'config_section': 'DEFAULT'}),
65
{'config_class': config.LocationConfig,
67
'config_section': '.'}), ]
70
load_tests = scenarios.load_tests_apply_scenarios
72
# Register helpers to build stores
73
config.test_store_builder_registry.register(
74
'configobj', lambda test: config.TransportIniFileStore(
75
test.get_transport(), 'configobj.conf'))
76
config.test_store_builder_registry.register(
77
'breezy', lambda test: config.GlobalStore())
78
config.test_store_builder_registry.register(
79
'location', lambda test: config.LocationStore())
82
def build_backing_branch(test, relpath,
83
transport_class=None, server_class=None):
84
"""Test helper to create a backing branch only once.
86
Some tests needs multiple stores/stacks to check concurrent update
87
behaviours. As such, they need to build different branch *objects* even if
88
they share the branch on disk.
90
:param relpath: The relative path to the branch. (Note that the helper
91
should always specify the same relpath).
93
:param transport_class: The Transport class the test needs to use.
95
:param server_class: The server associated with the ``transport_class``
98
Either both or neither of ``transport_class`` and ``server_class`` should
101
if transport_class is not None and server_class is not None:
102
test.transport_class = transport_class
103
test.transport_server = server_class
104
elif not (transport_class is None and server_class is None):
105
raise AssertionError('Specify both ``transport_class`` and '
106
'``server_class`` or neither of them')
107
if getattr(test, 'backing_branch', None) is None:
108
# First call, let's build the branch on disk
109
test.backing_branch = test.make_branch(relpath)
112
def build_branch_store(test):
113
build_backing_branch(test, 'branch')
114
b = branch.Branch.open('branch')
115
return config.BranchStore(b)
118
config.test_store_builder_registry.register('branch', build_branch_store)
121
def build_control_store(test):
122
build_backing_branch(test, 'branch')
123
b = controldir.ControlDir.open('branch')
124
return config.ControlStore(b)
127
config.test_store_builder_registry.register('control', build_control_store)
130
def build_remote_branch_store(test):
131
# There is only one permutation (but we won't be able to handle more with
132
# this design anyway)
134
server_class) = transport_remote.get_test_permutations()[0]
135
build_backing_branch(test, 'branch', transport_class, server_class)
136
b = branch.Branch.open(test.get_url('branch'))
137
return config.BranchStore(b)
140
config.test_store_builder_registry.register('remote_branch',
141
build_remote_branch_store)
144
config.test_stack_builder_registry.register(
145
'breezy', lambda test: config.GlobalStack())
146
config.test_stack_builder_registry.register(
147
'location', lambda test: config.LocationStack('.'))
150
def build_branch_stack(test):
151
build_backing_branch(test, 'branch')
152
b = branch.Branch.open('branch')
153
return config.BranchStack(b)
156
config.test_stack_builder_registry.register('branch', build_branch_stack)
159
def build_branch_only_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
build_backing_branch(test, 'branch', transport_class, server_class)
165
b = branch.Branch.open(test.get_url('branch'))
166
return config.BranchOnlyStack(b)
169
config.test_stack_builder_registry.register('branch_only',
170
build_branch_only_stack)
173
def build_remote_control_stack(test):
174
# There is only one permutation (but we won't be able to handle more with
175
# this design anyway)
177
server_class) = transport_remote.get_test_permutations()[0]
178
# We need only a bzrdir for this, not a full branch, but it's not worth
179
# creating a dedicated helper to create only the bzrdir
180
build_backing_branch(test, 'branch', transport_class, server_class)
181
b = branch.Branch.open(test.get_url('branch'))
182
return config.RemoteControlStack(b.controldir)
185
config.test_stack_builder_registry.register('remote_control',
186
build_remote_control_stack)
189
sample_long_alias = "log -r-15..-1 --line"
190
sample_config_text = u"""
192
email=Erik B\u00e5gfors <erik@bagfors.nu>
194
change_editor=vimdiff -of {new_path} {old_path}
195
gpg_signing_key=DD4D5088
197
validate_signatures_in_log=true
199
user_global_option=something
200
bzr.mergetool.sometool=sometool {base} {this} {other} -o {result}
201
bzr.mergetool.funkytool=funkytool "arg with spaces" {this_temp}
202
bzr.mergetool.newtool='"newtool with spaces" {this_temp}'
203
bzr.default_mergetool=sometool
206
ll=""".encode('utf-8') + sample_long_alias.encode('utf-8') + b"\n"
209
sample_always_signatures = b"""
211
check_signatures=ignore
212
create_signatures=always
215
sample_ignore_signatures = b"""
217
check_signatures=require
218
create_signatures=never
221
sample_maybe_signatures = b"""
223
check_signatures=ignore
224
create_signatures=when-required
227
sample_branches_text = b"""
228
[http://www.example.com]
230
email=Robert Collins <robertc@example.org>
231
normal_option = normal
232
appendpath_option = append
233
appendpath_option:policy = appendpath
234
norecurse_option = norecurse
235
norecurse_option:policy = norecurse
236
[http://www.example.com/ignoreparent]
237
# different project: ignore parent dir config
239
[http://www.example.com/norecurse]
240
# configuration items that only apply to this dir
242
normal_option = norecurse
243
[http://www.example.com/dir]
244
appendpath_option = normal
246
check_signatures=require
247
# test trailing / matching with no children
249
check_signatures=check-available
250
gpg_signing_key=default
251
user_local_option=local
252
# test trailing / matching
254
#subdirs will match but not the parent
256
check_signatures=ignore
257
post_commit=breezy.tests.test_config.post_commit
258
#testing explicit beats globs
262
def create_configs(test):
263
"""Create configuration files for a given test.
265
This requires creating a tree (and populate the ``test.tree`` attribute)
266
and its associated branch and will populate the following attributes:
268
- branch_config: A BranchConfig for the associated branch.
270
- locations_config : A LocationConfig for the associated branch
272
- breezy_config: A GlobalConfig.
274
The tree and branch are created in a 'tree' subdirectory so the tests can
275
still use the test directory to stay outside of the branch.
277
tree = test.make_branch_and_tree('tree')
279
test.branch_config = config.BranchConfig(tree.branch)
280
test.locations_config = config.LocationConfig(tree.basedir)
281
test.breezy_config = config.GlobalConfig()
284
def create_configs_with_file_option(test):
285
"""Create configuration files with a ``file`` option set in each.
287
This builds on ``create_configs`` and add one ``file`` option in each
288
configuration with a value which allows identifying the configuration file.
291
test.breezy_config.set_user_option('file', 'breezy')
292
test.locations_config.set_user_option('file', 'locations')
293
test.branch_config.set_user_option('file', 'branch')
296
class TestOptionsMixin:
298
def assertOptions(self, expected, conf):
299
# We don't care about the parser (as it will make tests hard to write
300
# and error-prone anyway)
301
self.assertThat([opt[:4] for opt in conf._get_options()],
302
matchers.Equals(expected))
305
class InstrumentedConfigObj(object):
306
"""A config obj look-enough-alike to record calls made to it."""
308
def __contains__(self, thing):
309
self._calls.append(('__contains__', thing))
312
def __getitem__(self, key):
313
self._calls.append(('__getitem__', key))
316
def __init__(self, input, encoding=None):
317
self._calls = [('__init__', input, encoding)]
319
def __setitem__(self, key, value):
320
self._calls.append(('__setitem__', key, value))
322
def __delitem__(self, key):
323
self._calls.append(('__delitem__', key))
326
self._calls.append(('keys',))
330
self._calls.append(('reload',))
332
def write(self, arg):
333
self._calls.append(('write',))
335
def as_bool(self, value):
336
self._calls.append(('as_bool', value))
339
def get_value(self, section, name):
340
self._calls.append(('get_value', section, name))
344
class FakeBranch(object):
346
def __init__(self, base=None):
348
self.base = "http://example.com/branches/demo"
351
self._transport = self.control_files = \
352
FakeControlFilesAndTransport()
354
def _get_config(self):
355
return config.TransportConfig(self._transport, 'branch.conf')
357
def lock_write(self):
358
return lock.LogicalLockResult(self.unlock)
364
class FakeControlFilesAndTransport(object):
368
self._transport = self
370
def get(self, filename):
373
return BytesIO(self.files[filename])
375
raise errors.NoSuchFile(filename)
377
def get_bytes(self, filename):
380
return self.files[filename]
382
raise errors.NoSuchFile(filename)
384
def put(self, filename, fileobj):
385
self.files[filename] = fileobj.read()
387
def put_file(self, filename, fileobj):
388
return self.put(filename, fileobj)
391
class InstrumentedConfig(config.Config):
392
"""An instrumented config that supplies stubs for template methods."""
395
super(InstrumentedConfig, self).__init__()
397
self._signatures = config.CHECK_NEVER
398
self._change_editor = 'vimdiff -fo {new_path} {old_path}'
400
def _get_user_id(self):
401
self._calls.append('_get_user_id')
402
return "Robert Collins <robert.collins@example.org>"
404
def _get_signature_checking(self):
405
self._calls.append('_get_signature_checking')
406
return self._signatures
408
def _get_change_editor(self):
409
self._calls.append('_get_change_editor')
410
return self._change_editor
413
bool_config = b"""[DEFAULT]
422
class TestConfigObj(tests.TestCase):
424
def test_get_bool(self):
425
co = config.ConfigObj(BytesIO(bool_config))
426
self.assertIs(co.get_bool('DEFAULT', 'active'), True)
427
self.assertIs(co.get_bool('DEFAULT', 'inactive'), False)
428
self.assertIs(co.get_bool('UPPERCASE', 'active'), True)
429
self.assertIs(co.get_bool('UPPERCASE', 'nonactive'), False)
431
def test_hash_sign_in_value(self):
433
Before 4.5.0, ConfigObj did not quote # signs in values, so they'd be
434
treated as comments when read in again. (#86838)
436
co = config.ConfigObj()
437
co['test'] = 'foo#bar'
439
co.write(outfile=outfile)
440
lines = outfile.getvalue().splitlines()
441
self.assertEqual(lines, [b'test = "foo#bar"'])
442
co2 = config.ConfigObj(lines)
443
self.assertEqual(co2['test'], 'foo#bar')
445
def test_triple_quotes(self):
446
# Bug #710410: if the value string has triple quotes
447
# then ConfigObj versions up to 4.7.2 will quote them wrong
448
# and won't able to read them back
449
triple_quotes_value = '''spam
450
""" that's my spam """
452
co = config.ConfigObj()
453
co['test'] = triple_quotes_value
454
# While writing this test another bug in ConfigObj has been found:
455
# method co.write() without arguments produces list of lines
456
# one option per line, and multiline values are not split
457
# across multiple lines,
458
# and that breaks the parsing these lines back by ConfigObj.
459
# This issue only affects test, but it's better to avoid
460
# `co.write()` construct at all.
461
# [bialix 20110222] bug report sent to ConfigObj's author
463
co.write(outfile=outfile)
464
output = outfile.getvalue()
465
# now we're trying to read it back
466
co2 = config.ConfigObj(BytesIO(output))
467
self.assertEqual(triple_quotes_value, co2['test'])
470
erroneous_config = b"""[section] # line 1
473
whocares=notme # line 4
477
class TestConfigObjErrors(tests.TestCase):
479
def test_duplicate_section_name_error_line(self):
481
co = configobj.ConfigObj(BytesIO(erroneous_config),
483
except config.configobj.DuplicateError as e:
484
self.assertEqual(3, e.line_number)
486
self.fail('Error in config file not detected')
489
class TestConfig(tests.TestCase):
491
def test_constructs(self):
494
def test_user_email(self):
495
my_config = InstrumentedConfig()
496
self.assertEqual('robert.collins@example.org', my_config.user_email())
497
self.assertEqual(['_get_user_id'], my_config._calls)
499
def test_username(self):
500
my_config = InstrumentedConfig()
501
self.assertEqual('Robert Collins <robert.collins@example.org>',
502
my_config.username())
503
self.assertEqual(['_get_user_id'], my_config._calls)
505
def test_get_user_option_default(self):
506
my_config = config.Config()
507
self.assertEqual(None, my_config.get_user_option('no_option'))
509
def test_validate_signatures_in_log_default(self):
510
my_config = config.Config()
511
self.assertEqual(False, my_config.validate_signatures_in_log())
513
def test_get_change_editor(self):
514
my_config = InstrumentedConfig()
515
change_editor = my_config.get_change_editor('old_tree', 'new_tree')
516
self.assertEqual(['_get_change_editor'], my_config._calls)
517
self.assertIs(diff.DiffFromTool, change_editor.__class__)
518
self.assertEqual(['vimdiff', '-fo', '{new_path}', '{old_path}'],
519
change_editor.command_template)
521
def test_get_change_editor_implicit_args(self):
522
# If there are no substitution variables, then assume the
523
# old and new path are the last arguments.
524
my_config = InstrumentedConfig()
525
my_config._change_editor = 'vimdiff -o'
526
change_editor = my_config.get_change_editor('old_tree', 'new_tree')
527
self.assertEqual(['_get_change_editor'], my_config._calls)
528
self.assertIs(diff.DiffFromTool, change_editor.__class__)
529
self.assertEqual(['vimdiff', '-o', '{old_path}', '{new_path}'],
530
change_editor.command_template)
532
def test_get_change_editor_old_style(self):
533
# Test the old style format for the change_editor setting.
534
my_config = InstrumentedConfig()
535
my_config._change_editor = 'vimdiff -o @old_path @new_path'
536
change_editor = my_config.get_change_editor('old_tree', 'new_tree')
537
self.assertEqual(['_get_change_editor'], my_config._calls)
538
self.assertIs(diff.DiffFromTool, change_editor.__class__)
539
self.assertEqual(['vimdiff', '-o', '{old_path}', '{new_path}'],
540
change_editor.command_template)
543
class TestIniConfig(tests.TestCaseInTempDir):
545
def make_config_parser(self, s):
546
conf = config.IniBasedConfig.from_string(s)
547
return conf, conf._get_parser()
550
class TestIniConfigBuilding(TestIniConfig):
552
def test_contructs(self):
553
config.IniBasedConfig()
555
def test_from_fp(self):
556
my_config = config.IniBasedConfig.from_string(sample_config_text)
557
self.assertIsInstance(my_config._get_parser(), configobj.ConfigObj)
559
def test_cached(self):
560
my_config = config.IniBasedConfig.from_string(sample_config_text)
561
parser = my_config._get_parser()
562
self.assertTrue(my_config._get_parser() is parser)
564
def _dummy_chown(self, path, uid, gid):
565
self.path, self.uid, self.gid = path, uid, gid
567
def test_ini_config_ownership(self):
568
"""Ensure that chown is happening during _write_config_file"""
569
self.requireFeature(features.chown_feature)
570
self.overrideAttr(os, 'chown', self._dummy_chown)
571
self.path = self.uid = self.gid = None
572
conf = config.IniBasedConfig(file_name='./foo.conf')
573
conf._write_config_file()
574
self.assertEqual(self.path, './foo.conf')
575
self.assertTrue(isinstance(self.uid, int))
576
self.assertTrue(isinstance(self.gid, int))
579
class TestIniConfigSaving(tests.TestCaseInTempDir):
581
def test_cant_save_without_a_file_name(self):
582
conf = config.IniBasedConfig()
583
self.assertRaises(AssertionError, conf._write_config_file)
585
def test_saved_with_content(self):
586
content = b'foo = bar\n'
587
config.IniBasedConfig.from_string(content, file_name='./test.conf',
589
self.assertFileEqual(content, 'test.conf')
592
class TestIniConfigOptionExpansion(tests.TestCase):
593
"""Test option expansion from the IniConfig level.
595
What we really want here is to test the Config level, but the class being
596
abstract as far as storing values is concerned, this can't be done
599
# FIXME: This should be rewritten when all configs share a storage
600
# implementation -- vila 2011-02-18
602
def get_config(self, string=None):
605
c = config.IniBasedConfig.from_string(string)
608
def assertExpansion(self, expected, conf, string, env=None):
609
self.assertEqual(expected, conf.expand_options(string, env))
611
def test_no_expansion(self):
612
c = self.get_config('')
613
self.assertExpansion('foo', c, 'foo')
615
def test_env_adding_options(self):
616
c = self.get_config('')
617
self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})
619
def test_env_overriding_options(self):
620
c = self.get_config('foo=baz')
621
self.assertExpansion('bar', c, '{foo}', {'foo': 'bar'})
623
def test_simple_ref(self):
624
c = self.get_config('foo=xxx')
625
self.assertExpansion('xxx', c, '{foo}')
627
def test_unknown_ref(self):
628
c = self.get_config('')
629
self.assertRaises(config.ExpandingUnknownOption,
630
c.expand_options, '{foo}')
632
def test_indirect_ref(self):
633
c = self.get_config('''
637
self.assertExpansion('xxx', c, '{bar}')
639
def test_embedded_ref(self):
640
c = self.get_config('''
644
self.assertExpansion('xxx', c, '{{bar}}')
646
def test_simple_loop(self):
647
c = self.get_config('foo={foo}')
648
self.assertRaises(config.OptionExpansionLoop, c.expand_options,
651
def test_indirect_loop(self):
652
c = self.get_config('''
656
e = self.assertRaises(config.OptionExpansionLoop,
657
c.expand_options, '{foo}')
658
self.assertEqual('foo->bar->baz', e.refs)
659
self.assertEqual('{foo}', e.string)
662
conf = self.get_config('''
666
list={foo},{bar},{baz}
668
self.assertEqual(['start', 'middle', 'end'],
669
conf.get_user_option('list', expand=True))
671
def test_cascading_list(self):
672
conf = self.get_config('''
678
self.assertEqual(['start', 'middle', 'end'],
679
conf.get_user_option('list', expand=True))
681
def test_pathological_hidden_list(self):
682
conf = self.get_config('''
688
hidden={start}{middle}{end}
690
# Nope, it's either a string or a list, and the list wins as soon as a
691
# ',' appears, so the string concatenation never occur.
692
self.assertEqual(['{foo', '}', '{', 'bar}'],
693
conf.get_user_option('hidden', expand=True))
696
class TestLocationConfigOptionExpansion(tests.TestCaseInTempDir):
698
def get_config(self, location, string=None):
701
# Since we don't save the config we won't strictly require to inherit
702
# from TestCaseInTempDir, but an error occurs so quickly...
703
c = config.LocationConfig.from_string(string, location)
706
def test_dont_cross_unrelated_section(self):
707
c = self.get_config('/another/branch/path', '''
712
[/another/branch/path]
715
self.assertRaises(config.ExpandingUnknownOption,
716
c.get_user_option, 'bar', expand=True)
718
def test_cross_related_sections(self):
719
c = self.get_config('/project/branch/path', '''
723
[/project/branch/path]
726
self.assertEqual('quux', c.get_user_option('bar', expand=True))
729
class TestIniBaseConfigOnDisk(tests.TestCaseInTempDir):
731
def test_cannot_reload_without_name(self):
732
conf = config.IniBasedConfig.from_string(sample_config_text)
733
self.assertRaises(AssertionError, conf.reload)
735
def test_reload_see_new_value(self):
736
c1 = config.IniBasedConfig.from_string('editor=vim\n',
737
file_name='./test/conf')
738
c1._write_config_file()
739
c2 = config.IniBasedConfig.from_string('editor=emacs\n',
740
file_name='./test/conf')
741
c2._write_config_file()
742
self.assertEqual('vim', c1.get_user_option('editor'))
743
self.assertEqual('emacs', c2.get_user_option('editor'))
744
# Make sure we get the Right value
746
self.assertEqual('emacs', c1.get_user_option('editor'))
749
class TestLockableConfig(tests.TestCaseInTempDir):
751
scenarios = lockable_config_scenarios()
756
config_section = None
759
super(TestLockableConfig, self).setUp()
760
self._content = '[%s]\none=1\ntwo=2\n' % (self.config_section,)
761
self.config = self.create_config(self._content)
763
def get_existing_config(self):
764
return self.config_class(*self.config_args)
766
def create_config(self, content):
767
kwargs = dict(save=True)
768
c = self.config_class.from_string(content, *self.config_args, **kwargs)
771
def test_simple_read_access(self):
772
self.assertEqual('1', self.config.get_user_option('one'))
774
def test_simple_write_access(self):
775
self.config.set_user_option('one', 'one')
776
self.assertEqual('one', self.config.get_user_option('one'))
778
def test_listen_to_the_last_speaker(self):
780
c2 = self.get_existing_config()
781
c1.set_user_option('one', 'ONE')
782
c2.set_user_option('two', 'TWO')
783
self.assertEqual('ONE', c1.get_user_option('one'))
784
self.assertEqual('TWO', c2.get_user_option('two'))
785
# The second update respect the first one
786
self.assertEqual('ONE', c2.get_user_option('one'))
788
def test_last_speaker_wins(self):
789
# If the same config is not shared, the same variable modified twice
790
# can only see a single result.
792
c2 = self.get_existing_config()
793
c1.set_user_option('one', 'c1')
794
c2.set_user_option('one', 'c2')
795
self.assertEqual('c2', c2._get_user_option('one'))
796
# The first modification is still available until another refresh
798
self.assertEqual('c1', c1._get_user_option('one'))
799
c1.set_user_option('two', 'done')
800
self.assertEqual('c2', c1._get_user_option('one'))
802
def test_writes_are_serialized(self):
804
c2 = self.get_existing_config()
806
# We spawn a thread that will pause *during* the write
807
before_writing = threading.Event()
808
after_writing = threading.Event()
809
writing_done = threading.Event()
810
c1_orig = c1._write_config_file
812
def c1_write_config_file():
815
# The lock is held. We wait for the main thread to decide when to
818
c1._write_config_file = c1_write_config_file
821
c1.set_user_option('one', 'c1')
823
t1 = threading.Thread(target=c1_set_option)
824
# Collect the thread after the test
825
self.addCleanup(t1.join)
826
# Be ready to unblock the thread if the test goes wrong
827
self.addCleanup(after_writing.set)
829
before_writing.wait()
830
self.assertTrue(c1._lock.is_held)
831
self.assertRaises(errors.LockContention,
832
c2.set_user_option, 'one', 'c2')
833
self.assertEqual('c1', c1.get_user_option('one'))
834
# Let the lock be released
837
c2.set_user_option('one', 'c2')
838
self.assertEqual('c2', c2.get_user_option('one'))
840
def test_read_while_writing(self):
842
# We spawn a thread that will pause *during* the write
843
ready_to_write = threading.Event()
844
do_writing = threading.Event()
845
writing_done = threading.Event()
846
c1_orig = c1._write_config_file
848
def c1_write_config_file():
850
# The lock is held. We wait for the main thread to decide when to
855
c1._write_config_file = c1_write_config_file
858
c1.set_user_option('one', 'c1')
859
t1 = threading.Thread(target=c1_set_option)
860
# Collect the thread after the test
861
self.addCleanup(t1.join)
862
# Be ready to unblock the thread if the test goes wrong
863
self.addCleanup(do_writing.set)
865
# Ensure the thread is ready to write
866
ready_to_write.wait()
867
self.assertTrue(c1._lock.is_held)
868
self.assertEqual('c1', c1.get_user_option('one'))
869
# If we read during the write, we get the old value
870
c2 = self.get_existing_config()
871
self.assertEqual('1', c2.get_user_option('one'))
872
# Let the writing occur and ensure it occurred
875
# Now we get the updated value
876
c3 = self.get_existing_config()
877
self.assertEqual('c1', c3.get_user_option('one'))
880
class TestGetUserOptionAs(TestIniConfig):
882
def test_get_user_option_as_bool(self):
883
conf, parser = self.make_config_parser("""
886
an_invalid_bool = maybe
887
a_list = hmm, who knows ? # This is interpreted as a list !
889
get_bool = conf.get_user_option_as_bool
890
self.assertEqual(True, get_bool('a_true_bool'))
891
self.assertEqual(False, get_bool('a_false_bool'))
895
warnings.append(args[0] % args[1:])
896
self.overrideAttr(trace, 'warning', warning)
897
msg = 'Value "%s" is not a boolean for "%s"'
898
self.assertIs(None, get_bool('an_invalid_bool'))
899
self.assertEqual(msg % ('maybe', 'an_invalid_bool'), warnings[0])
901
self.assertIs(None, get_bool('not_defined_in_this_config'))
902
self.assertEqual([], warnings)
904
def test_get_user_option_as_list(self):
905
conf, parser = self.make_config_parser("""
910
get_list = conf.get_user_option_as_list
911
self.assertEqual(['a', 'b', 'c'], get_list('a_list'))
912
self.assertEqual(['1'], get_list('length_1'))
913
self.assertEqual('x', conf.get_user_option('one_item'))
914
# automatically cast to list
915
self.assertEqual(['x'], get_list('one_item'))
918
class TestSupressWarning(TestIniConfig):
920
def make_warnings_config(self, s):
921
conf, parser = self.make_config_parser(s)
922
return conf.suppress_warning
924
def test_suppress_warning_unknown(self):
925
suppress_warning = self.make_warnings_config('')
926
self.assertEqual(False, suppress_warning('unknown_warning'))
928
def test_suppress_warning_known(self):
929
suppress_warning = self.make_warnings_config('suppress_warnings=a,b')
930
self.assertEqual(False, suppress_warning('c'))
931
self.assertEqual(True, suppress_warning('a'))
932
self.assertEqual(True, suppress_warning('b'))
935
class TestGetConfig(tests.TestCaseInTempDir):
937
def test_constructs(self):
938
config.GlobalConfig()
940
def test_calls_read_filenames(self):
941
# replace the class that is constructed, to check its parameters
942
oldparserclass = config.ConfigObj
943
config.ConfigObj = InstrumentedConfigObj
944
my_config = config.GlobalConfig()
946
parser = my_config._get_parser()
948
config.ConfigObj = oldparserclass
949
self.assertIsInstance(parser, InstrumentedConfigObj)
950
self.assertEqual(parser._calls, [('__init__', bedding.config_path(),
954
class TestBranchConfig(tests.TestCaseWithTransport):
956
def test_constructs_valid(self):
957
branch = FakeBranch()
958
my_config = config.BranchConfig(branch)
959
self.assertIsNot(None, my_config)
961
def test_constructs_error(self):
962
self.assertRaises(TypeError, config.BranchConfig)
964
def test_get_location_config(self):
965
branch = FakeBranch()
966
my_config = config.BranchConfig(branch)
967
location_config = my_config._get_location_config()
968
self.assertEqual(branch.base, location_config.location)
969
self.assertIs(location_config, my_config._get_location_config())
971
def test_get_config(self):
972
"""The Branch.get_config method works properly"""
973
b = controldir.ControlDir.create_standalone_workingtree('.').branch
974
my_config = b.get_config()
975
self.assertIs(my_config.get_user_option('wacky'), None)
976
my_config.set_user_option('wacky', 'unlikely')
977
self.assertEqual(my_config.get_user_option('wacky'), 'unlikely')
979
# Ensure we get the same thing if we start again
980
b2 = branch.Branch.open('.')
981
my_config2 = b2.get_config()
982
self.assertEqual(my_config2.get_user_option('wacky'), 'unlikely')
984
def test_has_explicit_nickname(self):
985
b = self.make_branch('.')
986
self.assertFalse(b.get_config().has_explicit_nickname())
988
self.assertTrue(b.get_config().has_explicit_nickname())
990
def test_config_url(self):
991
"""The Branch.get_config will use section that uses a local url"""
992
branch = self.make_branch('branch')
993
self.assertEqual('branch', branch.nick)
995
local_url = urlutils.local_path_to_url('branch')
996
conf = config.LocationConfig.from_string(
997
'[%s]\nnickname = foobar' % (local_url,),
998
local_url, save=True)
999
self.assertIsNot(None, conf)
1000
self.assertEqual('foobar', branch.nick)
1002
def test_config_local_path(self):
1003
"""The Branch.get_config will use a local system path"""
1004
branch = self.make_branch('branch')
1005
self.assertEqual('branch', branch.nick)
1007
local_path = osutils.getcwd().encode('utf8')
1008
config.LocationConfig.from_string(
1009
b'[%s/branch]\nnickname = barry' % (local_path,),
1010
'branch', save=True)
1011
# Now the branch will find its nick via the location config
1012
self.assertEqual('barry', branch.nick)
1014
def test_config_creates_local(self):
1015
"""Creating a new entry in config uses a local path."""
1016
branch = self.make_branch('branch', format='knit')
1017
branch.set_push_location('http://foobar')
1018
local_path = osutils.getcwd().encode('utf8')
1019
# Surprisingly ConfigObj doesn't create a trailing newline
1020
self.check_file_contents(bedding.locations_config_path(),
1022
b'push_location = http://foobar\n'
1023
b'push_location:policy = norecurse\n'
1026
def test_autonick_urlencoded(self):
1027
b = self.make_branch('!repo')
1028
self.assertEqual('!repo', b.get_config().get_nickname())
1030
def test_autonick_uses_branch_name(self):
1031
b = self.make_branch('foo', name='bar')
1032
self.assertEqual('bar', b.get_config().get_nickname())
1034
def test_warn_if_masked(self):
1038
warnings.append(args[0] % args[1:])
1039
self.overrideAttr(trace, 'warning', warning)
1041
def set_option(store, warn_masked=True):
1043
conf.set_user_option('example_option', repr(store), store=store,
1044
warn_masked=warn_masked)
1046
def assertWarning(warning):
1048
self.assertEqual(0, len(warnings))
1050
self.assertEqual(1, len(warnings))
1051
self.assertEqual(warning, warnings[0])
1052
branch = self.make_branch('.')
1053
conf = branch.get_config()
1054
set_option(config.STORE_GLOBAL)
1056
set_option(config.STORE_BRANCH)
1058
set_option(config.STORE_GLOBAL)
1059
assertWarning('Value "4" is masked by "3" from branch.conf')
1060
set_option(config.STORE_GLOBAL, warn_masked=False)
1062
set_option(config.STORE_LOCATION)
1064
set_option(config.STORE_BRANCH)
1065
assertWarning('Value "3" is masked by "0" from locations.conf')
1066
set_option(config.STORE_BRANCH, warn_masked=False)
1070
class TestGlobalConfigItems(tests.TestCaseInTempDir):
1072
def _get_empty_config(self):
1073
my_config = config.GlobalConfig()
1076
def _get_sample_config(self):
1077
my_config = config.GlobalConfig.from_string(sample_config_text)
1080
def test_user_id(self):
1081
my_config = config.GlobalConfig.from_string(sample_config_text)
1082
self.assertEqual(u"Erik B\u00e5gfors <erik@bagfors.nu>",
1083
my_config._get_user_id())
1085
def test_absent_user_id(self):
1086
my_config = config.GlobalConfig()
1087
self.assertEqual(None, my_config._get_user_id())
1089
def test_get_user_option_default(self):
1090
my_config = self._get_empty_config()
1091
self.assertEqual(None, my_config.get_user_option('no_option'))
1093
def test_get_user_option_global(self):
1094
my_config = self._get_sample_config()
1095
self.assertEqual("something",
1096
my_config.get_user_option('user_global_option'))
1098
def test_configured_validate_signatures_in_log(self):
1099
my_config = self._get_sample_config()
1100
self.assertEqual(True, my_config.validate_signatures_in_log())
1102
def test_get_alias(self):
1103
my_config = self._get_sample_config()
1104
self.assertEqual('help', my_config.get_alias('h'))
1106
def test_get_aliases(self):
1107
my_config = self._get_sample_config()
1108
aliases = my_config.get_aliases()
1109
self.assertEqual(2, len(aliases))
1110
sorted_keys = sorted(aliases)
1111
self.assertEqual('help', aliases[sorted_keys[0]])
1112
self.assertEqual(sample_long_alias, aliases[sorted_keys[1]])
1114
def test_get_no_alias(self):
1115
my_config = self._get_sample_config()
1116
self.assertEqual(None, my_config.get_alias('foo'))
1118
def test_get_long_alias(self):
1119
my_config = self._get_sample_config()
1120
self.assertEqual(sample_long_alias, my_config.get_alias('ll'))
1122
def test_get_change_editor(self):
1123
my_config = self._get_sample_config()
1124
change_editor = my_config.get_change_editor('old', 'new')
1125
self.assertIs(diff.DiffFromTool, change_editor.__class__)
1126
self.assertEqual('vimdiff -of {new_path} {old_path}',
1127
' '.join(change_editor.command_template))
1129
def test_get_no_change_editor(self):
1130
my_config = self._get_empty_config()
1131
change_editor = my_config.get_change_editor('old', 'new')
1132
self.assertIs(None, change_editor)
1134
def test_get_merge_tools(self):
1135
conf = self._get_sample_config()
1136
tools = conf.get_merge_tools()
1137
self.log(repr(tools))
1139
{u'funkytool': u'funkytool "arg with spaces" {this_temp}',
1140
u'sometool': u'sometool {base} {this} {other} -o {result}',
1141
u'newtool': u'"newtool with spaces" {this_temp}'},
1144
def test_get_merge_tools_empty(self):
1145
conf = self._get_empty_config()
1146
tools = conf.get_merge_tools()
1147
self.assertEqual({}, tools)
1149
def test_find_merge_tool(self):
1150
conf = self._get_sample_config()
1151
cmdline = conf.find_merge_tool('sometool')
1152
self.assertEqual('sometool {base} {this} {other} -o {result}', cmdline)
1154
def test_find_merge_tool_not_found(self):
1155
conf = self._get_sample_config()
1156
cmdline = conf.find_merge_tool('DOES NOT EXIST')
1157
self.assertIs(cmdline, None)
1159
def test_find_merge_tool_known(self):
1160
conf = self._get_empty_config()
1161
cmdline = conf.find_merge_tool('kdiff3')
1162
self.assertEqual('kdiff3 {base} {this} {other} -o {result}', cmdline)
1164
def test_find_merge_tool_override_known(self):
1165
conf = self._get_empty_config()
1166
conf.set_user_option('bzr.mergetool.kdiff3', 'kdiff3 blah')
1167
cmdline = conf.find_merge_tool('kdiff3')
1168
self.assertEqual('kdiff3 blah', cmdline)
1171
class TestGlobalConfigSavingOptions(tests.TestCaseInTempDir):
1173
def test_empty(self):
1174
my_config = config.GlobalConfig()
1175
self.assertEqual(0, len(my_config.get_aliases()))
1177
def test_set_alias(self):
1178
my_config = config.GlobalConfig()
1179
alias_value = 'commit --strict'
1180
my_config.set_alias('commit', alias_value)
1181
new_config = config.GlobalConfig()
1182
self.assertEqual(alias_value, new_config.get_alias('commit'))
1184
def test_remove_alias(self):
1185
my_config = config.GlobalConfig()
1186
my_config.set_alias('commit', 'commit --strict')
1187
# Now remove the alias again.
1188
my_config.unset_alias('commit')
1189
new_config = config.GlobalConfig()
1190
self.assertIs(None, new_config.get_alias('commit'))
1193
class TestLocationConfig(tests.TestCaseInTempDir, TestOptionsMixin):
1195
def test_constructs_valid(self):
1196
config.LocationConfig('http://example.com')
1198
def test_constructs_error(self):
1199
self.assertRaises(TypeError, config.LocationConfig)
1201
def test_branch_calls_read_filenames(self):
1202
# This is testing the correct file names are provided.
1203
# TODO: consolidate with the test for GlobalConfigs filename checks.
1205
# replace the class that is constructed, to check its parameters
1206
oldparserclass = config.ConfigObj
1207
config.ConfigObj = InstrumentedConfigObj
1209
my_config = config.LocationConfig('http://www.example.com')
1210
parser = my_config._get_parser()
1212
config.ConfigObj = oldparserclass
1213
self.assertIsInstance(parser, InstrumentedConfigObj)
1214
self.assertEqual(parser._calls,
1215
[('__init__', bedding.locations_config_path(),
1218
def test_get_global_config(self):
1219
my_config = config.BranchConfig(FakeBranch('http://example.com'))
1220
global_config = my_config._get_global_config()
1221
self.assertIsInstance(global_config, config.GlobalConfig)
1222
self.assertIs(global_config, my_config._get_global_config())
1224
def assertLocationMatching(self, expected):
1225
self.assertEqual(expected,
1226
list(self.my_location_config._get_matching_sections()))
1228
def test__get_matching_sections_no_match(self):
1229
self.get_branch_config('/')
1230
self.assertLocationMatching([])
1232
def test__get_matching_sections_exact(self):
1233
self.get_branch_config('http://www.example.com')
1234
self.assertLocationMatching([('http://www.example.com', '')])
1236
def test__get_matching_sections_suffix_does_not(self):
1237
self.get_branch_config('http://www.example.com-com')
1238
self.assertLocationMatching([])
1240
def test__get_matching_sections_subdir_recursive(self):
1241
self.get_branch_config('http://www.example.com/com')
1242
self.assertLocationMatching([('http://www.example.com', 'com')])
1244
def test__get_matching_sections_ignoreparent(self):
1245
self.get_branch_config('http://www.example.com/ignoreparent')
1246
self.assertLocationMatching([('http://www.example.com/ignoreparent',
1249
def test__get_matching_sections_ignoreparent_subdir(self):
1250
self.get_branch_config(
1251
'http://www.example.com/ignoreparent/childbranch')
1252
self.assertLocationMatching([('http://www.example.com/ignoreparent',
1255
def test__get_matching_sections_subdir_trailing_slash(self):
1256
self.get_branch_config('/b')
1257
self.assertLocationMatching([('/b/', '')])
1259
def test__get_matching_sections_subdir_child(self):
1260
self.get_branch_config('/a/foo')
1261
self.assertLocationMatching([('/a/*', ''), ('/a/', 'foo')])
1263
def test__get_matching_sections_subdir_child_child(self):
1264
self.get_branch_config('/a/foo/bar')
1265
self.assertLocationMatching([('/a/*', 'bar'), ('/a/', 'foo/bar')])
1267
def test__get_matching_sections_trailing_slash_with_children(self):
1268
self.get_branch_config('/a/')
1269
self.assertLocationMatching([('/a/', '')])
1271
def test__get_matching_sections_explicit_over_glob(self):
1272
# XXX: 2006-09-08 jamesh
1273
# This test only passes because ord('c') > ord('*'). If there
1274
# was a config section for '/a/?', it would get precedence
1276
self.get_branch_config('/a/c')
1277
self.assertLocationMatching([('/a/c', ''), ('/a/*', ''), ('/a/', 'c')])
1279
def test__get_option_policy_normal(self):
1280
self.get_branch_config('http://www.example.com')
1282
self.my_location_config._get_config_policy(
1283
'http://www.example.com', 'normal_option'),
1286
def test__get_option_policy_norecurse(self):
1287
self.get_branch_config('http://www.example.com')
1289
self.my_location_config._get_option_policy(
1290
'http://www.example.com', 'norecurse_option'),
1291
config.POLICY_NORECURSE)
1292
# Test old recurse=False setting:
1294
self.my_location_config._get_option_policy(
1295
'http://www.example.com/norecurse', 'normal_option'),
1296
config.POLICY_NORECURSE)
1298
def test__get_option_policy_normal(self):
1299
self.get_branch_config('http://www.example.com')
1301
self.my_location_config._get_option_policy(
1302
'http://www.example.com', 'appendpath_option'),
1303
config.POLICY_APPENDPATH)
1305
def test__get_options_with_policy(self):
1306
self.get_branch_config('/dir/subdir',
1307
location_config="""\
1309
other_url = /other-dir
1310
other_url:policy = appendpath
1312
other_url = /other-subdir
1315
[(u'other_url', u'/other-subdir', u'/dir/subdir', 'locations'),
1316
(u'other_url', u'/other-dir', u'/dir', 'locations'),
1317
(u'other_url:policy', u'appendpath', u'/dir', 'locations')],
1318
self.my_location_config)
1320
def test_location_without_username(self):
1321
self.get_branch_config('http://www.example.com/ignoreparent')
1322
self.assertEqual(u'Erik B\u00e5gfors <erik@bagfors.nu>',
1323
self.my_config.username())
1325
def test_location_not_listed(self):
1326
"""Test that the global username is used when no location matches"""
1327
self.get_branch_config('/home/robertc/sources')
1328
self.assertEqual(u'Erik B\u00e5gfors <erik@bagfors.nu>',
1329
self.my_config.username())
1331
def test_overriding_location(self):
1332
self.get_branch_config('http://www.example.com/foo')
1333
self.assertEqual('Robert Collins <robertc@example.org>',
1334
self.my_config.username())
1336
def test_get_user_option_global(self):
1337
self.get_branch_config('/a')
1338
self.assertEqual('something',
1339
self.my_config.get_user_option('user_global_option'))
1341
def test_get_user_option_local(self):
1342
self.get_branch_config('/a')
1343
self.assertEqual('local',
1344
self.my_config.get_user_option('user_local_option'))
1346
def test_get_user_option_appendpath(self):
1347
# returned as is for the base path:
1348
self.get_branch_config('http://www.example.com')
1349
self.assertEqual('append',
1350
self.my_config.get_user_option('appendpath_option'))
1351
# Extra path components get appended:
1352
self.get_branch_config('http://www.example.com/a/b/c')
1353
self.assertEqual('append/a/b/c',
1354
self.my_config.get_user_option('appendpath_option'))
1355
# Overriden for http://www.example.com/dir, where it is a
1357
self.get_branch_config('http://www.example.com/dir/a/b/c')
1358
self.assertEqual('normal',
1359
self.my_config.get_user_option('appendpath_option'))
1361
def test_get_user_option_norecurse(self):
1362
self.get_branch_config('http://www.example.com')
1363
self.assertEqual('norecurse',
1364
self.my_config.get_user_option('norecurse_option'))
1365
self.get_branch_config('http://www.example.com/dir')
1366
self.assertEqual(None,
1367
self.my_config.get_user_option('norecurse_option'))
1368
# http://www.example.com/norecurse is a recurse=False section
1369
# that redefines normal_option. Subdirectories do not pick up
1370
# this redefinition.
1371
self.get_branch_config('http://www.example.com/norecurse')
1372
self.assertEqual('norecurse',
1373
self.my_config.get_user_option('normal_option'))
1374
self.get_branch_config('http://www.example.com/norecurse/subdir')
1375
self.assertEqual('normal',
1376
self.my_config.get_user_option('normal_option'))
1378
def test_set_user_option_norecurse(self):
1379
self.get_branch_config('http://www.example.com')
1380
self.my_config.set_user_option('foo', 'bar',
1381
store=config.STORE_LOCATION_NORECURSE)
1383
self.my_location_config._get_option_policy(
1384
'http://www.example.com', 'foo'),
1385
config.POLICY_NORECURSE)
1387
def test_set_user_option_appendpath(self):
1388
self.get_branch_config('http://www.example.com')
1389
self.my_config.set_user_option('foo', 'bar',
1390
store=config.STORE_LOCATION_APPENDPATH)
1392
self.my_location_config._get_option_policy(
1393
'http://www.example.com', 'foo'),
1394
config.POLICY_APPENDPATH)
1396
def test_set_user_option_change_policy(self):
1397
self.get_branch_config('http://www.example.com')
1398
self.my_config.set_user_option('norecurse_option', 'normal',
1399
store=config.STORE_LOCATION)
1401
self.my_location_config._get_option_policy(
1402
'http://www.example.com', 'norecurse_option'),
1405
def get_branch_config(self, location, global_config=None,
1406
location_config=None):
1407
my_branch = FakeBranch(location)
1408
if global_config is None:
1409
global_config = sample_config_text
1410
if location_config is None:
1411
location_config = sample_branches_text
1413
config.GlobalConfig.from_string(global_config, save=True)
1414
config.LocationConfig.from_string(location_config, my_branch.base,
1416
my_config = config.BranchConfig(my_branch)
1417
self.my_config = my_config
1418
self.my_location_config = my_config._get_location_config()
1420
def test_set_user_setting_sets_and_saves2(self):
1421
self.get_branch_config('/a/c')
1422
self.assertIs(self.my_config.get_user_option('foo'), None)
1423
self.my_config.set_user_option('foo', 'bar')
1425
self.my_config.branch.control_files.files['branch.conf'].strip(),
1427
self.assertEqual(self.my_config.get_user_option('foo'), 'bar')
1428
self.my_config.set_user_option('foo', 'baz',
1429
store=config.STORE_LOCATION)
1430
self.assertEqual(self.my_config.get_user_option('foo'), 'baz')
1431
self.my_config.set_user_option('foo', 'qux')
1432
self.assertEqual(self.my_config.get_user_option('foo'), 'baz')
1434
def test_get_bzr_remote_path(self):
1435
my_config = config.LocationConfig('/a/c')
1436
self.assertEqual('bzr', my_config.get_bzr_remote_path())
1437
my_config.set_user_option('bzr_remote_path', '/path-bzr')
1438
self.assertEqual('/path-bzr', my_config.get_bzr_remote_path())
1439
self.overrideEnv('BZR_REMOTE_PATH', '/environ-bzr')
1440
self.assertEqual('/environ-bzr', my_config.get_bzr_remote_path())
1443
precedence_global = b'option = global'
1444
precedence_branch = b'option = branch'
1445
precedence_location = b"""
1449
[http://example.com/specific]
1454
class TestBranchConfigItems(tests.TestCaseInTempDir):
1456
def get_branch_config(self, global_config=None, location=None,
1457
location_config=None, branch_data_config=None):
1458
my_branch = FakeBranch(location)
1459
if global_config is not None:
1460
config.GlobalConfig.from_string(global_config, save=True)
1461
if location_config is not None:
1462
config.LocationConfig.from_string(location_config, my_branch.base,
1464
my_config = config.BranchConfig(my_branch)
1465
if branch_data_config is not None:
1466
my_config.branch.control_files.files['branch.conf'] = \
1470
def test_user_id(self):
1471
branch = FakeBranch()
1472
my_config = config.BranchConfig(branch)
1473
self.assertIsNot(None, my_config.username())
1474
my_config.branch.control_files.files['email'] = "John"
1475
my_config.set_user_option('email',
1476
"Robert Collins <robertc@example.org>")
1477
self.assertEqual("Robert Collins <robertc@example.org>",
1478
my_config.username())
1480
def test_BRZ_EMAIL_OVERRIDES(self):
1481
self.overrideEnv('BRZ_EMAIL', "Robert Collins <robertc@example.org>")
1482
branch = FakeBranch()
1483
my_config = config.BranchConfig(branch)
1484
self.assertEqual("Robert Collins <robertc@example.org>",
1485
my_config.username())
1487
def test_get_user_option_global(self):
1488
my_config = self.get_branch_config(global_config=sample_config_text)
1489
self.assertEqual('something',
1490
my_config.get_user_option('user_global_option'))
1492
def test_config_precedence(self):
1493
# FIXME: eager test, luckily no persitent config file makes it fail
1495
my_config = self.get_branch_config(global_config=precedence_global)
1496
self.assertEqual(my_config.get_user_option('option'), 'global')
1497
my_config = self.get_branch_config(global_config=precedence_global,
1498
branch_data_config=precedence_branch)
1499
self.assertEqual(my_config.get_user_option('option'), 'branch')
1500
my_config = self.get_branch_config(
1501
global_config=precedence_global,
1502
branch_data_config=precedence_branch,
1503
location_config=precedence_location)
1504
self.assertEqual(my_config.get_user_option('option'), 'recurse')
1505
my_config = self.get_branch_config(
1506
global_config=precedence_global,
1507
branch_data_config=precedence_branch,
1508
location_config=precedence_location,
1509
location='http://example.com/specific')
1510
self.assertEqual(my_config.get_user_option('option'), 'exact')
1513
class TestMailAddressExtraction(tests.TestCase):
1515
def test_extract_email_address(self):
1516
self.assertEqual('jane@test.com',
1517
config.extract_email_address('Jane <jane@test.com>'))
1518
self.assertRaises(config.NoEmailInUsername,
1519
config.extract_email_address, 'Jane Tester')
1521
def test_parse_username(self):
1522
self.assertEqual(('', 'jdoe@example.com'),
1523
config.parse_username('jdoe@example.com'))
1524
self.assertEqual(('', 'jdoe@example.com'),
1525
config.parse_username('<jdoe@example.com>'))
1526
self.assertEqual(('John Doe', 'jdoe@example.com'),
1527
config.parse_username('John Doe <jdoe@example.com>'))
1528
self.assertEqual(('John Doe', ''),
1529
config.parse_username('John Doe'))
1530
self.assertEqual(('John Doe', 'jdoe@example.com'),
1531
config.parse_username('John Doe jdoe@example.com'))
1534
class TestTreeConfig(tests.TestCaseWithTransport):
1536
def test_get_value(self):
1537
"""Test that retreiving a value from a section is possible"""
1538
branch = self.make_branch('.')
1539
tree_config = config.TreeConfig(branch)
1540
tree_config.set_option('value', 'key', 'SECTION')
1541
tree_config.set_option('value2', 'key2')
1542
tree_config.set_option('value3-top', 'key3')
1543
tree_config.set_option('value3-section', 'key3', 'SECTION')
1544
value = tree_config.get_option('key', 'SECTION')
1545
self.assertEqual(value, 'value')
1546
value = tree_config.get_option('key2')
1547
self.assertEqual(value, 'value2')
1548
self.assertEqual(tree_config.get_option('non-existant'), None)
1549
value = tree_config.get_option('non-existant', 'SECTION')
1550
self.assertEqual(value, None)
1551
value = tree_config.get_option('non-existant', default='default')
1552
self.assertEqual(value, 'default')
1553
self.assertEqual(tree_config.get_option('key2', 'NOSECTION'), None)
1554
value = tree_config.get_option('key2', 'NOSECTION', default='default')
1555
self.assertEqual(value, 'default')
1556
value = tree_config.get_option('key3')
1557
self.assertEqual(value, 'value3-top')
1558
value = tree_config.get_option('key3', 'SECTION')
1559
self.assertEqual(value, 'value3-section')
1562
class TestTransportConfig(tests.TestCaseWithTransport):
1564
def test_load_utf8(self):
1565
"""Ensure we can load an utf8-encoded file."""
1566
t = self.get_transport()
1567
unicode_user = u'b\N{Euro Sign}ar'
1568
unicode_content = u'user=%s' % (unicode_user,)
1569
utf8_content = unicode_content.encode('utf8')
1570
# Store the raw content in the config file
1571
t.put_bytes('foo.conf', utf8_content)
1572
conf = config.TransportConfig(t, 'foo.conf')
1573
self.assertEqual(unicode_user, conf.get_option('user'))
1575
def test_load_non_ascii(self):
1576
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
1577
t = self.get_transport()
1578
t.put_bytes('foo.conf', b'user=foo\n#\xff\n')
1579
conf = config.TransportConfig(t, 'foo.conf')
1580
self.assertRaises(config.ConfigContentError, conf._get_configobj)
1582
def test_load_erroneous_content(self):
1583
"""Ensure we display a proper error on content that can't be parsed."""
1584
t = self.get_transport()
1585
t.put_bytes('foo.conf', b'[open_section\n')
1586
conf = config.TransportConfig(t, 'foo.conf')
1587
self.assertRaises(config.ParseConfigError, conf._get_configobj)
1589
def test_load_permission_denied(self):
1590
"""Ensure we get an empty config file if the file is inaccessible."""
1594
warnings.append(args[0] % args[1:])
1595
self.overrideAttr(trace, 'warning', warning)
1597
class DenyingTransport(object):
1599
def __init__(self, base):
1602
def get_bytes(self, relpath):
1603
raise errors.PermissionDenied(relpath, "")
1605
cfg = config.TransportConfig(
1606
DenyingTransport("nonexisting://"), 'control.conf')
1607
self.assertIs(None, cfg.get_option('non-existant', 'SECTION'))
1610
[u'Permission denied while trying to open configuration file '
1611
u'nonexisting:///control.conf.'])
1613
def test_get_value(self):
1614
"""Test that retreiving a value from a section is possible"""
1615
bzrdir_config = config.TransportConfig(self.get_transport('.'),
1617
bzrdir_config.set_option('value', 'key', 'SECTION')
1618
bzrdir_config.set_option('value2', 'key2')
1619
bzrdir_config.set_option('value3-top', 'key3')
1620
bzrdir_config.set_option('value3-section', 'key3', 'SECTION')
1621
value = bzrdir_config.get_option('key', 'SECTION')
1622
self.assertEqual(value, 'value')
1623
value = bzrdir_config.get_option('key2')
1624
self.assertEqual(value, 'value2')
1625
self.assertEqual(bzrdir_config.get_option('non-existant'), None)
1626
value = bzrdir_config.get_option('non-existant', 'SECTION')
1627
self.assertEqual(value, None)
1628
value = bzrdir_config.get_option('non-existant', default='default')
1629
self.assertEqual(value, 'default')
1630
self.assertEqual(bzrdir_config.get_option('key2', 'NOSECTION'), None)
1631
value = bzrdir_config.get_option('key2', 'NOSECTION',
1633
self.assertEqual(value, 'default')
1634
value = bzrdir_config.get_option('key3')
1635
self.assertEqual(value, 'value3-top')
1636
value = bzrdir_config.get_option('key3', 'SECTION')
1637
self.assertEqual(value, 'value3-section')
1639
def test_set_unset_default_stack_on(self):
1640
my_dir = self.make_controldir('.')
1641
bzrdir_config = config.BzrDirConfig(my_dir)
1642
self.assertIs(None, bzrdir_config.get_default_stack_on())
1643
bzrdir_config.set_default_stack_on('Foo')
1644
self.assertEqual('Foo', bzrdir_config._config.get_option(
1645
'default_stack_on'))
1646
self.assertEqual('Foo', bzrdir_config.get_default_stack_on())
1647
bzrdir_config.set_default_stack_on(None)
1648
self.assertIs(None, bzrdir_config.get_default_stack_on())
1651
class TestOldConfigHooks(tests.TestCaseWithTransport):
1654
super(TestOldConfigHooks, self).setUp()
1655
create_configs_with_file_option(self)
1657
def assertGetHook(self, conf, name, value):
1662
config.OldConfigHooks.install_named_hook('get', hook, None)
1664
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1665
self.assertLength(0, calls)
1666
actual_value = conf.get_user_option(name)
1667
self.assertEqual(value, actual_value)
1668
self.assertLength(1, calls)
1669
self.assertEqual((conf, name, value), calls[0])
1671
def test_get_hook_breezy(self):
1672
self.assertGetHook(self.breezy_config, 'file', 'breezy')
1674
def test_get_hook_locations(self):
1675
self.assertGetHook(self.locations_config, 'file', 'locations')
1677
def test_get_hook_branch(self):
1678
# Since locations masks branch, we define a different option
1679
self.branch_config.set_user_option('file2', 'branch')
1680
self.assertGetHook(self.branch_config, 'file2', 'branch')
1682
def assertSetHook(self, conf, name, value):
1687
config.OldConfigHooks.install_named_hook('set', hook, None)
1689
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1690
self.assertLength(0, calls)
1691
conf.set_user_option(name, value)
1692
self.assertLength(1, calls)
1693
# We can't assert the conf object below as different configs use
1694
# different means to implement set_user_option and we care only about
1696
self.assertEqual((name, value), calls[0][1:])
1698
def test_set_hook_breezy(self):
1699
self.assertSetHook(self.breezy_config, 'foo', 'breezy')
1701
def test_set_hook_locations(self):
1702
self.assertSetHook(self.locations_config, 'foo', 'locations')
1704
def test_set_hook_branch(self):
1705
self.assertSetHook(self.branch_config, 'foo', 'branch')
1707
def assertRemoveHook(self, conf, name, section_name=None):
1712
config.OldConfigHooks.install_named_hook('remove', hook, None)
1714
config.OldConfigHooks.uninstall_named_hook, 'remove', None)
1715
self.assertLength(0, calls)
1716
conf.remove_user_option(name, section_name)
1717
self.assertLength(1, calls)
1718
# We can't assert the conf object below as different configs use
1719
# different means to implement remove_user_option and we care only about
1721
self.assertEqual((name,), calls[0][1:])
1723
def test_remove_hook_breezy(self):
1724
self.assertRemoveHook(self.breezy_config, 'file')
1726
def test_remove_hook_locations(self):
1727
self.assertRemoveHook(self.locations_config, 'file',
1728
self.locations_config.location)
1730
def test_remove_hook_branch(self):
1731
self.assertRemoveHook(self.branch_config, 'file')
1733
def assertLoadHook(self, name, conf_class, *conf_args):
1738
config.OldConfigHooks.install_named_hook('load', hook, None)
1740
config.OldConfigHooks.uninstall_named_hook, 'load', None)
1741
self.assertLength(0, calls)
1743
conf = conf_class(*conf_args)
1744
# Access an option to trigger a load
1745
conf.get_user_option(name)
1746
self.assertLength(1, calls)
1747
# Since we can't assert about conf, we just use the number of calls ;-/
1749
def test_load_hook_breezy(self):
1750
self.assertLoadHook('file', config.GlobalConfig)
1752
def test_load_hook_locations(self):
1753
self.assertLoadHook('file', config.LocationConfig, self.tree.basedir)
1755
def test_load_hook_branch(self):
1756
self.assertLoadHook('file', config.BranchConfig, self.tree.branch)
1758
def assertSaveHook(self, conf):
1763
config.OldConfigHooks.install_named_hook('save', hook, None)
1765
config.OldConfigHooks.uninstall_named_hook, 'save', None)
1766
self.assertLength(0, calls)
1767
# Setting an option triggers a save
1768
conf.set_user_option('foo', 'bar')
1769
self.assertLength(1, calls)
1770
# Since we can't assert about conf, we just use the number of calls ;-/
1772
def test_save_hook_breezy(self):
1773
self.assertSaveHook(self.breezy_config)
1775
def test_save_hook_locations(self):
1776
self.assertSaveHook(self.locations_config)
1778
def test_save_hook_branch(self):
1779
self.assertSaveHook(self.branch_config)
1782
class TestOldConfigHooksForRemote(tests.TestCaseWithTransport):
1783
"""Tests config hooks for remote configs.
1785
No tests for the remove hook as this is not implemented there.
1789
super(TestOldConfigHooksForRemote, self).setUp()
1790
self.transport_server = test_server.SmartTCPServer_for_testing
1791
create_configs_with_file_option(self)
1793
def assertGetHook(self, conf, name, value):
1798
config.OldConfigHooks.install_named_hook('get', hook, None)
1800
config.OldConfigHooks.uninstall_named_hook, 'get', None)
1801
self.assertLength(0, calls)
1802
actual_value = conf.get_option(name)
1803
self.assertEqual(value, actual_value)
1804
self.assertLength(1, calls)
1805
self.assertEqual((conf, name, value), calls[0])
1807
def test_get_hook_remote_branch(self):
1808
remote_branch = branch.Branch.open(self.get_url('tree'))
1809
self.assertGetHook(remote_branch._get_config(), 'file', 'branch')
1811
def test_get_hook_remote_bzrdir(self):
1812
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1813
conf = remote_bzrdir._get_config()
1814
conf.set_option('remotedir', 'file')
1815
self.assertGetHook(conf, 'file', 'remotedir')
1817
def assertSetHook(self, conf, name, value):
1822
config.OldConfigHooks.install_named_hook('set', hook, None)
1824
config.OldConfigHooks.uninstall_named_hook, 'set', None)
1825
self.assertLength(0, calls)
1826
conf.set_option(value, name)
1827
self.assertLength(1, calls)
1828
# We can't assert the conf object below as different configs use
1829
# different means to implement set_user_option and we care only about
1831
self.assertEqual((name, value), calls[0][1:])
1833
def test_set_hook_remote_branch(self):
1834
remote_branch = branch.Branch.open(self.get_url('tree'))
1835
self.addCleanup(remote_branch.lock_write().unlock)
1836
self.assertSetHook(remote_branch._get_config(), 'file', 'remote')
1838
def test_set_hook_remote_bzrdir(self):
1839
remote_branch = branch.Branch.open(self.get_url('tree'))
1840
self.addCleanup(remote_branch.lock_write().unlock)
1841
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1842
self.assertSetHook(remote_bzrdir._get_config(), 'file', 'remotedir')
1844
def assertLoadHook(self, expected_nb_calls, name, conf_class, *conf_args):
1849
config.OldConfigHooks.install_named_hook('load', hook, None)
1851
config.OldConfigHooks.uninstall_named_hook, 'load', None)
1852
self.assertLength(0, calls)
1854
conf = conf_class(*conf_args)
1855
# Access an option to trigger a load
1856
conf.get_option(name)
1857
self.assertLength(expected_nb_calls, calls)
1858
# Since we can't assert about conf, we just use the number of calls ;-/
1860
def test_load_hook_remote_branch(self):
1861
remote_branch = branch.Branch.open(self.get_url('tree'))
1862
self.assertLoadHook(
1863
1, 'file', remote.RemoteBranchConfig, remote_branch)
1865
def test_load_hook_remote_bzrdir(self):
1866
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1867
# The config file doesn't exist, set an option to force its creation
1868
conf = remote_bzrdir._get_config()
1869
conf.set_option('remotedir', 'file')
1870
# We get one call for the server and one call for the client, this is
1871
# caused by the differences in implementations betwen
1872
# SmartServerBzrDirRequestConfigFile (in smart/bzrdir.py) and
1873
# SmartServerBranchGetConfigFile (in smart/branch.py)
1874
self.assertLoadHook(
1875
2, 'file', remote.RemoteBzrDirConfig, remote_bzrdir)
1877
def assertSaveHook(self, conf):
1882
config.OldConfigHooks.install_named_hook('save', hook, None)
1884
config.OldConfigHooks.uninstall_named_hook, 'save', None)
1885
self.assertLength(0, calls)
1886
# Setting an option triggers a save
1887
conf.set_option('foo', 'bar')
1888
self.assertLength(1, calls)
1889
# Since we can't assert about conf, we just use the number of calls ;-/
1891
def test_save_hook_remote_branch(self):
1892
remote_branch = branch.Branch.open(self.get_url('tree'))
1893
self.addCleanup(remote_branch.lock_write().unlock)
1894
self.assertSaveHook(remote_branch._get_config())
1896
def test_save_hook_remote_bzrdir(self):
1897
remote_branch = branch.Branch.open(self.get_url('tree'))
1898
self.addCleanup(remote_branch.lock_write().unlock)
1899
remote_bzrdir = controldir.ControlDir.open(self.get_url('tree'))
1900
self.assertSaveHook(remote_bzrdir._get_config())
1903
class TestOptionNames(tests.TestCase):
1905
def is_valid(self, name):
1906
return config._option_ref_re.match('{%s}' % name) is not None
1908
def test_valid_names(self):
1909
self.assertTrue(self.is_valid('foo'))
1910
self.assertTrue(self.is_valid('foo.bar'))
1911
self.assertTrue(self.is_valid('f1'))
1912
self.assertTrue(self.is_valid('_'))
1913
self.assertTrue(self.is_valid('__bar__'))
1914
self.assertTrue(self.is_valid('a_'))
1915
self.assertTrue(self.is_valid('a1'))
1916
# Don't break bzr-svn for no good reason
1917
self.assertTrue(self.is_valid('guessed-layout'))
1919
def test_invalid_names(self):
1920
self.assertFalse(self.is_valid(' foo'))
1921
self.assertFalse(self.is_valid('foo '))
1922
self.assertFalse(self.is_valid('1'))
1923
self.assertFalse(self.is_valid('1,2'))
1924
self.assertFalse(self.is_valid('foo$'))
1925
self.assertFalse(self.is_valid('!foo'))
1926
self.assertFalse(self.is_valid('foo.'))
1927
self.assertFalse(self.is_valid('foo..bar'))
1928
self.assertFalse(self.is_valid('{}'))
1929
self.assertFalse(self.is_valid('{a}'))
1930
self.assertFalse(self.is_valid('a\n'))
1931
self.assertFalse(self.is_valid('-'))
1932
self.assertFalse(self.is_valid('-a'))
1933
self.assertFalse(self.is_valid('a-'))
1934
self.assertFalse(self.is_valid('a--a'))
1936
def assertSingleGroup(self, reference):
1937
# the regexp is used with split and as such should match the reference
1938
# *only*, if more groups needs to be defined, (?:...) should be used.
1939
m = config._option_ref_re.match('{a}')
1940
self.assertLength(1, m.groups())
1942
def test_valid_references(self):
1943
self.assertSingleGroup('{a}')
1944
self.assertSingleGroup('{{a}}')
1947
class TestOption(tests.TestCase):
1949
def test_default_value(self):
1950
opt = config.Option('foo', default='bar')
1951
self.assertEqual('bar', opt.get_default())
1953
def test_callable_default_value(self):
1954
def bar_as_unicode():
1956
opt = config.Option('foo', default=bar_as_unicode)
1957
self.assertEqual('bar', opt.get_default())
1959
def test_default_value_from_env(self):
1960
opt = config.Option('foo', default='bar', default_from_env=['FOO'])
1961
self.overrideEnv('FOO', 'quux')
1962
# Env variable provides a default taking over the option one
1963
self.assertEqual('quux', opt.get_default())
1965
def test_first_default_value_from_env_wins(self):
1966
opt = config.Option('foo', default='bar',
1967
default_from_env=['NO_VALUE', 'FOO', 'BAZ'])
1968
self.overrideEnv('FOO', 'foo')
1969
self.overrideEnv('BAZ', 'baz')
1970
# The first env var set wins
1971
self.assertEqual('foo', opt.get_default())
1973
def test_not_supported_list_default_value(self):
1974
self.assertRaises(AssertionError, config.Option, 'foo', default=[1])
1976
def test_not_supported_object_default_value(self):
1977
self.assertRaises(AssertionError, config.Option, 'foo',
1980
def test_not_supported_callable_default_value_not_unicode(self):
1981
def bar_not_unicode():
1983
opt = config.Option('foo', default=bar_not_unicode)
1984
self.assertRaises(AssertionError, opt.get_default)
1986
def test_get_help_topic(self):
1987
opt = config.Option('foo')
1988
self.assertEqual('foo', opt.get_help_topic())
1991
class TestOptionConverter(tests.TestCase):
1993
def assertConverted(self, expected, opt, value):
1994
self.assertEqual(expected, opt.convert_from_unicode(None, value))
1996
def assertCallsWarning(self, opt, value):
2000
warnings.append(args[0] % args[1:])
2001
self.overrideAttr(trace, 'warning', warning)
2002
self.assertEqual(None, opt.convert_from_unicode(None, value))
2003
self.assertLength(1, warnings)
2005
'Value "%s" is not valid for "%s"' % (value, opt.name),
2008
def assertCallsError(self, opt, value):
2009
self.assertRaises(config.ConfigOptionValueError,
2010
opt.convert_from_unicode, None, value)
2012
def assertConvertInvalid(self, opt, invalid_value):
2014
self.assertEqual(None, opt.convert_from_unicode(None, invalid_value))
2015
opt.invalid = 'warning'
2016
self.assertCallsWarning(opt, invalid_value)
2017
opt.invalid = 'error'
2018
self.assertCallsError(opt, invalid_value)
2021
class TestOptionWithBooleanConverter(TestOptionConverter):
2023
def get_option(self):
2024
return config.Option('foo', help='A boolean.',
2025
from_unicode=config.bool_from_store)
2027
def test_convert_invalid(self):
2028
opt = self.get_option()
2029
# A string that is not recognized as a boolean
2030
self.assertConvertInvalid(opt, u'invalid-boolean')
2031
# A list of strings is never recognized as a boolean
2032
self.assertConvertInvalid(opt, [u'not', u'a', u'boolean'])
2034
def test_convert_valid(self):
2035
opt = self.get_option()
2036
self.assertConverted(True, opt, u'True')
2037
self.assertConverted(True, opt, u'1')
2038
self.assertConverted(False, opt, u'False')
2041
class TestOptionWithIntegerConverter(TestOptionConverter):
2043
def get_option(self):
2044
return config.Option('foo', help='An integer.',
2045
from_unicode=config.int_from_store)
2047
def test_convert_invalid(self):
2048
opt = self.get_option()
2049
# A string that is not recognized as an integer
2050
self.assertConvertInvalid(opt, u'forty-two')
2051
# A list of strings is never recognized as an integer
2052
self.assertConvertInvalid(opt, [u'a', u'list'])
2054
def test_convert_valid(self):
2055
opt = self.get_option()
2056
self.assertConverted(16, opt, u'16')
2059
class TestOptionWithSIUnitConverter(TestOptionConverter):
2061
def get_option(self):
2062
return config.Option('foo', help='An integer in SI units.',
2063
from_unicode=config.int_SI_from_store)
2065
def test_convert_invalid(self):
2066
opt = self.get_option()
2067
self.assertConvertInvalid(opt, u'not-a-unit')
2068
self.assertConvertInvalid(opt, u'Gb') # Forgot the value
2069
self.assertConvertInvalid(opt, u'1b') # Forgot the unit
2070
self.assertConvertInvalid(opt, u'1GG')
2071
self.assertConvertInvalid(opt, u'1Mbb')
2072
self.assertConvertInvalid(opt, u'1MM')
2074
def test_convert_valid(self):
2075
opt = self.get_option()
2076
self.assertConverted(int(5e3), opt, u'5kb')
2077
self.assertConverted(int(5e6), opt, u'5M')
2078
self.assertConverted(int(5e6), opt, u'5MB')
2079
self.assertConverted(int(5e9), opt, u'5g')
2080
self.assertConverted(int(5e9), opt, u'5gB')
2081
self.assertConverted(100, opt, u'100')
2084
class TestListOption(TestOptionConverter):
2086
def get_option(self):
2087
return config.ListOption('foo', help='A list.')
2089
def test_convert_invalid(self):
2090
opt = self.get_option()
2091
# We don't even try to convert a list into a list, we only expect
2093
self.assertConvertInvalid(opt, [1])
2094
# No string is invalid as all forms can be converted to a list
2096
def test_convert_valid(self):
2097
opt = self.get_option()
2098
# An empty string is an empty list
2099
self.assertConverted([], opt, '') # Using a bare str() just in case
2100
self.assertConverted([], opt, u'')
2102
self.assertConverted([u'True'], opt, u'True')
2104
self.assertConverted([u'42'], opt, u'42')
2106
self.assertConverted([u'bar'], opt, u'bar')
2109
class TestRegistryOption(TestOptionConverter):
2111
def get_option(self, registry):
2112
return config.RegistryOption('foo', registry,
2113
help='A registry option.')
2115
def test_convert_invalid(self):
2116
registry = _mod_registry.Registry()
2117
opt = self.get_option(registry)
2118
self.assertConvertInvalid(opt, [1])
2119
self.assertConvertInvalid(opt, u"notregistered")
2121
def test_convert_valid(self):
2122
registry = _mod_registry.Registry()
2123
registry.register("someval", 1234)
2124
opt = self.get_option(registry)
2125
# Using a bare str() just in case
2126
self.assertConverted(1234, opt, "someval")
2127
self.assertConverted(1234, opt, u'someval')
2128
self.assertConverted(None, opt, None)
2130
def test_help(self):
2131
registry = _mod_registry.Registry()
2132
registry.register("someval", 1234, help="some option")
2133
registry.register("dunno", 1234, help="some other option")
2134
opt = self.get_option(registry)
2136
'A registry option.\n'
2138
'The following values are supported:\n'
2139
' dunno - some other option\n'
2140
' someval - some option\n',
2143
def test_get_help_text(self):
2144
registry = _mod_registry.Registry()
2145
registry.register("someval", 1234, help="some option")
2146
registry.register("dunno", 1234, help="some other option")
2147
opt = self.get_option(registry)
2149
'A registry option.\n'
2151
'The following values are supported:\n'
2152
' dunno - some other option\n'
2153
' someval - some option\n',
2154
opt.get_help_text())
2157
class TestOptionRegistry(tests.TestCase):
2160
super(TestOptionRegistry, self).setUp()
2161
# Always start with an empty registry
2162
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
2163
self.registry = config.option_registry
2165
def test_register(self):
2166
opt = config.Option('foo')
2167
self.registry.register(opt)
2168
self.assertIs(opt, self.registry.get('foo'))
2170
def test_registered_help(self):
2171
opt = config.Option('foo', help='A simple option')
2172
self.registry.register(opt)
2173
self.assertEqual('A simple option', self.registry.get_help('foo'))
2175
def test_dont_register_illegal_name(self):
2176
self.assertRaises(config.IllegalOptionName,
2177
self.registry.register, config.Option(' foo'))
2178
self.assertRaises(config.IllegalOptionName,
2179
self.registry.register, config.Option('bar,'))
2181
lazy_option = config.Option('lazy_foo', help='Lazy help')
2183
def test_register_lazy(self):
2184
self.registry.register_lazy('lazy_foo', self.__module__,
2185
'TestOptionRegistry.lazy_option')
2186
self.assertIs(self.lazy_option, self.registry.get('lazy_foo'))
2188
def test_registered_lazy_help(self):
2189
self.registry.register_lazy('lazy_foo', self.__module__,
2190
'TestOptionRegistry.lazy_option')
2191
self.assertEqual('Lazy help', self.registry.get_help('lazy_foo'))
2193
def test_dont_lazy_register_illegal_name(self):
2194
# This is where the root cause of http://pad.lv/1235099 is better
2195
# understood: 'register_lazy' doc string mentions that key should match
2196
# the option name which indirectly requires that the option name is a
2197
# valid python identifier. We violate that rule here (using a key that
2198
# doesn't match the option name) to test the option name checking.
2199
self.assertRaises(config.IllegalOptionName,
2200
self.registry.register_lazy, ' foo', self.__module__,
2201
'TestOptionRegistry.lazy_option')
2202
self.assertRaises(config.IllegalOptionName,
2203
self.registry.register_lazy, '1,2', self.__module__,
2204
'TestOptionRegistry.lazy_option')
2207
class TestRegisteredOptions(tests.TestCase):
2208
"""All registered options should verify some constraints."""
2210
scenarios = [(key, {'option_name': key, 'option': option}) for key, option
2211
in config.option_registry.iteritems()]
2214
super(TestRegisteredOptions, self).setUp()
2215
self.registry = config.option_registry
2217
def test_proper_name(self):
2218
# An option should be registered under its own name, this can't be
2219
# checked at registration time for the lazy ones.
2220
self.assertEqual(self.option_name, self.option.name)
2222
def test_help_is_set(self):
2223
option_help = self.registry.get_help(self.option_name)
2224
# Come on, think about the user, he really wants to know what the
2226
self.assertIsNot(None, option_help)
2227
self.assertNotEqual('', option_help)
2230
class TestSection(tests.TestCase):
2232
# FIXME: Parametrize so that all sections produced by Stores run these
2233
# tests -- vila 2011-04-01
2235
def test_get_a_value(self):
2236
a_dict = dict(foo='bar')
2237
section = config.Section('myID', a_dict)
2238
self.assertEqual('bar', section.get('foo'))
2240
def test_get_unknown_option(self):
2242
section = config.Section(None, a_dict)
2243
self.assertEqual('out of thin air',
2244
section.get('foo', 'out of thin air'))
2246
def test_options_is_shared(self):
2248
section = config.Section(None, a_dict)
2249
self.assertIs(a_dict, section.options)
2252
class TestMutableSection(tests.TestCase):
2254
scenarios = [('mutable',
2256
lambda opts: config.MutableSection('myID', opts)},),
2260
a_dict = dict(foo='bar')
2261
section = self.get_section(a_dict)
2262
section.set('foo', 'new_value')
2263
self.assertEqual('new_value', section.get('foo'))
2264
# The change appears in the shared section
2265
self.assertEqual('new_value', a_dict.get('foo'))
2266
# We keep track of the change
2267
self.assertTrue('foo' in section.orig)
2268
self.assertEqual('bar', section.orig.get('foo'))
2270
def test_set_preserve_original_once(self):
2271
a_dict = dict(foo='bar')
2272
section = self.get_section(a_dict)
2273
section.set('foo', 'first_value')
2274
section.set('foo', 'second_value')
2275
# We keep track of the original value
2276
self.assertTrue('foo' in section.orig)
2277
self.assertEqual('bar', section.orig.get('foo'))
2279
def test_remove(self):
2280
a_dict = dict(foo='bar')
2281
section = self.get_section(a_dict)
2282
section.remove('foo')
2283
# We get None for unknown options via the default value
2284
self.assertEqual(None, section.get('foo'))
2285
# Or we just get the default value
2286
self.assertEqual('unknown', section.get('foo', 'unknown'))
2287
self.assertFalse('foo' in section.options)
2288
# We keep track of the deletion
2289
self.assertTrue('foo' in section.orig)
2290
self.assertEqual('bar', section.orig.get('foo'))
2292
def test_remove_new_option(self):
2294
section = self.get_section(a_dict)
2295
section.set('foo', 'bar')
2296
section.remove('foo')
2297
self.assertFalse('foo' in section.options)
2298
# The option didn't exist initially so it we need to keep track of it
2299
# with a special value
2300
self.assertTrue('foo' in section.orig)
2301
self.assertEqual(config._NewlyCreatedOption, section.orig['foo'])
2304
class TestCommandLineStore(tests.TestCase):
2307
super(TestCommandLineStore, self).setUp()
2308
self.store = config.CommandLineStore()
2309
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
2311
def get_section(self):
2312
"""Get the unique section for the command line overrides."""
2313
sections = list(self.store.get_sections())
2314
self.assertLength(1, sections)
2315
store, section = sections[0]
2316
self.assertEqual(self.store, store)
2319
def test_no_override(self):
2320
self.store._from_cmdline([])
2321
section = self.get_section()
2322
self.assertLength(0, list(section.iter_option_names()))
2324
def test_simple_override(self):
2325
self.store._from_cmdline(['a=b'])
2326
section = self.get_section()
2327
self.assertEqual('b', section.get('a'))
2329
def test_list_override(self):
2330
opt = config.ListOption('l')
2331
config.option_registry.register(opt)
2332
self.store._from_cmdline(['l=1,2,3'])
2333
val = self.get_section().get('l')
2334
self.assertEqual('1,2,3', val)
2335
# Reminder: lists should be registered as such explicitely, otherwise
2336
# the conversion needs to be done afterwards.
2337
self.assertEqual(['1', '2', '3'],
2338
opt.convert_from_unicode(self.store, val))
2340
def test_multiple_overrides(self):
2341
self.store._from_cmdline(['a=b', 'x=y'])
2342
section = self.get_section()
2343
self.assertEqual('b', section.get('a'))
2344
self.assertEqual('y', section.get('x'))
2346
def test_wrong_syntax(self):
2347
self.assertRaises(errors.BzrCommandError,
2348
self.store._from_cmdline, ['a=b', 'c'])
2351
class TestStoreMinimalAPI(tests.TestCaseWithTransport):
2353
scenarios = [(key, {'get_store': builder}) for key, builder
2354
in config.test_store_builder_registry.iteritems()] + [
2355
('cmdline', {'get_store': lambda test: config.CommandLineStore()})]
2358
store = self.get_store(self)
2359
if isinstance(store, config.TransportIniFileStore):
2360
raise tests.TestNotApplicable(
2361
"%s is not a concrete Store implementation"
2362
" so it doesn't need an id" % (store.__class__.__name__,))
2363
self.assertIsNot(None, store.id)
2366
class TestStore(tests.TestCaseWithTransport):
2368
def assertSectionContent(self, expected, store_and_section):
2369
"""Assert that some options have the proper values in a section."""
2370
_, section = store_and_section
2371
expected_name, expected_options = expected
2372
self.assertEqual(expected_name, section.id)
2375
dict([(k, section.get(k)) for k in expected_options.keys()]))
2378
class TestReadonlyStore(TestStore):
2380
scenarios = [(key, {'get_store': builder}) for key, builder
2381
in config.test_store_builder_registry.iteritems()]
2383
def test_building_delays_load(self):
2384
store = self.get_store(self)
2385
self.assertEqual(False, store.is_loaded())
2386
store._load_from_string(b'')
2387
self.assertEqual(True, store.is_loaded())
2389
def test_get_no_sections_for_empty(self):
2390
store = self.get_store(self)
2391
store._load_from_string(b'')
2392
self.assertEqual([], list(store.get_sections()))
2394
def test_get_default_section(self):
2395
store = self.get_store(self)
2396
store._load_from_string(b'foo=bar')
2397
sections = list(store.get_sections())
2398
self.assertLength(1, sections)
2399
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2401
def test_get_named_section(self):
2402
store = self.get_store(self)
2403
store._load_from_string(b'[baz]\nfoo=bar')
2404
sections = list(store.get_sections())
2405
self.assertLength(1, sections)
2406
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2408
def test_load_from_string_fails_for_non_empty_store(self):
2409
store = self.get_store(self)
2410
store._load_from_string(b'foo=bar')
2411
self.assertRaises(AssertionError, store._load_from_string, b'bar=baz')
2414
class TestStoreQuoting(TestStore):
2416
scenarios = [(key, {'get_store': builder}) for key, builder
2417
in config.test_store_builder_registry.iteritems()]
2420
super(TestStoreQuoting, self).setUp()
2421
self.store = self.get_store(self)
2422
# We need a loaded store but any content will do
2423
self.store._load_from_string(b'')
2425
def assertIdempotent(self, s):
2426
"""Assert that quoting an unquoted string is a no-op and vice-versa.
2428
What matters here is that option values, as they appear in a store, can
2429
be safely round-tripped out of the store and back.
2431
:param s: A string, quoted if required.
2433
self.assertEqual(s, self.store.quote(self.store.unquote(s)))
2434
self.assertEqual(s, self.store.unquote(self.store.quote(s)))
2436
def test_empty_string(self):
2437
if isinstance(self.store, config.IniFileStore):
2438
# configobj._quote doesn't handle empty values
2439
self.assertRaises(AssertionError,
2440
self.assertIdempotent, '')
2442
self.assertIdempotent('')
2443
# But quoted empty strings are ok
2444
self.assertIdempotent('""')
2446
def test_embedded_spaces(self):
2447
self.assertIdempotent('" a b c "')
2449
def test_embedded_commas(self):
2450
self.assertIdempotent('" a , b c "')
2452
def test_simple_comma(self):
2453
if isinstance(self.store, config.IniFileStore):
2454
# configobj requires that lists are special-cased
2455
self.assertRaises(AssertionError,
2456
self.assertIdempotent, ',')
2458
self.assertIdempotent(',')
2459
# When a single comma is required, quoting is also required
2460
self.assertIdempotent('","')
2462
def test_list(self):
2463
if isinstance(self.store, config.IniFileStore):
2464
# configobj requires that lists are special-cased
2465
self.assertRaises(AssertionError,
2466
self.assertIdempotent, 'a,b')
2468
self.assertIdempotent('a,b')
2471
class TestDictFromStore(tests.TestCase):
2473
def test_unquote_not_string(self):
2474
conf = config.MemoryStack(b'x=2\n[a_section]\na=1\n')
2475
value = conf.get('a_section')
2476
# Urgh, despite 'conf' asking for the no-name section, we get the
2477
# content of another section as a dict o_O
2478
self.assertEqual({'a': '1'}, value)
2479
unquoted = conf.store.unquote(value)
2480
# Which cannot be unquoted but shouldn't crash either (the use cases
2481
# are getting the value or displaying it. In the later case, '%s' will
2483
self.assertEqual({'a': '1'}, unquoted)
2484
self.assertIn('%s' % (unquoted,), ("{u'a': u'1'}", "{'a': '1'}"))
2487
class TestIniFileStoreContent(tests.TestCaseWithTransport):
2488
"""Simulate loading a config store with content of various encodings.
2490
All files produced by bzr are in utf8 content.
2492
Users may modify them manually and end up with a file that can't be
2493
loaded. We need to issue proper error messages in this case.
2496
invalid_utf8_char = b'\xff'
2498
def test_load_utf8(self):
2499
"""Ensure we can load an utf8-encoded file."""
2500
t = self.get_transport()
2501
# From http://pad.lv/799212
2502
unicode_user = u'b\N{Euro Sign}ar'
2503
unicode_content = u'user=%s' % (unicode_user,)
2504
utf8_content = unicode_content.encode('utf8')
2505
# Store the raw content in the config file
2506
t.put_bytes('foo.conf', utf8_content)
2507
store = config.TransportIniFileStore(t, 'foo.conf')
2509
stack = config.Stack([store.get_sections], store)
2510
self.assertEqual(unicode_user, stack.get('user'))
2512
def test_load_non_ascii(self):
2513
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2514
t = self.get_transport()
2515
t.put_bytes('foo.conf', b'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2516
store = config.TransportIniFileStore(t, 'foo.conf')
2517
self.assertRaises(config.ConfigContentError, store.load)
2519
def test_load_erroneous_content(self):
2520
"""Ensure we display a proper error on content that can't be parsed."""
2521
t = self.get_transport()
2522
t.put_bytes('foo.conf', b'[open_section\n')
2523
store = config.TransportIniFileStore(t, 'foo.conf')
2524
self.assertRaises(config.ParseConfigError, store.load)
2526
def test_load_permission_denied(self):
2527
"""Ensure we get warned when trying to load an inaccessible file."""
2531
warnings.append(args[0] % args[1:])
2532
self.overrideAttr(trace, 'warning', warning)
2534
t = self.get_transport()
2536
def get_bytes(relpath):
2537
raise errors.PermissionDenied(relpath, "")
2538
t.get_bytes = get_bytes
2539
store = config.TransportIniFileStore(t, 'foo.conf')
2540
self.assertRaises(errors.PermissionDenied, store.load)
2543
[u'Permission denied while trying to load configuration store %s.'
2544
% store.external_url()])
2547
class TestIniConfigContent(tests.TestCaseWithTransport):
2548
"""Simulate loading a IniBasedConfig with content of various encodings.
2550
All files produced by bzr are in utf8 content.
2552
Users may modify them manually and end up with a file that can't be
2553
loaded. We need to issue proper error messages in this case.
2556
invalid_utf8_char = b'\xff'
2558
def test_load_utf8(self):
2559
"""Ensure we can load an utf8-encoded file."""
2560
# From http://pad.lv/799212
2561
unicode_user = u'b\N{Euro Sign}ar'
2562
unicode_content = u'user=%s' % (unicode_user,)
2563
utf8_content = unicode_content.encode('utf8')
2564
# Store the raw content in the config file
2565
with open('foo.conf', 'wb') as f:
2566
f.write(utf8_content)
2567
conf = config.IniBasedConfig(file_name='foo.conf')
2568
self.assertEqual(unicode_user, conf.get_user_option('user'))
2570
def test_load_badly_encoded_content(self):
2571
"""Ensure we display a proper error on non-ascii, non utf-8 content."""
2572
with open('foo.conf', 'wb') as f:
2573
f.write(b'user=foo\n#%s\n' % (self.invalid_utf8_char,))
2574
conf = config.IniBasedConfig(file_name='foo.conf')
2575
self.assertRaises(config.ConfigContentError, conf._get_parser)
2577
def test_load_erroneous_content(self):
2578
"""Ensure we display a proper error on content that can't be parsed."""
2579
with open('foo.conf', 'wb') as f:
2580
f.write(b'[open_section\n')
2581
conf = config.IniBasedConfig(file_name='foo.conf')
2582
self.assertRaises(config.ParseConfigError, conf._get_parser)
2585
class TestMutableStore(TestStore):
2587
scenarios = [(key, {'store_id': key, 'get_store': builder}) for key, builder
2588
in config.test_store_builder_registry.iteritems()]
2591
super(TestMutableStore, self).setUp()
2592
self.transport = self.get_transport()
2594
def has_store(self, store):
2595
store_basename = urlutils.relative_url(self.transport.external_url(),
2596
store.external_url())
2597
return self.transport.has(store_basename)
2599
def test_save_empty_creates_no_file(self):
2600
# FIXME: There should be a better way than relying on the test
2601
# parametrization to identify branch.conf -- vila 2011-0526
2602
if self.store_id in ('branch', 'remote_branch'):
2603
raise tests.TestNotApplicable(
2604
'branch.conf is *always* created when a branch is initialized')
2605
store = self.get_store(self)
2607
self.assertEqual(False, self.has_store(store))
2609
def test_mutable_section_shared(self):
2610
store = self.get_store(self)
2611
store._load_from_string(b'foo=bar\n')
2612
# FIXME: There should be a better way than relying on the test
2613
# parametrization to identify branch.conf -- vila 2011-0526
2614
if self.store_id in ('branch', 'remote_branch'):
2615
# branch stores requires write locked branches
2616
self.addCleanup(store.branch.lock_write().unlock)
2617
section1 = store.get_mutable_section(None)
2618
section2 = store.get_mutable_section(None)
2619
# If we get different sections, different callers won't share the
2621
self.assertIs(section1, section2)
2623
def test_save_emptied_succeeds(self):
2624
store = self.get_store(self)
2625
store._load_from_string(b'foo=bar\n')
2626
# FIXME: There should be a better way than relying on the test
2627
# parametrization to identify branch.conf -- vila 2011-0526
2628
if self.store_id in ('branch', 'remote_branch'):
2629
# branch stores requires write locked branches
2630
self.addCleanup(store.branch.lock_write().unlock)
2631
section = store.get_mutable_section(None)
2632
section.remove('foo')
2634
self.assertEqual(True, self.has_store(store))
2635
modified_store = self.get_store(self)
2636
sections = list(modified_store.get_sections())
2637
self.assertLength(0, sections)
2639
def test_save_with_content_succeeds(self):
2640
# FIXME: There should be a better way than relying on the test
2641
# parametrization to identify branch.conf -- vila 2011-0526
2642
if self.store_id in ('branch', 'remote_branch'):
2643
raise tests.TestNotApplicable(
2644
'branch.conf is *always* created when a branch is initialized')
2645
store = self.get_store(self)
2646
store._load_from_string(b'foo=bar\n')
2647
self.assertEqual(False, self.has_store(store))
2649
self.assertEqual(True, self.has_store(store))
2650
modified_store = self.get_store(self)
2651
sections = list(modified_store.get_sections())
2652
self.assertLength(1, sections)
2653
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2655
def test_set_option_in_empty_store(self):
2656
store = self.get_store(self)
2657
# FIXME: There should be a better way than relying on the test
2658
# parametrization to identify branch.conf -- vila 2011-0526
2659
if self.store_id in ('branch', 'remote_branch'):
2660
# branch stores requires write locked branches
2661
self.addCleanup(store.branch.lock_write().unlock)
2662
section = store.get_mutable_section(None)
2663
section.set('foo', 'bar')
2665
modified_store = self.get_store(self)
2666
sections = list(modified_store.get_sections())
2667
self.assertLength(1, sections)
2668
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2670
def test_set_option_in_default_section(self):
2671
store = self.get_store(self)
2672
store._load_from_string(b'')
2673
# FIXME: There should be a better way than relying on the test
2674
# parametrization to identify branch.conf -- vila 2011-0526
2675
if self.store_id in ('branch', 'remote_branch'):
2676
# branch stores requires write locked branches
2677
self.addCleanup(store.branch.lock_write().unlock)
2678
section = store.get_mutable_section(None)
2679
section.set('foo', 'bar')
2681
modified_store = self.get_store(self)
2682
sections = list(modified_store.get_sections())
2683
self.assertLength(1, sections)
2684
self.assertSectionContent((None, {'foo': 'bar'}), sections[0])
2686
def test_set_option_in_named_section(self):
2687
store = self.get_store(self)
2688
store._load_from_string(b'')
2689
# FIXME: There should be a better way than relying on the test
2690
# parametrization to identify branch.conf -- vila 2011-0526
2691
if self.store_id in ('branch', 'remote_branch'):
2692
# branch stores requires write locked branches
2693
self.addCleanup(store.branch.lock_write().unlock)
2694
section = store.get_mutable_section('baz')
2695
section.set('foo', 'bar')
2697
modified_store = self.get_store(self)
2698
sections = list(modified_store.get_sections())
2699
self.assertLength(1, sections)
2700
self.assertSectionContent(('baz', {'foo': 'bar'}), sections[0])
2702
def test_load_hook(self):
2703
# First, we need to ensure that the store exists
2704
store = self.get_store(self)
2705
# FIXME: There should be a better way than relying on the test
2706
# parametrization to identify branch.conf -- vila 2011-0526
2707
if self.store_id in ('branch', 'remote_branch'):
2708
# branch stores requires write locked branches
2709
self.addCleanup(store.branch.lock_write().unlock)
2710
section = store.get_mutable_section('baz')
2711
section.set('foo', 'bar')
2713
# Now we can try to load it
2714
store = self.get_store(self)
2719
config.ConfigHooks.install_named_hook('load', hook, None)
2720
self.assertLength(0, calls)
2722
self.assertLength(1, calls)
2723
self.assertEqual((store,), calls[0])
2725
def test_save_hook(self):
2730
config.ConfigHooks.install_named_hook('save', hook, None)
2731
self.assertLength(0, calls)
2732
store = self.get_store(self)
2733
# FIXME: There should be a better way than relying on the test
2734
# parametrization to identify branch.conf -- vila 2011-0526
2735
if self.store_id in ('branch', 'remote_branch'):
2736
# branch stores requires write locked branches
2737
self.addCleanup(store.branch.lock_write().unlock)
2738
section = store.get_mutable_section('baz')
2739
section.set('foo', 'bar')
2741
self.assertLength(1, calls)
2742
self.assertEqual((store,), calls[0])
2744
def test_set_mark_dirty(self):
2745
stack = config.MemoryStack(b'')
2746
self.assertLength(0, stack.store.dirty_sections)
2747
stack.set('foo', 'baz')
2748
self.assertLength(1, stack.store.dirty_sections)
2749
self.assertTrue(stack.store._need_saving())
2751
def test_remove_mark_dirty(self):
2752
stack = config.MemoryStack(b'foo=bar')
2753
self.assertLength(0, stack.store.dirty_sections)
2755
self.assertLength(1, stack.store.dirty_sections)
2756
self.assertTrue(stack.store._need_saving())
2759
class TestStoreSaveChanges(tests.TestCaseWithTransport):
2760
"""Tests that config changes are kept in memory and saved on-demand."""
2763
super(TestStoreSaveChanges, self).setUp()
2764
self.transport = self.get_transport()
2765
# Most of the tests involve two stores pointing to the same persistent
2766
# storage to observe the effects of concurrent changes
2767
self.st1 = config.TransportIniFileStore(self.transport, 'foo.conf')
2768
self.st2 = config.TransportIniFileStore(self.transport, 'foo.conf')
2772
self.warnings.append(args[0] % args[1:])
2773
self.overrideAttr(trace, 'warning', warning)
2775
def has_store(self, store):
2776
store_basename = urlutils.relative_url(self.transport.external_url(),
2777
store.external_url())
2778
return self.transport.has(store_basename)
2780
def get_stack(self, store):
2781
# Any stack will do as long as it uses the right store, just a single
2782
# no-name section is enough
2783
return config.Stack([store.get_sections], store)
2785
def test_no_changes_no_save(self):
2786
s = self.get_stack(self.st1)
2787
s.store.save_changes()
2788
self.assertEqual(False, self.has_store(self.st1))
2790
def test_unrelated_concurrent_update(self):
2791
s1 = self.get_stack(self.st1)
2792
s2 = self.get_stack(self.st2)
2793
s1.set('foo', 'bar')
2794
s2.set('baz', 'quux')
2796
# Changes don't propagate magically
2797
self.assertEqual(None, s1.get('baz'))
2798
s2.store.save_changes()
2799
self.assertEqual('quux', s2.get('baz'))
2800
# Changes are acquired when saving
2801
self.assertEqual('bar', s2.get('foo'))
2802
# Since there is no overlap, no warnings are emitted
2803
self.assertLength(0, self.warnings)
2805
def test_concurrent_update_modified(self):
2806
s1 = self.get_stack(self.st1)
2807
s2 = self.get_stack(self.st2)
2808
s1.set('foo', 'bar')
2809
s2.set('foo', 'baz')
2812
s2.store.save_changes()
2813
self.assertEqual('baz', s2.get('foo'))
2814
# But the user get a warning
2815
self.assertLength(1, self.warnings)
2816
warning = self.warnings[0]
2817
self.assertStartsWith(warning, 'Option foo in section None')
2818
self.assertEndsWith(warning, 'was changed from <CREATED> to bar.'
2819
' The baz value will be saved.')
2821
def test_concurrent_deletion(self):
2822
self.st1._load_from_string(b'foo=bar')
2824
s1 = self.get_stack(self.st1)
2825
s2 = self.get_stack(self.st2)
2828
s1.store.save_changes()
2830
self.assertLength(0, self.warnings)
2831
s2.store.save_changes()
2833
self.assertLength(1, self.warnings)
2834
warning = self.warnings[0]
2835
self.assertStartsWith(warning, 'Option foo in section None')
2836
self.assertEndsWith(warning, 'was changed from bar to <CREATED>.'
2837
' The <DELETED> value will be saved.')
2840
class TestQuotingIniFileStore(tests.TestCaseWithTransport):
2842
def get_store(self):
2843
return config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2845
def test_get_quoted_string(self):
2846
store = self.get_store()
2847
store._load_from_string(b'foo= " abc "')
2848
stack = config.Stack([store.get_sections])
2849
self.assertEqual(' abc ', stack.get('foo'))
2851
def test_set_quoted_string(self):
2852
store = self.get_store()
2853
stack = config.Stack([store.get_sections], store)
2854
stack.set('foo', ' a b c ')
2856
self.assertFileEqual(b'foo = " a b c "' +
2857
os.linesep.encode('ascii'), 'foo.conf')
2860
class TestTransportIniFileStore(TestStore):
2862
def test_loading_unknown_file_fails(self):
2863
store = config.TransportIniFileStore(self.get_transport(),
2865
self.assertRaises(errors.NoSuchFile, store.load)
2867
def test_invalid_content(self):
2868
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2869
self.assertEqual(False, store.is_loaded())
2870
exc = self.assertRaises(
2871
config.ParseConfigError, store._load_from_string,
2872
b'this is invalid !')
2873
self.assertEndsWith(exc.filename, 'foo.conf')
2874
# And the load failed
2875
self.assertEqual(False, store.is_loaded())
2877
def test_get_embedded_sections(self):
2878
# A more complicated example (which also shows that section names and
2879
# option names share the same name space...)
2880
# FIXME: This should be fixed by forbidding dicts as values ?
2881
# -- vila 2011-04-05
2882
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
2883
store._load_from_string(b'''
2887
foo_in_DEFAULT=foo_DEFAULT
2895
sections = list(store.get_sections())
2896
self.assertLength(4, sections)
2897
# The default section has no name.
2898
# List values are provided as strings and need to be explicitly
2899
# converted by specifying from_unicode=list_from_store at option
2901
self.assertSectionContent((None, {'foo': 'bar', 'l': u'1,2'}),
2903
self.assertSectionContent(
2904
('DEFAULT', {'foo_in_DEFAULT': 'foo_DEFAULT'}), sections[1])
2905
self.assertSectionContent(
2906
('bar', {'foo_in_bar': 'barbar'}), sections[2])
2907
# sub sections are provided as embedded dicts.
2908
self.assertSectionContent(
2909
('baz', {'foo_in_baz': 'barbaz', 'qux': {'foo_in_qux': 'quux'}}),
2913
class TestLockableIniFileStore(TestStore):
2915
def test_create_store_in_created_dir(self):
2916
self.assertPathDoesNotExist('dir')
2917
t = self.get_transport('dir/subdir')
2918
store = config.LockableIniFileStore(t, 'foo.conf')
2919
store.get_mutable_section(None).set('foo', 'bar')
2921
self.assertPathExists('dir/subdir')
2924
class TestConcurrentStoreUpdates(TestStore):
2925
"""Test that Stores properly handle conccurent updates.
2927
New Store implementation may fail some of these tests but until such
2928
implementations exist it's hard to properly filter them from the scenarios
2929
applied here. If you encounter such a case, contact the bzr devs.
2932
scenarios = [(key, {'get_stack': builder}) for key, builder
2933
in config.test_stack_builder_registry.iteritems()]
2936
super(TestConcurrentStoreUpdates, self).setUp()
2937
self.stack = self.get_stack(self)
2938
if not isinstance(self.stack, config._CompatibleStack):
2939
raise tests.TestNotApplicable(
2940
'%s is not meant to be compatible with the old config design'
2942
self.stack.set('one', '1')
2943
self.stack.set('two', '2')
2945
self.stack.store.save()
2947
def test_simple_read_access(self):
2948
self.assertEqual('1', self.stack.get('one'))
2950
def test_simple_write_access(self):
2951
self.stack.set('one', 'one')
2952
self.assertEqual('one', self.stack.get('one'))
2954
def test_listen_to_the_last_speaker(self):
2956
c2 = self.get_stack(self)
2957
c1.set('one', 'ONE')
2958
c2.set('two', 'TWO')
2959
self.assertEqual('ONE', c1.get('one'))
2960
self.assertEqual('TWO', c2.get('two'))
2961
# The second update respect the first one
2962
self.assertEqual('ONE', c2.get('one'))
2964
def test_last_speaker_wins(self):
2965
# If the same config is not shared, the same variable modified twice
2966
# can only see a single result.
2968
c2 = self.get_stack(self)
2971
self.assertEqual('c2', c2.get('one'))
2972
# The first modification is still available until another refresh
2974
self.assertEqual('c1', c1.get('one'))
2975
c1.set('two', 'done')
2976
self.assertEqual('c2', c1.get('one'))
2978
def test_writes_are_serialized(self):
2980
c2 = self.get_stack(self)
2982
# We spawn a thread that will pause *during* the config saving.
2983
before_writing = threading.Event()
2984
after_writing = threading.Event()
2985
writing_done = threading.Event()
2986
c1_save_without_locking_orig = c1.store.save_without_locking
2988
def c1_save_without_locking():
2989
before_writing.set()
2990
c1_save_without_locking_orig()
2991
# The lock is held. We wait for the main thread to decide when to
2993
after_writing.wait()
2994
c1.store.save_without_locking = c1_save_without_locking
2999
t1 = threading.Thread(target=c1_set)
3000
# Collect the thread after the test
3001
self.addCleanup(t1.join)
3002
# Be ready to unblock the thread if the test goes wrong
3003
self.addCleanup(after_writing.set)
3005
before_writing.wait()
3006
self.assertRaises(errors.LockContention,
3007
c2.set, 'one', 'c2')
3008
self.assertEqual('c1', c1.get('one'))
3009
# Let the lock be released
3013
self.assertEqual('c2', c2.get('one'))
3015
def test_read_while_writing(self):
3017
# We spawn a thread that will pause *during* the write
3018
ready_to_write = threading.Event()
3019
do_writing = threading.Event()
3020
writing_done = threading.Event()
3021
# We override the _save implementation so we know the store is locked
3022
c1_save_without_locking_orig = c1.store.save_without_locking
3024
def c1_save_without_locking():
3025
ready_to_write.set()
3026
# The lock is held. We wait for the main thread to decide when to
3029
c1_save_without_locking_orig()
3031
c1.store.save_without_locking = c1_save_without_locking
3035
t1 = threading.Thread(target=c1_set)
3036
# Collect the thread after the test
3037
self.addCleanup(t1.join)
3038
# Be ready to unblock the thread if the test goes wrong
3039
self.addCleanup(do_writing.set)
3041
# Ensure the thread is ready to write
3042
ready_to_write.wait()
3043
self.assertEqual('c1', c1.get('one'))
3044
# If we read during the write, we get the old value
3045
c2 = self.get_stack(self)
3046
self.assertEqual('1', c2.get('one'))
3047
# Let the writing occur and ensure it occurred
3050
# Now we get the updated value
3051
c3 = self.get_stack(self)
3052
self.assertEqual('c1', c3.get('one'))
3054
# FIXME: It may be worth looking into removing the lock dir when it's not
3055
# needed anymore and look at possible fallouts for concurrent lockers. This
3056
# will matter if/when we use config files outside of breezy directories
3057
# (.config/breezy or .bzr) -- vila 20110-04-111
3060
class TestSectionMatcher(TestStore):
3062
scenarios = [('location', {'matcher': config.LocationMatcher}),
3063
('id', {'matcher': config.NameMatcher}), ]
3066
super(TestSectionMatcher, self).setUp()
3067
# Any simple store is good enough
3068
self.get_store = config.test_store_builder_registry.get('configobj')
3070
def test_no_matches_for_empty_stores(self):
3071
store = self.get_store(self)
3072
store._load_from_string(b'')
3073
matcher = self.matcher(store, '/bar')
3074
self.assertEqual([], list(matcher.get_sections()))
3076
def test_build_doesnt_load_store(self):
3077
store = self.get_store(self)
3078
self.matcher(store, '/bar')
3079
self.assertFalse(store.is_loaded())
3082
class TestLocationSection(tests.TestCase):
3084
def get_section(self, options, extra_path):
3085
section = config.Section('foo', options)
3086
return config.LocationSection(section, extra_path)
3088
def test_simple_option(self):
3089
section = self.get_section({'foo': 'bar'}, '')
3090
self.assertEqual('bar', section.get('foo'))
3092
def test_option_with_extra_path(self):
3093
section = self.get_section({'foo': 'bar', 'foo:policy': 'appendpath'},
3095
self.assertEqual('bar/baz', section.get('foo'))
3097
def test_invalid_policy(self):
3098
section = self.get_section({'foo': 'bar', 'foo:policy': 'die'},
3100
# invalid policies are ignored
3101
self.assertEqual('bar', section.get('foo'))
3104
class TestLocationMatcher(TestStore):
3107
super(TestLocationMatcher, self).setUp()
3108
# Any simple store is good enough
3109
self.get_store = config.test_store_builder_registry.get('configobj')
3111
def test_unrelated_section_excluded(self):
3112
store = self.get_store(self)
3113
store._load_from_string(b'''
3121
section=/foo/bar/baz
3125
self.assertEqual(['/foo', '/foo/baz', '/foo/bar', '/foo/bar/baz',
3127
[section.id for _, section in store.get_sections()])
3128
matcher = config.LocationMatcher(store, '/foo/bar/quux')
3129
sections = [section for _, section in matcher.get_sections()]
3130
self.assertEqual(['/foo/bar', '/foo'],
3131
[section.id for section in sections])
3132
self.assertEqual(['quux', 'bar/quux'],
3133
[section.extra_path for section in sections])
3135
def test_more_specific_sections_first(self):
3136
store = self.get_store(self)
3137
store._load_from_string(b'''
3143
self.assertEqual(['/foo', '/foo/bar'],
3144
[section.id for _, section in store.get_sections()])
3145
matcher = config.LocationMatcher(store, '/foo/bar/baz')
3146
sections = [section for _, section in matcher.get_sections()]
3147
self.assertEqual(['/foo/bar', '/foo'],
3148
[section.id for section in sections])
3149
self.assertEqual(['baz', 'bar/baz'],
3150
[section.extra_path for section in sections])
3152
def test_appendpath_in_no_name_section(self):
3153
# It's a bit weird to allow appendpath in a no-name section, but
3154
# someone may found a use for it
3155
store = self.get_store(self)
3156
store._load_from_string(b'''
3158
foo:policy = appendpath
3160
matcher = config.LocationMatcher(store, 'dir/subdir')
3161
sections = list(matcher.get_sections())
3162
self.assertLength(1, sections)
3163
self.assertEqual('bar/dir/subdir', sections[0][1].get('foo'))
3165
def test_file_urls_are_normalized(self):
3166
store = self.get_store(self)
3167
if sys.platform == 'win32':
3168
expected_url = 'file:///C:/dir/subdir'
3169
expected_location = 'C:/dir/subdir'
3171
expected_url = 'file:///dir/subdir'
3172
expected_location = '/dir/subdir'
3173
matcher = config.LocationMatcher(store, expected_url)
3174
self.assertEqual(expected_location, matcher.location)
3176
def test_branch_name_colo(self):
3177
store = self.get_store(self)
3178
store._load_from_string(dedent("""\
3180
push_location=my{branchname}
3181
""").encode('ascii'))
3182
matcher = config.LocationMatcher(store, 'file:///,branch=example%3c')
3183
self.assertEqual('example<', matcher.branch_name)
3184
((_, section),) = matcher.get_sections()
3185
self.assertEqual('example<', section.locals['branchname'])
3187
def test_branch_name_basename(self):
3188
store = self.get_store(self)
3189
store._load_from_string(dedent("""\
3191
push_location=my{branchname}
3192
""").encode('ascii'))
3193
matcher = config.LocationMatcher(store, 'file:///parent/example%3c')
3194
self.assertEqual('example<', matcher.branch_name)
3195
((_, section),) = matcher.get_sections()
3196
self.assertEqual('example<', section.locals['branchname'])
3199
class TestStartingPathMatcher(TestStore):
3202
super(TestStartingPathMatcher, self).setUp()
3203
# Any simple store is good enough
3204
self.store = config.IniFileStore()
3206
def assertSectionIDs(self, expected, location, content):
3207
self.store._load_from_string(content)
3208
matcher = config.StartingPathMatcher(self.store, location)
3209
sections = list(matcher.get_sections())
3210
self.assertLength(len(expected), sections)
3211
self.assertEqual(expected, [section.id for _, section in sections])
3214
def test_empty(self):
3215
self.assertSectionIDs([], self.get_url(), b'')
3217
def test_url_vs_local_paths(self):
3218
# The matcher location is an url and the section names are local paths
3219
self.assertSectionIDs(['/foo/bar', '/foo'],
3220
'file:///foo/bar/baz', b'''\
3225
def test_local_path_vs_url(self):
3226
# The matcher location is a local path and the section names are urls
3227
self.assertSectionIDs(['file:///foo/bar', 'file:///foo'],
3228
'/foo/bar/baz', b'''\
3233
def test_no_name_section_included_when_present(self):
3234
# Note that other tests will cover the case where the no-name section
3235
# is empty and as such, not included.
3236
sections = self.assertSectionIDs(['/foo/bar', '/foo', None],
3237
'/foo/bar/baz', b'''\
3238
option = defined so the no-name section exists
3242
self.assertEqual(['baz', 'bar/baz', '/foo/bar/baz'],
3243
[s.locals['relpath'] for _, s in sections])
3245
def test_order_reversed(self):
3246
self.assertSectionIDs(['/foo/bar', '/foo'], '/foo/bar/baz', b'''\
3251
def test_unrelated_section_excluded(self):
3252
self.assertSectionIDs(['/foo/bar', '/foo'], '/foo/bar/baz', b'''\
3258
def test_glob_included(self):
3259
sections = self.assertSectionIDs(['/foo/*/baz', '/foo/b*', '/foo'],
3260
'/foo/bar/baz', b'''\
3266
# Note that 'baz' as a relpath for /foo/b* is not fully correct, but
3267
# nothing really is... as far using {relpath} to append it to something
3268
# else, this seems good enough though.
3269
self.assertEqual(['', 'baz', 'bar/baz'],
3270
[s.locals['relpath'] for _, s in sections])
3272
def test_respect_order(self):
3273
self.assertSectionIDs(['/foo', '/foo/b*', '/foo/*/baz'],
3274
'/foo/bar/baz', b'''\
3282
class TestNameMatcher(TestStore):
3285
super(TestNameMatcher, self).setUp()
3286
self.matcher = config.NameMatcher
3287
# Any simple store is good enough
3288
self.get_store = config.test_store_builder_registry.get('configobj')
3290
def get_matching_sections(self, name):
3291
store = self.get_store(self)
3292
store._load_from_string(b'''
3300
matcher = self.matcher(store, name)
3301
return list(matcher.get_sections())
3303
def test_matching(self):
3304
sections = self.get_matching_sections('foo')
3305
self.assertLength(1, sections)
3306
self.assertSectionContent(('foo', {'option': 'foo'}), sections[0])
3308
def test_not_matching(self):
3309
sections = self.get_matching_sections('baz')
3310
self.assertLength(0, sections)
3313
class TestBaseStackGet(tests.TestCase):
3316
super(TestBaseStackGet, self).setUp()
3317
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3319
def test_get_first_definition(self):
3320
store1 = config.IniFileStore()
3321
store1._load_from_string(b'foo=bar')
3322
store2 = config.IniFileStore()
3323
store2._load_from_string(b'foo=baz')
3324
conf = config.Stack([store1.get_sections, store2.get_sections])
3325
self.assertEqual('bar', conf.get('foo'))
3327
def test_get_with_registered_default_value(self):
3328
config.option_registry.register(config.Option('foo', default='bar'))
3329
conf_stack = config.Stack([])
3330
self.assertEqual('bar', conf_stack.get('foo'))
3332
def test_get_without_registered_default_value(self):
3333
config.option_registry.register(config.Option('foo'))
3334
conf_stack = config.Stack([])
3335
self.assertEqual(None, conf_stack.get('foo'))
3337
def test_get_without_default_value_for_not_registered(self):
3338
conf_stack = config.Stack([])
3339
self.assertEqual(None, conf_stack.get('foo'))
3341
def test_get_for_empty_section_callable(self):
3342
conf_stack = config.Stack([lambda: []])
3343
self.assertEqual(None, conf_stack.get('foo'))
3345
def test_get_for_broken_callable(self):
3346
# Trying to use and invalid callable raises an exception on first use
3347
conf_stack = config.Stack([object])
3348
self.assertRaises(TypeError, conf_stack.get, 'foo')
3351
class TestStackWithSimpleStore(tests.TestCase):
3354
super(TestStackWithSimpleStore, self).setUp()
3355
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3356
self.registry = config.option_registry
3358
def get_conf(self, content=None):
3359
return config.MemoryStack(content)
3361
def test_override_value_from_env(self):
3362
self.overrideEnv('FOO', None)
3363
self.registry.register(
3364
config.Option('foo', default='bar', override_from_env=['FOO']))
3365
self.overrideEnv('FOO', 'quux')
3366
# Env variable provides a default taking over the option one
3367
conf = self.get_conf(b'foo=store')
3368
self.assertEqual('quux', conf.get('foo'))
3370
def test_first_override_value_from_env_wins(self):
3371
self.overrideEnv('NO_VALUE', None)
3372
self.overrideEnv('FOO', None)
3373
self.overrideEnv('BAZ', None)
3374
self.registry.register(
3375
config.Option('foo', default='bar',
3376
override_from_env=['NO_VALUE', 'FOO', 'BAZ']))
3377
self.overrideEnv('FOO', 'foo')
3378
self.overrideEnv('BAZ', 'baz')
3379
# The first env var set wins
3380
conf = self.get_conf(b'foo=store')
3381
self.assertEqual('foo', conf.get('foo'))
3384
class TestMemoryStack(tests.TestCase):
3387
conf = config.MemoryStack(b'foo=bar')
3388
self.assertEqual('bar', conf.get('foo'))
3391
conf = config.MemoryStack(b'foo=bar')
3392
conf.set('foo', 'baz')
3393
self.assertEqual('baz', conf.get('foo'))
3395
def test_no_content(self):
3396
conf = config.MemoryStack()
3397
# No content means no loading
3398
self.assertFalse(conf.store.is_loaded())
3399
self.assertRaises(NotImplementedError, conf.get, 'foo')
3400
# But a content can still be provided
3401
conf.store._load_from_string(b'foo=bar')
3402
self.assertEqual('bar', conf.get('foo'))
3405
class TestStackIterSections(tests.TestCase):
3407
def test_empty_stack(self):
3408
conf = config.Stack([])
3409
sections = list(conf.iter_sections())
3410
self.assertLength(0, sections)
3412
def test_empty_store(self):
3413
store = config.IniFileStore()
3414
store._load_from_string(b'')
3415
conf = config.Stack([store.get_sections])
3416
sections = list(conf.iter_sections())
3417
self.assertLength(0, sections)
3419
def test_simple_store(self):
3420
store = config.IniFileStore()
3421
store._load_from_string(b'foo=bar')
3422
conf = config.Stack([store.get_sections])
3423
tuples = list(conf.iter_sections())
3424
self.assertLength(1, tuples)
3425
(found_store, found_section) = tuples[0]
3426
self.assertIs(store, found_store)
3428
def test_two_stores(self):
3429
store1 = config.IniFileStore()
3430
store1._load_from_string(b'foo=bar')
3431
store2 = config.IniFileStore()
3432
store2._load_from_string(b'bar=qux')
3433
conf = config.Stack([store1.get_sections, store2.get_sections])
3434
tuples = list(conf.iter_sections())
3435
self.assertLength(2, tuples)
3436
self.assertIs(store1, tuples[0][0])
3437
self.assertIs(store2, tuples[1][0])
3440
class TestStackWithTransport(tests.TestCaseWithTransport):
3442
scenarios = [(key, {'get_stack': builder}) for key, builder
3443
in config.test_stack_builder_registry.iteritems()]
3446
class TestConcreteStacks(TestStackWithTransport):
3448
def test_build_stack(self):
3449
# Just a smoke test to help debug builders
3450
self.get_stack(self)
3453
class TestStackGet(TestStackWithTransport):
3456
super(TestStackGet, self).setUp()
3457
self.conf = self.get_stack(self)
3459
def test_get_for_empty_stack(self):
3460
self.assertEqual(None, self.conf.get('foo'))
3462
def test_get_hook(self):
3463
self.conf.set('foo', 'bar')
3468
config.ConfigHooks.install_named_hook('get', hook, None)
3469
self.assertLength(0, calls)
3470
value = self.conf.get('foo')
3471
self.assertEqual('bar', value)
3472
self.assertLength(1, calls)
3473
self.assertEqual((self.conf, 'foo', 'bar'), calls[0])
3476
class TestStackGetWithConverter(tests.TestCase):
3479
super(TestStackGetWithConverter, self).setUp()
3480
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3481
self.registry = config.option_registry
3483
def get_conf(self, content=None):
3484
return config.MemoryStack(content)
3486
def register_bool_option(self, name, default=None, default_from_env=None):
3487
b = config.Option(name, help='A boolean.',
3488
default=default, default_from_env=default_from_env,
3489
from_unicode=config.bool_from_store)
3490
self.registry.register(b)
3492
def test_get_default_bool_None(self):
3493
self.register_bool_option('foo')
3494
conf = self.get_conf(b'')
3495
self.assertEqual(None, conf.get('foo'))
3497
def test_get_default_bool_True(self):
3498
self.register_bool_option('foo', u'True')
3499
conf = self.get_conf(b'')
3500
self.assertEqual(True, conf.get('foo'))
3502
def test_get_default_bool_False(self):
3503
self.register_bool_option('foo', False)
3504
conf = self.get_conf(b'')
3505
self.assertEqual(False, conf.get('foo'))
3507
def test_get_default_bool_False_as_string(self):
3508
self.register_bool_option('foo', u'False')
3509
conf = self.get_conf(b'')
3510
self.assertEqual(False, conf.get('foo'))
3512
def test_get_default_bool_from_env_converted(self):
3513
self.register_bool_option('foo', u'True', default_from_env=['FOO'])
3514
self.overrideEnv('FOO', 'False')
3515
conf = self.get_conf(b'')
3516
self.assertEqual(False, conf.get('foo'))
3518
def test_get_default_bool_when_conversion_fails(self):
3519
self.register_bool_option('foo', default='True')
3520
conf = self.get_conf(b'foo=invalid boolean')
3521
self.assertEqual(True, conf.get('foo'))
3523
def register_integer_option(self, name,
3524
default=None, default_from_env=None):
3525
i = config.Option(name, help='An integer.',
3526
default=default, default_from_env=default_from_env,
3527
from_unicode=config.int_from_store)
3528
self.registry.register(i)
3530
def test_get_default_integer_None(self):
3531
self.register_integer_option('foo')
3532
conf = self.get_conf(b'')
3533
self.assertEqual(None, conf.get('foo'))
3535
def test_get_default_integer(self):
3536
self.register_integer_option('foo', 42)
3537
conf = self.get_conf(b'')
3538
self.assertEqual(42, conf.get('foo'))
3540
def test_get_default_integer_as_string(self):
3541
self.register_integer_option('foo', u'42')
3542
conf = self.get_conf(b'')
3543
self.assertEqual(42, conf.get('foo'))
3545
def test_get_default_integer_from_env(self):
3546
self.register_integer_option('foo', default_from_env=['FOO'])
3547
self.overrideEnv('FOO', '18')
3548
conf = self.get_conf(b'')
3549
self.assertEqual(18, conf.get('foo'))
3551
def test_get_default_integer_when_conversion_fails(self):
3552
self.register_integer_option('foo', default='12')
3553
conf = self.get_conf(b'foo=invalid integer')
3554
self.assertEqual(12, conf.get('foo'))
3556
def register_list_option(self, name, default=None, default_from_env=None):
3557
l = config.ListOption(name, help='A list.', default=default,
3558
default_from_env=default_from_env)
3559
self.registry.register(l)
3561
def test_get_default_list_None(self):
3562
self.register_list_option('foo')
3563
conf = self.get_conf(b'')
3564
self.assertEqual(None, conf.get('foo'))
3566
def test_get_default_list_empty(self):
3567
self.register_list_option('foo', '')
3568
conf = self.get_conf(b'')
3569
self.assertEqual([], conf.get('foo'))
3571
def test_get_default_list_from_env(self):
3572
self.register_list_option('foo', default_from_env=['FOO'])
3573
self.overrideEnv('FOO', '')
3574
conf = self.get_conf(b'')
3575
self.assertEqual([], conf.get('foo'))
3577
def test_get_with_list_converter_no_item(self):
3578
self.register_list_option('foo', None)
3579
conf = self.get_conf(b'foo=,')
3580
self.assertEqual([], conf.get('foo'))
3582
def test_get_with_list_converter_many_items(self):
3583
self.register_list_option('foo', None)
3584
conf = self.get_conf(b'foo=m,o,r,e')
3585
self.assertEqual(['m', 'o', 'r', 'e'], conf.get('foo'))
3587
def test_get_with_list_converter_embedded_spaces_many_items(self):
3588
self.register_list_option('foo', None)
3589
conf = self.get_conf(b'foo=" bar", "baz "')
3590
self.assertEqual([' bar', 'baz '], conf.get('foo'))
3592
def test_get_with_list_converter_stripped_spaces_many_items(self):
3593
self.register_list_option('foo', None)
3594
conf = self.get_conf(b'foo= bar , baz ')
3595
self.assertEqual(['bar', 'baz'], conf.get('foo'))
3598
class TestIterOptionRefs(tests.TestCase):
3599
"""iter_option_refs is a bit unusual, document some cases."""
3601
def assertRefs(self, expected, string):
3602
self.assertEqual(expected, list(config.iter_option_refs(string)))
3604
def test_empty(self):
3605
self.assertRefs([(False, '')], '')
3607
def test_no_refs(self):
3608
self.assertRefs([(False, 'foo bar')], 'foo bar')
3610
def test_single_ref(self):
3611
self.assertRefs([(False, ''), (True, '{foo}'), (False, '')], '{foo}')
3613
def test_broken_ref(self):
3614
self.assertRefs([(False, '{foo')], '{foo')
3616
def test_embedded_ref(self):
3617
self.assertRefs([(False, '{'), (True, '{foo}'), (False, '}')],
3620
def test_two_refs(self):
3621
self.assertRefs([(False, ''), (True, '{foo}'),
3622
(False, ''), (True, '{bar}'),
3626
def test_newline_in_refs_are_not_matched(self):
3627
self.assertRefs([(False, '{\nxx}{xx\n}{{\n}}')], '{\nxx}{xx\n}{{\n}}')
3630
class TestStackExpandOptions(tests.TestCaseWithTransport):
3633
super(TestStackExpandOptions, self).setUp()
3634
self.overrideAttr(config, 'option_registry', config.OptionRegistry())
3635
self.registry = config.option_registry
3636
store = config.TransportIniFileStore(self.get_transport(), 'foo.conf')
3637
self.conf = config.Stack([store.get_sections], store)
3639
def assertExpansion(self, expected, string, env=None):
3640
self.assertEqual(expected, self.conf.expand_options(string, env))
3642
def test_no_expansion(self):
3643
self.assertExpansion('foo', 'foo')
3645
def test_expand_default_value(self):
3646
self.conf.store._load_from_string(b'bar=baz')
3647
self.registry.register(config.Option('foo', default=u'{bar}'))
3648
self.assertEqual('baz', self.conf.get('foo', expand=True))
3650
def test_expand_default_from_env(self):
3651
self.conf.store._load_from_string(b'bar=baz')
3652
self.registry.register(config.Option('foo', default_from_env=['FOO']))
3653
self.overrideEnv('FOO', '{bar}')
3654
self.assertEqual('baz', self.conf.get('foo', expand=True))
3656
def test_expand_default_on_failed_conversion(self):
3657
self.conf.store._load_from_string(b'baz=bogus\nbar=42\nfoo={baz}')
3658
self.registry.register(
3659
config.Option('foo', default=u'{bar}',
3660
from_unicode=config.int_from_store))
3661
self.assertEqual(42, self.conf.get('foo', expand=True))
3663
def test_env_adding_options(self):
3664
self.assertExpansion('bar', '{foo}', {'foo': 'bar'})
3666
def test_env_overriding_options(self):
3667
self.conf.store._load_from_string(b'foo=baz')
3668
self.assertExpansion('bar', '{foo}', {'foo': 'bar'})
3670
def test_simple_ref(self):
3671
self.conf.store._load_from_string(b'foo=xxx')
3672
self.assertExpansion('xxx', '{foo}')
3674
def test_unknown_ref(self):
3675
self.assertRaises(config.ExpandingUnknownOption,
3676
self.conf.expand_options, '{foo}')
3678
def test_illegal_def_is_ignored(self):
3679
self.assertExpansion('{1,2}', '{1,2}')
3680
self.assertExpansion('{ }', '{ }')
3681
self.assertExpansion('${Foo,f}', '${Foo,f}')
3683
def test_indirect_ref(self):
3684
self.conf.store._load_from_string(b'''
3688
self.assertExpansion('xxx', '{bar}')
3690
def test_embedded_ref(self):
3691
self.conf.store._load_from_string(b'''
3695
self.assertExpansion('xxx', '{{bar}}')
3697
def test_simple_loop(self):
3698
self.conf.store._load_from_string(b'foo={foo}')
3699
self.assertRaises(config.OptionExpansionLoop,
3700
self.conf.expand_options, '{foo}')
3702
def test_indirect_loop(self):
3703
self.conf.store._load_from_string(b'''
3707
e = self.assertRaises(config.OptionExpansionLoop,
3708
self.conf.expand_options, '{foo}')
3709
self.assertEqual('foo->bar->baz', e.refs)
3710
self.assertEqual('{foo}', e.string)
3712
def test_list(self):
3713
self.conf.store._load_from_string(b'''
3717
list={foo},{bar},{baz}
3719
self.registry.register(
3720
config.ListOption('list'))
3721
self.assertEqual(['start', 'middle', 'end'],
3722
self.conf.get('list', expand=True))
3724
def test_cascading_list(self):
3725
self.conf.store._load_from_string(b'''
3731
self.registry.register(config.ListOption('list'))
3732
# Register an intermediate option as a list to ensure no conversion
3733
# happen while expanding. Conversion should only occur for the original
3734
# option ('list' here).
3735
self.registry.register(config.ListOption('baz'))
3736
self.assertEqual(['start', 'middle', 'end'],
3737
self.conf.get('list', expand=True))
3739
def test_pathologically_hidden_list(self):
3740
self.conf.store._load_from_string(b'''
3746
hidden={start}{middle}{end}
3748
# What matters is what the registration says, the conversion happens
3749
# only after all expansions have been performed
3750
self.registry.register(config.ListOption('hidden'))
3751
self.assertEqual(['bin', 'go'],
3752
self.conf.get('hidden', expand=True))
3755
class TestStackCrossSectionsExpand(tests.TestCaseWithTransport):
3758
super(TestStackCrossSectionsExpand, self).setUp()
3760
def get_config(self, location, string):
3763
# Since we don't save the config we won't strictly require to inherit
3764
# from TestCaseInTempDir, but an error occurs so quickly...
3765
c = config.LocationStack(location)
3766
c.store._load_from_string(string)
3769
def test_dont_cross_unrelated_section(self):
3770
c = self.get_config('/another/branch/path', b'''
3775
[/another/branch/path]
3778
self.assertRaises(config.ExpandingUnknownOption,
3779
c.get, 'bar', expand=True)
3781
def test_cross_related_sections(self):
3782
c = self.get_config('/project/branch/path', b'''
3786
[/project/branch/path]
3789
self.assertEqual('quux', c.get('bar', expand=True))
3792
class TestStackCrossStoresExpand(tests.TestCaseWithTransport):
3794
def test_cross_global_locations(self):
3795
l_store = config.LocationStore()
3796
l_store._load_from_string(b'''
3802
g_store = config.GlobalStore()
3803
g_store._load_from_string(b'''
3809
stack = config.LocationStack('/branch')
3810
self.assertEqual('glob-bar', stack.get('lbar', expand=True))
3811
self.assertEqual('loc-foo', stack.get('gfoo', expand=True))
3814
class TestStackExpandSectionLocals(tests.TestCaseWithTransport):
3816
def test_expand_locals_empty(self):
3817
l_store = config.LocationStore()
3818
l_store._load_from_string(b'''
3819
[/home/user/project]
3824
stack = config.LocationStack('/home/user/project/')
3825
self.assertEqual('', stack.get('base', expand=True))
3826
self.assertEqual('', stack.get('rel', expand=True))
3828
def test_expand_basename_locally(self):
3829
l_store = config.LocationStore()
3830
l_store._load_from_string(b'''
3831
[/home/user/project]
3835
stack = config.LocationStack('/home/user/project/branch')
3836
self.assertEqual('branch', stack.get('bfoo', expand=True))
3838
def test_expand_basename_locally_longer_path(self):
3839
l_store = config.LocationStore()
3840
l_store._load_from_string(b'''
3845
stack = config.LocationStack('/home/user/project/dir/branch')
3846
self.assertEqual('branch', stack.get('bfoo', expand=True))
3848
def test_expand_relpath_locally(self):
3849
l_store = config.LocationStore()
3850
l_store._load_from_string(b'''
3851
[/home/user/project]
3852
lfoo = loc-foo/{relpath}
3855
stack = config.LocationStack('/home/user/project/branch')
3856
self.assertEqual('loc-foo/branch', stack.get('lfoo', expand=True))
3858
def test_expand_relpath_unknonw_in_global(self):
3859
g_store = config.GlobalStore()
3860
g_store._load_from_string(b'''
3865
stack = config.LocationStack('/home/user/project/branch')
3866
self.assertRaises(config.ExpandingUnknownOption,
3867
stack.get, 'gfoo', expand=True)
3869
def test_expand_local_option_locally(self):
3870
l_store = config.LocationStore()
3871
l_store._load_from_string(b'''
3872
[/home/user/project]
3873
lfoo = loc-foo/{relpath}
3877
g_store = config.GlobalStore()
3878
g_store._load_from_string(b'''
3884
stack = config.LocationStack('/home/user/project/branch')
3885
self.assertEqual('glob-bar', stack.get('lbar', expand=True))
3886
self.assertEqual('loc-foo/branch', stack.get('gfoo', expand=True))
3888
def test_locals_dont_leak(self):
3889
"""Make sure we chose the right local in presence of several sections.
3891
l_store = config.LocationStore()
3892
l_store._load_from_string(b'''
3894
lfoo = loc-foo/{relpath}
3895
[/home/user/project]
3896
lfoo = loc-foo/{relpath}
3899
stack = config.LocationStack('/home/user/project/branch')
3900
self.assertEqual('loc-foo/branch', stack.get('lfoo', expand=True))
3901
stack = config.LocationStack('/home/user/bar/baz')
3902
self.assertEqual('loc-foo/bar/baz', stack.get('lfoo', expand=True))
3905
class TestStackSet(TestStackWithTransport):
3907
def test_simple_set(self):
3908
conf = self.get_stack(self)
3909
self.assertEqual(None, conf.get('foo'))
3910
conf.set('foo', 'baz')
3911
# Did we get it back ?
3912
self.assertEqual('baz', conf.get('foo'))
3914
def test_set_creates_a_new_section(self):
3915
conf = self.get_stack(self)
3916
conf.set('foo', 'baz')
3917
self.assertEqual, 'baz', conf.get('foo')
3919
def test_set_hook(self):
3924
config.ConfigHooks.install_named_hook('set', hook, None)
3925
self.assertLength(0, calls)
3926
conf = self.get_stack(self)
3927
conf.set('foo', 'bar')
3928
self.assertLength(1, calls)
3929
self.assertEqual((conf, 'foo', 'bar'), calls[0])
3932
class TestStackRemove(TestStackWithTransport):
3934
def test_remove_existing(self):
3935
conf = self.get_stack(self)
3936
conf.set('foo', 'bar')
3937
self.assertEqual('bar', conf.get('foo'))
3939
# Did we get it back ?
3940
self.assertEqual(None, conf.get('foo'))
3942
def test_remove_unknown(self):
3943
conf = self.get_stack(self)
3944
self.assertRaises(KeyError, conf.remove, 'I_do_not_exist')
3946
def test_remove_hook(self):
3951
config.ConfigHooks.install_named_hook('remove', hook, None)
3952
self.assertLength(0, calls)
3953
conf = self.get_stack(self)
3954
conf.set('foo', 'bar')
3956
self.assertLength(1, calls)
3957
self.assertEqual((conf, 'foo'), calls[0])
3960
class TestConfigGetOptions(tests.TestCaseWithTransport, TestOptionsMixin):
3963
super(TestConfigGetOptions, self).setUp()
3964
create_configs(self)
3966
def test_no_variable(self):
3967
# Using branch should query branch, locations and breezy
3968
self.assertOptions([], self.branch_config)
3970
def test_option_in_breezy(self):
3971
self.breezy_config.set_user_option('file', 'breezy')
3972
self.assertOptions([('file', 'breezy', 'DEFAULT', 'breezy')],
3975
def test_option_in_locations(self):
3976
self.locations_config.set_user_option('file', 'locations')
3978
[('file', 'locations', self.tree.basedir, 'locations')],
3979
self.locations_config)
3981
def test_option_in_branch(self):
3982
self.branch_config.set_user_option('file', 'branch')
3983
self.assertOptions([('file', 'branch', 'DEFAULT', 'branch')],
3986
def test_option_in_breezy_and_branch(self):
3987
self.breezy_config.set_user_option('file', 'breezy')
3988
self.branch_config.set_user_option('file', 'branch')
3989
self.assertOptions([('file', 'branch', 'DEFAULT', 'branch'),
3990
('file', 'breezy', 'DEFAULT', 'breezy'), ],
3993
def test_option_in_branch_and_locations(self):
3994
# Hmm, locations override branch :-/
3995
self.locations_config.set_user_option('file', 'locations')
3996
self.branch_config.set_user_option('file', 'branch')
3998
[('file', 'locations', self.tree.basedir, 'locations'),
3999
('file', 'branch', 'DEFAULT', 'branch'), ],
4002
def test_option_in_breezy_locations_and_branch(self):
4003
self.breezy_config.set_user_option('file', 'breezy')
4004
self.locations_config.set_user_option('file', 'locations')
4005
self.branch_config.set_user_option('file', 'branch')
4007
[('file', 'locations', self.tree.basedir, 'locations'),
4008
('file', 'branch', 'DEFAULT', 'branch'),
4009
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4013
class TestConfigRemoveOption(tests.TestCaseWithTransport, TestOptionsMixin):
4016
super(TestConfigRemoveOption, self).setUp()
4017
create_configs_with_file_option(self)
4019
def test_remove_in_locations(self):
4020
self.locations_config.remove_user_option('file', self.tree.basedir)
4022
[('file', 'branch', 'DEFAULT', 'branch'),
4023
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4026
def test_remove_in_branch(self):
4027
self.branch_config.remove_user_option('file')
4029
[('file', 'locations', self.tree.basedir, 'locations'),
4030
('file', 'breezy', 'DEFAULT', 'breezy'), ],
4033
def test_remove_in_breezy(self):
4034
self.breezy_config.remove_user_option('file')
4036
[('file', 'locations', self.tree.basedir, 'locations'),
4037
('file', 'branch', 'DEFAULT', 'branch'), ],
4041
class TestConfigGetSections(tests.TestCaseWithTransport):
4044
super(TestConfigGetSections, self).setUp()
4045
create_configs(self)
4047
def assertSectionNames(self, expected, conf, name=None):
4048
"""Check which sections are returned for a given config.
4050
If fallback configurations exist their sections can be included.
4052
:param expected: A list of section names.
4054
:param conf: The configuration that will be queried.
4056
:param name: An optional section name that will be passed to
4059
sections = list(conf._get_sections(name))
4060
self.assertLength(len(expected), sections)
4061
self.assertEqual(expected, [n for n, _, _ in sections])
4063
def test_breezy_default_section(self):
4064
self.assertSectionNames(['DEFAULT'], self.breezy_config)
4066
def test_locations_default_section(self):
4067
# No sections are defined in an empty file
4068
self.assertSectionNames([], self.locations_config)
4070
def test_locations_named_section(self):
4071
self.locations_config.set_user_option('file', 'locations')
4072
self.assertSectionNames([self.tree.basedir], self.locations_config)
4074
def test_locations_matching_sections(self):
4075
loc_config = self.locations_config
4076
loc_config.set_user_option('file', 'locations')
4077
# We need to cheat a bit here to create an option in sections above and
4078
# below the 'location' one.
4079
parser = loc_config._get_parser()
4080
# locations.cong deals with '/' ignoring native os.sep
4081
location_names = self.tree.basedir.split('/')
4082
parent = '/'.join(location_names[:-1])
4083
child = '/'.join(location_names + ['child'])
4085
parser[parent]['file'] = 'parent'
4087
parser[child]['file'] = 'child'
4088
self.assertSectionNames([self.tree.basedir, parent], loc_config)
4090
def test_branch_data_default_section(self):
4091
self.assertSectionNames([None],
4092
self.branch_config._get_branch_data_config())
4094
def test_branch_default_sections(self):
4095
# No sections are defined in an empty locations file
4096
self.assertSectionNames([None, 'DEFAULT'],
4098
# Unless we define an option
4099
self.branch_config._get_location_config().set_user_option(
4100
'file', 'locations')
4101
self.assertSectionNames([self.tree.basedir, None, 'DEFAULT'],
4104
def test_breezy_named_section(self):
4105
# We need to cheat as the API doesn't give direct access to sections
4106
# other than DEFAULT.
4107
self.breezy_config.set_alias('breezy', 'bzr')
4108
self.assertSectionNames(['ALIASES'], self.breezy_config, 'ALIASES')
4111
class TestSharedStores(tests.TestCaseInTempDir):
4113
def test_breezy_conf_shared(self):
4114
g1 = config.GlobalStack()
4115
g2 = config.GlobalStack()
4116
# The two stacks share the same store
4117
self.assertIs(g1.store, g2.store)
4120
class TestAuthenticationConfigFilePermissions(tests.TestCaseInTempDir):
4121
"""Test warning for permissions of authentication.conf."""
4124
super(TestAuthenticationConfigFilePermissions, self).setUp()
4125
self.path = osutils.pathjoin(self.test_dir, 'authentication.conf')
4126
with open(self.path, 'wb') as f:
4127
f.write(b"""[broken]
4130
port=port # Error: Not an int
4132
self.overrideAttr(bedding, 'authentication_config_path',
4134
osutils.chmod_if_possible(self.path, 0o755)
4136
def test_check_warning(self):
4137
conf = config.AuthenticationConfig()
4138
self.assertEqual(conf._filename, self.path)
4139
self.assertContainsRe(self.get_log(),
4140
'Saved passwords may be accessible by other users.')
4142
def test_check_suppressed_warning(self):
4143
global_config = config.GlobalConfig()
4144
global_config.set_user_option('suppress_warnings',
4145
'insecure_permissions')
4146
conf = config.AuthenticationConfig()
4147
self.assertEqual(conf._filename, self.path)
4148
self.assertNotContainsRe(self.get_log(),
4149
'Saved passwords may be accessible by other users.')
4152
class TestAuthenticationConfigFile(tests.TestCase):
4153
"""Test the authentication.conf file matching"""
4155
def _got_user_passwd(self, expected_user, expected_password,
4156
config, *args, **kwargs):
4157
credentials = config.get_credentials(*args, **kwargs)
4158
if credentials is None:
4162
user = credentials['user']
4163
password = credentials['password']
4164
self.assertEqual(expected_user, user)
4165
self.assertEqual(expected_password, password)
4167
def test_empty_config(self):
4168
conf = config.AuthenticationConfig(_file=BytesIO())
4169
self.assertEqual({}, conf._get_config())
4170
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
4172
def test_non_utf8_config(self):
4173
conf = config.AuthenticationConfig(_file=BytesIO(b'foo = bar\xff'))
4174
self.assertRaises(config.ConfigContentError, conf._get_config)
4176
def test_missing_auth_section_header(self):
4177
conf = config.AuthenticationConfig(_file=BytesIO(b'foo = bar'))
4178
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4180
def test_auth_section_header_not_closed(self):
4181
conf = config.AuthenticationConfig(_file=BytesIO(b'[DEF'))
4182
self.assertRaises(config.ParseConfigError, conf._get_config)
4184
def test_auth_value_not_boolean(self):
4185
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4189
verify_certificates=askme # Error: Not a boolean
4191
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4193
def test_auth_value_not_int(self):
4194
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4198
port=port # Error: Not an int
4200
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
4202
def test_unknown_password_encoding(self):
4203
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4207
password_encoding=unknown
4209
self.assertRaises(ValueError, conf.get_password,
4210
'ftp', 'foo.net', 'joe')
4212
def test_credentials_for_scheme_host(self):
4213
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4214
# Identity on foo.net
4219
password=secret-pass
4222
self._got_user_passwd('joe', 'secret-pass', conf, 'ftp', 'foo.net')
4224
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
4226
self._got_user_passwd(None, None, conf, 'ftp', 'bar.net')
4228
def test_credentials_for_host_port(self):
4229
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4230
# Identity on foo.net
4236
password=secret-pass
4239
self._got_user_passwd('joe', 'secret-pass',
4240
conf, 'ftp', 'foo.net', port=10021)
4242
self._got_user_passwd(None, None, conf, 'ftp', 'foo.net')
4244
def test_for_matching_host(self):
4245
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4246
# Identity on foo.net
4252
[sourceforge domain]
4259
self._got_user_passwd('georges', 'bendover',
4260
conf, 'bzr', 'foo.bzr.sf.net')
4262
self._got_user_passwd(None, None,
4263
conf, 'bzr', 'bbzr.sf.net')
4265
def test_for_matching_host_None(self):
4266
conf = config.AuthenticationConfig(_file=BytesIO(b"""\
4267
# Identity on foo.net
4277
self._got_user_passwd('joe', 'joepass',
4278
conf, 'bzr', 'quux.net')
4279
# no host but different scheme
4280
self._got_user_passwd('georges', 'bendover',
4281
conf, 'ftp', 'quux.net')
4283
def test_credentials_for_path(self):
4284
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4299
self._got_user_passwd(None, None,
4300
conf, 'http', host='bar.org', path='/dir3')
4302
self._got_user_passwd('georges', 'bendover',
4303
conf, 'http', host='bar.org', path='/dir2')
4305
self._got_user_passwd('jim', 'jimpass',
4306
conf, 'http', host='bar.org', path='/dir1/subdir')
4308
def test_credentials_for_user(self):
4309
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4317
self._got_user_passwd('jim', 'jimpass',
4318
conf, 'http', 'bar.org')
4320
self._got_user_passwd('jim', 'jimpass',
4321
conf, 'http', 'bar.org', user='jim')
4322
# Don't get a different user if one is specified
4323
self._got_user_passwd(None, None,
4324
conf, 'http', 'bar.org', user='georges')
4326
def test_credentials_for_user_without_password(self):
4327
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4333
# Get user but no password
4334
self._got_user_passwd('jim', None,
4335
conf, 'http', 'bar.org')
4337
def test_verify_certificates(self):
4338
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4344
verify_certificates=False
4351
credentials = conf.get_credentials('https', 'bar.org')
4352
self.assertEqual(False, credentials.get('verify_certificates'))
4353
credentials = conf.get_credentials('https', 'foo.net')
4354
self.assertEqual(True, credentials.get('verify_certificates'))
4357
class TestAuthenticationStorage(tests.TestCaseInTempDir):
4359
def test_set_credentials(self):
4360
conf = config.AuthenticationConfig()
4361
conf.set_credentials('name', 'host', 'user', 'scheme', 'password',
4362
99, path='/foo', verify_certificates=False, realm='realm')
4363
credentials = conf.get_credentials(host='host', scheme='scheme',
4364
port=99, path='/foo',
4366
CREDENTIALS = {'name': 'name', 'user': 'user', 'password': 'password',
4367
'verify_certificates': False, 'scheme': 'scheme',
4368
'host': 'host', 'port': 99, 'path': '/foo',
4370
self.assertEqual(CREDENTIALS, credentials)
4371
credentials_from_disk = config.AuthenticationConfig().get_credentials(
4372
host='host', scheme='scheme', port=99, path='/foo', realm='realm')
4373
self.assertEqual(CREDENTIALS, credentials_from_disk)
4375
def test_reset_credentials_different_name(self):
4376
conf = config.AuthenticationConfig()
4377
conf.set_credentials('name', 'host', 'user', 'scheme', 'password'),
4378
conf.set_credentials('name2', 'host', 'user2', 'scheme', 'password'),
4379
self.assertIs(None, conf._get_config().get('name'))
4380
credentials = conf.get_credentials(host='host', scheme='scheme')
4381
CREDENTIALS = {'name': 'name2', 'user': 'user2', 'password':
4382
'password', 'verify_certificates': True,
4383
'scheme': 'scheme', 'host': 'host', 'port': None,
4384
'path': None, 'realm': None}
4385
self.assertEqual(CREDENTIALS, credentials)
4388
class TestAuthenticationConfig(tests.TestCaseInTempDir):
4389
"""Test AuthenticationConfig behaviour"""
4391
def _check_default_password_prompt(self, expected_prompt_format, scheme,
4392
host=None, port=None, realm=None,
4396
user, password = 'jim', 'precious'
4397
expected_prompt = expected_prompt_format % {
4398
'scheme': scheme, 'host': host, 'port': port,
4399
'user': user, 'realm': realm}
4401
ui.ui_factory = tests.TestUIFactory(stdin=password + '\n')
4402
# We use an empty conf so that the user is always prompted
4403
conf = config.AuthenticationConfig()
4404
self.assertEqual(password,
4405
conf.get_password(scheme, host, user, port=port,
4406
realm=realm, path=path))
4407
self.assertEqual(expected_prompt, ui.ui_factory.stderr.getvalue())
4408
self.assertEqual('', ui.ui_factory.stdout.getvalue())
4410
def _check_default_username_prompt(self, expected_prompt_format, scheme,
4411
host=None, port=None, realm=None,
4416
expected_prompt = expected_prompt_format % {
4417
'scheme': scheme, 'host': host, 'port': port,
4419
ui.ui_factory = tests.TestUIFactory(stdin=username + '\n')
4420
# We use an empty conf so that the user is always prompted
4421
conf = config.AuthenticationConfig()
4422
self.assertEqual(username, conf.get_user(scheme, host, port=port,
4423
realm=realm, path=path, ask=True))
4424
self.assertEqual(expected_prompt, ui.ui_factory.stderr.getvalue())
4425
self.assertEqual('', ui.ui_factory.stdout.getvalue())
4427
def test_username_defaults_prompts(self):
4428
# HTTP prompts can't be tested here, see test_http.py
4429
self._check_default_username_prompt(u'FTP %(host)s username: ', 'ftp')
4430
self._check_default_username_prompt(
4431
u'FTP %(host)s:%(port)d username: ', 'ftp', port=10020)
4432
self._check_default_username_prompt(
4433
u'SSH %(host)s:%(port)d username: ', 'ssh', port=12345)
4435
def test_username_default_no_prompt(self):
4436
conf = config.AuthenticationConfig()
4437
self.assertEqual(None,
4438
conf.get_user('ftp', 'example.com'))
4439
self.assertEqual("explicitdefault",
4440
conf.get_user('ftp', 'example.com', default="explicitdefault"))
4442
def test_password_default_prompts(self):
4443
# HTTP prompts can't be tested here, see test_http.py
4444
self._check_default_password_prompt(
4445
u'FTP %(user)s@%(host)s password: ', 'ftp')
4446
self._check_default_password_prompt(
4447
u'FTP %(user)s@%(host)s:%(port)d password: ', 'ftp', port=10020)
4448
self._check_default_password_prompt(
4449
u'SSH %(user)s@%(host)s:%(port)d password: ', 'ssh', port=12345)
4450
# SMTP port handling is a bit special (it's handled if embedded in the
4452
# FIXME: should we: forbid that, extend it to other schemes, leave
4453
# things as they are that's fine thank you ?
4454
self._check_default_password_prompt(
4455
u'SMTP %(user)s@%(host)s password: ', 'smtp')
4456
self._check_default_password_prompt(
4457
u'SMTP %(user)s@%(host)s password: ', 'smtp', host='bar.org:10025')
4458
self._check_default_password_prompt(
4459
u'SMTP %(user)s@%(host)s:%(port)d password: ', 'smtp', port=10025)
4461
def test_ssh_password_emits_warning(self):
4462
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4469
entered_password = 'typed-by-hand'
4470
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n')
4472
# Since the password defined in the authentication config is ignored,
4473
# the user is prompted
4474
self.assertEqual(entered_password,
4475
conf.get_password('ssh', 'bar.org', user='jim'))
4476
self.assertContainsRe(
4478
'password ignored in section \\[ssh with password\\]')
4480
def test_ssh_without_password_doesnt_emit_warning(self):
4481
conf = config.AuthenticationConfig(_file=BytesIO(b"""
4487
entered_password = 'typed-by-hand'
4488
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n')
4490
# Since the password defined in the authentication config is ignored,
4491
# the user is prompted
4492
self.assertEqual(entered_password,
4493
conf.get_password('ssh', 'bar.org', user='jim'))
4494
# No warning shoud be emitted since there is no password. We are only
4496
self.assertNotContainsRe(
4498
'password ignored in section \\[ssh with password\\]')
4500
def test_uses_fallback_stores(self):
4501
self.overrideAttr(config, 'credential_store_registry',
4502
config.CredentialStoreRegistry())
4503
store = StubCredentialStore()
4504
store.add_credentials("http", "example.com", "joe", "secret")
4505
config.credential_store_registry.register("stub", store, fallback=True)
4506
conf = config.AuthenticationConfig(_file=BytesIO())
4507
creds = conf.get_credentials("http", "example.com")
4508
self.assertEqual("joe", creds["user"])
4509
self.assertEqual("secret", creds["password"])
4512
class StubCredentialStore(config.CredentialStore):
4518
def add_credentials(self, scheme, host, user, password=None):
4519
self._username[(scheme, host)] = user
4520
self._password[(scheme, host)] = password
4522
def get_credentials(self, scheme, host, port=None, user=None,
4523
path=None, realm=None):
4524
key = (scheme, host)
4525
if key not in self._username:
4527
return {"scheme": scheme, "host": host, "port": port,
4528
"user": self._username[key], "password": self._password[key]}
4531
class CountingCredentialStore(config.CredentialStore):
4536
def get_credentials(self, scheme, host, port=None, user=None,
4537
path=None, realm=None):
4542
class TestCredentialStoreRegistry(tests.TestCase):
4544
def _get_cs_registry(self):
4545
return config.credential_store_registry
4547
def test_default_credential_store(self):
4548
r = self._get_cs_registry()
4549
default = r.get_credential_store(None)
4550
self.assertIsInstance(default, config.PlainTextCredentialStore)
4552
def test_unknown_credential_store(self):
4553
r = self._get_cs_registry()
4554
# It's hard to imagine someone creating a credential store named
4555
# 'unknown' so we use that as an never registered key.
4556
self.assertRaises(KeyError, r.get_credential_store, 'unknown')
4558
def test_fallback_none_registered(self):
4559
r = config.CredentialStoreRegistry()
4560
self.assertEqual(None,
4561
r.get_fallback_credentials("http", "example.com"))
4563
def test_register(self):
4564
r = config.CredentialStoreRegistry()
4565
r.register("stub", StubCredentialStore(), fallback=False)
4566
r.register("another", StubCredentialStore(), fallback=True)
4567
self.assertEqual(["another", "stub"], r.keys())
4569
def test_register_lazy(self):
4570
r = config.CredentialStoreRegistry()
4571
r.register_lazy("stub", "breezy.tests.test_config",
4572
"StubCredentialStore", fallback=False)
4573
self.assertEqual(["stub"], r.keys())
4574
self.assertIsInstance(r.get_credential_store("stub"),
4575
StubCredentialStore)
4577
def test_is_fallback(self):
4578
r = config.CredentialStoreRegistry()
4579
r.register("stub1", None, fallback=False)
4580
r.register("stub2", None, fallback=True)
4581
self.assertEqual(False, r.is_fallback("stub1"))
4582
self.assertEqual(True, r.is_fallback("stub2"))
4584
def test_no_fallback(self):
4585
r = config.CredentialStoreRegistry()
4586
store = CountingCredentialStore()
4587
r.register("count", store, fallback=False)
4588
self.assertEqual(None,
4589
r.get_fallback_credentials("http", "example.com"))
4590
self.assertEqual(0, store._calls)
4592
def test_fallback_credentials(self):
4593
r = config.CredentialStoreRegistry()
4594
store = StubCredentialStore()
4595
store.add_credentials("http", "example.com",
4596
"somebody", "geheim")
4597
r.register("stub", store, fallback=True)
4598
creds = r.get_fallback_credentials("http", "example.com")
4599
self.assertEqual("somebody", creds["user"])
4600
self.assertEqual("geheim", creds["password"])
4602
def test_fallback_first_wins(self):
4603
r = config.CredentialStoreRegistry()
4604
stub1 = StubCredentialStore()
4605
stub1.add_credentials("http", "example.com",
4606
"somebody", "stub1")
4607
r.register("stub1", stub1, fallback=True)
4608
stub2 = StubCredentialStore()
4609
stub2.add_credentials("http", "example.com",
4610
"somebody", "stub2")
4611
r.register("stub2", stub1, fallback=True)
4612
creds = r.get_fallback_credentials("http", "example.com")
4613
self.assertEqual("somebody", creds["user"])
4614
self.assertEqual("stub1", creds["password"])
4617
class TestPlainTextCredentialStore(tests.TestCase):
4619
def test_decode_password(self):
4620
r = config.credential_store_registry
4621
plain_text = r.get_credential_store()
4622
decoded = plain_text.decode_password(dict(password='secret'))
4623
self.assertEqual('secret', decoded)
4626
class TestBase64CredentialStore(tests.TestCase):
4628
def test_decode_password(self):
4629
r = config.credential_store_registry
4630
plain_text = r.get_credential_store('base64')
4631
decoded = plain_text.decode_password(dict(password='c2VjcmV0'))
4632
self.assertEqual(b'secret', decoded)
4635
# FIXME: Once we have a way to declare authentication to all test servers, we
4636
# can implement generic tests.
4637
# test_user_password_in_url
4638
# test_user_in_url_password_from_config
4639
# test_user_in_url_password_prompted
4640
# test_user_in_config
4641
# test_user_getpass.getuser
4642
# test_user_prompted ?
4643
class TestAuthenticationRing(tests.TestCaseWithTransport):
4647
class EmailOptionTests(tests.TestCase):
4649
def test_default_email_uses_BRZ_EMAIL(self):
4650
conf = config.MemoryStack(b'email=jelmer@debian.org')
4651
# BRZ_EMAIL takes precedence over EMAIL
4652
self.overrideEnv('BRZ_EMAIL', 'jelmer@samba.org')
4653
self.overrideEnv('EMAIL', 'jelmer@apache.org')
4654
self.assertEqual('jelmer@samba.org', conf.get('email'))
4656
def test_default_email_uses_EMAIL(self):
4657
conf = config.MemoryStack(b'')
4658
self.overrideEnv('BRZ_EMAIL', None)
4659
self.overrideEnv('EMAIL', 'jelmer@apache.org')
4660
self.assertEqual('jelmer@apache.org', conf.get('email'))
4662
def test_BRZ_EMAIL_overrides(self):
4663
conf = config.MemoryStack(b'email=jelmer@debian.org')
4664
self.overrideEnv('BRZ_EMAIL', 'jelmer@apache.org')
4665
self.assertEqual('jelmer@apache.org', conf.get('email'))
4666
self.overrideEnv('BRZ_EMAIL', None)
4667
self.overrideEnv('EMAIL', 'jelmer@samba.org')
4668
self.assertEqual('jelmer@debian.org', conf.get('email'))
4671
class MailClientOptionTests(tests.TestCase):
4673
def test_default(self):
4674
conf = config.MemoryStack(b'')
4675
client = conf.get('mail_client')
4676
self.assertIs(client, mail_client.DefaultMail)
4678
def test_evolution(self):
4679
conf = config.MemoryStack(b'mail_client=evolution')
4680
client = conf.get('mail_client')
4681
self.assertIs(client, mail_client.Evolution)
4683
def test_kmail(self):
4684
conf = config.MemoryStack(b'mail_client=kmail')
4685
client = conf.get('mail_client')
4686
self.assertIs(client, mail_client.KMail)
4688
def test_mutt(self):
4689
conf = config.MemoryStack(b'mail_client=mutt')
4690
client = conf.get('mail_client')
4691
self.assertIs(client, mail_client.Mutt)
4693
def test_thunderbird(self):
4694
conf = config.MemoryStack(b'mail_client=thunderbird')
4695
client = conf.get('mail_client')
4696
self.assertIs(client, mail_client.Thunderbird)
4698
def test_explicit_default(self):
4699
conf = config.MemoryStack(b'mail_client=default')
4700
client = conf.get('mail_client')
4701
self.assertIs(client, mail_client.DefaultMail)
4703
def test_editor(self):
4704
conf = config.MemoryStack(b'mail_client=editor')
4705
client = conf.get('mail_client')
4706
self.assertIs(client, mail_client.Editor)
4708
def test_mapi(self):
4709
conf = config.MemoryStack(b'mail_client=mapi')
4710
client = conf.get('mail_client')
4711
self.assertIs(client, mail_client.MAPIClient)
4713
def test_xdg_email(self):
4714
conf = config.MemoryStack(b'mail_client=xdg-email')
4715
client = conf.get('mail_client')
4716
self.assertIs(client, mail_client.XDGEmail)
4718
def test_unknown(self):
4719
conf = config.MemoryStack(b'mail_client=firebird')
4720
self.assertRaises(config.ConfigOptionValueError, conf.get,
added
removed
breezy/tests/test_config.py
