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=False)
207 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
208 def test_detach_interface_no_bridge(self, dump_ports_func,
209 is_bridge_func, subprocess_func):
210 ifcfg_dir = con.TEST_DUMMY_CONFIG
211 jumphost.NET_CFG_PATH = ifcfg_dir
212 output = jumphost.detach_interface_from_ovs('admin')
213 assert output is None
215 @patch('subprocess.check_call')
216 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
217 @patch('apex.network.jumphost.dump_ovs_ports', return_value=[])
218 def test_detach_interface_no_int_to_remove(self, dump_ports_func,
221 ifcfg_dir = con.TEST_DUMMY_CONFIG
222 jumphost.NET_CFG_PATH = ifcfg_dir
223 output = jumphost.detach_interface_from_ovs('admin')
224 assert output is None
226 @patch('subprocess.check_call')
227 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
228 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0'])
229 def test_negative_detach_interface(self, dump_ports_func, is_bridge_func,
231 ifcfg_dir = con.TEST_DUMMY_CONFIG
232 jumphost.NET_CFG_PATH = ifcfg_dir
233 assert_raises(FileNotFoundError, jumphost.detach_interface_from_ovs,
236 @patch('subprocess.check_call', side_effect=subprocess_exception)
237 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
238 @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0'])
239 def test_negative_detach_interface_process_error(
240 self, dump_ports_func, is_bridge_func, subprocess_func):
241 ifcfg_dir = con.TEST_DUMMY_CONFIG
242 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
243 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
244 jumphost.NET_CFG_PATH = ifcfg_dir
245 assert_raises(subprocess.CalledProcessError,
246 jumphost.detach_interface_from_ovs, 'admin')
247 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0'))
248 assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
250 for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
252 ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
253 if os.path.isfile(ifcfg_path):
254 os.remove(ifcfg_path)
256 @patch('subprocess.check_call')
257 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
258 def test_remove_ovs_bridge(self, is_bridge_func, subprocess_func):
259 ifcfg_dir = con.TEST_DUMMY_CONFIG
260 jumphost.NET_CFG_PATH = ifcfg_dir
261 shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'),
262 os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
263 assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None
264 assert not os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin'))
267 assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None
269 @patch('subprocess.check_call', side_effect=subprocess_exception)
270 @patch('apex.network.jumphost.is_ovs_bridge', return_value=True)
271 def test_negative_remove_ovs_bridge(self, is_bridge_func, subprocess_func):
272 ifcfg_dir = con.TEST_DUMMY_CONFIG
273 jumphost.NET_CFG_PATH = ifcfg_dir
274 assert_raises(subprocess.CalledProcessError,
275 jumphost.remove_ovs_bridge,
276 apex_constants.ADMIN_NETWORK)