microvm tests: stream OpenVMM transcripts to disk as they are read

scripts/nvx_tools/openvmm_process.py

@@ -83,7 +83,15 @@ def __init__(
         process_environment["OPENVMM_LOG"] = "off"
         if environment is not None:
             process_environment.update(environment)
-        self._interaction = InteractiveProcess(command, process_environment)
+        # The transcript is open before OpenVMM starts, so a failure to create it
+        # stops the run here, and every consumed byte reaches it as it is read.
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        self._log = log_path.open("wb")
+        try:
+            self._interaction = InteractiveProcess(command, process_environment)
+        except BaseException:
+            self._log.close()
+            raise
         self._chunks: queue.Queue[bytes | None] = queue.Queue()
         self._reader = threading.Thread(
             target=self._interaction.read_output,
@@ -97,6 +105,7 @@ def __init__(
         except BaseException:
             terminate(self._interaction.process)
             self._interaction.close()
+            self._log.close()
             raise
         self._output = bytearray()
         self._search_offset = 0
@@ -154,7 +163,7 @@ def _query_status_first(self) -> None:
             self._consume(chunk)
 
     def _consume(self, chunk: bytes) -> None:
-        self._output.extend(chunk)
+        self._record(chunk)
         try:
             self._monitor.feed(chunk)
         except TimeAbiFailure as error:
@@ -266,7 +275,7 @@ def wait(self, timeout: float) -> OpenvmmProcessResult:
         except TimeAbiFailure as error:
             self._fail(error)
         self._finished = True
-        self._write_log()
+        self._close_log()
         return OpenvmmProcessResult(returncode, bytes(self._output))
 
     def close(self) -> None:
@@ -279,24 +288,52 @@ def close(self) -> None:
         """
         if not self._finished and self.process.poll() is None:
             terminate(self.process)
-        self._drain_available()
-        self._write_log()
-        self._interaction.close()
-        self._finished = True
+        try:
+            # The reader queues its last chunks while the interaction shuts down, so
+            # drain only after it has stopped. Drain even if that shutdown fails, and
+            # release the transcript either way.
+            try:
+                self._interaction.close()
+            finally:
+                if not self._log.closed:
+                    self._drain_available()
+        finally:
+            self._close_log()
+            self._finished = True
 
     def _drain_available(self) -> None:
-        """Keep the output still queued; failure paths only, so no scan."""
+        """Keep the output still queued; failure paths only, so no scan.
+
+        A transcript write error does not stop the drain: every queued chunk is
+        kept in memory, and the first error is re-raised once the queue is empty.
+        """
+        error: Exception | None = None
         while True:
             try:
                 chunk = self._chunks.get_nowait()
             except queue.Empty:
-                return
-            if chunk is not None:
-                self._output.extend(chunk)
+                break
+            if chunk is None:
+                continue
+            self._output.extend(chunk)
+            if error is None:
+                try:
+                    self._log.write(chunk)
+                    self._log.flush()
+                except Exception as write_error:
+                    error = write_error
+        if error is not None:
+            raise error
+
+    def _record(self, chunk: bytes) -> None:
+        """Keep a consumed chunk in memory and in the transcript, flushed for readers."""
+        self._output.extend(chunk)
+        self._log.write(chunk)
+        self._log.flush()
 
-    def _write_log(self) -> None:
-        self._log_path.parent.mkdir(parents=True, exist_ok=True)
-        self._log_path.write_bytes(self._output)
+    def _close_log(self) -> None:
+        if not self._log.closed:
+            self._log.close()
 
     def _fail(self, error: Exception):
         self.close()

scripts/test_microvm_tests.py

@@ -4827,6 +4827,139 @@ def test_process_wait_reads_final_chunks_after_process_exit(self):
                 self.assertEqual(log_path.read_bytes(), result.output)
                 self.assertEqual(queues.return_value.get.call_count, 4)
 
+    def test_process_transcript_grows_while_the_process_runs(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            with (
+                patch.object(openvmm_process, "InteractiveProcess") as interaction,
+                patch.object(openvmm_process.threading, "Thread"),
+                patch.object(openvmm_process.queue, "Queue") as queues,
+            ):
+                interaction.return_value.process.poll.return_value = None
+                interaction.return_value.process.wait.return_value = 0
+                queues.return_value.get.side_effect = [b"BOOT\n", None]
+                queues.return_value.get_nowait.side_effect = queue.Empty
+                with openvmm_process.OpenvmmProcess(["openvmm"], log_path) as process:
+                    process.wait_for(b"BOOT", 1)
+                    # Consumed bytes are on disk while OpenVMM is still running.
+                    self.assertEqual(log_path.read_bytes(), b"BOOT\n")
+                    process.wait(1)
+                self.assertEqual(log_path.read_bytes(), b"BOOT\n")
+
+    def test_process_transcript_open_failure_surfaces_before_start(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            log_path.mkdir()
+            with patch.object(openvmm_process, "InteractiveProcess") as interaction:
+                with self.assertRaises(OSError):
+                    openvmm_process.OpenvmmProcess(["openvmm"], log_path)
+            interaction.assert_not_called()
+
+    def test_process_close_writes_output_delivered_after_the_last_wait(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            with (
+                patch.object(openvmm_process, "InteractiveProcess") as interaction,
+                patch.object(openvmm_process.threading, "Thread"),
+                patch.object(openvmm_process.queue, "Queue") as queues,
+            ):
+                interaction.return_value.process.poll.return_value = None
+                queues.return_value.get_nowait.side_effect = [b"LATE\n", queue.Empty]
+                process = openvmm_process.OpenvmmProcess(["openvmm"], log_path)
+                process.close()
+            self.assertEqual(log_path.read_bytes(), b"LATE\n")
+
+    def test_process_close_keeps_output_queued_during_interaction_close(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            closed: list[bool] = []
+            with (
+                patch.object(openvmm_process, "InteractiveProcess") as interaction,
+                patch.object(openvmm_process.threading, "Thread"),
+                patch.object(openvmm_process.queue, "Queue") as queues,
+            ):
+                interaction.return_value.process.poll.return_value = None
+                interaction.return_value.close.side_effect = lambda: closed.append(True)
+
+                pending = [b"LAST\n"]
+
+                def get_nowait() -> bytes:
+                    # The reader queues its last chunk only while the interaction shuts down.
+                    if closed and pending:
+                        return pending.pop()
+                    raise queue.Empty
+
+                queues.return_value.get_nowait.side_effect = get_nowait
+                process = openvmm_process.OpenvmmProcess(["openvmm"], log_path)
+                process.close()
+                self.assertEqual(process.output, b"LAST\n")
+            self.assertEqual(log_path.read_bytes(), b"LAST\n")
+
+    def test_process_close_keeps_queued_output_when_interaction_close_fails(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            with (
+                patch.object(openvmm_process, "InteractiveProcess") as interaction,
+                patch.object(openvmm_process.threading, "Thread"),
+                patch.object(openvmm_process.queue, "Queue") as queues,
+            ):
+                interaction.return_value.process.poll.return_value = None
+                interaction.return_value.close.side_effect = OSError("reader stuck")
+                queues.return_value.get_nowait.side_effect = [b"QUEUED\n", queue.Empty]
+                process = openvmm_process.OpenvmmProcess(["openvmm"], log_path)
+                with self.assertRaises(OSError):
+                    process.close()
+                self.assertEqual(process.output, b"QUEUED\n")
+            self.assertEqual(log_path.read_bytes(), b"QUEUED\n")
+
+    def test_process_close_releases_resources_when_the_transcript_write_fails(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            with (
+                patch.object(openvmm_process, "InteractiveProcess") as interaction,
+                patch.object(openvmm_process.threading, "Thread"),
+                patch.object(openvmm_process.queue, "Queue") as queues,
+            ):
+                interaction.return_value.process.poll.return_value = None
+                queues.return_value.get_nowait.side_effect = [b"LATE\n", queue.Empty]
+                process = openvmm_process.OpenvmmProcess(["openvmm"], log_path)
+                failing = MagicMock()
+                failing.closed = False
+                failing.write.side_effect = OSError("no space left on device")
+                process._log.close()
+                process._log = failing
+                with self.assertRaises(OSError):
+                    process.close()
+            interaction.return_value.close.assert_called_once_with()
+            failing.close.assert_called_once_with()
+
+    def test_process_close_keeps_every_queued_chunk_when_the_transcript_write_fails(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            log_path = Path(temporary) / "output.log"
+            with (
+                patch.object(openvmm_process, "InteractiveProcess") as interaction,
+                patch.object(openvmm_process.threading, "Thread"),
+                patch.object(openvmm_process.queue, "Queue") as queues,
+            ):
+                interaction.return_value.process.poll.return_value = None
+                queues.return_value.get_nowait.side_effect = [
+                    b"FIRST\n",
+                    b"SECOND\n",
+                    b"THIRD\n",
+                    queue.Empty,
+                ]
+                process = openvmm_process.OpenvmmProcess(["openvmm"], log_path)
+                failing = MagicMock()
+                failing.closed = False
+                failing.write.side_effect = OSError("no space left on device")
+                process._log.close()
+                process._log = failing
+                with self.assertRaises(OSError):
+                    process.close()
+                self.assertEqual(bytes(process._output), b"FIRST\nSECOND\nTHIRD\n")
+            interaction.return_value.close.assert_called_once_with()
+            failing.close.assert_called_once_with()
+
     def test_process_wait_clamps_elapsed_process_timeout(self):
         with tempfile.TemporaryDirectory() as temporary:
             with (