handle FileNotFoundError in FIFO checks and add FIFO load test

Author: Sidharth Sudhir

src/dotenv/main.py

@@ -62,7 +62,7 @@ def __init__(
 
     @contextmanager
     def _get_stream(self) -> Iterator[IO[str]]:
-        if self.dotenv_path and os.path.isfile(self.dotenv_path) or stat.S_ISFIFO(os.stat(self.dotenv_path).st_mode):
+        if self.dotenv_path and _is_file_or_fifo(self.dotenv_path):
             with open(self.dotenv_path, encoding=self.encoding) as stream:
                 yield stream
         elif self.stream is not None:
@@ -326,7 +326,7 @@ def _is_debugger():
 
     for dirname in _walk_to_root(path):
         check_path = os.path.join(dirname, filename)
-        if os.path.isfile(check_path) or stat.S_ISFIFO(os.stat(check_path).st_mode):
+        if _is_file_or_fifo(check_path):
             return check_path
 
     if raise_error_if_not_found:
@@ -418,3 +418,18 @@ def dotenv_values(
         override=True,
         encoding=encoding,
     ).dict()
+
+
+def _is_file_or_fifo(path: StrPath) -> bool:
+    """
+    Return True if `path` exists and is either a regular file or a FIFO.
+    """
+    if os.path.isfile(path):
+        return True
+
+    try:
+        st = os.stat(path)
+    except (FileNotFoundError, OSError):
+        return False
+
+    return stat.S_ISFIFO(st.st_mode)

tests/test_fifo_dotenv.py

@@ -0,0 +1,33 @@
+import os
+import pathlib
+import sys
+import threading
+
+import pytest
+
+from dotenv import load_dotenv
+
+pytestmark = pytest.mark.skipif(
+    sys.platform.startswith("win"), reason="FIFOs are Unix-only"
+)
+
+
+def test_load_dotenv_from_fifo(tmp_path: pathlib.Path, monkeypatch):
+    fifo = tmp_path / ".env"
+    os.mkfifo(fifo)  # create named pipe
+
+    def writer():
+        with open(fifo, "w", encoding="utf-8") as w:
+            w.write("MY_PASSWORD=pipe-secret\n")
+
+    t = threading.Thread(target=writer)
+    t.start()
+
+    # Ensure env is clean
+    monkeypatch.delenv("MY_PASSWORD", raising=False)
+
+    ok = load_dotenv(dotenv_path=str(fifo), override=True)
+    t.join(timeout=2)
+
+    assert ok is True
+    assert os.getenv("MY_PASSWORD") == "pipe-secret"

tests/test_main.py

@@ -263,7 +263,9 @@ def test_load_dotenv_existing_file(dotenv_path):
 )
 def test_load_dotenv_disabled(dotenv_path, flag_value):
     expected_environ = {"PYTHON_DOTENV_DISABLED": flag_value}
-    with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True):
+    with mock.patch.dict(
+        os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True
+    ):
         dotenv_path.write_text("a=b")
 
         result = dotenv.load_dotenv(dotenv_path)
@@ -289,7 +291,9 @@ def test_load_dotenv_disabled(dotenv_path, flag_value):
     ],
 )
 def test_load_dotenv_disabled_notification(dotenv_path, flag_value):
-    with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True):
+    with mock.patch.dict(
+        os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True
+    ):
         dotenv_path.write_text("a=b")
 
         logger = logging.getLogger("dotenv.main")
@@ -298,7 +302,7 @@ def test_load_dotenv_disabled_notification(dotenv_path, flag_value):
 
         assert result is False
         mock_debug.assert_called_once_with(
-			"python-dotenv: .env loading disabled by PYTHON_DOTENV_DISABLED environment variable"
+            "python-dotenv: .env loading disabled by PYTHON_DOTENV_DISABLED environment variable"
         )
 
 
@@ -321,7 +325,9 @@ def test_load_dotenv_disabled_notification(dotenv_path, flag_value):
 )
 def test_load_dotenv_enabled(dotenv_path, flag_value):
     expected_environ = {"PYTHON_DOTENV_DISABLED": flag_value, "a": "b"}
-    with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True):
+    with mock.patch.dict(
+        os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True
+    ):
         dotenv_path.write_text("a=b")
 
         result = dotenv.load_dotenv(dotenv_path)
@@ -348,7 +354,9 @@ def test_load_dotenv_enabled(dotenv_path, flag_value):
     ],
 )
 def test_load_dotenv_enabled_no_notification(dotenv_path, flag_value):
-    with mock.patch.dict(os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True):
+    with mock.patch.dict(
+        os.environ, {"PYTHON_DOTENV_DISABLED": flag_value}, clear=True
+    ):
         dotenv_path.write_text("a=b")
 
         logger = logging.getLogger("dotenv.main")