| 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
|