Make is_stdout work when sys.stdout does not have a fd

quentin-sommer · quentin-sommer · commit a4b30593ee1a · 2025-05-08T12:54:00.000-04:00
diff --git a/airflow-core/src/airflow/cli/utils.py b/airflow-core/src/airflow/cli/utils.py
@@ -44,13 +44,12 @@ def is_stdout(fileio: IOBase) -> bool:
     with argparse.FileType points to stdout (by setting the path to ``-``). This
     is why there is no equivalent for stderr; argparse does not allow using it.
 
-    .. warning:: *fileio* must be open for this check to be successful.
     """
-    return fileio.fileno() == sys.stdout.fileno()
+    return fileio is sys.stdout
 
 
 def print_export_output(command_type: str, exported_items: Collection, file: TextIOWrapper):
-    if not file.closed and is_stdout(file):
+    if is_stdout(file):
         print(f"\n{len(exported_items)} {command_type} successfully exported.", file=sys.stderr)
     else:
         print(f"{len(exported_items)} {command_type} successfully exported to {file.name}.")
diff --git a/airflow-core/tests/unit/cli/commands/test_connection_command.py b/airflow-core/tests/unit/cli/commands/test_connection_command.py
@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import io
 import json
 import os
 import re
@@ -333,6 +334,38 @@ def test_cli_connections_export_should_force_export_as_specified_format(self, tm
         }
         assert json.loads(output_filepath.read_text()) == expected_connections
 
+    def test_cli_connections_export_should_work_when_stdout_is_not_a_real_fd(self, tmp_path):
+        class FakeFileStringIO(io.StringIO):
+            """
+            Buffer the contents of a StringIO
+            to make them accessible after close
+            """
+
+            def __init__(self):
+                super().__init__()
+                self.content = ""
+
+            def write(self, s: str) -> int:
+                self.content += s
+                return len(s)
+
+        tmp_stdout = FakeFileStringIO()
+        with redirect_stdout(tmp_stdout):
+            args = self.parser.parse_args(
+                [
+                    "connections",
+                    "export",
+                    "--format",
+                    "env",
+                    "-",
+                ]
+            )
+            connection_command.connections_export(args)
+        assert tmp_stdout.content.splitlines() == [
+            "airflow_db=mysql://root:plainpassword@mysql/airflow",
+            "druid_broker_default=druid://druid-broker:8082/?endpoint=druid%2Fv2%2Fsql",
+        ]
+
 
 TEST_URL = "postgresql://airflow:airflow@host:5432/airflow"
 TEST_JSON = json.dumps(
