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
|
#!.venv3/bin/python3
# Copyright 2015 Canonical Ltd.
#
# This file is part of the Cassandra Charm for Juju.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3, as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from itertools import chain
import functools
import unittest
from unittest.mock import patch
from charmhelpers.core import hookenv
from charmhelpers.core.services import ServiceManager
from tests.base import TestCaseBase
import definitions
patch = functools.partial(patch, autospec=True)
class TestDefinitions(TestCaseBase):
def test_get_service_definitions(self):
# We can't really test this in unit tests, but at least we can
# ensure the basic data structure is returned and accepted.
defs = definitions.get_service_definitions()
self.assertIsInstance(defs, list)
for d in defs:
with self.subTest(d=d):
self.assertIsInstance(d, dict)
def test_get_service_definitions_open_ports(self):
config = hookenv.config()
defs = definitions.get_service_definitions()
expected_ports = set([config['rpc_port'],
config['native_transport_port'],
config['storage_port'],
config['ssl_storage_port']])
opened_ports = set(chain(*(d.get('ports', []) for d in defs)))
self.assertSetEqual(opened_ports, expected_ports)
def test_get_service_manager(self):
self.assertIsInstance(definitions.get_service_manager(),
ServiceManager)
@patch('helpers.get_unit_superusers')
@patch('helpers.is_decommissioned')
@patch('helpers.is_cassandra_running')
def test_requires_live_node(self, is_running, is_decommissioned, get_sup):
is_decommissioned.return_value = False # Is not decommissioned.
is_running.return_value = True # Is running.
get_sup.return_value = set([hookenv.local_unit()]) # Creds exist.
self.assertTrue(bool(definitions.RequiresLiveNode()))
@patch('helpers.get_unit_superusers')
@patch('helpers.is_decommissioned')
@patch('helpers.is_cassandra_running')
def test_requires_live_node_decommissioned(self, is_running,
is_decommissioned, get_sup):
is_decommissioned.return_value = True # Is decommissioned.
is_running.return_value = True # Is running.
get_sup.return_value = set([hookenv.local_unit()]) # Creds exist.
self.assertFalse(bool(definitions.RequiresLiveNode()))
@patch('helpers.get_unit_superusers')
@patch('helpers.is_decommissioned')
@patch('helpers.is_cassandra_running')
def test_requires_live_node_down(self, is_running,
is_decommissioned, get_sup):
is_decommissioned.return_value = False # Is not decommissioned.
is_running.return_value = False # Is not running.
get_sup.return_value = set([hookenv.local_unit()]) # Creds exist.
self.assertFalse(bool(definitions.RequiresLiveNode()))
@patch('helpers.get_unit_superusers')
@patch('helpers.is_decommissioned')
@patch('helpers.is_cassandra_running')
def test_requires_live_node_creds(self, is_running,
is_decommissioned, get_sup):
is_decommissioned.return_value = False # Is not decommissioned.
is_running.return_value = True # Is running.
get_sup.return_value = set() # Creds do not exist.
self.assertFalse(bool(definitions.RequiresLiveNode()))
if __name__ == '__main__':
unittest.main(verbosity=2)
|
