# -*- coding: utf-8 -*- ''' Test salt.utils.zeromq ''' # Import Python libs from __future__ import absolute_import, print_function, unicode_literals import zmq from salt._compat import ipaddress # Import Salt Testing libs from tests.support.unit import TestCase, skipIf from tests.support.mock import ( patch, ) # Import salt libs import salt.utils.zeromq from salt.exceptions import SaltSystemExit class UtilsTestCase(TestCase): def test_ip_bracket(self): test_ipv4 = '127.0.0.1' test_ipv6 = '::1' test_ipv6_uri = '[::1]' self.assertEqual(test_ipv4, salt.utils.zeromq.ip_bracket(test_ipv4)) self.assertEqual('[{0}]'.format(test_ipv6), salt.utils.zeromq.ip_bracket(test_ipv6)) self.assertEqual('[{0}]'.format(test_ipv6), salt.utils.zeromq.ip_bracket(test_ipv6_uri)) ip_addr_obj = ipaddress.ip_address(test_ipv4) self.assertEqual(test_ipv4, salt.utils.zeromq.ip_bracket(ip_addr_obj)) @skipIf(not hasattr(zmq, 'IPC_PATH_MAX_LEN'), "ZMQ does not have max length support.") def test_check_ipc_length(self): ''' Ensure we throw an exception if we have a too-long IPC URI ''' with patch('zmq.IPC_PATH_MAX_LEN', 1): self.assertRaises(SaltSystemExit, salt.utils.zeromq.check_ipc_path_max_len, '1' * 1024)