From cb606f45e3852432787ed895dc55665caa950161 Mon Sep 17 00:00:00 2001 From: Tim Rozet Date: Fri, 8 Sep 2017 16:57:36 -0400 Subject: Migrates clean to python ci/clean.sh will be removed in a future patch after releng is updated to use python. JIRA: APEX-509 JIRA: APEX-319 Change-Id: If890db2fc5a31833ad28ec6f04589e25457bd380 Signed-off-by: Tim Rozet --- apex/tests/test_apex_network_jumphost.py | 276 +++++++++++++++++++++++++++++++ 1 file changed, 276 insertions(+) create mode 100644 apex/tests/test_apex_network_jumphost.py (limited to 'apex/tests/test_apex_network_jumphost.py') diff --git a/apex/tests/test_apex_network_jumphost.py b/apex/tests/test_apex_network_jumphost.py new file mode 100644 index 00000000..a23f1c56 --- /dev/null +++ b/apex/tests/test_apex_network_jumphost.py @@ -0,0 +1,276 @@ +############################################################################## +# Copyright (c) 2016 Dan Radez (Red Hat) +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################## + +import os +import shutil +import subprocess + +from apex import NetworkSettings +from apex.tests import constants as con +from apex.common import constants as apex_constants +from apex.network import jumphost +from apex.common.exceptions import JumpHostNetworkException +from ipaddress import IPv4Interface +from mock import patch +from nose.tools import ( + assert_is_instance, + assert_dict_equal, + assert_raises, + assert_true, + assert_false +) + + +def bridge_show_output(*args, **kwargs): + return b""" + b6f1b54a-b8ba-4e86-9c5b-733ab71b5712 + Bridge br-admin + Port br-admin + Interface br-admin + type: internal + ovs_version: "2.5.0" +""" + + +def bridge_port_list(*args, **kwargs): + return b""" +enp6s0 +vnet1 +""" + + +def subprocess_exception(*args, **kwargs): + raise subprocess.CalledProcessError(returncode=2, cmd='dummy') + + +class TestNetworkJumpHost: + @classmethod + def setup_class(cls): + """This method is run once for each class before any tests are run""" + + @classmethod + def teardown_class(cls): + """This method is run once for each class _after_ all tests are run""" + + def setup(self): + """This method is run once before _each_ test method is executed""" + + def teardown(self): + """This method is run once after _each_ test method is executed""" + + @patch('subprocess.check_output', side_effect=bridge_show_output) + def test_is_ovs_bridge(self, bridge_output_function): + assert_true(jumphost.is_ovs_bridge('br-admin')) + assert_false(jumphost.is_ovs_bridge('br-blah')) + + @patch('subprocess.check_output', side_effect=bridge_port_list) + def test_dump_ovs_ports(self, bridge_function): + output = jumphost.dump_ovs_ports('br-admin') + assert_is_instance(output, list) + assert 'enp6s0' in output + + def test_generate_ifcfg_params(self): + output = jumphost.generate_ifcfg_params( + os.path.join(con.TEST_DUMMY_CONFIG, 'ifcfg-br-external'), + apex_constants.EXTERNAL_NETWORK) + assert_is_instance(output, dict) + assert output['IPADDR'] == '172.30.9.66' + assert output['PEERDNS'] == 'no' + + def test_negative_generate_ifcfg_params(self): + assert_raises(JumpHostNetworkException, jumphost.generate_ifcfg_params, + os.path.join(con.TEST_DUMMY_CONFIG, + 'bad_ifcfg-br-external'), + apex_constants.EXTERNAL_NETWORK) + + @patch('subprocess.check_call') + @patch('apex.network.ip_utils.get_interface', return_value=IPv4Interface( + '10.10.10.2')) + def test_configure_bridges_ip_exists(self, interface_function, + subprocess_func): + ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR, + 'network', 'network_settings.yaml')) + assert jumphost.configure_bridges(ns) is None + + @patch('subprocess.check_call') + @patch('apex.network.ip_utils.get_interface', return_value=None) + def test_configure_bridges_no_ip(self, interface_function, + subprocess_func): + ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR, + 'network', 'network_settings.yaml')) + assert jumphost.configure_bridges(ns) is None + + @patch('subprocess.check_call', side_effect=subprocess_exception) + @patch('apex.network.ip_utils.get_interface', return_value=None) + def test_negative_configure_bridges(self, interface_function, + subprocess_func): + ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR, + 'network', 'network_settings.yaml')) + assert_raises(subprocess.CalledProcessError, + jumphost.configure_bridges, ns) + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=[]) + def test_attach_interface(self, dump_ports_func, is_bridge_func, + subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-enpfakes0')) + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + jumphost.NET_CFG_PATH = ifcfg_dir + output = jumphost.attach_interface_to_ovs('br-admin', 'enpfakes0', + 'admin') + assert output is None + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0')) + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0.orig')) + + for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig', + 'ifcfg-br-admin'): + ifcfg_path = os.path.join(ifcfg_dir, ifcfg) + if os.path.isfile(ifcfg_path): + os.remove(ifcfg_path) + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=['dummy_int']) + def test_already_attached_interface(self, dump_ports_func, is_bridge_func, + subprocess_func): + output = jumphost.attach_interface_to_ovs('br-dummy', 'dummy_int', + 'admin') + assert output is None + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=[]) + def test_negative_attach_interface(self, dump_ports_func, is_bridge_func, + subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + jumphost.NET_CFG_PATH = ifcfg_dir + assert_raises(FileNotFoundError, jumphost.attach_interface_to_ovs, + 'br-dummy', 'dummy_int', 'admin') + + @patch('subprocess.check_call', side_effect=subprocess_exception) + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=[]) + def test_negative_attach_interface_process_error( + self, dump_ports_func, is_bridge_func, subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-enpfakes0')) + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + jumphost.NET_CFG_PATH = ifcfg_dir + assert_raises(subprocess.CalledProcessError, + jumphost.attach_interface_to_ovs, + 'br-admin', 'enpfakes0', 'admin') + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0')) + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0.orig')) + + for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig', + 'ifcfg-br-admin'): + ifcfg_path = os.path.join(ifcfg_dir, ifcfg) + if os.path.isfile(ifcfg_path): + os.remove(ifcfg_path) + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0']) + def test_detach_interface(self, dump_ports_func, is_bridge_func, + subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + jumphost.NET_CFG_PATH = ifcfg_dir + output = jumphost.detach_interface_from_ovs('admin') + assert output is None + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0')) + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + + for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig', + 'ifcfg-br-admin'): + ifcfg_path = os.path.join(ifcfg_dir, ifcfg) + if os.path.isfile(ifcfg_path): + os.remove(ifcfg_path) + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=False) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=[]) + def test_detach_interface_no_bridge(self, dump_ports_func, + is_bridge_func, subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + jumphost.NET_CFG_PATH = ifcfg_dir + output = jumphost.detach_interface_from_ovs('admin') + assert output is None + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=[]) + def test_detach_interface_no_int_to_remove(self, dump_ports_func, + is_bridge_func, + subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + jumphost.NET_CFG_PATH = ifcfg_dir + output = jumphost.detach_interface_from_ovs('admin') + assert output is None + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0']) + def test_negative_detach_interface(self, dump_ports_func, is_bridge_func, + subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + jumphost.NET_CFG_PATH = ifcfg_dir + assert_raises(FileNotFoundError, jumphost.detach_interface_from_ovs, + 'admin') + + @patch('subprocess.check_call', side_effect=subprocess_exception) + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + @patch('apex.network.jumphost.dump_ovs_ports', return_value=['enpfakes0']) + def test_negative_detach_interface_process_error( + self, dump_ports_func, is_bridge_func, subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + jumphost.NET_CFG_PATH = ifcfg_dir + assert_raises(subprocess.CalledProcessError, + jumphost.detach_interface_from_ovs, 'admin') + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-enpfakes0')) + assert os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + + for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig', + 'ifcfg-br-admin'): + ifcfg_path = os.path.join(ifcfg_dir, ifcfg) + if os.path.isfile(ifcfg_path): + os.remove(ifcfg_path) + + @patch('subprocess.check_call') + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + def test_remove_ovs_bridge(self, is_bridge_func, subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + jumphost.NET_CFG_PATH = ifcfg_dir + shutil.copyfile(os.path.join(ifcfg_dir, 'ifcfg-br-dummy'), + os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None + assert not os.path.isfile(os.path.join(ifcfg_dir, 'ifcfg-br-admin')) + + # test without file + assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None + + @patch('subprocess.check_call', side_effect=subprocess_exception) + @patch('apex.network.jumphost.is_ovs_bridge', return_value=True) + def test_negative_remove_ovs_bridge(self, is_bridge_func, subprocess_func): + ifcfg_dir = con.TEST_DUMMY_CONFIG + jumphost.NET_CFG_PATH = ifcfg_dir + assert_raises(subprocess.CalledProcessError, + jumphost.remove_ovs_bridge, + apex_constants.ADMIN_NETWORK) -- cgit 1.2.3-korg