apex/tests/test_apex_network_jumphost.py

   1 ##############################################################################
   2 # Copyright (c) 2016 Dan Radez (Red Hat)
   3 #
   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 ##############################################################################
   9
  10 import os
  11 import shutil
  12 import subprocess
  13
  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 (
  22     assert_is_instance,
  23     assert_dict_equal,
  24     assert_raises,
  25     assert_true,
  26     assert_false
  27 )
  28
  29
  30 def bridge_show_output(*args, **kwargs):
  31     return b"""
  32     b6f1b54a-b8ba-4e86-9c5b-733ab71b5712
  33     Bridge br-admin
  34         Port br-admin
  35             Interface br-admin
  36                 type: internal
  37     ovs_version: "2.5.0"
  38 """
  39
  40
  41 def bridge_port_list(*args, **kwargs):
  42     return b"""
  43 enp6s0
  44 vnet1
  45 """
  46
  47
  48 def subprocess_exception(*args, **kwargs):
  49     raise subprocess.CalledProcessError(returncode=2, cmd='dummy')
  50
  51
  52 class TestNetworkJumpHost:
  53     @classmethod
  54     def setup_class(cls):
  55         """This method is run once for each class before any tests are run"""
  56
  57     @classmethod
  58     def teardown_class(cls):
  59         """This method is run once for each class _after_ all tests are run"""
  60
  61     def setup(self):
  62         """This method is run once before _each_ test method is executed"""
  63
  64     def teardown(self):
  65         """This method is run once after _each_ test method is executed"""
  66
  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'))
  71
  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
  77
  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'
  85
  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)
  91
  92     @patch('subprocess.check_call')
  93     @patch('apex.network.ip_utils.get_interface', return_value=IPv4Interface(
  94         '10.10.10.2'))
  95     def test_configure_bridges_ip_exists(self, interface_function,
  96                                          subprocess_func):
  97         ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR,
  98                                           'network', 'network_settings.yaml'))
  99         assert jumphost.configure_bridges(ns) is None
 100
 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,
 104                                      subprocess_func):
 105         ns = NetworkSettings(os.path.join(con.TEST_CONFIG_DIR,
 106                                           'network', 'network_settings.yaml'))
 107         assert jumphost.configure_bridges(ns) is None
 108
 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,
 112                                         subprocess_func):
 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)
 117
 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,
 122                               subprocess_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',
 130                                                   'admin')
 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'))
 135
 136         for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
 137                       'ifcfg-br-admin'):
 138             ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
 139             if os.path.isfile(ifcfg_path):
 140                 os.remove(ifcfg_path)
 141
 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,
 146                                         subprocess_func):
 147         output = jumphost.attach_interface_to_ovs('br-dummy', 'dummy_int',
 148                                                   'admin')
 149         assert output is None
 150
 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,
 155                                        subprocess_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')
 160
 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'))
 178
 179         for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
 180                       'ifcfg-br-admin'):
 181             ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
 182             if os.path.isfile(ifcfg_path):
 183                 os.remove(ifcfg_path)
 184
 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,
 189                               subprocess_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'))
 198
 199         for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
 200                       'ifcfg-br-admin'):
 201             ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
 202             if os.path.isfile(ifcfg_path):
 203                 os.remove(ifcfg_path)
 204
 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',
 223                       'ifcfg-br-admin'):
 224             ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
 225             if os.path.isfile(ifcfg_path):
 226                 os.remove(ifcfg_path)
 227
 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
 237
 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,
 242                                                is_bridge_func,
 243                                                subprocess_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
 248
 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,
 253                                        subprocess_func):
 254         ifcfg_dir = con.TEST_DUMMY_CONFIG
 255         jumphost.NET_CFG_PATH = ifcfg_dir
 256         assert_raises(FileNotFoundError, jumphost.detach_interface_from_ovs,
 257                       'admin')
 258
 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'))
 272
 273         for ifcfg in ('ifcfg-enpfakes0', 'ifcfg-enpfakes0.orig',
 274                       'ifcfg-br-admin'):
 275             ifcfg_path = os.path.join(ifcfg_dir, ifcfg)
 276             if os.path.isfile(ifcfg_path):
 277                 os.remove(ifcfg_path)
 278
 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'))
 288
 289         # test without file
 290         assert jumphost.remove_ovs_bridge(apex_constants.ADMIN_NETWORK) is None
 291
 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)