Coverage for tests/io/test_source.py: 100%

1"""Unit tests for the ByteSource hierarchy (io/_source.py). 

2 

3Covers BufferSource, StreamSource, MmapSource and the make_source() factory: 

4sequential reads, non-consuming peek, seek/tell clamping, zero-copy mapping, 

5lifetime rules (views must be dropped before close), and factory dispatch. 

6""" 

7import io 

8 

9import numpy as np 

10import pytest 

11 

12from typing import Any 

13 

14from evutils.io._source import ( 

15 BufferSource, 

16 MmapSource, 

17 StreamSource, 

18 make_source, 

19) 

20 

21PAYLOAD = b"0123456789abcdef" 

22 

23 

24#################################### 

25# BufferSource 

26#################################### 

27 

28def test_buffersource_sequential_read() -> None: 

29 src = BufferSource(PAYLOAD) 

30 assert src.read(4) == b"0123" 

31 assert src.read(4) == b"4567" 

32 assert src.read(-1) == b"89abcdef" 

33 assert src.read(4) == b"" # EOF 

34 

35 

36def test_buffersource_peek_does_not_consume() -> None: 

37 src = BufferSource(PAYLOAD) 

38 assert src.peek(4) == b"0123" 

39 assert src.peek(4) == b"0123" 

40 assert src.read(4) == b"0123" 

41 assert src.peek(4) == b"4567" 

42 

43 

44def test_buffersource_peek_past_end() -> None: 

45 src = BufferSource(b"ab") 

46 assert src.peek(10) == b"ab" 

47 

48 

49def test_buffersource_seek_tell() -> None: 

50 src = BufferSource(PAYLOAD) 

51 assert src.seekable() 

52 assert src.tell() == 0 

53 assert src.seek(8) == 8 

54 assert src.tell() == 8 

55 assert src.read(2) == b"89" 

56 assert src.seek(-2, io.SEEK_CUR) == 8 

57 assert src.seek(-4, io.SEEK_END) == len(PAYLOAD) - 4 

58 assert src.read(-1) == b"cdef" 

59 

60 

61def test_buffersource_seek_clamps() -> None: 

62 src = BufferSource(PAYLOAD) 

63 assert src.seek(-100) == 0 

64 assert src.seek(1000) == len(PAYLOAD) 

65 assert src.read(1) == b"" 

66 with pytest.raises(ValueError): 

67 src.seek(0, whence=42) 

68 

69 

70def test_buffersource_mappable_zero_copy() -> None: 

71 data = bytearray(PAYLOAD) 

72 src = BufferSource(data) 

73 assert src.mappable() 

74 mv = src.buffer() 

75 assert bytes(mv) == PAYLOAD 

76 data[0] = ord(b"X") # same storage: mutation is visible through the view 

77 assert bytes(mv[:1]) == b"X" 

78 

79 

80def test_buffersource_accepts_memoryview_and_bytesio_buffer() -> None: 

81 assert BufferSource(memoryview(PAYLOAD)).read(-1) == PAYLOAD 

82 bio = io.BytesIO(PAYLOAD) 

83 assert BufferSource(bio.getbuffer()).read(-1) == PAYLOAD 

84 

85 

86def test_buffersource_readline_generic() -> None: 

87 src = BufferSource(b"line1\nline2\nno-newline") 

88 assert src.readline() == b"line1\n" 

89 assert src.readline() == b"line2\n" 

90 assert src.readline() == b"no-newline" 

91 assert src.readline() == b"" 

92 

93 

94def test_buffersource_reset() -> None: 

95 src = BufferSource(PAYLOAD) 

96 src.read(8) 

97 src.reset() 

98 assert src.tell() == 0 

99 assert src.read(4) == b"0123" 

100 

101 

102#################################### 

103# StreamSource 

104#################################### 

105 

106class _ReadOnlyStream: 

107 """Minimal non-seekable, non-peekable binary stream (pipe-like).""" 

108 

109 def __init__(self, data: bytes) -> None: 

110 self._bio = io.BytesIO(data) 

111 

112 def read(self, size: int = -1) -> bytes: 

113 return self._bio.read(size) 

114 

115 def seekable(self) -> bool: 

116 return False 

117 

118 

119def test_streamsource_requires_read() -> None: 

120 with pytest.raises(TypeError): 

121 StreamSource(object()) 

122 

123 

124def test_streamsource_read_and_seek_via_bytesio() -> None: 

125 src = StreamSource(io.BytesIO(PAYLOAD)) 

126 assert src.read(4) == b"0123" 

127 assert src.seekable() 

128 assert src.tell() == 4 

129 src.seek(0) 

130 assert src.read(-1) == PAYLOAD 

131 assert not src.mappable() 

132 with pytest.raises(io.UnsupportedOperation): 

133 src.buffer() 

134 

135 

136def test_streamsource_peek_fallback_via_seek() -> None: 

137 # BytesIO has no peek() but is seekable: peek must restore the position. 

138 src = StreamSource(io.BytesIO(PAYLOAD)) 

139 src.read(2) 

140 assert src.peek(4) == b"2345" 

141 assert src.tell() == 2 

142 assert src.read(4) == b"2345" 

143 

144 

145def test_streamsource_peek_native(tmp_path: Any) -> None: 

146 # BufferedReader exposes peek() directly. 

147 p = tmp_path / "data.bin" 

148 p.write_bytes(PAYLOAD) 

149 with open(p, "rb") as f: 

150 src = StreamSource(f) 

151 assert src.peek(4) == b"0123" 

152 assert src.read(4) == b"0123" 

153 assert src.name == str(p) 

154 

155 

156def test_streamsource_nonseekable_peek_raises() -> None: 

157 src = StreamSource(_ReadOnlyStream(PAYLOAD)) 

158 assert not src.seekable() 

159 with pytest.raises(io.UnsupportedOperation): 

160 src.peek(4) 

161 # sequential reading still works 

162 assert src.read(4) == b"0123" 

163 

164 

165def test_streamsource_readline_delegates() -> None: 

166 src = StreamSource(io.BytesIO(b"a\nb\n")) 

167 assert src.readline() == b"a\n" 

168 assert src.readline() == b"b\n" 

169 

170 

171def test_streamsource_close_ownership() -> None: 

172 owned = io.BytesIO(PAYLOAD) 

173 StreamSource(owned, owns=True).close() 

174 assert owned.closed 

175 

176 borrowed = io.BytesIO(PAYLOAD) 

177 StreamSource(borrowed, owns=False).close() 

178 assert not borrowed.closed 

179 

180 

181#################################### 

182# MmapSource 

183#################################### 

184 

185@pytest.fixture 

186def payload_file(tmp_path: Any) -> Any: 

187 p = tmp_path / "payload.bin" 

188 p.write_bytes(PAYLOAD) 

189 return p 

190 

191 

192def test_mmapsource_read_peek_seek(payload_file: Any) -> None: 

193 with MmapSource(payload_file) as src: 

194 assert src.mappable() and src.seekable() 

195 assert src.name == payload_file.name 

196 assert src.peek(4) == b"0123" 

197 assert src.read(4) == b"0123" 

198 assert src.tell() == 4 

199 src.seek(-4, io.SEEK_END) 

200 assert src.read(-1) == b"cdef" 

201 

202 

203def test_mmapsource_buffer_zero_copy(payload_file: Any) -> None: 

204 src = MmapSource(payload_file) 

205 mv = src.buffer() 

206 arr = np.frombuffer(mv, dtype=np.uint8) 

207 assert bytes(arr.tobytes()) == PAYLOAD 

208 # Views alias the mapping: they must be dropped before close(). 

209 del arr 

210 mv.release() 

211 src.close() 

212 

213 

214def test_mmapsource_close_with_live_view_raises(payload_file: Any) -> None: 

215 src = MmapSource(payload_file) 

216 mv = src.buffer() 

217 with pytest.raises(BufferError): 

218 src.close() 

219 mv.release() 

220 src.close() 

221 

222 

223def test_mmapsource_empty_file_raises(tmp_path: Any) -> None: 

224 p = tmp_path / "empty.bin" 

225 p.touch() 

226 with pytest.raises((ValueError, OSError)): 

227 MmapSource(p) 

228 

229 

230#################################### 

231# make_source factory 

232#################################### 

233 

234def test_make_source_path_maps(payload_file: Any) -> None: 

235 src = make_source(payload_file) 

236 assert isinstance(src, MmapSource) 

237 assert src.read(-1) == PAYLOAD 

238 src.close() 

239 # str path works too 

240 src = make_source(str(payload_file)) 

241 assert isinstance(src, MmapSource) 

242 src.close() 

243 

244 

245def test_make_source_path_no_mmap(payload_file: Any) -> None: 

246 src = make_source(payload_file, mmap_files=False) 

247 assert isinstance(src, StreamSource) 

248 assert src.read(-1) == PAYLOAD 

249 src.close() 

250 

251 

252def test_make_source_empty_file_falls_back_to_stream(tmp_path: Any) -> None: 

253 p = tmp_path / "empty.bin" 

254 p.touch() 

255 src = make_source(p) 

256 assert isinstance(src, StreamSource) 

257 assert src.read(-1) == b"" 

258 src.close() 

259 

260 

261def test_make_source_missing_file() -> None: 

262 with pytest.raises(FileNotFoundError): 

263 make_source("/nonexistent/definitely_missing.raw") 

264 

265 

266def test_make_source_buffers() -> None: 

267 assert isinstance(make_source(PAYLOAD), BufferSource) 

268 assert isinstance(make_source(bytearray(PAYLOAD)), BufferSource) 

269 assert isinstance(make_source(memoryview(PAYLOAD)), BufferSource) 

270 

271 

272def test_make_source_bytesio_zero_copy() -> None: 

273 src = make_source(io.BytesIO(PAYLOAD)) 

274 assert isinstance(src, BufferSource) 

275 assert src.read(-1) == PAYLOAD 

276 

277 

278def test_make_source_readable_object() -> None: 

279 src = make_source(_ReadOnlyStream(PAYLOAD)) 

280 assert isinstance(src, StreamSource) 

281 assert src.read(-1) == PAYLOAD 

282 

283 

284def test_make_source_passthrough() -> None: 

285 inner = BufferSource(PAYLOAD) 

286 assert make_source(inner) is inner 

287 

288 

289def test_make_source_rejects_unknown_type() -> None: 

290 with pytest.raises(TypeError): 

291 make_source(12345)