Skip to content

Commit 8c74db5

Browse files
committed
test: add tests for FIFO file content waiting logic
Add tests for _wait_for_file_content() to verify it correctly handles both regular files and FIFOs. The existing FIFO test worked because it started a writer thread before reading, ensuring the FIFO was ready. However, with 1Password's asynchronously mounted FIFO files, the FIFO may exist but not be immediately readable when first accessed. The old code would open FIFOs directly in text mode, which could block indefinitely waiting for a writer. The new _wait_for_file_content() function addresses this by: - Detecting FIFOs using stat.S_ISFIFO() - Using select.select() to wait for the FIFO to become readable - Reading content only when the FIFO is ready These tests verify: - Regular files continue to work correctly (no regression) - FIFOs are properly handled with waiting logic for async-mounted files
1 parent ad86b66 commit 8c74db5

File tree

1 file changed

+33
-0
lines changed

1 file changed

+33
-0
lines changed

tests/test_fifo_dotenv.py

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import pytest
77

88
from dotenv import load_dotenv
9+
from dotenv.main import _wait_for_file_content
910

1011
pytestmark = pytest.mark.skipif(
1112
sys.platform.startswith("win"), reason="FIFOs are Unix-only"
@@ -31,3 +32,35 @@ def writer():
3132

3233
assert ok is True
3334
assert os.getenv("MY_PASSWORD") == "pipe-secret"
35+
36+
37+
def test_wait_for_file_content_with_regular_file(tmp_path: pathlib.Path):
38+
"""Test that _wait_for_file_content handles regular files correctly."""
39+
regular_file = tmp_path / ".env"
40+
regular_file.write_text("KEY=value\n", encoding="utf-8")
41+
42+
stream = _wait_for_file_content(str(regular_file), encoding="utf-8")
43+
content = stream.read()
44+
45+
assert content == "KEY=value\n"
46+
stream.close()
47+
48+
49+
def test_wait_for_file_content_with_fifo(tmp_path: pathlib.Path):
50+
"""Test that _wait_for_file_content handles FIFOs correctly."""
51+
fifo = tmp_path / ".env"
52+
os.mkfifo(fifo)
53+
54+
def writer():
55+
with open(fifo, "w", encoding="utf-8") as w:
56+
w.write("FIFO_KEY=fifo-value\n")
57+
58+
t = threading.Thread(target=writer)
59+
t.start()
60+
61+
stream = _wait_for_file_content(str(fifo), encoding="utf-8")
62+
content = stream.read()
63+
64+
t.join(timeout=2)
65+
66+
assert content == "FIFO_KEY=fifo-value\n"

0 commit comments

Comments
 (0)