mirror of
https://github.com/valitydev/salt.git
synced 2024-11-08 09:23:56 +00:00
fc27c1a948
017fbdb changed the function calls in such a way that tests in 2017.7
now fail. This updates the mocking to account for new code paths
introduced in this commit.
368 lines
14 KiB
Python
368 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
'''
|
|
:codeauthor: :email:`Jayesh Kariya <jayeshk@saltstack.com>`
|
|
'''
|
|
|
|
# Import Python Libs
|
|
from __future__ import absolute_import
|
|
import socket
|
|
import os.path
|
|
|
|
# Import Salt Testing Libs
|
|
from tests.support.mixins import LoaderModuleMockMixin
|
|
from tests.support.unit import TestCase, skipIf
|
|
from tests.support.mock import (
|
|
mock_open,
|
|
MagicMock,
|
|
patch,
|
|
NO_MOCK,
|
|
NO_MOCK_REASON
|
|
)
|
|
|
|
# Import Salt Libs
|
|
import salt.ext.six as six
|
|
import salt.utils
|
|
import salt.modules.network as network
|
|
from salt.exceptions import CommandExecutionError
|
|
if six.PY2:
|
|
import salt.ext.ipaddress as ipaddress
|
|
HAS_IPADDRESS = True
|
|
else:
|
|
try:
|
|
import ipaddress
|
|
HAS_IPADDRESS = True
|
|
except ImportError:
|
|
HAS_IPADDRESS = False
|
|
|
|
|
|
@skipIf(NO_MOCK, NO_MOCK_REASON)
|
|
class NetworkTestCase(TestCase, LoaderModuleMockMixin):
|
|
'''
|
|
Test cases for salt.modules.network
|
|
'''
|
|
def setup_loader_modules(self):
|
|
return {network: {}}
|
|
|
|
def test_wol_bad_mac(self):
|
|
'''
|
|
tests network.wol with bad mac
|
|
'''
|
|
bad_mac = '31337'
|
|
self.assertRaises(ValueError, network.wol, bad_mac)
|
|
|
|
def test_wol_success(self):
|
|
'''
|
|
tests network.wol success
|
|
'''
|
|
mac = '080027136977'
|
|
bcast = '255.255.255.255 7'
|
|
|
|
class MockSocket(object):
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def setsockopt(self, *args, **kwargs):
|
|
pass
|
|
|
|
def sendto(self, *args, **kwargs):
|
|
pass
|
|
|
|
with patch('socket.socket', MockSocket):
|
|
self.assertTrue(network.wol(mac, bcast))
|
|
|
|
def test_ping(self):
|
|
'''
|
|
Test for Performs a ping to a host
|
|
'''
|
|
with patch.object(salt.utils.network, 'sanitize_host',
|
|
return_value='A'):
|
|
mock_all = MagicMock(side_effect=[{'retcode': 1}, {'retcode': 0}])
|
|
with patch.dict(network.__salt__, {'cmd.run_all': mock_all}):
|
|
self.assertFalse(network.ping('host', return_boolean=True))
|
|
|
|
self.assertTrue(network.ping('host', return_boolean=True))
|
|
|
|
with patch.dict(network.__salt__, {'cmd.run':
|
|
MagicMock(return_value='A')}):
|
|
self.assertEqual(network.ping('host'), 'A')
|
|
|
|
def test_netstat(self):
|
|
'''
|
|
Test for return information on open ports and states
|
|
'''
|
|
with patch.dict(network.__grains__, {'kernel': 'Linux'}):
|
|
with patch.object(network, '_netstat_linux', return_value='A'):
|
|
with patch.object(network, '_ss_linux', return_value='A'):
|
|
self.assertEqual(network.netstat(), 'A')
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'OpenBSD'}):
|
|
with patch.object(network, '_netstat_bsd', return_value='A'):
|
|
self.assertEqual(network.netstat(), 'A')
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'A'}):
|
|
self.assertRaises(CommandExecutionError, network.netstat)
|
|
|
|
def test_active_tcp(self):
|
|
'''
|
|
Test for return a dict containing information on all
|
|
of the running TCP connections
|
|
'''
|
|
with patch.object(salt.utils.network, 'active_tcp', return_value='A'):
|
|
with patch.dict(network.__grains__, {'kernel': 'Linux'}):
|
|
self.assertEqual(network.active_tcp(), 'A')
|
|
|
|
def test_traceroute(self):
|
|
'''
|
|
Test for Performs a traceroute to a 3rd party host
|
|
'''
|
|
with patch.object(salt.utils, 'which', side_effect=[False, True]):
|
|
self.assertListEqual(network.traceroute('host'), [])
|
|
|
|
with patch.object(salt.utils.network, 'sanitize_host',
|
|
return_value='A'):
|
|
with patch.dict(network.__salt__, {'cmd.run':
|
|
MagicMock(return_value="")}):
|
|
self.assertListEqual(network.traceroute('host'), [])
|
|
|
|
def test_dig(self):
|
|
'''
|
|
Test for Performs a DNS lookup with dig
|
|
'''
|
|
with patch('salt.utils.which', MagicMock(return_value='dig')), \
|
|
patch.object(salt.utils.network, 'sanitize_host',
|
|
return_value='A'), \
|
|
patch.dict(network.__salt__, {'cmd.run':
|
|
MagicMock(return_value='A')}):
|
|
self.assertEqual(network.dig('host'), 'A')
|
|
|
|
def test_arp(self):
|
|
'''
|
|
Test for return the arp table from the minion
|
|
'''
|
|
with patch.dict(network.__salt__,
|
|
{'cmd.run':
|
|
MagicMock(return_value='A,B,C,D\nE,F,G,H\n')}), \
|
|
patch('salt.utils.which', MagicMock(return_value='')):
|
|
self.assertDictEqual(network.arp(), {})
|
|
|
|
def test_interfaces(self):
|
|
'''
|
|
Test for return a dictionary of information about
|
|
all the interfaces on the minion
|
|
'''
|
|
with patch.object(salt.utils.network, 'interfaces', return_value={}):
|
|
self.assertDictEqual(network.interfaces(), {})
|
|
|
|
def test_hw_addr(self):
|
|
'''
|
|
Test for return the hardware address (a.k.a. MAC address)
|
|
for a given interface
|
|
'''
|
|
with patch.object(salt.utils.network, 'hw_addr', return_value={}):
|
|
self.assertDictEqual(network.hw_addr('iface'), {})
|
|
|
|
def test_interface(self):
|
|
'''
|
|
Test for return the inet address for a given interface
|
|
'''
|
|
with patch.object(salt.utils.network, 'interface', return_value={}):
|
|
self.assertDictEqual(network.interface('iface'), {})
|
|
|
|
def test_interface_ip(self):
|
|
'''
|
|
Test for return the inet address for a given interface
|
|
'''
|
|
with patch.object(salt.utils.network, 'interface_ip', return_value={}):
|
|
self.assertDictEqual(network.interface_ip('iface'), {})
|
|
|
|
def test_subnets(self):
|
|
'''
|
|
Test for returns a list of subnets to which the host belongs
|
|
'''
|
|
with patch.object(salt.utils.network, 'subnets', return_value={}):
|
|
self.assertDictEqual(network.subnets(), {})
|
|
|
|
def test_in_subnet(self):
|
|
'''
|
|
Test for returns True if host is within specified
|
|
subnet, otherwise False.
|
|
'''
|
|
with patch.object(salt.utils.network, 'in_subnet', return_value={}):
|
|
self.assertDictEqual(network.in_subnet('iface'), {})
|
|
|
|
def test_ip_addrs(self):
|
|
'''
|
|
Test for returns a list of IPv4 addresses assigned to the host.
|
|
'''
|
|
with patch.object(salt.utils.network, 'ip_addrs',
|
|
return_value=['0.0.0.0']):
|
|
with patch.object(salt.utils.network, 'in_subnet',
|
|
return_value=True):
|
|
self.assertListEqual(network.ip_addrs('interface',
|
|
'include_loopback',
|
|
'cidr'), ['0.0.0.0'])
|
|
|
|
self.assertListEqual(network.ip_addrs('interface',
|
|
'include_loopback'),
|
|
['0.0.0.0'])
|
|
|
|
def test_ip_addrs6(self):
|
|
'''
|
|
Test for returns a list of IPv6 addresses assigned to the host.
|
|
'''
|
|
with patch.object(salt.utils.network, 'ip_addrs6',
|
|
return_value=['A']):
|
|
self.assertListEqual(network.ip_addrs6('int', 'include'), ['A'])
|
|
|
|
def test_get_hostname(self):
|
|
'''
|
|
Test for Get hostname
|
|
'''
|
|
with patch.object(network.socket, 'gethostname', return_value='A'):
|
|
self.assertEqual(network.get_hostname(), 'A')
|
|
|
|
def test_mod_hostname(self):
|
|
'''
|
|
Test for Modify hostname
|
|
'''
|
|
self.assertFalse(network.mod_hostname(None))
|
|
|
|
with patch.object(salt.utils, 'which', return_value='hostname'):
|
|
with patch.dict(network.__salt__,
|
|
{'cmd.run': MagicMock(return_value=None)}):
|
|
file_d = '\n'.join(['#', 'A B C D,E,F G H'])
|
|
with patch('salt.utils.fopen', mock_open(read_data=file_d),
|
|
create=True) as mfi:
|
|
mfi.return_value.__iter__.return_value = file_d.splitlines()
|
|
with patch.dict(network.__grains__, {'os_family': 'A'}):
|
|
self.assertTrue(network.mod_hostname('hostname'))
|
|
|
|
def test_connect(self):
|
|
'''
|
|
Test for Test connectivity to a host using a particular
|
|
port from the minion.
|
|
'''
|
|
with patch('socket.socket') as mock_socket:
|
|
self.assertDictEqual(network.connect(False, 'port'),
|
|
{'comment': 'Required argument, host, is missing.',
|
|
'result': False})
|
|
|
|
self.assertDictEqual(network.connect('host', False),
|
|
{'comment': 'Required argument, port, is missing.',
|
|
'result': False})
|
|
|
|
ret = 'Unable to connect to host (0) on tcp port port'
|
|
mock_socket.side_effect = Exception('foo')
|
|
with patch.object(salt.utils.network, 'sanitize_host',
|
|
return_value='A'):
|
|
with patch.object(socket, 'getaddrinfo',
|
|
return_value=[['ipv4', 'A', 6, 'B', '0.0.0.0']]):
|
|
self.assertDictEqual(network.connect('host', 'port'),
|
|
{'comment': ret, 'result': False})
|
|
|
|
ret = 'Successfully connected to host (0) on tcp port port'
|
|
mock_socket.side_effect = MagicMock()
|
|
mock_socket.settimeout().return_value = None
|
|
mock_socket.connect().return_value = None
|
|
mock_socket.shutdown().return_value = None
|
|
with patch.object(salt.utils.network, 'sanitize_host',
|
|
return_value='A'):
|
|
with patch.object(socket,
|
|
'getaddrinfo',
|
|
return_value=[['ipv4',
|
|
'A', 6, 'B', '0.0.0.0']]):
|
|
self.assertDictEqual(network.connect('host', 'port'),
|
|
{'comment': ret, 'result': True})
|
|
|
|
@skipIf(HAS_IPADDRESS is False, 'unable to import \'ipaddress\'')
|
|
def test_is_private(self):
|
|
'''
|
|
Test for Check if the given IP address is a private address
|
|
'''
|
|
with patch.object(ipaddress.IPv4Address, 'is_private',
|
|
return_value=True):
|
|
self.assertTrue(network.is_private('0.0.0.0'))
|
|
with patch.object(ipaddress.IPv6Address, 'is_private',
|
|
return_value=True):
|
|
self.assertTrue(network.is_private('::1'))
|
|
|
|
@skipIf(HAS_IPADDRESS is False, 'unable to import \'ipaddress\'')
|
|
def test_is_loopback(self):
|
|
'''
|
|
Test for Check if the given IP address is a loopback address
|
|
'''
|
|
with patch.object(ipaddress.IPv4Address, 'is_loopback',
|
|
return_value=True):
|
|
self.assertTrue(network.is_loopback('127.0.0.1'))
|
|
with patch.object(ipaddress.IPv6Address, 'is_loopback',
|
|
return_value=True):
|
|
self.assertTrue(network.is_loopback('::1'))
|
|
|
|
def test_get_bufsize(self):
|
|
'''
|
|
Test for return network buffer sizes as a dict
|
|
'''
|
|
with patch.dict(network.__grains__, {'kernel': 'Linux'}):
|
|
with patch.object(os.path, 'exists', return_value=True):
|
|
with patch.object(network, '_get_bufsize_linux',
|
|
return_value={'size': 1}):
|
|
self.assertDictEqual(network.get_bufsize('iface'),
|
|
{'size': 1})
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'A'}):
|
|
self.assertDictEqual(network.get_bufsize('iface'), {})
|
|
|
|
def test_mod_bufsize(self):
|
|
'''
|
|
Test for Modify network interface buffers (currently linux only)
|
|
'''
|
|
with patch.dict(network.__grains__, {'kernel': 'Linux'}):
|
|
with patch.object(os.path, 'exists', return_value=True):
|
|
with patch.object(network, '_mod_bufsize_linux',
|
|
return_value={'size': 1}):
|
|
self.assertDictEqual(network.mod_bufsize('iface'),
|
|
{'size': 1})
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'A'}):
|
|
self.assertFalse(network.mod_bufsize('iface'))
|
|
|
|
def test_routes(self):
|
|
'''
|
|
Test for return currently configured routes from routing table
|
|
'''
|
|
self.assertRaises(CommandExecutionError, network.routes, 'family')
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'A', 'os': 'B'}):
|
|
self.assertRaises(CommandExecutionError, network.routes, 'inet')
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'Linux'}):
|
|
with patch.object(network, '_netstat_route_linux',
|
|
side_effect=['A', [{'addr_family': 'inet'}]]):
|
|
with patch.object(network, '_ip_route_linux',
|
|
side_effect=['A', [{'addr_family': 'inet'}]]):
|
|
self.assertEqual(network.routes(None), 'A')
|
|
|
|
self.assertListEqual(network.routes('inet'),
|
|
[{'addr_family': 'inet'}])
|
|
|
|
def test_default_route(self):
|
|
'''
|
|
Test for return default route(s) from routing table
|
|
'''
|
|
self.assertRaises(CommandExecutionError, network.default_route,
|
|
'family')
|
|
|
|
with patch.object(network, 'routes',
|
|
side_effect=[[{'addr_family': 'inet'},
|
|
{'destination': 'A'}], []]):
|
|
with patch.dict(network.__grains__, {'kernel': 'A',
|
|
'os': 'B'}):
|
|
self.assertRaises(CommandExecutionError,
|
|
network.default_route, 'inet')
|
|
|
|
with patch.dict(network.__grains__, {'kernel': 'Linux'}):
|
|
self.assertListEqual(network.default_route('inet'), [])
|