1 ##############################################################################
2 # Copyright (c) 2016 Dan Radez (Red Hat)
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
14 from apex import NetworkSettings
15 from apex.tests import constants as con
16 from apex.common import constants as apex_constants
17 from apex.network import jumphost
18 from apex.common.exceptions import JumpHostNetworkException
19 from ipaddress import IPv4Interface
20 from mock import patch
21 from nose.tools import (
30 def bridge_show_output(*args, **kwargs):
32 b6f1b54a-b8ba-4e86-9c5b-733ab71b5712
41 def bridge_port_list(*args, **kwargs):
48 def subprocess_exception(*args, **kwargs):
49 raise subprocess.CalledProcessError(returncode=2, cmd='dummy')
52 class TestNetworkJumpHost:
55 """This method is run once for each class before any tests are run"""
58 def teardown_class(cls):
59 """This method is run once for each class _after_ all tests are run"""
62 """This method is run once before _each_ test method is executed"""
65 """This method is run once after _each_ test method is executed"""
67 @patch('subprocess.check_output', side_effect=bridge_show_output)
68 def test_is_ovs_bridge(self, bridge_output_function):
69 assert_true(jumphost.is_ovs_bridge('br-admin'))
70 assert_false(jumphost.is_ovs_bridge('br-blah'))
72 @patch('subprocess.check_output', side_effect=bridge_port_list)
73 def test_dump_ovs_ports(self, bridge_function):
74 output = jumphost.dump_ovs_ports('br-admin')
75 assert_is_instance(output, list)
76 assert 'enp6s0' in output
78 def test_generate_ifcfg_params(self):
79 output = jumphost.generate_ifcfg_params(
80 os.path.join(con.TEST_DUMMY_CONFIG, 'ifcfg-br-external'),
81 apex_constants.EXTERNAL_NETWORK)
82 assert_is_instance(output, dict)
83 assert output['IPADDR'] == '172.30.9.66'
84 assert output['PEERDNS'] == 'no'
86 def test_negative_generate_ifcfg_params(self):
87 assert_raises(JumpHostNetworkException, jumphost.generate_ifcfg_params,
88 os.path.join(con.TEST_DUMMY_CONFIG,
89 'bad_ifcfg-br-external'),
90 apex_constants.EXTERNAL_NETWORK)
92 @patch('subprocess.check_call')
93 @patch('apex.network.ip_utils.get_interface', return_value=IPv4Interface(
95 def test_configure_bridges_ip_exists(self, interface_function,
97 ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR,
98 'network', 'network_settings.yaml'))
99 assert jumphost.configure_bridges(ns) is None
101 @patch('subprocess.check_call')
102 @patch('apex.network.ip_utils.get_interface', return_value=None)
103 def test_configure_bridges_no_ip(self, interface_function,
105 ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR,
106 'network', 'network_settings.yaml'))
107 assert jumphost.configure_bridges(ns) is None
109 @patch('subprocess.check_call', side_effect=subprocess_exception)
110 @patch('apex.network.ip_utils.get_interface', return_value=None)
111 def test_negative_configure_bridges(self, interface_function,
113 ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR,
114 'network', 'network_settings.yaml'))
115 assert_raises(subprocess.CalledProcessError,
116 jumphost.configure_bridges, ns)
118 @patch('subprocess.check_call')
119 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
120 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
121 def test_attach_interface(self, dump_ports_func, is_bridge_func,
123 ifcfg_dir = con.TEST_DUMMY_CONFIG
124 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-dummy'),
125 os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
126 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
127 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
128 jumphost.NET_CFG_PATH = ifcfg_dir
129 output = jumphost.attach_interface_to_ovs('br-admin', 'enpfakes0',
131 assert output is None
132 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
133 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
134 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0.orig'))
136 for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
138 ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
139 if os.path.isfile(ifcfg_path):
140 os.remove(ifcfg_path)
142 @patch('subprocess.check_call')
143 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
144 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['dummy_int'])
145 def test_already_attached_interface(self, dump_ports_func, is_bridge_func,
147 output = jumphost.attach_interface_to_ovs('br-dummy', 'dummy_int',
149 assert output is None
151 @patch('subprocess.check_call')
152 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
153 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
154 def test_negative_attach_interface(self, dump_ports_func, is_bridge_func,
156 ifcfg_dir = con.TEST_DUMMY_CONFIG
157 jumphost.NET_CFG_PATH = ifcfg_dir
158 assert_raises(FileNotFoundError, jumphost.attach_interface_to_ovs,
159 'br-dummy', 'dummy_int', 'admin')
161 @patch('subprocess.check_call', side_effect=subprocess_exception)
162 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
163 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
164 def test_negative_attach_interface_process_error(
165 self, dump_ports_func, is_bridge_func, subprocess_func):
166 ifcfg_dir = con.TEST_DUMMY_CONFIG
167 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-dummy'),
168 os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
169 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
170 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
171 jumphost.NET_CFG_PATH = ifcfg_dir
172 assert_raises(subprocess.CalledProcessError,
173 jumphost.attach_interface_to_ovs,
174 'br-admin', 'enpfakes0', 'admin')
175 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
176 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
177 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0.orig'))
179 for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
181 ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
182 if os.path.isfile(ifcfg_path):
183 os.remove(ifcfg_path)
185 @patch('subprocess.check_call')
186 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
187 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0'])
188 def test_detach_interface(self, dump_ports_func, is_bridge_func,
190 ifcfg_dir = con.TEST_DUMMY_CONFIG
191 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
192 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
193 jumphost.NET_CFG_PATH = ifcfg_dir
194 output = jumphost.detach_interface_from_ovs('admin')
195 assert output is None
196 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
197 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
199 for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
201 ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
202 if os.path.isfile(ifcfg_path):
203 os.remove(ifcfg_path)
205 @patch('subprocess.check_call')
206 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
207 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0'])
208 def test_detach_interface_orig_exists(self, dump_ports_func,
209 is_bridge_func, subprocess_func):
210 ifcfg_dir = con.TEST_DUMMY_CONFIG
211 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
212 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
213 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-dummy'),
214 os.path.join(ifcfg_dir, 'ifcfg-enpfakes0.orig'))
215 jumphost.NET_CFG_PATH = ifcfg_dir
216 output = jumphost.detach_interface_from_ovs('admin')
217 assert output is None
218 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
219 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
220 assert not os.path.isfile(os.path.join(ifcfg_dir,
221 'ifcfg-enpfakes0.orig'))
222 for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
224 ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
225 if os.path.isfile(ifcfg_path):
226 os.remove(ifcfg_path)
228 @patch('subprocess.check_call')
229 @patch('apex.network.jumphost.is_ovs_bridge', return_value=False)
230 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
231 def test_detach_interface_no_bridge(self, dump_ports_func,
232 is_bridge_func, subprocess_func):
233 ifcfg_dir = con.TEST_DUMMY_CONFIG
234 jumphost.NET_CFG_PATH = ifcfg_dir
235 output = jumphost.detach_interface_from_ovs('admin')
236 assert output is None
238 @patch('subprocess.check_call')
239 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
240 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
241 def test_detach_interface_no_int_to_remove(self, dump_ports_func,
244 ifcfg_dir = con.TEST_DUMMY_CONFIG
245 jumphost.NET_CFG_PATH = ifcfg_dir
246 output = jumphost.detach_interface_from_ovs('admin')
247 assert output is None
249 @patch('subprocess.check_call')
250 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
251 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0'])
252 def test_negative_detach_interface(self, dump_ports_func, is_bridge_func,
254 ifcfg_dir = con.TEST_DUMMY_CONFIG
255 jumphost.NET_CFG_PATH = ifcfg_dir
256 assert_raises(FileNotFoundError, jumphost.detach_interface_from_ovs,
259 @patch('subprocess.check_call', side_effect=subprocess_exception)
260 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
261 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0'])
262 def test_negative_detach_interface_process_error(
263 self, dump_ports_func, is_bridge_func, subprocess_func):
264 ifcfg_dir = con.TEST_DUMMY_CONFIG
265 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
266 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
267 jumphost.NET_CFG_PATH = ifcfg_dir
268 assert_raises(subprocess.CalledProcessError,
269 jumphost.detach_interface_from_ovs, 'admin')
270 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
271 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
273 for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
275 ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
276 if os.path.isfile(ifcfg_path):
277 os.remove(ifcfg_path)
279 @patch('subprocess.check_call')
280 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
281 def test_remove_ovs_bridge(self, is_bridge_func, subprocess_func):
282 ifcfg_dir = con.TEST_DUMMY_CONFIG
283 jumphost.NET_CFG_PATH = ifcfg_dir
284 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
285 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
286 assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None
287 assert not os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
290 assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None
292 @patch('subprocess.check_call', side_effect=subprocess_exception)
293 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
294 def test_negative_remove_ovs_bridge(self, is_bridge_func, subprocess_func):
295 ifcfg_dir = con.TEST_DUMMY_CONFIG
296 jumphost.NET_CFG_PATH = ifcfg_dir
297 assert_raises(subprocess.CalledProcessError,
298 jumphost.remove_ovs_bridge,
299 apex_constants.ADMIN_NETWORK)