Saddala commented on code in PR #65002:
URL: https://github.com/apache/airflow/pull/65002#discussion_r3104973742
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -3147,6 +3147,22 @@ def
test_process_log_messages_from_subprocess(monkeypatch, caplog):
]
+def test_process_log_messages_tolerates_closed_log_handle():
+ """ValueError from a closed log handle must not propagate out of the
generator."""
+ broken_logger = MagicMock()
+ broken_logger.log.side_effect = ValueError("write to closed file")
+
+ with mock.patch(
+ "airflow.sdk.execution_time.supervisor.reconfigure_logger",
+ return_value=broken_logger,
+ ):
+ gen = process_log_messages_from_subprocess(loggers=(broken_logger,))
+ next(gen)
Review Comment:
Added tests with spec=FilteringBoundLogger / spec=socket
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -831,6 +831,79 @@ def test_kill_timed_out_processors_no_kill(self):
manager._kill_timed_out_processors()
mock_kill.assert_not_called()
+ def test_terminate_orphan_processes_deregisters_sockets(self):
+ manager = DagFileProcessorManager(max_runs=1)
+ processor, _ = self.mock_processor()
+
+ sock1, sock2 = MagicMock(), MagicMock()
+ processor._open_sockets[sock1] = "log"
+ processor._open_sockets[sock2] = "log"
+
+ file_info = DagFileInfo(
+ bundle_name="testing", rel_path=Path("removed.py"),
bundle_path=TEST_DAGS_FOLDER
+ )
+ manager._processors = {file_info: processor}
+ manager.selector = MagicMock()
+
+ with mock.patch.object(type(processor), "kill"):
+ manager.terminate_orphan_processes(present=set())
+
+ manager.selector.unregister.assert_any_call(sock1)
+ manager.selector.unregister.assert_any_call(sock2)
+ sock1.close.assert_called_once()
+ sock2.close.assert_called_once()
+ assert not processor._open_sockets
+ processor.logger_filehandle.close.assert_called_once()
+
+ def test_kill_timed_out_processors_deregisters_sockets(self):
+ manager = DagFileProcessorManager(max_runs=1, processor_timeout=5)
+ start_time = time.monotonic() - manager.processor_timeout - 1
+ processor, _ = self.mock_processor(start_time=start_time)
+
+ sock = MagicMock()
+ processor._open_sockets[sock] = "log"
+
Review Comment:
Added tests with spec=FilteringBoundLogger / spec=socket
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
