Merge pull request #5704 from grondo/issue#5702

python: fix `flux-watch: TypeError: Object of type 'bytes' is not JSON serializable`

Author: mergify[bot]

src/bindings/python/flux/job/output.py

@@ -113,7 +113,9 @@ def __init__(self, entry, labelio=False):
             data = self.context["data"]
             if "encoding" in self.context:
                 if self.context["encoding"] == "base64":
-                    data = base64.b64decode(data)
+                    data = base64.b64decode(data).decode(
+                        "utf-8", errors="surrogateescape"
+                    )
             if "repeat" in self.context:
                 data *= self.context["repeat"]
             if labelio:

src/bindings/python/flux/job/watcher.py

@@ -367,8 +367,8 @@ def __init__(
         self.t0 = starttime
         self.log_events = log_events
         self.log_status = log_status
-        self.stdout = stdout
-        self.stderr = stderr
+        self.stdout = self._reopen(stdout)
+        self.stderr = self._reopen(stderr)
         self.labelio = labelio
         self.exitcode = 0
         self.show_progress = progress
@@ -379,6 +379,22 @@ def __init__(
         if jobs:
             self.add_jobs(*jobs)
 
+    @staticmethod
+    def _reopen(stream):
+        """reconfigure/reopen stream with correct encoding and error handling"""
+        try:
+            # reconfigure() only available in >=3.7
+            stream.reconfigure(encoding="utf-8", errors="surrogateescape")
+            return stream
+        except AttributeError:
+            return open(
+                stream.fileno(),
+                mode="w",
+                encoding="utf-8",
+                errors="surrogateescape",
+                closefd=False,
+            )
+
     @staticmethod
     def _status_to_exitcode(status):
         """Calculate exitcode from job status"""
@@ -420,9 +436,9 @@ def add_jobs(self, *jobs, stdout=None, stderr=None, wait="clean"):
         self.progress.add_jobs(*jobs)
 
         if stdout is None:
-            stdout = self.stdout
+            stdout = self._reopen(self.stdout)
         if stderr is None:
-            stderr = self.stderr
+            stderr = self._reopen(self.stderr)
 
         for job in jobs:
             if not self.t0 or job.t_submit < self.t0:

t/t2813-flux-watch.t

@@ -130,4 +130,11 @@ test_expect_success 'flux-watch: --filter works' '
 	test_debug "cat failed.out" &&
 	grep "Watching ${nfailed} job" failed.out
 '
+test_expect_success 'flux-watch: handles binary data' '
+	id=$(flux submit dd if=/dev/urandom count=1) &&
+	test_debug "flux job eventlog -p guest.output -HL $id" &&
+	flux job attach $id >binary.expected &&
+	flux watch $id >binary.output &&
+	test_cmp binary.expected binary.output
+'
 test_done