349
234
my_config = config.Config()
350
235
self.assertEqual('long', my_config.log_format())
352
def test_get_change_editor(self):
353
my_config = InstrumentedConfig()
354
change_editor = my_config.get_change_editor('old_tree', 'new_tree')
355
self.assertEqual(['_get_change_editor'], my_config._calls)
356
self.assertIs(diff.DiffFromTool, change_editor.__class__)
357
self.assertEqual(['vimdiff', '-fo', '@new_path', '@old_path'],
358
change_editor.command_template)
361
class TestConfigPath(tests.TestCase):
238
class TestConfigPath(TestCase):
364
241
super(TestConfigPath, self).setUp()
242
self.old_home = os.environ.get('HOME', None)
243
self.old_appdata = os.environ.get('APPDATA', None)
365
244
os.environ['HOME'] = '/home/bogus'
366
os.environ['XDG_CACHE_DIR'] = ''
367
if sys.platform == 'win32':
368
os.environ['BZR_HOME'] = \
369
r'C:\Documents and Settings\bogus\Application Data'
371
'C:/Documents and Settings/bogus/Application Data/bazaar/2.0'
373
self.bzr_home = '/home/bogus/.bazaar'
245
os.environ['APPDATA'] = \
246
r'C:\Documents and Settings\bogus\Application Data'
249
if self.old_home is None:
250
del os.environ['HOME']
252
os.environ['HOME'] = self.old_home
253
if self.old_appdata is None:
254
del os.environ['APPDATA']
256
os.environ['APPDATA'] = self.old_appdata
257
super(TestConfigPath, self).tearDown()
375
259
def test_config_dir(self):
376
self.assertEqual(config.config_dir(), self.bzr_home)
260
if sys.platform == 'win32':
261
self.assertEqual(config.config_dir(),
262
'C:/Documents and Settings/bogus/Application Data/bazaar/2.0')
264
self.assertEqual(config.config_dir(), '/home/bogus/.bazaar')
378
266
def test_config_filename(self):
379
self.assertEqual(config.config_filename(),
380
self.bzr_home + '/bazaar.conf')
267
if sys.platform == 'win32':
268
self.assertEqual(config.config_filename(),
269
'C:/Documents and Settings/bogus/Application Data/bazaar/2.0/bazaar.conf')
271
self.assertEqual(config.config_filename(),
272
'/home/bogus/.bazaar/bazaar.conf')
274
def test_branches_config_filename(self):
275
if sys.platform == 'win32':
276
self.assertEqual(config.branches_config_filename(),
277
'C:/Documents and Settings/bogus/Application Data/bazaar/2.0/branches.conf')
279
self.assertEqual(config.branches_config_filename(),
280
'/home/bogus/.bazaar/branches.conf')
382
282
def test_locations_config_filename(self):
383
self.assertEqual(config.locations_config_filename(),
384
self.bzr_home + '/locations.conf')
386
def test_authentication_config_filename(self):
387
self.assertEqual(config.authentication_config_filename(),
388
self.bzr_home + '/authentication.conf')
390
def test_xdg_cache_dir(self):
391
self.assertEqual(config.xdg_cache_dir(),
392
'/home/bogus/.cache')
395
class TestIniConfig(tests.TestCaseInTempDir):
397
def make_config_parser(self, s):
398
conf = config.IniBasedConfig.from_string(s)
399
return conf, conf._get_parser()
402
class TestIniConfigBuilding(TestIniConfig):
283
if sys.platform == 'win32':
284
self.assertEqual(config.locations_config_filename(),
285
'C:/Documents and Settings/bogus/Application Data/bazaar/2.0/locations.conf')
287
self.assertEqual(config.locations_config_filename(),
288
'/home/bogus/.bazaar/locations.conf')
290
class TestIniConfig(TestCase):
404
292
def test_contructs(self):
405
my_config = config.IniBasedConfig()
293
my_config = config.IniBasedConfig("nothing")
407
295
def test_from_fp(self):
408
my_config = config.IniBasedConfig.from_string(sample_config_text)
409
self.assertIsInstance(my_config._get_parser(), configobj.ConfigObj)
296
config_file = StringIO(sample_config_text.encode('utf-8'))
297
my_config = config.IniBasedConfig(None)
299
isinstance(my_config._get_parser(file=config_file),
411
302
def test_cached(self):
412
my_config = config.IniBasedConfig.from_string(sample_config_text)
413
parser = my_config._get_parser()
303
config_file = StringIO(sample_config_text.encode('utf-8'))
304
my_config = config.IniBasedConfig(None)
305
parser = my_config._get_parser(file=config_file)
414
306
self.failUnless(my_config._get_parser() is parser)
416
def _dummy_chown(self, path, uid, gid):
417
self.path, self.uid, self.gid = path, uid, gid
419
def test_ini_config_ownership(self):
420
"""Ensure that chown is happening during _write_config_file"""
421
self.requireFeature(features.chown_feature)
422
self.overrideAttr(os, 'chown', self._dummy_chown)
423
self.path = self.uid = self.gid = None
424
conf = config.IniBasedConfig(file_name='./foo.conf')
425
conf._write_config_file()
426
self.assertEquals(self.path, './foo.conf')
427
self.assertTrue(isinstance(self.uid, int))
428
self.assertTrue(isinstance(self.gid, int))
430
def test_get_filename_parameter_is_deprecated_(self):
431
conf = self.callDeprecated([
432
'IniBasedConfig.__init__(get_filename) was deprecated in 2.3.'
433
' Use file_name instead.'],
434
config.IniBasedConfig, lambda: 'ini.conf')
435
self.assertEqual('ini.conf', conf.file_name)
437
def test_get_parser_file_parameter_is_deprecated_(self):
438
config_file = StringIO(sample_config_text.encode('utf-8'))
439
conf = config.IniBasedConfig.from_string(sample_config_text)
440
conf = self.callDeprecated([
441
'IniBasedConfig._get_parser(file=xxx) was deprecated in 2.3.'
442
' Use IniBasedConfig(_content=xxx) instead.'],
443
conf._get_parser, file=config_file)
445
class TestIniConfigSaving(tests.TestCaseInTempDir):
447
def test_cant_save_without_a_file_name(self):
448
conf = config.IniBasedConfig()
449
self.assertRaises(AssertionError, conf._write_config_file)
451
def test_saved_with_content(self):
452
content = 'foo = bar\n'
453
conf = config.IniBasedConfig.from_string(
454
content, file_name='./test.conf', save=True)
455
self.assertFileEqual(content, 'test.conf')
458
class TestIniBaseConfigOnDisk(tests.TestCaseInTempDir):
460
def test_cannot_reload_without_name(self):
461
conf = config.IniBasedConfig.from_string(sample_config_text)
462
self.assertRaises(AssertionError, conf.reload)
464
def test_reload_see_new_value(self):
465
c1 = config.IniBasedConfig.from_string('editor=vim\n',
466
file_name='./test/conf')
467
c1._write_config_file()
468
c2 = config.IniBasedConfig.from_string('editor=emacs\n',
469
file_name='./test/conf')
470
c2._write_config_file()
471
self.assertEqual('vim', c1.get_user_option('editor'))
472
self.assertEqual('emacs', c2.get_user_option('editor'))
473
# Make sure we get the Right value
475
self.assertEqual('emacs', c1.get_user_option('editor'))
478
class TestLockableConfig(tests.TestCaseInTempDir):
483
config_section = None
486
super(TestLockableConfig, self).setUp()
487
self._content = '[%s]\none=1\ntwo=2\n' % (self.config_section,)
488
self.config = self.create_config(self._content)
490
def get_existing_config(self):
491
return self.config_class(*self.config_args)
493
def create_config(self, content):
494
kwargs = dict(save=True)
495
c = self.config_class.from_string(content, *self.config_args, **kwargs)
498
def test_simple_read_access(self):
499
self.assertEquals('1', self.config.get_user_option('one'))
501
def test_simple_write_access(self):
502
self.config.set_user_option('one', 'one')
503
self.assertEquals('one', self.config.get_user_option('one'))
505
def test_listen_to_the_last_speaker(self):
507
c2 = self.get_existing_config()
508
c1.set_user_option('one', 'ONE')
509
c2.set_user_option('two', 'TWO')
510
self.assertEquals('ONE', c1.get_user_option('one'))
511
self.assertEquals('TWO', c2.get_user_option('two'))
512
# The second update respect the first one
513
self.assertEquals('ONE', c2.get_user_option('one'))
515
def test_last_speaker_wins(self):
516
# If the same config is not shared, the same variable modified twice
517
# can only see a single result.
519
c2 = self.get_existing_config()
520
c1.set_user_option('one', 'c1')
521
c2.set_user_option('one', 'c2')
522
self.assertEquals('c2', c2._get_user_option('one'))
523
# The first modification is still available until another refresh
525
self.assertEquals('c1', c1._get_user_option('one'))
526
c1.set_user_option('two', 'done')
527
self.assertEquals('c2', c1._get_user_option('one'))
529
def test_writes_are_serialized(self):
531
c2 = self.get_existing_config()
533
# We spawn a thread that will pause *during* the write
534
before_writing = threading.Event()
535
after_writing = threading.Event()
536
writing_done = threading.Event()
537
c1_orig = c1._write_config_file
538
def c1_write_config_file():
541
# The lock is held we wait for the main thread to decide when to
544
c1._write_config_file = c1_write_config_file
546
c1.set_user_option('one', 'c1')
548
t1 = threading.Thread(target=c1_set_option)
549
# Collect the thread after the test
550
self.addCleanup(t1.join)
551
# Be ready to unblock the thread if the test goes wrong
552
self.addCleanup(after_writing.set)
554
before_writing.wait()
555
self.assertTrue(c1._lock.is_held)
556
self.assertRaises(errors.LockContention,
557
c2.set_user_option, 'one', 'c2')
558
self.assertEquals('c1', c1.get_user_option('one'))
559
# Let the lock be released
562
c2.set_user_option('one', 'c2')
563
self.assertEquals('c2', c2.get_user_option('one'))
565
def test_read_while_writing(self):
567
# We spawn a thread that will pause *during* the write
568
ready_to_write = threading.Event()
569
do_writing = threading.Event()
570
writing_done = threading.Event()
571
c1_orig = c1._write_config_file
572
def c1_write_config_file():
574
# The lock is held we wait for the main thread to decide when to
579
c1._write_config_file = c1_write_config_file
581
c1.set_user_option('one', 'c1')
582
t1 = threading.Thread(target=c1_set_option)
583
# Collect the thread after the test
584
self.addCleanup(t1.join)
585
# Be ready to unblock the thread if the test goes wrong
586
self.addCleanup(do_writing.set)
588
# Ensure the thread is ready to write
589
ready_to_write.wait()
590
self.assertTrue(c1._lock.is_held)
591
self.assertEquals('c1', c1.get_user_option('one'))
592
# If we read during the write, we get the old value
593
c2 = self.get_existing_config()
594
self.assertEquals('1', c2.get_user_option('one'))
595
# Let the writing occur and ensure it occurred
598
# Now we get the updated value
599
c3 = self.get_existing_config()
600
self.assertEquals('c1', c3.get_user_option('one'))
603
class TestGetUserOptionAs(TestIniConfig):
605
def test_get_user_option_as_bool(self):
606
conf, parser = self.make_config_parser("""
609
an_invalid_bool = maybe
610
a_list = hmm, who knows ? # This is interpreted as a list !
612
get_bool = conf.get_user_option_as_bool
613
self.assertEqual(True, get_bool('a_true_bool'))
614
self.assertEqual(False, get_bool('a_false_bool'))
617
warnings.append(args[0] % args[1:])
618
self.overrideAttr(trace, 'warning', warning)
619
msg = 'Value "%s" is not a boolean for "%s"'
620
self.assertIs(None, get_bool('an_invalid_bool'))
621
self.assertEquals(msg % ('maybe', 'an_invalid_bool'), warnings[0])
623
self.assertIs(None, get_bool('not_defined_in_this_config'))
624
self.assertEquals([], warnings)
626
def test_get_user_option_as_list(self):
627
conf, parser = self.make_config_parser("""
632
get_list = conf.get_user_option_as_list
633
self.assertEqual(['a', 'b', 'c'], get_list('a_list'))
634
self.assertEqual(['1'], get_list('length_1'))
635
self.assertEqual('x', conf.get_user_option('one_item'))
636
# automatically cast to list
637
self.assertEqual(['x'], get_list('one_item'))
640
class TestSupressWarning(TestIniConfig):
642
def make_warnings_config(self, s):
643
conf, parser = self.make_config_parser(s)
644
return conf.suppress_warning
646
def test_suppress_warning_unknown(self):
647
suppress_warning = self.make_warnings_config('')
648
self.assertEqual(False, suppress_warning('unknown_warning'))
650
def test_suppress_warning_known(self):
651
suppress_warning = self.make_warnings_config('suppress_warnings=a,b')
652
self.assertEqual(False, suppress_warning('c'))
653
self.assertEqual(True, suppress_warning('a'))
654
self.assertEqual(True, suppress_warning('b'))
657
class TestGetConfig(tests.TestCase):
309
class TestGetConfig(TestCase):
659
311
def test_constructs(self):
660
312
my_config = config.GlobalConfig()
662
314
def test_calls_read_filenames(self):
663
# replace the class that is constructed, to check its parameters
315
# replace the class that is constructured, to check its parameters
664
316
oldparserclass = config.ConfigObj
665
317
config.ConfigObj = InstrumentedConfigObj
666
318
my_config = config.GlobalConfig()
1076
641
self.get_branch_config('/a')
1077
642
self.assertEqual('local',
1078
643
self.my_config.get_user_option('user_local_option'))
1080
def test_get_user_option_appendpath(self):
1081
# returned as is for the base path:
1082
self.get_branch_config('http://www.example.com')
1083
self.assertEqual('append',
1084
self.my_config.get_user_option('appendpath_option'))
1085
# Extra path components get appended:
1086
self.get_branch_config('http://www.example.com/a/b/c')
1087
self.assertEqual('append/a/b/c',
1088
self.my_config.get_user_option('appendpath_option'))
1089
# Overriden for http://www.example.com/dir, where it is a
1091
self.get_branch_config('http://www.example.com/dir/a/b/c')
1092
self.assertEqual('normal',
1093
self.my_config.get_user_option('appendpath_option'))
1095
def test_get_user_option_norecurse(self):
1096
self.get_branch_config('http://www.example.com')
1097
self.assertEqual('norecurse',
1098
self.my_config.get_user_option('norecurse_option'))
1099
self.get_branch_config('http://www.example.com/dir')
1100
self.assertEqual(None,
1101
self.my_config.get_user_option('norecurse_option'))
1102
# http://www.example.com/norecurse is a recurse=False section
1103
# that redefines normal_option. Subdirectories do not pick up
1104
# this redefinition.
1105
self.get_branch_config('http://www.example.com/norecurse')
1106
self.assertEqual('norecurse',
1107
self.my_config.get_user_option('normal_option'))
1108
self.get_branch_config('http://www.example.com/norecurse/subdir')
1109
self.assertEqual('normal',
1110
self.my_config.get_user_option('normal_option'))
1112
def test_set_user_option_norecurse(self):
1113
self.get_branch_config('http://www.example.com')
1114
self.my_config.set_user_option('foo', 'bar',
1115
store=config.STORE_LOCATION_NORECURSE)
1117
self.my_location_config._get_option_policy(
1118
'http://www.example.com', 'foo'),
1119
config.POLICY_NORECURSE)
1121
def test_set_user_option_appendpath(self):
1122
self.get_branch_config('http://www.example.com')
1123
self.my_config.set_user_option('foo', 'bar',
1124
store=config.STORE_LOCATION_APPENDPATH)
1126
self.my_location_config._get_option_policy(
1127
'http://www.example.com', 'foo'),
1128
config.POLICY_APPENDPATH)
1130
def test_set_user_option_change_policy(self):
1131
self.get_branch_config('http://www.example.com')
1132
self.my_config.set_user_option('norecurse_option', 'normal',
1133
store=config.STORE_LOCATION)
1135
self.my_location_config._get_option_policy(
1136
'http://www.example.com', 'norecurse_option'),
1139
def test_set_user_option_recurse_false_section(self):
1140
# The following section has recurse=False set. The test is to
1141
# make sure that a normal option can be added to the section,
1142
# converting recurse=False to the norecurse policy.
1143
self.get_branch_config('http://www.example.com/norecurse')
1144
self.callDeprecated(['The recurse option is deprecated as of 0.14. '
1145
'The section "http://www.example.com/norecurse" '
1146
'has been converted to use policies.'],
1147
self.my_config.set_user_option,
1148
'foo', 'bar', store=config.STORE_LOCATION)
1150
self.my_location_config._get_option_policy(
1151
'http://www.example.com/norecurse', 'foo'),
1153
# The previously existing option is still norecurse:
1155
self.my_location_config._get_option_policy(
1156
'http://www.example.com/norecurse', 'normal_option'),
1157
config.POLICY_NORECURSE)
1159
645
def test_post_commit_default(self):
1160
646
self.get_branch_config('/a/c')
1161
647
self.assertEqual('bzrlib.tests.test_config.post_commit',
1162
648
self.my_config.post_commit())
1164
650
def get_branch_config(self, location, global_config=None):
1165
my_branch = FakeBranch(location)
1166
651
if global_config is None:
1167
global_config = sample_config_text
1169
my_global_config = config.GlobalConfig.from_string(global_config,
1171
my_location_config = config.LocationConfig.from_string(
1172
sample_branches_text, my_branch.base, save=True)
1173
my_config = config.BranchConfig(my_branch)
1174
self.my_config = my_config
1175
self.my_location_config = my_config._get_location_config()
652
global_file = StringIO(sample_config_text.encode('utf-8'))
654
global_file = StringIO(global_config.encode('utf-8'))
655
branches_file = StringIO(sample_branches_text.encode('utf-8'))
656
self.my_config = config.BranchConfig(FakeBranch(location))
657
# Force location config to use specified file
658
self.my_location_config = self.my_config._get_location_config()
659
self.my_location_config._get_parser(branches_file)
660
# Force global config to use specified file
661
self.my_config._get_global_config()._get_parser(global_file)
1177
663
def test_set_user_setting_sets_and_saves(self):
1178
664
self.get_branch_config('/a/c')
1179
665
record = InstrumentedConfigObj("foo")
1180
666
self.my_location_config._parser = record
1182
self.callDeprecated(['The recurse option is deprecated as of '
1183
'0.14. The section "/a/c" has been '
1184
'converted to use policies.'],
1185
self.my_config.set_user_option,
1186
'foo', 'bar', store=config.STORE_LOCATION)
1187
self.assertEqual([('reload',),
1188
('__contains__', '/a/c'),
668
real_mkdir = os.mkdir
670
def checked_mkdir(path, mode=0777):
671
self.log('making directory: %s', path)
672
real_mkdir(path, mode)
675
os.mkdir = checked_mkdir
677
self.my_config.set_user_option('foo', 'bar', local=True)
679
os.mkdir = real_mkdir
681
self.failUnless(self.created, 'Failed to create ~/.bazaar')
682
self.assertEqual([('__contains__', '/a/c'),
1189
683
('__contains__', '/a/c/'),
1190
684
('__setitem__', '/a/c', {}),
1191
685
('__getitem__', '/a/c'),
1192
686
('__setitem__', 'foo', 'bar'),
1193
('__getitem__', '/a/c'),
1194
('as_bool', 'recurse'),
1195
('__getitem__', '/a/c'),
1196
('__delitem__', 'recurse'),
1197
('__getitem__', '/a/c'),
1199
('__getitem__', '/a/c'),
1200
('__contains__', 'foo:policy'),
1202
688
record._calls[1:])
1296
775
def test_gpg_signing_command(self):
1297
776
my_config = self.get_branch_config(
1298
global_config=sample_config_text,
1299
777
# branch data cannot set gpg_signing_command
1300
778
branch_data_config="gpg_signing_command=pgp")
779
config_file = StringIO(sample_config_text.encode('utf-8'))
780
my_config._get_global_config()._get_parser(config_file)
1301
781
self.assertEqual('gnome-gpg', my_config.gpg_signing_command())
1303
783
def test_get_user_option_global(self):
1304
my_config = self.get_branch_config(global_config=sample_config_text)
784
branch = FakeBranch()
785
my_config = config.BranchConfig(branch)
786
config_file = StringIO(sample_config_text.encode('utf-8'))
787
(my_config._get_global_config()._get_parser(config_file))
1305
788
self.assertEqual('something',
1306
789
my_config.get_user_option('user_global_option'))
1308
791
def test_post_commit_default(self):
1309
my_config = self.get_branch_config(global_config=sample_config_text,
1311
location_config=sample_branches_text)
792
branch = FakeBranch()
793
my_config = self.get_branch_config(sample_config_text, '/a/c',
794
sample_branches_text)
1312
795
self.assertEqual(my_config.branch.base, '/a/c')
1313
796
self.assertEqual('bzrlib.tests.test_config.post_commit',
1314
797
my_config.post_commit())
1315
798
my_config.set_user_option('post_commit', 'rmtree_root')
1316
# post-commit is ignored when present in branch data
799
# post-commit is ignored when bresent in branch data
1317
800
self.assertEqual('bzrlib.tests.test_config.post_commit',
1318
801
my_config.post_commit())
1319
my_config.set_user_option('post_commit', 'rmtree_root',
1320
store=config.STORE_LOCATION)
802
my_config.set_user_option('post_commit', 'rmtree_root', local=True)
1321
803
self.assertEqual('rmtree_root', my_config.post_commit())
1323
805
def test_config_precedence(self):
1324
# FIXME: eager test, luckily no persitent config file makes it fail
1326
806
my_config = self.get_branch_config(global_config=precedence_global)
1327
807
self.assertEqual(my_config.get_user_option('option'), 'global')
1328
my_config = self.get_branch_config(global_config=precedence_global,
1329
branch_data_config=precedence_branch)
808
my_config = self.get_branch_config(global_config=precedence_global,
809
branch_data_config=precedence_branch)
1330
810
self.assertEqual(my_config.get_user_option('option'), 'branch')
1331
my_config = self.get_branch_config(
1332
global_config=precedence_global,
1333
branch_data_config=precedence_branch,
1334
location_config=precedence_location)
811
my_config = self.get_branch_config(global_config=precedence_global,
812
branch_data_config=precedence_branch,
813
location_config=precedence_location)
1335
814
self.assertEqual(my_config.get_user_option('option'), 'recurse')
1336
my_config = self.get_branch_config(
1337
global_config=precedence_global,
1338
branch_data_config=precedence_branch,
1339
location_config=precedence_location,
1340
location='http://example.com/specific')
815
my_config = self.get_branch_config(global_config=precedence_global,
816
branch_data_config=precedence_branch,
817
location_config=precedence_location,
818
location='http://example.com/specific')
1341
819
self.assertEqual(my_config.get_user_option('option'), 'exact')
1343
def test_get_mail_client(self):
1344
config = self.get_branch_config()
1345
client = config.get_mail_client()
1346
self.assertIsInstance(client, mail_client.DefaultMail)
1349
config.set_user_option('mail_client', 'evolution')
1350
client = config.get_mail_client()
1351
self.assertIsInstance(client, mail_client.Evolution)
1353
config.set_user_option('mail_client', 'kmail')
1354
client = config.get_mail_client()
1355
self.assertIsInstance(client, mail_client.KMail)
1357
config.set_user_option('mail_client', 'mutt')
1358
client = config.get_mail_client()
1359
self.assertIsInstance(client, mail_client.Mutt)
1361
config.set_user_option('mail_client', 'thunderbird')
1362
client = config.get_mail_client()
1363
self.assertIsInstance(client, mail_client.Thunderbird)
1366
config.set_user_option('mail_client', 'default')
1367
client = config.get_mail_client()
1368
self.assertIsInstance(client, mail_client.DefaultMail)
1370
config.set_user_option('mail_client', 'editor')
1371
client = config.get_mail_client()
1372
self.assertIsInstance(client, mail_client.Editor)
1374
config.set_user_option('mail_client', 'mapi')
1375
client = config.get_mail_client()
1376
self.assertIsInstance(client, mail_client.MAPIClient)
1378
config.set_user_option('mail_client', 'xdg-email')
1379
client = config.get_mail_client()
1380
self.assertIsInstance(client, mail_client.XDGEmail)
1382
config.set_user_option('mail_client', 'firebird')
1383
self.assertRaises(errors.UnknownMailClient, config.get_mail_client)
1386
class TestMailAddressExtraction(tests.TestCase):
822
class TestMailAddressExtraction(TestCase):
1388
824
def test_extract_email_address(self):
1389
825
self.assertEqual('jane@test.com',
1390
826
config.extract_email_address('Jane <jane@test.com>'))
1391
self.assertRaises(errors.NoEmailInUsername,
827
self.assertRaises(errors.BzrError,
1392
828
config.extract_email_address, 'Jane Tester')
1394
def test_parse_username(self):
1395
self.assertEqual(('', 'jdoe@example.com'),
1396
config.parse_username('jdoe@example.com'))
1397
self.assertEqual(('', 'jdoe@example.com'),
1398
config.parse_username('<jdoe@example.com>'))
1399
self.assertEqual(('John Doe', 'jdoe@example.com'),
1400
config.parse_username('John Doe <jdoe@example.com>'))
1401
self.assertEqual(('John Doe', ''),
1402
config.parse_username('John Doe'))
1403
self.assertEqual(('John Doe', 'jdoe@example.com'),
1404
config.parse_username('John Doe jdoe@example.com'))
1406
class TestTreeConfig(tests.TestCaseWithTransport):
1408
def test_get_value(self):
1409
"""Test that retreiving a value from a section is possible"""
1410
branch = self.make_branch('.')
1411
tree_config = config.TreeConfig(branch)
1412
tree_config.set_option('value', 'key', 'SECTION')
1413
tree_config.set_option('value2', 'key2')
1414
tree_config.set_option('value3-top', 'key3')
1415
tree_config.set_option('value3-section', 'key3', 'SECTION')
1416
value = tree_config.get_option('key', 'SECTION')
1417
self.assertEqual(value, 'value')
1418
value = tree_config.get_option('key2')
1419
self.assertEqual(value, 'value2')
1420
self.assertEqual(tree_config.get_option('non-existant'), None)
1421
value = tree_config.get_option('non-existant', 'SECTION')
1422
self.assertEqual(value, None)
1423
value = tree_config.get_option('non-existant', default='default')
1424
self.assertEqual(value, 'default')
1425
self.assertEqual(tree_config.get_option('key2', 'NOSECTION'), None)
1426
value = tree_config.get_option('key2', 'NOSECTION', default='default')
1427
self.assertEqual(value, 'default')
1428
value = tree_config.get_option('key3')
1429
self.assertEqual(value, 'value3-top')
1430
value = tree_config.get_option('key3', 'SECTION')
1431
self.assertEqual(value, 'value3-section')
1434
class TestTransportConfig(tests.TestCaseWithTransport):
1436
def test_get_value(self):
1437
"""Test that retreiving a value from a section is possible"""
1438
bzrdir_config = config.TransportConfig(transport.get_transport('.'),
1440
bzrdir_config.set_option('value', 'key', 'SECTION')
1441
bzrdir_config.set_option('value2', 'key2')
1442
bzrdir_config.set_option('value3-top', 'key3')
1443
bzrdir_config.set_option('value3-section', 'key3', 'SECTION')
1444
value = bzrdir_config.get_option('key', 'SECTION')
1445
self.assertEqual(value, 'value')
1446
value = bzrdir_config.get_option('key2')
1447
self.assertEqual(value, 'value2')
1448
self.assertEqual(bzrdir_config.get_option('non-existant'), None)
1449
value = bzrdir_config.get_option('non-existant', 'SECTION')
1450
self.assertEqual(value, None)
1451
value = bzrdir_config.get_option('non-existant', default='default')
1452
self.assertEqual(value, 'default')
1453
self.assertEqual(bzrdir_config.get_option('key2', 'NOSECTION'), None)
1454
value = bzrdir_config.get_option('key2', 'NOSECTION',
1456
self.assertEqual(value, 'default')
1457
value = bzrdir_config.get_option('key3')
1458
self.assertEqual(value, 'value3-top')
1459
value = bzrdir_config.get_option('key3', 'SECTION')
1460
self.assertEqual(value, 'value3-section')
1462
def test_set_unset_default_stack_on(self):
1463
my_dir = self.make_bzrdir('.')
1464
bzrdir_config = config.BzrDirConfig(my_dir)
1465
self.assertIs(None, bzrdir_config.get_default_stack_on())
1466
bzrdir_config.set_default_stack_on('Foo')
1467
self.assertEqual('Foo', bzrdir_config._config.get_option(
1468
'default_stack_on'))
1469
self.assertEqual('Foo', bzrdir_config.get_default_stack_on())
1470
bzrdir_config.set_default_stack_on(None)
1471
self.assertIs(None, bzrdir_config.get_default_stack_on())
1474
class TestAuthenticationConfigFile(tests.TestCase):
1475
"""Test the authentication.conf file matching"""
1477
def _got_user_passwd(self, expected_user, expected_password,
1478
config, *args, **kwargs):
1479
credentials = config.get_credentials(*args, **kwargs)
1480
if credentials is None:
1484
user = credentials['user']
1485
password = credentials['password']
1486
self.assertEquals(expected_user, user)
1487
self.assertEquals(expected_password, password)
1489
def test_empty_config(self):
1490
conf = config.AuthenticationConfig(_file=StringIO())
1491
self.assertEquals({}, conf._get_config())
1492
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
1494
def test_missing_auth_section_header(self):
1495
conf = config.AuthenticationConfig(_file=StringIO('foo = bar'))
1496
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
1498
def test_auth_section_header_not_closed(self):
1499
conf = config.AuthenticationConfig(_file=StringIO('[DEF'))
1500
self.assertRaises(errors.ParseConfigError, conf._get_config)
1502
def test_auth_value_not_boolean(self):
1503
conf = config.AuthenticationConfig(_file=StringIO(
1507
verify_certificates=askme # Error: Not a boolean
1509
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
1511
def test_auth_value_not_int(self):
1512
conf = config.AuthenticationConfig(_file=StringIO(
1516
port=port # Error: Not an int
1518
self.assertRaises(ValueError, conf.get_credentials, 'ftp', 'foo.net')
1520
def test_unknown_password_encoding(self):
1521
conf = config.AuthenticationConfig(_file=StringIO(
1525
password_encoding=unknown
1527
self.assertRaises(ValueError, conf.get_password,
1528
'ftp', 'foo.net', 'joe')
1530
def test_credentials_for_scheme_host(self):
1531
conf = config.AuthenticationConfig(_file=StringIO(
1532
"""# Identity on foo.net
1537
password=secret-pass
1540
self._got_user_passwd('joe', 'secret-pass', conf, 'ftp', 'foo.net')
1542
self._got_user_passwd(None, None, conf, 'http', 'foo.net')
1544
self._got_user_passwd(None, None, conf, 'ftp', 'bar.net')
1546
def test_credentials_for_host_port(self):
1547
conf = config.AuthenticationConfig(_file=StringIO(
1548
"""# Identity on foo.net
1554
password=secret-pass
1557
self._got_user_passwd('joe', 'secret-pass',
1558
conf, 'ftp', 'foo.net', port=10021)
1560
self._got_user_passwd(None, None, conf, 'ftp', 'foo.net')
1562
def test_for_matching_host(self):
1563
conf = config.AuthenticationConfig(_file=StringIO(
1564
"""# Identity on foo.net
1570
[sourceforge domain]
1577
self._got_user_passwd('georges', 'bendover',
1578
conf, 'bzr', 'foo.bzr.sf.net')
1580
self._got_user_passwd(None, None,
1581
conf, 'bzr', 'bbzr.sf.net')
1583
def test_for_matching_host_None(self):
1584
conf = config.AuthenticationConfig(_file=StringIO(
1585
"""# Identity on foo.net
1595
self._got_user_passwd('joe', 'joepass',
1596
conf, 'bzr', 'quux.net')
1597
# no host but different scheme
1598
self._got_user_passwd('georges', 'bendover',
1599
conf, 'ftp', 'quux.net')
1601
def test_credentials_for_path(self):
1602
conf = config.AuthenticationConfig(_file=StringIO(
1618
self._got_user_passwd(None, None,
1619
conf, 'http', host='bar.org', path='/dir3')
1621
self._got_user_passwd('georges', 'bendover',
1622
conf, 'http', host='bar.org', path='/dir2')
1624
self._got_user_passwd('jim', 'jimpass',
1625
conf, 'http', host='bar.org',path='/dir1/subdir')
1627
def test_credentials_for_user(self):
1628
conf = config.AuthenticationConfig(_file=StringIO(
1637
self._got_user_passwd('jim', 'jimpass',
1638
conf, 'http', 'bar.org')
1640
self._got_user_passwd('jim', 'jimpass',
1641
conf, 'http', 'bar.org', user='jim')
1642
# Don't get a different user if one is specified
1643
self._got_user_passwd(None, None,
1644
conf, 'http', 'bar.org', user='georges')
1646
def test_credentials_for_user_without_password(self):
1647
conf = config.AuthenticationConfig(_file=StringIO(
1654
# Get user but no password
1655
self._got_user_passwd('jim', None,
1656
conf, 'http', 'bar.org')
1658
def test_verify_certificates(self):
1659
conf = config.AuthenticationConfig(_file=StringIO(
1666
verify_certificates=False
1673
credentials = conf.get_credentials('https', 'bar.org')
1674
self.assertEquals(False, credentials.get('verify_certificates'))
1675
credentials = conf.get_credentials('https', 'foo.net')
1676
self.assertEquals(True, credentials.get('verify_certificates'))
1679
class TestAuthenticationStorage(tests.TestCaseInTempDir):
1681
def test_set_credentials(self):
1682
conf = config.AuthenticationConfig()
1683
conf.set_credentials('name', 'host', 'user', 'scheme', 'password',
1684
99, path='/foo', verify_certificates=False, realm='realm')
1685
credentials = conf.get_credentials(host='host', scheme='scheme',
1686
port=99, path='/foo',
1688
CREDENTIALS = {'name': 'name', 'user': 'user', 'password': 'password',
1689
'verify_certificates': False, 'scheme': 'scheme',
1690
'host': 'host', 'port': 99, 'path': '/foo',
1692
self.assertEqual(CREDENTIALS, credentials)
1693
credentials_from_disk = config.AuthenticationConfig().get_credentials(
1694
host='host', scheme='scheme', port=99, path='/foo', realm='realm')
1695
self.assertEqual(CREDENTIALS, credentials_from_disk)
1697
def test_reset_credentials_different_name(self):
1698
conf = config.AuthenticationConfig()
1699
conf.set_credentials('name', 'host', 'user', 'scheme', 'password'),
1700
conf.set_credentials('name2', 'host', 'user2', 'scheme', 'password'),
1701
self.assertIs(None, conf._get_config().get('name'))
1702
credentials = conf.get_credentials(host='host', scheme='scheme')
1703
CREDENTIALS = {'name': 'name2', 'user': 'user2', 'password':
1704
'password', 'verify_certificates': True,
1705
'scheme': 'scheme', 'host': 'host', 'port': None,
1706
'path': None, 'realm': None}
1707
self.assertEqual(CREDENTIALS, credentials)
1710
class TestAuthenticationConfig(tests.TestCase):
1711
"""Test AuthenticationConfig behaviour"""
1713
def _check_default_password_prompt(self, expected_prompt_format, scheme,
1714
host=None, port=None, realm=None,
1718
user, password = 'jim', 'precious'
1719
expected_prompt = expected_prompt_format % {
1720
'scheme': scheme, 'host': host, 'port': port,
1721
'user': user, 'realm': realm}
1723
stdout = tests.StringIOWrapper()
1724
stderr = tests.StringIOWrapper()
1725
ui.ui_factory = tests.TestUIFactory(stdin=password + '\n',
1726
stdout=stdout, stderr=stderr)
1727
# We use an empty conf so that the user is always prompted
1728
conf = config.AuthenticationConfig()
1729
self.assertEquals(password,
1730
conf.get_password(scheme, host, user, port=port,
1731
realm=realm, path=path))
1732
self.assertEquals(expected_prompt, stderr.getvalue())
1733
self.assertEquals('', stdout.getvalue())
1735
def _check_default_username_prompt(self, expected_prompt_format, scheme,
1736
host=None, port=None, realm=None,
1741
expected_prompt = expected_prompt_format % {
1742
'scheme': scheme, 'host': host, 'port': port,
1744
stdout = tests.StringIOWrapper()
1745
stderr = tests.StringIOWrapper()
1746
ui.ui_factory = tests.TestUIFactory(stdin=username+ '\n',
1747
stdout=stdout, stderr=stderr)
1748
# We use an empty conf so that the user is always prompted
1749
conf = config.AuthenticationConfig()
1750
self.assertEquals(username, conf.get_user(scheme, host, port=port,
1751
realm=realm, path=path, ask=True))
1752
self.assertEquals(expected_prompt, stderr.getvalue())
1753
self.assertEquals('', stdout.getvalue())
1755
def test_username_defaults_prompts(self):
1756
# HTTP prompts can't be tested here, see test_http.py
1757
self._check_default_username_prompt('FTP %(host)s username: ', 'ftp')
1758
self._check_default_username_prompt(
1759
'FTP %(host)s:%(port)d username: ', 'ftp', port=10020)
1760
self._check_default_username_prompt(
1761
'SSH %(host)s:%(port)d username: ', 'ssh', port=12345)
1763
def test_username_default_no_prompt(self):
1764
conf = config.AuthenticationConfig()
1765
self.assertEquals(None,
1766
conf.get_user('ftp', 'example.com'))
1767
self.assertEquals("explicitdefault",
1768
conf.get_user('ftp', 'example.com', default="explicitdefault"))
1770
def test_password_default_prompts(self):
1771
# HTTP prompts can't be tested here, see test_http.py
1772
self._check_default_password_prompt(
1773
'FTP %(user)s@%(host)s password: ', 'ftp')
1774
self._check_default_password_prompt(
1775
'FTP %(user)s@%(host)s:%(port)d password: ', 'ftp', port=10020)
1776
self._check_default_password_prompt(
1777
'SSH %(user)s@%(host)s:%(port)d password: ', 'ssh', port=12345)
1778
# SMTP port handling is a bit special (it's handled if embedded in the
1780
# FIXME: should we: forbid that, extend it to other schemes, leave
1781
# things as they are that's fine thank you ?
1782
self._check_default_password_prompt('SMTP %(user)s@%(host)s password: ',
1784
self._check_default_password_prompt('SMTP %(user)s@%(host)s password: ',
1785
'smtp', host='bar.org:10025')
1786
self._check_default_password_prompt(
1787
'SMTP %(user)s@%(host)s:%(port)d password: ',
1790
def test_ssh_password_emits_warning(self):
1791
conf = config.AuthenticationConfig(_file=StringIO(
1799
entered_password = 'typed-by-hand'
1800
stdout = tests.StringIOWrapper()
1801
stderr = tests.StringIOWrapper()
1802
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n',
1803
stdout=stdout, stderr=stderr)
1805
# Since the password defined in the authentication config is ignored,
1806
# the user is prompted
1807
self.assertEquals(entered_password,
1808
conf.get_password('ssh', 'bar.org', user='jim'))
1809
self.assertContainsRe(
1811
'password ignored in section \[ssh with password\]')
1813
def test_ssh_without_password_doesnt_emit_warning(self):
1814
conf = config.AuthenticationConfig(_file=StringIO(
1821
entered_password = 'typed-by-hand'
1822
stdout = tests.StringIOWrapper()
1823
stderr = tests.StringIOWrapper()
1824
ui.ui_factory = tests.TestUIFactory(stdin=entered_password + '\n',
1828
# Since the password defined in the authentication config is ignored,
1829
# the user is prompted
1830
self.assertEquals(entered_password,
1831
conf.get_password('ssh', 'bar.org', user='jim'))
1832
# No warning shoud be emitted since there is no password. We are only
1834
self.assertNotContainsRe(
1836
'password ignored in section \[ssh with password\]')
1838
def test_uses_fallback_stores(self):
1839
self.overrideAttr(config, 'credential_store_registry',
1840
config.CredentialStoreRegistry())
1841
store = StubCredentialStore()
1842
store.add_credentials("http", "example.com", "joe", "secret")
1843
config.credential_store_registry.register("stub", store, fallback=True)
1844
conf = config.AuthenticationConfig(_file=StringIO())
1845
creds = conf.get_credentials("http", "example.com")
1846
self.assertEquals("joe", creds["user"])
1847
self.assertEquals("secret", creds["password"])
1850
class StubCredentialStore(config.CredentialStore):
1856
def add_credentials(self, scheme, host, user, password=None):
1857
self._username[(scheme, host)] = user
1858
self._password[(scheme, host)] = password
1860
def get_credentials(self, scheme, host, port=None, user=None,
1861
path=None, realm=None):
1862
key = (scheme, host)
1863
if not key in self._username:
1865
return { "scheme": scheme, "host": host, "port": port,
1866
"user": self._username[key], "password": self._password[key]}
1869
class CountingCredentialStore(config.CredentialStore):
1874
def get_credentials(self, scheme, host, port=None, user=None,
1875
path=None, realm=None):
1880
class TestCredentialStoreRegistry(tests.TestCase):
1882
def _get_cs_registry(self):
1883
return config.credential_store_registry
1885
def test_default_credential_store(self):
1886
r = self._get_cs_registry()
1887
default = r.get_credential_store(None)
1888
self.assertIsInstance(default, config.PlainTextCredentialStore)
1890
def test_unknown_credential_store(self):
1891
r = self._get_cs_registry()
1892
# It's hard to imagine someone creating a credential store named
1893
# 'unknown' so we use that as an never registered key.
1894
self.assertRaises(KeyError, r.get_credential_store, 'unknown')
1896
def test_fallback_none_registered(self):
1897
r = config.CredentialStoreRegistry()
1898
self.assertEquals(None,
1899
r.get_fallback_credentials("http", "example.com"))
1901
def test_register(self):
1902
r = config.CredentialStoreRegistry()
1903
r.register("stub", StubCredentialStore(), fallback=False)
1904
r.register("another", StubCredentialStore(), fallback=True)
1905
self.assertEquals(["another", "stub"], r.keys())
1907
def test_register_lazy(self):
1908
r = config.CredentialStoreRegistry()
1909
r.register_lazy("stub", "bzrlib.tests.test_config",
1910
"StubCredentialStore", fallback=False)
1911
self.assertEquals(["stub"], r.keys())
1912
self.assertIsInstance(r.get_credential_store("stub"),
1913
StubCredentialStore)
1915
def test_is_fallback(self):
1916
r = config.CredentialStoreRegistry()
1917
r.register("stub1", None, fallback=False)
1918
r.register("stub2", None, fallback=True)
1919
self.assertEquals(False, r.is_fallback("stub1"))
1920
self.assertEquals(True, r.is_fallback("stub2"))
1922
def test_no_fallback(self):
1923
r = config.CredentialStoreRegistry()
1924
store = CountingCredentialStore()
1925
r.register("count", store, fallback=False)
1926
self.assertEquals(None,
1927
r.get_fallback_credentials("http", "example.com"))
1928
self.assertEquals(0, store._calls)
1930
def test_fallback_credentials(self):
1931
r = config.CredentialStoreRegistry()
1932
store = StubCredentialStore()
1933
store.add_credentials("http", "example.com",
1934
"somebody", "geheim")
1935
r.register("stub", store, fallback=True)
1936
creds = r.get_fallback_credentials("http", "example.com")
1937
self.assertEquals("somebody", creds["user"])
1938
self.assertEquals("geheim", creds["password"])
1940
def test_fallback_first_wins(self):
1941
r = config.CredentialStoreRegistry()
1942
stub1 = StubCredentialStore()
1943
stub1.add_credentials("http", "example.com",
1944
"somebody", "stub1")
1945
r.register("stub1", stub1, fallback=True)
1946
stub2 = StubCredentialStore()
1947
stub2.add_credentials("http", "example.com",
1948
"somebody", "stub2")
1949
r.register("stub2", stub1, fallback=True)
1950
creds = r.get_fallback_credentials("http", "example.com")
1951
self.assertEquals("somebody", creds["user"])
1952
self.assertEquals("stub1", creds["password"])
1955
class TestPlainTextCredentialStore(tests.TestCase):
1957
def test_decode_password(self):
1958
r = config.credential_store_registry
1959
plain_text = r.get_credential_store()
1960
decoded = plain_text.decode_password(dict(password='secret'))
1961
self.assertEquals('secret', decoded)
1964
# FIXME: Once we have a way to declare authentication to all test servers, we
1965
# can implement generic tests.
1966
# test_user_password_in_url
1967
# test_user_in_url_password_from_config
1968
# test_user_in_url_password_prompted
1969
# test_user_in_config
1970
# test_user_getpass.getuser
1971
# test_user_prompted ?
1972
class TestAuthenticationRing(tests.TestCaseWithTransport):