py3: fix salt/modules/network.py and friends

- dry: move mac address string-to-bytes conversion into a single routine
- add unit tests for said conversion routine
This commit is contained in:
Michael Steed 2015-06-08 16:03:19 -06:00
parent 24626b7aec
commit f163e50368
4 changed files with 39 additions and 35 deletions

View File

@ -21,8 +21,11 @@ from salt.exceptions import CommandExecutionError
# Import 3rd-party libs
import salt.ext.six as six
import salt.ext.ipaddress as ipaddress
from salt.ext.six.moves import range # pylint: disable=import-error,no-name-in-module,redefined-builtin
try:
import ipaddress
except ImportError:
import salt.ext.ipaddress as ipaddress
log = logging.getLogger(__name__)
@ -51,22 +54,10 @@ def wol(mac, bcast='255.255.255.255', destport=9):
salt '*' network.wol 080027136977 255.255.255.255 7
salt '*' network.wol 08:00:27:13:69:77 255.255.255.255 7
'''
if len(mac) == 12:
pass
elif len(mac) == 17:
sep = mac[2]
mac = mac.replace(sep, '')
else:
raise ValueError('Invalid MAC address')
dest = salt.utils.mac_str_to_bytes(mac)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
dest = ('\\x' + mac[0:2]).decode('string_escape') + \
('\\x' + mac[2:4]).decode('string_escape') + \
('\\x' + mac[4:6]).decode('string_escape') + \
('\\x' + mac[6:8]).decode('string_escape') + \
('\\x' + mac[8:10]).decode('string_escape') + \
('\\x' + mac[10:12]).decode('string_escape')
sock.sendto('\xff' * 6 + dest * 16, (bcast, int(destport)))
sock.sendto(b'\xff' * 6 + dest * 16, (bcast, int(destport)))
return True
@ -952,12 +943,7 @@ def connect(host, port=None, **kwargs):
skt.shutdown(2)
except Exception as exc:
ret['result'] = False
try:
errno, errtxt = exc
except ValueError:
ret['comment'] = 'Unable to connect to {0} ({1}) on {2} port {3}'.format(host, _address[0], proto, port)
else:
ret['comment'] = '{0}'.format(errtxt)
ret['comment'] = 'Unable to connect to {0} ({1}) on {2} port {3}'.format(host, _address[0], proto, port)
return ret
ret['result'] = True

View File

@ -50,20 +50,8 @@ def wol(mac, bcast='255.255.255.255', destport=9):
salt-run network.wol 080027136977 255.255.255.255 7
salt-run network.wol 08:00:27:13:69:77 255.255.255.255 7
'''
if len(mac) == 12:
pass
elif len(mac) == 17:
sep = mac[2]
mac = mac.replace(sep, '')
else:
raise ValueError('Invalid MAC address')
dest = salt.utils.mac_str_to_bytes(mac)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
dest = ('\\x' + mac[0:2]).decode('string_escape') + \
('\\x' + mac[2:4]).decode('string_escape') + \
('\\x' + mac[4:6]).decode('string_escape') + \
('\\x' + mac[6:8]).decode('string_escape') + \
('\\x' + mac[8:10]).decode('string_escape') + \
('\\x' + mac[10:12]).decode('string_escape')
sock.sendto('\xff' * 6 + dest * 16, (bcast, int(destport)))
sock.sendto(b'\xff' * 6 + dest * 16, (bcast, int(destport)))
return True

View File

@ -635,6 +635,28 @@ def gen_mac(prefix='AC:DE:48'):
random.randint(0, 0xff))
def mac_str_to_bytes(mac_str):
'''
Convert a MAC address string into bytes. Works with or without delimiters,
e.g.:
mac_str_to_bytes('08:00:27:13:69:77')
mac_str_to_bytes('080027136977')
'''
if len(mac_str) == 12:
pass
elif len(mac_str) == 17:
sep = mac_str[2]
mac_str = mac_str.replace(sep, '')
else:
raise ValueError('Invalid MAC address')
if six.PY3:
mac_bytes = bytes(int(mac_str[s:s+2], 16) for s in range(0, 12, 2))
else:
mac_bytes = ''.join(chr(int(mac_str[s:s+2], 16)) for s in range(0, 12, 2))
return mac_bytes
def ip_bracket(addr):
'''
Convert IP address representation to ZMQ (URL) format. ZMQ expects

View File

@ -78,6 +78,14 @@ class UtilsTestCase(TestCase):
expected_mac = '00:16:3E:01:01:01'
self.assertEqual(ret, expected_mac)
def test_mac_str_to_bytes(self):
self.assertRaises(ValueError, utils.mac_str_to_bytes, '31337')
self.assertRaises(ValueError, utils.mac_str_to_bytes, '0001020304056')
self.assertRaises(ValueError, utils.mac_str_to_bytes, '00:01:02:03:04:056')
self.assertRaises(ValueError, utils.mac_str_to_bytes, 'a0:b0:c0:d0:e0:fg')
self.assertEqual(b'\x10\x08\x06\x04\x02\x00', utils.mac_str_to_bytes('100806040200'))
self.assertEqual(b'\xf8\xe7\xd6\xc5\xb4\xa3', utils.mac_str_to_bytes('f8e7d6c5b4a3'))
def test_ip_bracket(self):
test_ipv4 = '127.0.0.1'
test_ipv6 = '::1'