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
121
122
123
124
125
126
127
128
129
130
131
132
|
from unittest import TestCase
from mock import patch, call
from chaos_monkey_base import Chaos
from chaos.net import Net
class TestNet(TestCase):
def _run_test(self, method_call, assert_args, **kwargs):
net = Net()
with patch('utility.check_output', autospec=True) as mock:
getattr(net, method_call)(**kwargs)
mock.assert_called_once_with(assert_args)
def _run_mock_calls(self, method_call, call_list):
net = Net()
with patch('utility.check_output', autospec=True) as mock:
getattr(net, method_call)()
self.assertEqual(mock.mock_calls, call_list)
def test_reset(self):
self._run_test('reset', ['ufw', 'reset'])
def test_default_deny(self):
self._run_test('default_deny', ['ufw', 'default', 'deny'])
def test_default_allow(self):
self._run_test('default_allow', ['ufw', 'default', 'allow'])
def test_allow_ssh(self):
self._run_test('allow_ssh', ['ufw', 'allow', 'ssh'])
def test_deny_ssh(self):
self._run_test('deny_ssh', ['ufw', 'deny', 'ssh'])
def test_deny_all_incoming_and_outgoing_except_ssh(self):
self._run_mock_calls(
'deny_all_incoming_and_outgoing_except_ssh',
[self._allow_ssh_call(), self._default_deny_call(),
self._allow_ssh_call(),
call(['ufw', 'deny', 'out', 'to', 'any'])])
def test_allow_all_incoming_and_outgoing(self):
self._run_mock_calls(
'allow_all_incoming_and_outgoing',
[self._default_allow_call(),
call(['ufw', 'delete', 'deny', 'out', 'to', 'any'])])
def test_deny_all_incoming_except_ssh(self):
self._run_mock_calls(
'deny_all_incoming_except_ssh',
[self._allow_ssh_call(), self._default_deny_call()])
def test_allow_all_incoming(self):
self._run_test('allow_all_incoming', ['ufw', 'default', 'allow'])
def test_deny_all_outgoing_except_ssh(self):
self._run_mock_calls(
'deny_all_outgoing_except_ssh',
[self._allow_ssh_call(),
call(['ufw', 'deny', 'out', 'to', 'any'])])
def test_allow_all_outgoing(self):
self._run_test(
'allow_all_outgoing',
['ufw', 'delete', 'deny', 'out', 'to', 'any'])
def test_deny_port(self):
self._run_test(
'deny_port', ['ufw', 'deny', '8080'], port=8080)
def test_allow_port(self):
self._run_test(
'allow_port', ['ufw', 'delete', 'deny', '8080'], port=8080)
def test_deny_state_server(self):
self._run_test('deny_state_server', ['ufw', 'deny', '37017'])
def test_allow_state_server(self):
self._run_test(
'allow_state_server', ['ufw', 'delete', 'deny', '37017'])
def test_deny_api_server(self):
self._run_test('deny_api_server', ['ufw', 'deny', '17070'])
def test_allow_api_server(self):
self._run_test('allow_api_server', ['ufw', 'delete', 'deny', '17070'])
def test_deny_sys_log(self):
self._run_test('deny_sys_log', ['ufw', 'deny', '6514'])
def test_allow_sys_log(self):
self._run_test('allow_sys_log', ['ufw', 'delete', 'deny', '6514'])
def test_get_chaos(self):
net = Net()
chaos = net.get_chaos()
self.assertEqual(len(chaos), 7)
self.assertItemsEqual(
self._get_all_command_str(chaos), self._command_strings())
for c in chaos:
self.assertEqual('net', c.group)
def test_add_chaos(self):
net = Net()
chaos = net._add_chaos('enable', 'disable', 'command')
self.assertIs(type(chaos), Chaos)
self.assertEqual(chaos.enable, 'enable')
self.assertEqual(chaos.disable, 'disable')
self.assertEqual(chaos.group, 'net')
self.assertEqual(chaos.command_str, 'command')
def _get_all_command_str(self, chaos):
return [c.command_str for c in chaos]
def _command_strings(self):
return ['deny-all', 'deny-incoming', 'deny-outgoing', 'allow-ssh',
'deny-state-server', 'deny-api-server', 'deny-sys-log']
def test_shutdown(self):
self._run_test('reset', ['ufw', 'reset'])
def _allow_ssh_call(self):
return call(['ufw', 'allow', 'ssh'])
def _default_deny_call(self):
return call(['ufw', 'default', 'deny'])
def _default_allow_call(self):
return call(['ufw', 'default', 'allow'])
|