diff options
author | John Koehl <jkoehl@gmail.com> | 2018-01-28 02:21:05 -0500 |
---|---|---|
committer | Omer Katz <omer.drow@gmail.com> | 2018-01-28 09:21:05 +0200 |
commit | 39e733c14347f982e38e27c9f9edd7c400798e69 (patch) | |
tree | df1094c8c2cd9c17b34c6cb433a007680cf42873 | |
parent | c13d91634daaf138586dd37f6123d6aeb15f4aa5 (diff) | |
download | kombu-39e733c14347f982e38e27c9f9edd7c400798e69.tar.gz |
Fixes #791 - SQS queue name gets mangled in Python 2.7 environment (#794)
* Fixes #791
* Changing to recommended patch by @georgepsarakis
* Revert "Fixes #791"
This reverts commit 5593505dd9deea5d0089d03cddfb3728f09a2048.
* Updated to make tests pass
* Made _ensure_str a private function
* Code formatting for flake8
* Added a mock of the newstr and newbytes classes to create a failing test that simulates the issue with using python-future under 2.7.
-rw-r--r-- | kombu/utils/encoding.py | 9 | ||||
-rw-r--r-- | t/unit/utils/test_encoding.py | 26 |
2 files changed, 31 insertions, 4 deletions
diff --git a/kombu/utils/encoding.py b/kombu/utils/encoding.py index 21b6d903..b1ddeeb3 100644 --- a/kombu/utils/encoding.py +++ b/kombu/utils/encoding.py @@ -122,11 +122,18 @@ if is_py3k: # pragma: no cover return '<Unrepresentable {0!r}: {1!r} {2!r}>'.format( type(s), exc, '\n'.join(traceback.format_stack())) else: + def _ensure_str(s, encoding, errors): + if isinstance(s, bytes): + return s.decode(encoding, errors) + return s + + def _safe_str(s, errors='replace', file=None): # noqa encoding = default_encoding(file) try: if isinstance(s, unicode): - return s.encode(encoding, errors) + return _ensure_str(s.encode(encoding, errors), + encoding, errors) return unicode(s, encoding, errors) except Exception as exc: return '<Unrepresentable {0!r}: {1!r} {2!r}>'.format( diff --git a/t/unit/utils/test_encoding.py b/t/unit/utils/test_encoding.py index a3d6bdc8..44c81a73 100644 --- a/t/unit/utils/test_encoding.py +++ b/t/unit/utils/test_encoding.py @@ -7,7 +7,7 @@ from contextlib import contextmanager from case import patch, skip -from kombu.five import bytes_t, string_t +from kombu.five import bytes_t, string_t, string from kombu.utils.encoding import ( get_default_encoding_file, safe_str, set_default_encoding_file, default_encoding, @@ -65,6 +65,22 @@ def test_default_encode(): assert e.default_encode(b'foo') +class newbytes(bytes): + """Mock class to simulate python-future newbytes class""" + def __repr__(self): + return 'b' + super(newbytes, self).__repr__() + + def __str__(self): + return 'b' + "'{0}'".format(super(newbytes, self).__str__()) + + +class newstr(string): + """Mock class to simulate python-future newstr class""" + + def encode(self, encoding=None, errors=None): + return newbytes(super(newstr, self).encode(encoding, errors)) + + class test_safe_str: def setup(self): @@ -74,6 +90,10 @@ class test_safe_str: def test_when_bytes(self): assert safe_str('foo') == 'foo' + def test_when_newstr(self): + """Simulates using python-future package under 2.7""" + assert str(safe_str(newstr('foo'))) == 'foo' + def test_when_unicode(self): assert isinstance(safe_str('foo'), string_t) @@ -82,13 +102,13 @@ class test_safe_str: assert default_encoding() == 'utf-8' s = 'The quiæk fåx jømps øver the lazy dåg' res = safe_str(s) - assert isinstance(res, str) + assert isinstance(res, string_t) def test_when_containing_high_chars(self): self._encoding.return_value = 'ascii' s = 'The quiæk fåx jømps øver the lazy dåg' res = safe_str(s) - assert isinstance(res, str) + assert isinstance(res, string_t) assert len(s) == len(res) def test_when_not_string(self): |