1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
import json
import mock
import pkg_resources
import unittest
from cloudfoundry import utils
class TestUtils(unittest.TestCase):
@mock.patch('cloudfoundry.utils.sh.juju')
def test_current_env(self, check_output):
utils.current_env()
check_output.assert_called_once_with('switch')
def test_flatten_relations(self):
r = utils.flatten_relations([
['mysql', ['a', 'b']],
['website', 'c']])
self.assertEqual(r, set((
('a', 'mysql'),
('b', 'mysql'),
('c', 'website'))))
def test_flatten_reality(self):
data = json.loads(
pkg_resources.resource_string(__name__, 'status.json'))
r = utils.flatten_reality(data)
self.assertEqual(r,
set([(u'etcd', u'etcd:cluster'),
(u'mysql', u'mysql:cluster'),
(u'nats', u'router:nats'),
(u'nats:nats', u'router')])
)
def test_rel_exists(self):
data = json.loads(
pkg_resources.resource_string(__name__, 'status.json'))
self.assertTrue(utils.rel_exists(data, 'etcd', 'etcd:cluster'))
self.assertTrue(utils.rel_exists(data, 'etcd:cluster', 'etcd'))
self.assertTrue(utils.rel_exists(data, 'nats', 'router:nats'))
self.assertTrue(utils.rel_exists(data, 'router:nats', 'nats'))
self.assertFalse(utils.rel_exists(data, 'nats', 'rabbitmq'))
def test_deep_merge(self):
initial = {'properties': {'job1': {'prop1': 'val1'}, 'job2': None}}
additional = {'properties': {'job1': {'prop2': 'val2'}, 'job2': {'prop3': 'val3'}}}
expected = {
'properties': {
'job1': {'prop1': 'val1', 'prop2': 'val2'},
'job2': {'prop3': 'val3'},
},
}
actual = utils.deepmerge(initial, additional)
self.assertEqual(actual, expected)
def test_nested_dict(self):
nd = utils.NestedDict()
nd['foo.bar.baz'] = 'qux'
nd['foo.moo'] = 'mux'
self.assertEqual(nd, {
'foo': {
'bar': {'baz': 'qux'},
'moo': 'mux',
}
})
self.assertEqual(nd['foo.bar.baz'], 'qux')
def test_nested_dict_iterable(self):
nd = utils.NestedDict((('a', 'b'),
('b.c.d', 1)))
self.assertEqual(nd['a'], 'b')
self.assertEqual(nd['b.c.d'], 1)
def test_nested_dict_init(self):
nd = utils.NestedDict({'a': 'b', 'b.c.d': 1})
self.assertEqual(nd['a'], 'b')
self.assertEqual(nd['b.c.d'], 1)
self.assertEqual(nd['b']['c']['d'], 1)
def test_parse_config(self):
fn = pkg_resources.resource_filename(__name__, 'server.conf')
conf = utils.parse_config(fn)
self.assertEqual(conf['credentials.user'], 'user-admin')
self.assertEqual(conf['credentials.password'], 'xxx-admin')
self.assertEqual(conf['server.port'], 8888)
self.assertEqual(conf['server.repository'], 'build')
def test_parse_config_with_defaults(self):
fn = pkg_resources.resource_filename(__name__, 'server.conf')
conf = utils.parse_config(fn, {
'server.port': 8000,
'server.address': '127.0.0.1',
})
self.assertEqual(conf['credentials.user'], 'user-admin')
self.assertEqual(conf['credentials.password'], 'xxx-admin')
self.assertEqual(conf['server.port'], 8888)
self.assertEqual(conf['server.address'], '127.0.0.1')
self.assertEqual(conf['server.repository'], 'build')
def test_to_ip_already_ip(self):
self.assertEqual(utils.to_ip('127.0.0.1'), '127.0.0.1')
def test_to_ip_invalid(self):
self.assertEqual(utils.to_ip(''), None)
self.assertEqual(utils.to_ip(None), None)
@mock.patch('subprocess.check_output')
def test_to_ip(self, mcheck_output):
mcheck_output.return_value = ' foo.maas\n 0.0.0.0\n'
self.assertEqual(utils.to_ip('foo'), '0.0.0.0')
@mock.patch('subprocess.check_output')
def test_to_ip_none(self, mcheck_output):
mcheck_output.return_value = ' foo.maas\n bar\n'
self.assertEqual(utils.to_ip('foo'), None)
|