[checkmk-commits] Fix integration tests ensuring correct conversion utf8 and bytestr

Module: check_mk Branch: master Commit: a7b244bc7ea8fbd0f75b2d5340fb51c5cdcd4441 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=a7b244bc7ea8fb… Author: Óscar Nájera <on(a)mathias-kettner.de> Date: Fri Feb 22 17:17:48 2019 +0100 Fix integration tests ensuring correct conversion utf8 and bytestr Change-Id: I87f4b849cbbae76194e1d0cb71a682df910c63cd --- livestatus/api/python/livestatus.py | 17 +++++++++++++---- tests/unit/livestatus/test_livestatus_unit.py | 19 +++++++++++++++++++ 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/livestatus/api/python/livestatus.py b/livestatus/api/python/livestatus.py index d36e6d3..53a8b62 100644 --- a/livestatus/api/python/livestatus.py +++ b/livestatus/api/python/livestatus.py @@ -54,10 +54,20 @@ remove_cache_regex = re.compile("\nCache:[^\n]*") # type: Pattern def ensure_unicode(text): if hasattr(text, "decode"): - return text.decode("utf-8") + try: + return text.decode("utf-8") + except UnicodeEncodeError: + return text return text +def ensure_bytestr(text): + try: + return text.encode("utf-8") + except UnicodeDecodeError: + return text + + class MKLivestatusException(Exception): def __init__(self, value): self.parameter = value @@ -445,9 +455,8 @@ class SingleSiteConnection(Helpers): query += "\n" try: - # socket.send() will implicitely cast to str(), we need ot - # convert to UTF-8 in order to avoid exceptions - query = query.encode("utf-8") + # socket.send() only works with byte strings + query = ensure_bytestr(query) self.socket.send(query) except IOError as e: if self.persist: diff --git a/tests/unit/livestatus/test_livestatus_unit.py b/tests/unit/livestatus/test_livestatus_unit.py index 24435e4..f6b13c7 100644 --- a/tests/unit/livestatus/test_livestatus_unit.py +++ b/tests/unit/livestatus/test_livestatus_unit.py @@ -1,3 +1,4 @@ +# -*- coding: utf-8 -*- import socket import ssl from contextlib import closing @@ -12,6 +13,24 @@ import omdlib.certs as certs import livestatus +(a)pytest.mark.parametrize("sourceource, utf8str", [ + ('hi', u'hi'), + ("há li", u"há li"), + (u"hé ßß", u"hé ßß"), +]) +def test_ensure_unicode(source, utf8str): + assert livestatus.ensure_unicode(source) == utf8str + + +(a)pytest.mark.parametrize("sourceource, bytestr", [ + ('hi', b'hi'), + ("há li", b"h\xc3\xa1 li"), + (u"hé ßß", b"h\xc3\xa9 \xc3\x9f\xc3\x9f"), +]) +def test_ensure_bytestr(source, bytestr): + assert livestatus.ensure_bytestr(source) == bytestr + + @pytest.fixture def ca(tmpdir, monkeypatch): p = Path("%s" % tmpdir) / "etc" / "ssl"