Revision 8556d269 kamaki/cli/config/test.py

b/kamaki/cli/config/test.py
38 38
from tempfile import NamedTemporaryFile
39 39
from io import StringIO
40 40

  
41
from kamaki.cli.config import HEADER
42

  
41 43

  
42 44
def _2steps_gen(limit=2):
43 45
    counter, ret = 0, None
......
54 56
class Config(TestCase):
55 57
    """Test Config methods"""
56 58

  
57
    config_file_content = [
58
        '#kamaki config file version 0.9\n',
59
        '[global]\n',
60
        'max_threads = 5\n',
61
        'default_cloud = ~mycloud\n',
62
        'file_cli = pithos\n',
63
        'history_file = /home/user/.kamaki.history\n',
64
        'colors = off\n',
65
        'config_cli = config\n',
66
        'history_cli = history\n',
67
        'log_token = off\n',
68
        'server_cli = cyclades\n',
69
        'user_cli = astakos\n',
70
        'log_data = off\n',
71
        'flavor_cli = cyclades\n',
72
        'image_cli = image\n',
73
        'log_file = /home/user/.kamaki.log\n',
74
        'network_cli = cyclades\n',
75
        'log_pid = off\n',
76
        '\n',
77
        '[cloud "demo"]\n',
78
        'url = https://demo.example.com\n',
79
        'token = t0k3n-0f-d3m0-3x4mp13\n',
80
        '\n',
81
        '[cloud "~mycloud"]\n',
82
        'url = https://example.com\n',
83
        'pithos_container = images\n']
84

  
85 59
    def setUp(self):
86 60
        self.f = NamedTemporaryFile()
87 61

  
88
    def readDown(self):
62
        from kamaki.cli.config import DEFAULTS
63

  
64
        self.DEFAULTS = dict()
65
        for k, v in DEFAULTS.items():
66
            self.DEFAULTS[k] = dict(v) if isinstance(v, dict) else v
67

  
68
        self.config_file_content = [
69
            HEADER,
70
            '[global]\n',
71
            'max_threads = 5\n',
72
            'default_cloud = ~mycloud\n',
73
            'file_cli = pithos\n',
74
            'history_file = /home/user/.kamaki.history\n',
75
            'colors = off\n',
76
            'config_cli = config\n',
77
            'history_cli = history\n',
78
            'log_token = off\n',
79
            'server_cli = cyclades\n',
80
            'user_cli = astakos\n',
81
            'log_data = off\n',
82
            'flavor_cli = cyclades\n',
83
            'image_cli = image\n',
84
            'log_file = /home/user/.kamaki.log\n',
85
            'network_cli = cyclades\n',
86
            'log_pid = off\n',
87
            '\n',
88
            '[cloud "demo"]\n',
89
            'url = https://demo.example.com\n',
90
            'token = t0k3n-0f-d3m0-3x4mp13\n',
91
            '\n',
92
            '[cloud "~mycloud"]\n',
93
            'url = https://example.com\n',
94
            'pithos_container = images\n']
95

  
96
    def tearDown(self):
89 97
        try:
90 98
            self.f.close()
91 99
        except Exception:
92 100
            pass
101
        finally:
102
            from kamaki.cli.config import DEFAULTS
103
            keys = DEFAULTS.keys()
104
            for k in keys:
105
                DEFAULTS.pop(k)
106
            for k, v in self.DEFAULTS.items():
107
                DEFAULTS[k] = v
93 108

  
94 109
    @patch('kamaki.cli.config.Config.remove_section')
95 110
    @patch('kamaki.cli.config.Config.items', return_value=(
......
171 186
        content1.insert(2, '%s\n' % sample)
172 187

  
173 188
        with make_file(content1) as f:
189
            f.seek(0)
174 190
            _cnf = Config(path=f.name)
175
            self.assertEqual(['global.%s' % sample], _cnf.rescue_old_file())
191
            self.assertEqual(
192
                sorted(['global.%s' % sample]), sorted(_cnf.rescue_old_file()))
176 193
        del _cnf
177 194

  
178 195
        content2, sample = list(content0), 'http://www.example2.org'
......
368 385
            _cnf.reload()
369 386
            i.assert_called_once_with(self.f.name)
370 387

  
388
    @patch('kamaki.cli.config.Config.get_cloud', return_value='get cloud')
389
    def test_get(self, get_cloud):
390
        from kamaki.cli.config import Config
391
        _cnf = Config(path=self.f.name)
392
        self.assertEqual('pithos', _cnf.get('global', 'file_cli'))
393
        self.assertEqual(get_cloud.mock_calls, [])
394
        for opt, sec in (('cloud', 'non-existing'), ('non-opt', 'exists')):
395
            self.assertEqual(None, _cnf.get(opt, sec))
396
            self.assertEqual(get_cloud.mock_calls, [])
397
        self.assertEqual('get cloud', _cnf.get('cloud.demo', 'url'))
398
        self.assertEqual(get_cloud.mock_calls[-1], call('demo', 'url'))
399

  
400
    def test_set(self):
401
        from kamaki.cli.config import Config, CLOUD_PREFIX
402
        _cnf = Config(path=self.f.name)
403

  
404
        with patch(
405
                'kamaki.cli.config.Config._cloud_name',
406
                return_value='cn') as _cloud_name:
407
            with patch(
408
                    'kamaki.cli.config.Config.set_cloud',
409
                    return_value='sc') as set_cloud:
410
                self.assertEqual(
411
                    'sc', _cnf.set('%s.sec' % CLOUD_PREFIX, 'opt', 'val'))
412
                self.assertEqual(
413
                    _cloud_name.mock_calls[-1],
414
                    call('%s "sec"' % CLOUD_PREFIX))
415
                self.assertEqual(
416
                    set_cloud.mock_calls[-1], call('cn', 'opt', 'val'))
417

  
418
                self.assertTrue(len(_cnf.items('global')) > 0)
419
                self.assertEqual(None, _cnf.set('global', 'opt', 'val'))
420
                self.assertTrue(('opt', 'val') in _cnf.items('global'))
421

  
422
                self.assertTrue(len(_cnf.items('new')) == 0)
423
                self.assertEqual(None, _cnf.set('new', 'opt', 'val'))
424
                self.assertTrue(('opt', 'val') in _cnf.items('new'))
425

  
426
    def test_remove_option(self):
427
        from kamaki.cli.config import Config
428
        _cnf = Config(path=self.f.name)
429

  
430
        self.assertEqual(len(_cnf.items('no-section')), 0)
431
        _cnf.remove_option('no-section', 'opt', False)
432
        self.assertEqual(len(_cnf.items('no-section')), 0)
433
        _cnf.remove_option('no-section', 'opt', True)
434
        self.assertEqual(len(_cnf.items('no-section')), 0)
435

  
436
        opt_num = len(_cnf.items('global'))
437
        self.assertTrue(opt_num > 0)
438
        _cnf.remove_option('global', 'file_cli', False)
439
        self.assertEqual(len(_cnf.items('global')), opt_num)
440
        _cnf.remove_option('global', 'file_cli', True)
441
        self.assertEqual(len(_cnf.items('global')), opt_num - 1)
442

  
443
        _cnf.set('global', 'server_cli', 'alt-server')
444
        self.assertTrue(('server_cli', 'alt-server') in _cnf.items('global'))
445
        self.assertFalse(('server_cli', 'cyclades') in _cnf.items('global'))
446
        _cnf.remove_option('global', 'server_cli', False)
447
        self.assertFalse(('server_cli', 'alt-server') in _cnf.items('global'))
448
        self.assertTrue(('server_cli', 'cyclades') in _cnf.items('global'))
449
        _cnf.remove_option('global', 'server_cli', True)
450
        self.assertFalse(('server_cli', 'alt-server') in _cnf.items('global'))
451
        self.assertFalse(('server_cli', 'cyclades') in _cnf.items('global'))
452

  
453
    def test_remove_from_cloud(self):
454
        from kamaki.cli.config import Config, CLOUD_PREFIX
455
        _cnf = Config(path=self.f.name)
456

  
457
        d = dict(k1='v1', k2='v2')
458
        with patch('kamaki.cli.config.Config.get', return_value=d) as get:
459
            _cnf.remove_from_cloud('cld', 'k1')
460
            self.assertEqual(d, dict(k2='v2'))
461
            self.assertRaises(KeyError, _cnf.remove_from_cloud, 'cld', 'opt')
462
            self.assertEqual(get.mock_calls, 2 * [call(CLOUD_PREFIX, 'cld')])
463

  
464
    @patch(
465
        'kamaki.cli.config.Config._get_dict',
466
        return_value={'k1': 'v1', 'k2': 'v2'})
467
    def test_keys(self, _get_dict):
468
        from kamaki.cli.config import Config
469
        _cnf = Config(path=self.f.name)
470

  
471
        self.assertEqual(
472
            sorted(['k1', 'k2']), sorted(_cnf.keys('opt', 'boolean')))
473
        _get_dict.assert_called_once_with('opt', 'boolean')
474

  
475
    @patch(
476
        'kamaki.cli.config.Config._get_dict',
477
        return_value={'k1': 'v1', 'k2': 'v2'})
478
    def test_items(self, _get_dict):
479
        from kamaki.cli.config import Config
480
        _cnf = Config(path=self.f.name)
481

  
482
        self.assertEqual(
483
            sorted([('k1', 'v1'), ('k2', 'v2')]),
484
            sorted(_cnf.items('opt', 'boolean')))
485
        _get_dict.assert_called_once_with('opt', 'boolean')
486

  
487
    def test_override(self):
488
        from kamaki.cli.config import Config
489
        _cnf = Config(path=self.f.name)
490

  
491
        _cnf.override('sec', 'opt', 'val')
492
        self.assertEqual(_cnf._overrides['sec']['opt'], 'val')
493

  
494
    def test_write(self):
495
        from kamaki.cli.config import Config, DEFAULTS
496
        _cnf = Config(path=self.f.name)
497

  
498
        exp = '%s[global]\n' % HEADER
499
        exp += ''.join([
500
            '%s = %s\n' % (k, v) for k, v in DEFAULTS['global'].items()])
501
        exp += '\n'
502

  
503
        _cnf.write()
504
        self.f.seek(0)
505
        self.assertEqual(self.f.read(), exp)
506

  
507
        del _cnf
508
        with NamedTemporaryFile() as f:
509
            f.write('\n'.join(self.config_file_content))
510
            f.flush()
511
            _cnf = Config(path=f.name)
512
            f.seek(0)
513
            self.assertEqual(f.read(), '\n'.join(self.config_file_content))
514
            _cnf.write()
515
            f.seek(0)
516
            file_contents = f.read()
517
            for line in self.config_file_content:
518
                self.assertTrue(line in file_contents)
519
            _cnf.set('sec', 'opt', 'val')
520
            _cnf.set('global', 'opt', 'val')
521
            _cnf.set('global', 'file_cli', 'val')
522
            _cnf.write()
523
            f.seek(0)
524
            file_contents = f.read()
525
            for line in ('file_cli = val\n', '[sec]\n', 'opt = val\n'):
526
                self.assertTrue(line in file_contents)
527

  
371 528

  
372 529
if __name__ == '__main__':
373 530
    from sys import argv

Also available in: Unified diff