fix: Handle non-UTF-8 lines in plugin output for logging (#8399)

* fix: Handle non-UTF-8 lines in plugin output for logging #8377

* Test change

---------

Co-authored-by: nKandel <npkand+3070@gmail.com>
Co-authored-by: Edgar Ramírez-Mondragón <edgar@meltano.com>

.github/workflows/test.yml

@@ -289,6 +289,8 @@ jobs:
 
     - name: Upload coverage report
       with:
+        fail_ci_if_error: true
+        files: ./coverage.xml
         token: ${{ secrets.CODECOV_TOKEN }}
       uses: codecov/codecov-action@v4.0.1
 

src/meltano/core/logging/utils.py

@@ -205,7 +205,7 @@ async def _write_line_writer(writer: SubprocessOutputWriter, line: bytes):
             await writer.wait_closed()
             return False
     else:
-        writer.writeline(line.decode())
+        writer.writeline(line.decode(errors="replace"))
 
     return True
 
@@ -214,7 +214,7 @@ async def capture_subprocess_output(
     reader: asyncio.StreamReader | None,
     *line_writers: SubprocessOutputWriter,
 ) -> None:
-    """Capture in real time the output stream of a suprocess that is run async.
+    """Capture in real time the output stream of a subprocess that is run async.
 
     The stream has been set to asyncio.subprocess.PIPE and is provided using
     reader to this function.

tests/meltano/core/logging/test_logging_utils.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from meltano.core.logging.utils import capture_subprocess_output
+
+
+class AsyncReader(asyncio.StreamReader):
+    def __init__(self, lines: list[bytes]):
+        self.lines = lines
+
+    def at_eof(self) -> bool:
+        return not self.lines
+
+    async def readline(self) -> bytes:
+        return self.lines.pop(0) if self.lines else b""
+
+
+@pytest.mark.asyncio()
+async def test_capture_subprocess_output():
+    INPUT_LINES = [b"LINE\n", b"LINE 2\n", b"\xed\n"]
+    OUTPUT_LINES = []
+
+    class LineWriter:
+        def writeline(self, line: str):
+            OUTPUT_LINES.append(line)
+
+    reader = AsyncReader(INPUT_LINES)
+
+    await capture_subprocess_output(reader, LineWriter())
+    assert ["LINE\n", "LINE 2\n", "�\n"] == OUTPUT_LINES