fix: avoid flushing a missing capture stream when tracking URLs

luigi/contrib/external_program.py

@@ -38,7 +38,6 @@
 import tempfile
 from contextlib import contextmanager
 from multiprocessing import Process
-from time import sleep
 
 import luigi
 from luigi.parameter import ParameterVisibility
@@ -189,16 +188,15 @@ def _track_url_by_pattern():
             If tmp_stdout is passed, also appends lines to this file.
             """
             pattern = re.compile(self.tracking_url_pattern)
-            for new_line in iter(pipe_to_read.readline, ""):
-                if new_line:
-                    if file_to_write:
-                        file_to_write.write(new_line)
-                    match = re.search(pattern, new_line.decode("utf-8"))
-                    if match:
-                        self.set_tracking_url(self.build_tracking_url(match.group(1)))
-                else:
-                    file_to_write.flush()
-                    sleep(time_to_sleep)
+            # PIPE is binary, so EOF is b""; the old "" sentinel never stopped the iterator.
+            for new_line in iter(pipe_to_read.readline, b""):
+                if file_to_write:
+                    file_to_write.write(new_line)
+                match = re.search(pattern, new_line.decode("utf-8"))
+                if match:
+                    self.set_tracking_url(self.build_tracking_url(match.group(1)))
+            if file_to_write:
+                file_to_write.flush()
 
         track_proc = Process(target=_track_url_by_pattern)
         try:

test/contrib/external_program_test.py

@@ -20,7 +20,7 @@
 import tempfile
 from functools import partial
 from io import BytesIO
-from multiprocessing import Value
+from multiprocessing import Process, Value
 from subprocess import Popen
 
 import mock
@@ -198,6 +198,24 @@ def fake_set_tracking_url(val, url):
             task.run()
             self.assertEqual(test_val.value, 1)
 
+    def test_tracking_process_exits_cleanly_when_capture_output_disabled(self):
+        # When output is not captured, there is no file to tee into. EOF used to call
+        # flush() on that None handle and crash the tracker (issue #3131).
+        procs = []
+
+        class RecordingProcess(Process):
+            def __init__(self, *args, **kwargs):
+                super().__init__(*args, **kwargs)
+                procs.append(self)
+
+        task = TestEchoTask(capture_output=False, stream_for_searching_tracking_url="stdout", tracking_url_pattern=r"Hello, (.*)!")
+        with mock.patch("luigi.contrib.external_program.Process", RecordingProcess):
+            with mock.patch.object(task, "set_tracking_url"):
+                task.run()
+
+        self.assertEqual(len(procs), 1)
+        self.assertEqual(procs[0].exitcode, 0)
+
     def test_tracking_url_pattern_works_with_capture_output_enabled(self):
         test_val = Value("i", 0)