diff --git a/supervisor/compat.py b/supervisor/compat.py index 004c00e0f..82752b2a6 100644 --- a/supervisor/compat.py +++ b/supervisor/compat.py @@ -15,13 +15,13 @@ def as_bytes(s, encoding='utf-8'): if isinstance(s, str): return s else: - return s.encode(encoding) + return s.encode(encoding, errors='replace') def as_string(s, encoding='utf-8'): if isinstance(s, unicode): return s else: - return s.decode(encoding) + return s.decode(encoding, errors='replace') def is_text_stream(stream): try: @@ -51,13 +51,13 @@ def as_bytes(s, encoding='utf8'): if isinstance(s, bytes): return s else: - return s.encode(encoding) + return s.encode(encoding, errors='replace') def as_string(s, encoding='utf8'): if isinstance(s, str): return s else: - return s.decode(encoding) + return s.decode(encoding, errors='replace') def is_text_stream(stream): import _io diff --git a/supervisor/tests/test_compat.py b/supervisor/tests/test_compat.py new file mode 100644 index 000000000..99f735abf --- /dev/null +++ b/supervisor/tests/test_compat.py @@ -0,0 +1,35 @@ +"""Test suite for supervisor.compat""" + +import unittest + +from supervisor.compat import as_bytes, as_string + + +class AsStringTests(unittest.TestCase): + def test_decodes_valid_utf8(self): + self.assertEqual(as_string(b'hello'), 'hello') + + def test_returns_string_unchanged(self): + self.assertEqual(as_string('hello'), 'hello') + + def test_replaces_invalid_bytes_instead_of_raising(self): + # Incomplete multi-byte sequence (e.g. truncated Korean character) + broken = b'\xec\x95' # incomplete 3-byte UTF-8 sequence + result = as_string(broken) + self.assertIn('�', result) + + def test_decodes_korean_utf8(self): + korean = '안녕하세요'.encode('utf-8') + self.assertEqual(as_string(korean), '안녕하세요') + + +class AsBytesTests(unittest.TestCase): + def test_encodes_valid_string(self): + self.assertEqual(as_bytes('hello'), b'hello') + + def test_returns_bytes_unchanged(self): + self.assertEqual(as_bytes(b'hello'), b'hello') + + def test_encodes_korean_string(self): + result = as_bytes('안녕하세요') + self.assertEqual(result, '안녕하세요'.encode('utf-8'))