add test

Author: wjddn279

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -303,6 +303,32 @@ def test_trigger_log(mock_monotonic, trigger, watcher_count, trigger_count, sess
     trigger_runner_supervisor.kill(force=False)
 
 
+def test_trigger_logger_fd_closed_when_removed(session):
+    import structlog
+
+    trigger = TimeDeltaTrigger(datetime.timedelta(seconds=0.5))
+
+    create_trigger_in_db(session, trigger)
+
+    mock_file = MagicMock()
+    mock_file.closed = False
+
+    mock_underlying_logger = MagicMock()
+    mock_underlying_logger._file = mock_file
+
+    with patch.object(structlog, "BytesLogger", return_value=mock_underlying_logger):
+        trigger_runner_supervisor = TriggerRunnerSupervisor.start(job=Job(id=123456), capacity=10)
+        trigger_runner_supervisor.load_triggers()
+
+        for _ in range(30):
+            trigger_runner_supervisor._service_subprocess(0.1)
+
+    mock_file.flush.assert_called_once()
+    mock_file.close.assert_called_once()
+
+    trigger_runner_supervisor.kill(force=False)
+
+
 class TestTriggerRunner:
     def test_run_inline_trigger_canceled(self, session) -> None:
         trigger_runner = TriggerRunner()