Change IOBuffer API to accept strings only

Bytes to strings conversion was an API flaw that this patch will fix. In
this way we threat every message coming from SUT as string and we won't
get crazy with decode/encode operations.

Author: acerv

ltp/dispatcher.py

@@ -174,31 +174,29 @@ def __init__(self, test: Test) -> None:
         self._test = test
         self._line = ""
 
-    def write(self, data: bytes) -> None:
-        data_str = data.decode(encoding="utf-8", errors="replace")
-
-        if len(data_str) == 1:
-            self._line += data_str
-            if data_str == "\n":
+    def write(self, data: str) -> None:
+        if len(data) == 1:
+            self._line += data
+            if data == "\n":
                 ltp.events.fire(
                     "test_stdout_line",
                     self._test,
                     self._line[:-1])
                 self._line = ""
         else:
-            lines = data_str.split('\n')
+            lines = data.split('\n')
             for line in lines[:-1]:
                 self._line += line
                 ltp.events.fire("test_stdout_line", self._test, self._line)
                 self._line = ""
 
             self._line = lines[-1]
 
-            if data_str.endswith('\n') and self._line:
+            if data.endswith('\n') and self._line:
                 ltp.events.fire("test_stdout_line", self._test, self._line)
                 self._line = ""
 
-        self.stdout += data_str
+        self.stdout += data
 
     def flush(self) -> None:
         pass
@@ -212,7 +210,7 @@ class RedirectStdout(IOBuffer):
     def __init__(self, sut: SUT) -> None:
         self._sut = sut
 
-    def write(self, data: bytes) -> None:
+    def write(self, data: str) -> None:
         if not self._sut:
             return
 

ltp/host.py

@@ -115,23 +115,22 @@ def force_stop(
             iobuffer: IOBuffer = None) -> None:
         self._inner_stop(signal.SIGKILL, timeout)
 
-    def _read_stdout(self, size: int, iobuffer: IOBuffer = None) -> bytes:
+    def _read_stdout(self, size: int, iobuffer: IOBuffer = None) -> str:
         """
         Read data from stdout.
         """
         if not self.is_running:
             return None
 
         data = os.read(self._proc.stdout.fileno(), size)
+        rdata = data.decode(encoding="utf-8", errors="replace")
+        rdata = rdata.replace('\r', '')
 
         # write on stdout buffers
         if iobuffer:
-            iobuffer.write(data)
+            iobuffer.write(rdata)
             iobuffer.flush()
 
-        rdata = data.decode(encoding="utf-8", errors="replace")
-        rdata = rdata.replace('\r', '')
-
         return rdata
 
     # pylint: disable=too-many-locals

ltp/qemu.py

@@ -220,23 +220,22 @@ def ping(self) -> float:
 
         return exec_time
 
-    def _read_stdout(self, size: int, iobuffer: IOBuffer) -> bytes:
+    def _read_stdout(self, size: int, iobuffer: IOBuffer) -> str:
         """
         Read data from stdout.
         """
         if not self.is_running:
             return None
 
         data = os.read(self._proc.stdout.fileno(), size)
+        rdata = data.decode(encoding="utf-8", errors="replace")
+        rdata = rdata.replace('\r', '')
 
         # write on stdout buffers
         if iobuffer:
-            iobuffer.write(data)
+            iobuffer.write(rdata)
             iobuffer.flush()
 
-        rdata = data.decode(encoding="utf-8", errors="replace")
-        rdata = rdata.replace('\r', '')
-
         return rdata
 
     def _write_stdin(self, data: str) -> None:

ltp/session.py

@@ -36,7 +36,7 @@ def __init__(self, sut: SUT, is_cmd: bool) -> None:
         self._sut = sut
         self._is_cmd = is_cmd
 
-    def write(self, data: bytes) -> None:
+    def write(self, data: str) -> None:
         if self._is_cmd:
             ltp.events.fire("run_cmd_stdout", data)
         else:

ltp/sut.py

@@ -32,7 +32,7 @@ class IOBuffer:
     IO stdout buffer. The API is similar to ``IO`` types.
     """
 
-    def write(self, data: bytes) -> None:
+    def write(self, data: str) -> None:
         """
         Write data.
         """

ltp/tests/sut.py

@@ -6,12 +6,13 @@
 import pytest
 import logging
 import threading
+from ltp.sut import IOBuffer
 from ltp.sut import SUTError
 from ltp.sut import SUTTimeoutError
 from ltp.utils import Timeout
 
 
-class Printer:
+class Printer(IOBuffer):
     """
     stdout printer.
     """
@@ -20,9 +21,8 @@ def __init__(self) -> None:
         self._logger = logging.getLogger("test.host")
         self._line = ""
 
-    def write(self, data):
-        data_str = data.decode(encoding="utf-8", errors="replace")
-        print(data_str, end="")
+    def write(self, data: str) -> None:
+        print(data, end="")
 
     def flush(self):
         pass

ltp/tests/test_ssh.py

@@ -78,9 +78,9 @@ def test_reset_command(self, config):
         sut.communicate()
 
         class MyBuffer(IOBuffer):
-            data = bytes()
+            data = ""
 
-            def write(self, data: bytes) -> None:
+            def write(self, data: str) -> None:
                 self.data = data
                 # wait for data inside the buffer
                 time.sleep(0.1)

ltp/ui.py

@@ -79,19 +79,17 @@ def _user_friendly_duration(duration: float) -> str:
 
         return uf_time
 
-    def _print_stdout(self, data: bytes) -> None:
+    def _print_stdout(self, data: str) -> None:
         """
         Print stdout coming from command run or test.
         """
-        data_str = data.decode(encoding="utf-8", errors="replace")
-
-        if len(data_str) == 1:
-            self._line += data_str
-            if data_str == "\n":
+        if len(data) == 1:
+            self._line += data
+            if data == "\n":
                 self._print(self._line, end='')
                 self._line = ""
         else:
-            lines = data_str.splitlines(True)
+            lines = data.splitlines(True)
             if len(lines) > 0:
                 for line in lines[:-1]:
                     self._line += line
@@ -101,7 +99,7 @@ def _print_stdout(self, data: bytes) -> None:
 
                 self._line = lines[-1]
 
-            if data_str.endswith('\n') and self._line:
+            if data.endswith('\n') and self._line:
                 self._print(self._line, end='')
                 self._line = ""
 
@@ -133,7 +131,7 @@ def sut_restart(self, sut: str) -> None:
     def run_cmd_start(self, cmd: str) -> None:
         self._print(f"{cmd}", color=self.CYAN)
 
-    def run_cmd_stdout(self, data: bytes) -> None:
+    def run_cmd_stdout(self, data: str) -> None:
         self._print_stdout(data)
 
     def run_cmd_stop(self, command: str, stdout: str, returncode: int) -> None:
@@ -274,7 +272,7 @@ def __init__(self, no_colors: bool = False) -> None:
         ltp.events.register("test_completed", self.test_completed)
         ltp.events.register("test_stdout_line", self.test_stdout_line)
 
-    def sut_stdout_line(self, _: str, data: bytes) -> None:
+    def sut_stdout_line(self, _: str, data: str) -> None:
         self._print_stdout(data)
 
     def kernel_tained(self, message: str) -> None: