Import patches-unapplied version 2.2.16-0ubuntu1 to ubuntu/artful-proposed

Imported using git-ubuntu import.

Changelog parent: 43bdf9debe5377216aed0086bff2aad864f6ba82

New changelog entries:
  * New upstream release (LP: #1714299).

3 files changed, 486 insertions, 29 deletions

diff --git a/tests/utils/test_file_util.py b/tests/utils/test_file_util.py
index 0b92513..87bce8c 100644
--- a/tests/utils/test_file_util.py
+++ b/tests/utils/test_file_util.py
@@ -15,6 +15,7 @@
 # Requires Python 2.4+ and Openssl 1.0+
 #
 
+import errno as errno
 import glob
 import random
 import string
@@ -64,6 +65,50 @@ class TestFileOperations(AgentTestCase):
 
         os.remove(test_file)
 
+    def test_findre_in_file(self):
+        fp = tempfile.mktemp()
+        with open(fp, 'w') as f:
+            f.write(
+'''
+First line
+Second line
+Third line with more words
+'''
+            )
+
+        self.assertNotEquals(
+            None,
+            fileutil.findre_in_file(fp, ".*rst line$"))
+        self.assertNotEquals(
+            None,
+            fileutil.findre_in_file(fp, ".*ond line$"))
+        self.assertNotEquals(
+            None,
+            fileutil.findre_in_file(fp, ".*with more.*"))
+        self.assertNotEquals(
+            None,
+            fileutil.findre_in_file(fp, "^Third.*"))
+        self.assertEquals(
+            None,
+            fileutil.findre_in_file(fp, "^Do not match.*"))
+
+    def test_findstr_in_file(self):
+        fp = tempfile.mktemp()
+        with open(fp, 'w') as f:
+            f.write(
+'''
+First line
+Second line
+Third line with more words
+'''
+            )
+
+        self.assertTrue(fileutil.findstr_in_file(fp, "First line"))
+        self.assertTrue(fileutil.findstr_in_file(fp, "Second line"))
+        self.assertTrue(
+            fileutil.findstr_in_file(fp, "Third line with more words"))
+        self.assertFalse(fileutil.findstr_in_file(fp, "Not a line"))
+
     def test_get_last_path_element(self):
         filepath = '/tmp/abc.def'
         filename = fileutil.base_name(filepath)
@@ -197,5 +242,75 @@ DHCP_HOSTNAME=test\n"
                 fileutil.update_conf_file(path, 'DHCP_HOSTNAME', 'DHCP_HOSTNAME=test')
                 patch_write.assert_called_once_with(path, updated_file)
 
+    def test_clean_ioerror_ignores_missing(self):
+        e = IOError()
+        e.errno = errno.ENOSPC
+
+        # Send no paths
+        fileutil.clean_ioerror(e)
+
+        # Send missing file(s) / directories
+        fileutil.clean_ioerror(e, paths=['/foo/not/here', None, '/bar/not/there'])
+
+    def test_clean_ioerror_ignores_unless_ioerror(self):
+        try:
+            d = tempfile.mkdtemp()
+            fd, f = tempfile.mkstemp()
+            os.close(fd)
+            fileutil.write_file(f, 'Not empty')
+
+            # Send non-IOError exception
+            e = Exception()
+            fileutil.clean_ioerror(e, paths=[d, f])
+            self.assertTrue(os.path.isdir(d))
+            self.assertTrue(os.path.isfile(f))
+
+            # Send unrecognized IOError
+            e = IOError()
+            e.errno = errno.EFAULT
+            self.assertFalse(e.errno in fileutil.KNOWN_IOERRORS)
+            fileutil.clean_ioerror(e, paths=[d, f])
+            self.assertTrue(os.path.isdir(d))
+            self.assertTrue(os.path.isfile(f))
+
+        finally:
+            shutil.rmtree(d)
+            os.remove(f)
+
+    def test_clean_ioerror_removes_files(self):
+        fd, f = tempfile.mkstemp()
+        os.close(fd)
+        fileutil.write_file(f, 'Not empty')
+
+        e = IOError()
+        e.errno = errno.ENOSPC
+        fileutil.clean_ioerror(e, paths=[f])
+        self.assertFalse(os.path.isdir(f))
+        self.assertFalse(os.path.isfile(f))
+
+    def test_clean_ioerror_removes_directories(self):
+        d1 = tempfile.mkdtemp()
+        d2 = tempfile.mkdtemp()
+        for n in ['foo', 'bar']:
+            fileutil.write_file(os.path.join(d2, n), 'Not empty')
+
+        e = IOError()
+        e.errno = errno.ENOSPC
+        fileutil.clean_ioerror(e, paths=[d1, d2])
+        self.assertFalse(os.path.isdir(d1))
+        self.assertFalse(os.path.isfile(d1))
+        self.assertFalse(os.path.isdir(d2))
+        self.assertFalse(os.path.isfile(d2))
+
+    def test_clean_ioerror_handles_a_range_of_errors(self):
+        for err in fileutil.KNOWN_IOERRORS:
+            e = IOError()
+            e.errno = err
+
+            d = tempfile.mkdtemp()
+            fileutil.clean_ioerror(e, paths=[d])
+            self.assertFalse(os.path.isdir(d))
+            self.assertFalse(os.path.isfile(d))
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/utils/test_rest_util.py b/tests/utils/test_rest_util.py
index 5f084a6..52674da 100644
--- a/tests/utils/test_rest_util.py
+++ b/tests/utils/test_rest_util.py
@@ -15,10 +15,16 @@
 # Requires Python 2.4+ and Openssl 1.0+
 #
 
+import os
 import unittest
+
+from azurelinuxagent.common.exception import HttpError, \
+                                            ProtocolError, \
+                                            ResourceGoneError
 import azurelinuxagent.common.utils.restutil as restutil
-from azurelinuxagent.common.future import httpclient
-from tests.tools import AgentTestCase, patch, Mock, MagicMock
+
+from azurelinuxagent.common.future import httpclient, ustr
+from tests.tools import *
 
 
 class TestHttpOperations(AgentTestCase):
@@ -50,45 +56,163 @@ class TestHttpOperations(AgentTestCase):
         self.assertEquals(None, host)
         self.assertEquals(rel_uri, "None")
 
+    @patch('azurelinuxagent.common.conf.get_httpproxy_port')
+    @patch('azurelinuxagent.common.conf.get_httpproxy_host')
+    def test_get_http_proxy_none_is_default(self, mock_host, mock_port):
+        mock_host.return_value = None
+        mock_port.return_value = None
+        h, p = restutil._get_http_proxy()
+        self.assertEqual(None, h)
+        self.assertEqual(None, p)
+
+    @patch('azurelinuxagent.common.conf.get_httpproxy_port')
+    @patch('azurelinuxagent.common.conf.get_httpproxy_host')
+    def test_get_http_proxy_configuration_overrides_env(self, mock_host, mock_port):
+        mock_host.return_value = "host"
+        mock_port.return_value = None
+        h, p = restutil._get_http_proxy()
+        self.assertEqual("host", h)
+        self.assertEqual(None, p)
+        mock_host.assert_called_once()
+        mock_port.assert_called_once()
+
+    @patch('azurelinuxagent.common.conf.get_httpproxy_port')
+    @patch('azurelinuxagent.common.conf.get_httpproxy_host')
+    def test_get_http_proxy_configuration_requires_host(self, mock_host, mock_port):
+        mock_host.return_value = None
+        mock_port.return_value = None
+        h, p = restutil._get_http_proxy()
+        self.assertEqual(None, h)
+        self.assertEqual(None, p)
+        mock_host.assert_called_once()
+        mock_port.assert_not_called()
+
+    @patch('azurelinuxagent.common.conf.get_httpproxy_host')
+    def test_get_http_proxy_http_uses_httpproxy(self, mock_host):
+        mock_host.return_value = None
+        with patch.dict(os.environ, {
+                                    'http_proxy' : 'http://foo.com:80',
+                                    'https_proxy' : 'https://bar.com:443'
+                                }):
+            h, p = restutil._get_http_proxy()
+            self.assertEqual("foo.com", h)
+            self.assertEqual(80, p)
+
+    @patch('azurelinuxagent.common.conf.get_httpproxy_host')
+    def test_get_http_proxy_https_uses_httpsproxy(self, mock_host):
+        mock_host.return_value = None
+        with patch.dict(os.environ, {
+                                    'http_proxy' : 'http://foo.com:80',
+                                    'https_proxy' : 'https://bar.com:443'
+                                }):
+            h, p = restutil._get_http_proxy(secure=True)
+            self.assertEqual("bar.com", h)
+            self.assertEqual(443, p)
+
+    @patch('azurelinuxagent.common.conf.get_httpproxy_host')
+    def test_get_http_proxy_ignores_user_in_httpproxy(self, mock_host):
+        mock_host.return_value = None
+        with patch.dict(os.environ, {
+                                    'http_proxy' : 'http://user:pw@foo.com:80'
+                                }):
+            h, p = restutil._get_http_proxy()
+            self.assertEqual("foo.com", h)
+            self.assertEqual(80, p)
+
     @patch("azurelinuxagent.common.future.httpclient.HTTPSConnection")
     @patch("azurelinuxagent.common.future.httpclient.HTTPConnection")
-    def test_http_request(self, HTTPConnection, HTTPSConnection):
-        mock_http_conn = MagicMock()
-        mock_http_resp = MagicMock()
-        mock_http_conn.getresponse = Mock(return_value=mock_http_resp)
-        HTTPConnection.return_value = mock_http_conn
-        HTTPSConnection.return_value = mock_http_conn
+    def test_http_request_direct(self, HTTPConnection, HTTPSConnection):
+        mock_conn = \
+            MagicMock(getresponse=\
+                Mock(return_value=\
+                    Mock(read=Mock(return_value="TheResults"))))
 
-        mock_http_resp.read = Mock(return_value="_(:3| <)_")
+        HTTPConnection.return_value = mock_conn
 
-        # Test http get
-        resp = restutil._http_request("GET", "foo", "bar")
-        self.assertNotEquals(None, resp)
-        self.assertEquals("_(:3| <)_", resp.read())
+        resp = restutil._http_request("GET", "foo", "/bar")
 
-        # Test https get
-        resp = restutil._http_request("GET", "foo", "bar", secure=True)
+        HTTPConnection.assert_has_calls([
+            call("foo", 80, timeout=10)
+        ])
+        HTTPSConnection.assert_not_called()
+        mock_conn.request.assert_has_calls([
+            call(method="GET", url="/bar", body=None, headers={})
+        ])
+        mock_conn.getresponse.assert_called_once()
         self.assertNotEquals(None, resp)
-        self.assertEquals("_(:3| <)_", resp.read())
+        self.assertEquals("TheResults", resp.read())
+
+    @patch("azurelinuxagent.common.future.httpclient.HTTPSConnection")
+    @patch("azurelinuxagent.common.future.httpclient.HTTPConnection")
+    def test_http_request_direct_secure(self, HTTPConnection, HTTPSConnection):
+        mock_conn = \
+            MagicMock(getresponse=\
+                Mock(return_value=\
+                    Mock(read=Mock(return_value="TheResults"))))
+
+        HTTPSConnection.return_value = mock_conn
+
+        resp = restutil._http_request("GET", "foo", "/bar", secure=True)
 
-        # Test http get with proxy
-        mock_http_resp.read = Mock(return_value="_(:3| <)_")
-        resp = restutil._http_request("GET", "foo", "bar", proxy_host="foo.bar",
-                                      proxy_port=23333)
+        HTTPConnection.assert_not_called()
+        HTTPSConnection.assert_has_calls([
+            call("foo", 443, timeout=10)
+        ])
+        mock_conn.request.assert_has_calls([
+            call(method="GET", url="/bar", body=None, headers={})
+        ])
+        mock_conn.getresponse.assert_called_once()
         self.assertNotEquals(None, resp)
-        self.assertEquals("_(:3| <)_", resp.read())
+        self.assertEquals("TheResults", resp.read())
 
-        # Test https get
-        resp = restutil._http_request("GET", "foo", "bar", secure=True)
+    @patch("azurelinuxagent.common.future.httpclient.HTTPSConnection")
+    @patch("azurelinuxagent.common.future.httpclient.HTTPConnection")
+    def test_http_request_proxy(self, HTTPConnection, HTTPSConnection):
+        mock_conn = \
+            MagicMock(getresponse=\
+                Mock(return_value=\
+                    Mock(read=Mock(return_value="TheResults"))))
+
+        HTTPConnection.return_value = mock_conn
+
+        resp = restutil._http_request("GET", "foo", "/bar",
+                            proxy_host="foo.bar", proxy_port=23333)
+
+        HTTPConnection.assert_has_calls([
+            call("foo.bar", 23333, timeout=10)
+        ])
+        HTTPSConnection.assert_not_called()
+        mock_conn.request.assert_has_calls([
+            call(method="GET", url="http://foo:80/bar", body=None, headers={})
+        ])
+        mock_conn.getresponse.assert_called_once()
         self.assertNotEquals(None, resp)
-        self.assertEquals("_(:3| <)_", resp.read())
+        self.assertEquals("TheResults", resp.read())
+
+    @patch("azurelinuxagent.common.future.httpclient.HTTPSConnection")
+    @patch("azurelinuxagent.common.future.httpclient.HTTPConnection")
+    def test_http_request_proxy_secure(self, HTTPConnection, HTTPSConnection):
+        mock_conn = \
+            MagicMock(getresponse=\
+                Mock(return_value=\
+                    Mock(read=Mock(return_value="TheResults"))))
+
+        HTTPSConnection.return_value = mock_conn
 
-        # Test https get with proxy
-        mock_http_resp.read = Mock(return_value="_(:3| <)_")
-        resp = restutil._http_request("GET", "foo", "bar", proxy_host="foo.bar",
-                                      proxy_port=23333, secure=True)
+        resp = restutil._http_request("GET", "foo", "/bar",
+                            proxy_host="foo.bar", proxy_port=23333,
+                            secure=True)
+
+        HTTPConnection.assert_not_called()
+        HTTPSConnection.assert_has_calls([
+            call("foo.bar", 23333, timeout=10)
+        ])
+        mock_conn.request.assert_has_calls([
+            call(method="GET", url="https://foo:443/bar", body=None, headers={})
+        ])
+        mock_conn.getresponse.assert_called_once()
         self.assertNotEquals(None, resp)
-        self.assertEquals("_(:3| <)_", resp.read())
+        self.assertEquals("TheResults", resp.read())
 
     @patch("time.sleep")
     @patch("azurelinuxagent.common.utils.restutil._http_request")
@@ -115,6 +239,180 @@ class TestHttpOperations(AgentTestCase):
         self.assertRaises(restutil.HttpError, restutil.http_get,
                           "http://foo.bar")
 
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_retries_status_codes(self, _http_request, _sleep):
+        _http_request.side_effect = [
+            Mock(status=httpclient.SERVICE_UNAVAILABLE),
+            Mock(status=httpclient.OK)
+        ]
+
+        restutil.http_get("https://foo.bar")
+        self.assertEqual(2, _http_request.call_count)
+        self.assertEqual(1, _sleep.call_count)
+
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_retries_passed_status_codes(self, _http_request, _sleep):
+        # Ensure the code is not part of the standard set
+        self.assertFalse(httpclient.UNAUTHORIZED in restutil.RETRY_CODES)
+
+        _http_request.side_effect = [
+            Mock(status=httpclient.UNAUTHORIZED),
+            Mock(status=httpclient.OK)
+        ]
+
+        restutil.http_get("https://foo.bar", retry_codes=[httpclient.UNAUTHORIZED])
+        self.assertEqual(2, _http_request.call_count)
+        self.assertEqual(1, _sleep.call_count)
+
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_raises_for_bad_request(self, _http_request, _sleep):
+        _http_request.side_effect = [
+            Mock(status=httpclient.BAD_REQUEST)
+        ]
+
+        self.assertRaises(ResourceGoneError, restutil.http_get, "https://foo.bar")
+        self.assertEqual(1, _http_request.call_count)
+
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_raises_for_resource_gone(self, _http_request, _sleep):
+        _http_request.side_effect = [
+            Mock(status=httpclient.GONE)
+        ]
+
+        self.assertRaises(ResourceGoneError, restutil.http_get, "https://foo.bar")
+        self.assertEqual(1, _http_request.call_count)
+
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_retries_exceptions(self, _http_request, _sleep):
+        # Testing each exception is difficult because they have varying
+        # signatures; for now, test one and ensure the set is unchanged
+        recognized_exceptions = [
+            httpclient.NotConnected,
+            httpclient.IncompleteRead,
+            httpclient.ImproperConnectionState,
+            httpclient.BadStatusLine
+        ]
+        self.assertEqual(recognized_exceptions, restutil.RETRY_EXCEPTIONS)
+
+        _http_request.side_effect = [
+            httpclient.IncompleteRead(''),
+            Mock(status=httpclient.OK)
+        ]
+
+        restutil.http_get("https://foo.bar")
+        self.assertEqual(2, _http_request.call_count)
+        self.assertEqual(1, _sleep.call_count)
+
+    @patch("time.sleep")
+    @patch("azurelinuxagent.common.utils.restutil._http_request")
+    def test_http_request_retries_ioerrors(self, _http_request, _sleep):
+        ioerror = IOError()
+        ioerror.errno = 42
+
+        _http_request.side_effect = [
+            ioerror,
+            Mock(status=httpclient.OK)
+        ]
+
+        restutil.http_get("https://foo.bar")
+        self.assertEqual(2, _http_request.call_count)
+        self.assertEqual(1, _sleep.call_count)
+
+    def test_request_failed(self):
+        self.assertTrue(restutil.request_failed(None))
+
+        resp = Mock()
+        for status in restutil.OK_CODES:
+            resp.status = status
+            self.assertFalse(restutil.request_failed(resp))
+
+        self.assertFalse(httpclient.BAD_REQUEST in restutil.OK_CODES)
+        resp.status = httpclient.BAD_REQUEST
+        self.assertTrue(restutil.request_failed(resp))
+
+        self.assertFalse(
+            restutil.request_failed(
+                resp, ok_codes=[httpclient.BAD_REQUEST]))
+
+    def test_request_succeeded(self):
+        self.assertFalse(restutil.request_succeeded(None))
+
+        resp = Mock()
+        for status in restutil.OK_CODES:
+            resp.status = status
+            self.assertTrue(restutil.request_succeeded(resp))
+
+        self.assertFalse(httpclient.BAD_REQUEST in restutil.OK_CODES)
+        resp.status = httpclient.BAD_REQUEST
+        self.assertFalse(restutil.request_succeeded(resp))
+
+        self.assertTrue(
+            restutil.request_succeeded(
+                resp, ok_codes=[httpclient.BAD_REQUEST]))
+
+    def test_read_response_error(self):
+        """
+        Validate the read_response_error method handles encoding correctly
+        """
+        responses = ['message', b'message', '\x80message\x80']
+        response = MagicMock()
+        response.status = 'status'
+        response.reason = 'reason'
+        with patch.object(response, 'read') as patch_response:
+            for s in responses:
+                patch_response.return_value = s
+                result = restutil.read_response_error(response)
+                print("RESPONSE: {0}".format(s))
+                print("RESULT: {0}".format(result))
+                print("PRESENT: {0}".format('[status: reason]' in result))
+                self.assertTrue('[status: reason]' in result)
+                self.assertTrue('message' in result)
+
+    def test_read_response_bytes(self):
+        response_bytes = '7b:0a:20:20:20:20:22:65:72:72:6f:72:43:6f:64:65:22:' \
+                         '3a:20:22:54:68:65:20:62:6c:6f:62:20:74:79:70:65:20:' \
+                         '69:73:20:69:6e:76:61:6c:69:64:20:66:6f:72:20:74:68:' \
+                         '69:73:20:6f:70:65:72:61:74:69:6f:6e:2e:22:2c:0a:20:' \
+                         '20:20:20:22:6d:65:73:73:61:67:65:22:3a:20:22:c3:af:' \
+                         'c2:bb:c2:bf:3c:3f:78:6d:6c:20:76:65:72:73:69:6f:6e:' \
+                         '3d:22:31:2e:30:22:20:65:6e:63:6f:64:69:6e:67:3d:22:' \
+                         '75:74:66:2d:38:22:3f:3e:3c:45:72:72:6f:72:3e:3c:43:' \
+                         '6f:64:65:3e:49:6e:76:61:6c:69:64:42:6c:6f:62:54:79:' \
+                         '70:65:3c:2f:43:6f:64:65:3e:3c:4d:65:73:73:61:67:65:' \
+                         '3e:54:68:65:20:62:6c:6f:62:20:74:79:70:65:20:69:73:' \
+                         '20:69:6e:76:61:6c:69:64:20:66:6f:72:20:74:68:69:73:' \
+                         '20:6f:70:65:72:61:74:69:6f:6e:2e:0a:52:65:71:75:65:' \
+                         '73:74:49:64:3a:63:37:34:32:39:30:63:62:2d:30:30:30:' \
+                         '31:2d:30:30:62:35:2d:30:36:64:61:2d:64:64:36:36:36:' \
+                         '61:30:30:30:22:2c:0a:20:20:20:20:22:64:65:74:61:69:' \
+                         '6c:73:22:3a:20:22:22:0a:7d'.split(':')
+        expected_response = '[HTTP Failed] [status: reason] {\n    "errorCode": "The blob ' \
+                            'type is invalid for this operation.",\n    ' \
+                            '"message": "<?xml version="1.0" ' \
+                            'encoding="utf-8"?>' \
+                            '<Error><Code>InvalidBlobType</Code><Message>The ' \
+                            'blob type is invalid for this operation.\n' \
+                            'RequestId:c74290cb-0001-00b5-06da-dd666a000",' \
+                            '\n    "details": ""\n}'
+
+        response_string = ''.join(chr(int(b, 16)) for b in response_bytes)
+        response = MagicMock()
+        response.status = 'status'
+        response.reason = 'reason'
+        with patch.object(response, 'read') as patch_response:
+            patch_response.return_value = response_string
+            result = restutil.read_response_error(response)
+            self.assertEqual(result, expected_response)
+            try:
+                raise HttpError("{0}".format(result))
+            except HttpError as e:
+                self.assertTrue(result in ustr(e))
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/utils/test_text_util.py b/tests/utils/test_text_util.py
index 6f204c7..d182a67 100644
--- a/tests/utils/test_text_util.py
+++ b/tests/utils/test_text_util.py
@@ -34,6 +34,19 @@ class TestTextUtil(AgentTestCase):
                 password_hash = textutil.gen_password_hash(data, 6, 10)
                 self.assertNotEquals(None, password_hash)
 
+    def test_replace_non_ascii(self):
+        data = ustr(b'\xef\xbb\xbfhehe', encoding='utf-8')
+        self.assertEqual('hehe', textutil.replace_non_ascii(data))
+
+        data = "abcd\xa0e\xf0fghijk\xbblm"
+        self.assertEqual("abcdefghijklm", textutil.replace_non_ascii(data))
+
+        data = "abcd\xa0e\xf0fghijk\xbblm"
+        self.assertEqual("abcdXeXfghijkXlm",
+            textutil.replace_non_ascii(data, replace_char='X'))
+
+        self.assertEqual('', textutil.replace_non_ascii(None))
+
     def test_remove_bom(self):
         #Test bom could be removed
         data = ustr(b'\xef\xbb\xbfhehe', encoding='utf-8')
@@ -94,6 +107,37 @@ class TestTextUtil(AgentTestCase):
                    "-----END PRIVATE Key-----\n")
         base64_bytes = textutil.get_bytes_from_pem(content)
         self.assertEquals("private key", base64_bytes)
+
+    def test_swap_hexstring(self):
+        data = [
+            ['12', 1, '21'],
+            ['12', 2, '12'],
+            ['12', 3, '012'],
+            ['12', 4, '0012'],
+
+            ['123', 1, '321'],
+            ['123', 2, '2301'],
+            ['123', 3, '123'],
+            ['123', 4, '0123'],
+
+            ['1234', 1, '4321'],
+            ['1234', 2, '3412'],
+            ['1234', 3, '234001'],
+            ['1234', 4, '1234'],
+
+            ['abcdef12', 1, '21fedcba'],
+            ['abcdef12', 2, '12efcdab'],
+            ['abcdef12', 3, 'f12cde0ab'],
+            ['abcdef12', 4, 'ef12abcd'],
+
+            ['aBcdEf12', 1, '21fEdcBa'],
+            ['aBcdEf12', 2, '12EfcdaB'],
+            ['aBcdEf12', 3, 'f12cdE0aB'],
+            ['aBcdEf12', 4, 'Ef12aBcd']
+        ]
+
+        for t in data:
+            self.assertEqual(t[2], textutil.swap_hexstring(t[0], width=t[1]))
         
 if __name__ == '__main__':
     unittest.main()