3
3
from mock import patch, call
5
from chaos_monkey_base import Chaos
5
6
from chaos.net import Net
8
9
class TestNet(TestCase):
10
def _run_test(self, method_call, assert_args):
11
def _run_test(self, method_call, assert_args, **kwargs):
12
13
with patch('utility.check_output', autospec=True) as mock:
13
getattr(net, method_call)()
14
getattr(net, method_call)(**kwargs)
14
15
mock.assert_called_once_with(assert_args)
16
17
def _run_mock_calls(self, method_call, call_list):
69
70
def test_deny_port(self):
71
'deny_port', ['ufw', 'deny', '8080'])
72
'deny_port', ['ufw', 'deny', '8080'], port=8080)
74
def test_allow_port(self):
76
'allow_port', ['ufw', 'delete', 'deny', '8080'], port=8080)
78
def test_deny_state_server(self):
79
self._run_test('deny_state_server', ['ufw', 'deny', '37017'])
81
def test_allow_state_server(self):
83
'allow_state_server', ['ufw', 'delete', 'deny', '37017'])
85
def test_deny_api_server(self):
86
self._run_test('deny_api_server', ['ufw', 'deny', '17070'])
88
def test_allow_api_server(self):
89
self._run_test('allow_api_server', ['ufw', 'delete', 'deny', '17070'])
91
def test_deny_sys_log(self):
92
self._run_test('deny_sys_log', ['ufw', 'deny', '6514'])
94
def test_allow_sys_log(self):
95
self._run_test('allow_sys_log', ['ufw', 'delete', 'deny', '6514'])
73
97
def test_get_chaos(self):
75
99
chaos = net.get_chaos()
76
self.assertEqual(len(chaos), 5)
100
self.assertEqual(len(chaos), 7)
77
101
self.assertItemsEqual(
78
102
self._get_all_command_str(chaos), self._command_strings())
80
104
self.assertEqual('net', c.group)
106
def test_add_chaos(self):
108
chaos = net._add_chaos('enable', 'disable', 'command')
109
self.assertIs(type(chaos), Chaos)
110
self.assertEqual(chaos.enable, 'enable')
111
self.assertEqual(chaos.disable, 'disable')
112
self.assertEqual(chaos.group, 'net')
113
self.assertEqual(chaos.command_str, 'command')
82
115
def _get_all_command_str(self, chaos):
83
116
return [c.command_str for c in chaos]
85
118
def _command_strings(self):
86
119
return ['deny-all', 'deny-incoming', 'deny-outgoing', 'allow-ssh',
120
'deny-state-server', 'deny-api-server', 'deny-sys-log']
89
122
def test_shutdown(self):
90
123
self._run_test('reset', ['ufw', 'reset'])