|
1 | 1 | """Tests for :py:mod:`cheroot.makefile`.""" |
2 | 2 |
|
| 3 | +import errno |
| 4 | +import io |
| 5 | +import math |
| 6 | + |
| 7 | +import pytest |
| 8 | + |
3 | 9 | from cheroot import makefile |
4 | 10 |
|
5 | 11 |
|
6 | 12 | class MockSocket: |
7 | | - """A mock socket.""" |
| 13 | + """A mock socket for emulating buffered I/O.""" |
8 | 14 |
|
9 | 15 | def __init__(self): |
10 | 16 | """Initialize :py:class:`MockSocket`.""" |
@@ -51,3 +57,116 @@ def test_bytes_written(): |
51 | 57 | wfile = makefile.MakeFile(sock, 'w') |
52 | 58 | wfile.write(b'bar') |
53 | 59 | assert wfile.bytes_written == 3 |
| 60 | + |
| 61 | + |
| 62 | +def test_close_is_idempotent(): |
| 63 | + """Test that double ``close()`` does not error out.""" |
| 64 | + raw_buffer = io.BytesIO() |
| 65 | + buffered_writer = makefile.BufferedWriter(raw_buffer) |
| 66 | + |
| 67 | + # Should not raise any exceptions |
| 68 | + buffered_writer.close() |
| 69 | + assert buffered_writer.closed |
| 70 | + |
| 71 | + buffered_writer.close() # Second call should be safe |
| 72 | + assert buffered_writer.closed |
| 73 | + |
| 74 | + |
| 75 | +def test_close_handles_already_closed_buffer(): |
| 76 | + """Test that ``close()`` handles already closed underlying buffer.""" |
| 77 | + raw_buffer = io.BytesIO() |
| 78 | + buffered_writer = makefile.BufferedWriter(raw_buffer) |
| 79 | + |
| 80 | + # Close the underlying buffer first |
| 81 | + raw_buffer.close() |
| 82 | + |
| 83 | + # This should not raise an exception |
| 84 | + assert raw_buffer.closed |
| 85 | + assert buffered_writer.closed |
| 86 | + |
| 87 | + |
| 88 | +def test_flush_unlocked_handles_blocking_io_error(mock_buffer_writer, mocker): |
| 89 | + """ |
| 90 | + Test that a BlockingIOError is handled correctly. |
| 91 | +
|
| 92 | + We extracting characters_written, |
| 93 | + and execution continues without raising the error. |
| 94 | + """ |
| 95 | + # 1. Create a mock object to replace the real 'write' method |
| 96 | + mock_write_method = mocker.Mock() |
| 97 | + |
| 98 | + # 2. Set the side effect on the mock object |
| 99 | + err = io.BlockingIOError(errno.EAGAIN, 'Resource temporarily unavailable') |
| 100 | + err.characters_written = 5 |
| 101 | + mock_write_method.side_effect = err |
| 102 | + |
| 103 | + # 3. Use mocker.patch.object to replace the 'write' method |
| 104 | + # with mock_write_method |
| 105 | + mocker.patch.object(mock_buffer_writer.raw, 'write', new=mock_write_method) |
| 106 | + |
| 107 | + # Check the initial state of the buffer |
| 108 | + initial_len = len(mock_buffer_writer._write_buf) |
| 109 | + |
| 110 | + # 4. Execute the code |
| 111 | + try: |
| 112 | + mock_buffer_writer._flush_unlocked() |
| 113 | + except Exception as exc: |
| 114 | + pytest.fail(f'Unexpected exception raised: {type(exc).__name__}') |
| 115 | + |
| 116 | + # 5. Verify the side-effect (buffer should be empty) |
| 117 | + assert len(mock_buffer_writer._write_buf) == 0 |
| 118 | + |
| 119 | + # 6 Check mock calls (Logic/Mechanism) |
| 120 | + # The number of calls should be |
| 121 | + # initial_len / bytes_written_per_call |
| 122 | + expected_calls = math.ceil(initial_len / 5) |
| 123 | + assert mock_write_method.call_count == expected_calls |
| 124 | + |
| 125 | + |
| 126 | +class MockRawSocket: |
| 127 | + """ |
| 128 | + A mock raw socket for emulating low level unbuffered I/O. |
| 129 | +
|
| 130 | + We use this mock with ``io.BufferedWriter``, which accesses it via |
| 131 | + the ``.raw`` attribute. |
| 132 | + """ |
| 133 | + |
| 134 | + def __init__(self, *args, **kwargs): |
| 135 | + """Initialize :py:class:`MockRawSocket`.""" |
| 136 | + # 1. Call the parent's init to set up self.messages |
| 137 | + super().__init__(*args, **kwargs) |
| 138 | + |
| 139 | + # 2. Rquired by the io.BufferedWriter base class |
| 140 | + self._is_closed = False |
| 141 | + |
| 142 | + def write(self, message): |
| 143 | + """Emulate ``io.RawIOBase write``.""" |
| 144 | + # Use the underlying send method implemented in MockSocket |
| 145 | + return self.send(message) |
| 146 | + |
| 147 | + def writable(self): |
| 148 | + """Indicate that the raw stream supports writing.""" |
| 149 | + return True |
| 150 | + |
| 151 | + def send(self, message): |
| 152 | + """Emulate a send.""" |
| 153 | + return len(message) |
| 154 | + |
| 155 | + def close(self): |
| 156 | + """Emulate close.""" |
| 157 | + self._is_closed = True |
| 158 | + |
| 159 | + @property |
| 160 | + def closed(self): |
| 161 | + """Emulate the required ``closed`` property.""" |
| 162 | + return self._is_closed |
| 163 | + |
| 164 | + |
| 165 | +@pytest.fixture |
| 166 | +def mock_buffer_writer(): |
| 167 | + """Fixture to create a BufferedWriter instance with a mock raw socket.""" |
| 168 | + # Create a BufferedWriter instance with a buffer that has content |
| 169 | + # to ensure _flush_unlocked attempts to write. |
| 170 | + writer = makefile.BufferedWriter(MockRawSocket()) |
| 171 | + writer._write_buf = bytearray(b'data to flush') |
| 172 | + return writer |
0 commit comments