8000 Revert "move the stream test to the main libvips suite" · libvips/pyvips@29b922c · GitHub
[go: up one dir, main page]
Skip to content

Commit 29b922c

Browse files
jcupittjcupitt
committed
Revert "move the stream test to the main libvips suite"
This reverts commit b25e6e1.
1 parent b25e6e1 commit 29b922c

File tree

1 file changed

+159
-0
lines changed

1 file changed

+159
-0
lines changed

‎tests/test_streams.py

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
# vim: set fileencoding=utf-8 :
2+
3+
import tempfile
4+
import pytest
5+
6+
import pyvips
7+
from helpers import JPEG_FILE, temp_filename, skip_if_no
8+
9+
if pyvips.at_least_libvips(8, 9):
10+
class Mystreami(pyvips.Streamiu):
11+
def __init__(self, pointer):
12+
super(Mystreami, self).__init__(pointer)
13+
14+
# these must be attached before we build the streamiu, see `new`
15+
self.signal_connect('read', self.read_cb)
16+
self.signal_connect('seek', self.seek_cb)
17+
18+
@staticmethod
19+
def new(name, pipe_mode=False):
20+
"""Make a new input stream from a filename.
21+
22+
"""
23+
24+
gtype = pyvips.type_from_name('VipsStreamiu')
25+
pointer = pyvips.GObject.new_pointer_from_gtype(gtype)
26+
self = Mystreami(pointer)
27+
28+
self.name = name
29+
self.pipe_mode = pipe_mode
30+
self.loaded_bytes = open(name, 'rb').read()
31+
self.memory = memoryview(self.loaded_bytes)
32+
self.length = len(self.loaded_bytes)
33+
self.read_point = 0
34+
35+
return self.build()
36+
37+
def read_cb(self, buf):
38+
# print('read: {0} bytes ...'.format(len(buf)))
39+
p = self.read_point
40+
bytes_available = self.length - p
41+
bytes_to_copy = min(bytes_available, len(buf))
42+
buf[:bytes_to_copy] = self.memory[p:p + bytes_to_copy]
43+
self.read_point += bytes_to_copy
44+
# print(' copied from position {0}'.format(p))
45+
46+
return bytes_to_copy
47+
48+
def seek_cb(self, offset, whence):
49+
# print('seek: offset = {0}, whence = {1} ...'
50+
# .format(offset, whence))
51+
52+
if self.pipe_mode:
53+
# print(' -1 (pipe mode)')
54+
return -1
55+
56+
if whence == 0:
57+
# SEEK_SET
58+
new_read_point = offset
59+
elif whence == 1:
60+
# SEEK_CUR
61+
new_read_point = self.read_point + offset
62+
elif whence == 2:
63+
# SEEK_END
64+
new_read_point = self.length + offset
65+
else:
66+
raise Exception('bad whence {0}'.format(whence))
67+
68+
self.read_point = max(0, min(self.length, new_read_point))
69+
# print(' new read_point = {0}'.format(self.read_point))
70+
71+
return self.read_point
72+
73+
74+
class Mystreamo(pyvips.Streamou):
75+
def __init__(self, pointer):
76+
super(Mystreamo, self).__init__(pointer)
77+
78+
# these must be attached before we build the streamou, see `new`
79+
self.signal_connect('write', self.write_cb)
80+
self.signal_connect('finish', self.finish_cb)
81+
82+
@staticmethod
83+
def new(name):
84+
"""Make a new output stream from a filename.
85+
86+
"""
87+
88+
gtype = pyvips.type_from_name('VipsStreamou')
89+
pointer = pyvips.GObject.new_pointer_from_gtype(gtype)
90+
self = Mystreamo(pointer)
91+
92+
self.name = name
93+
self.f = open(name, 'wb')
94+
95+
return self.build()
96+
97+
def write_cb(self, buf):
98+
# print('write: {0} bytes ...'.format(len(buf)))
99+
# py2 write does not return number of bytes written
100+
self.f.write(buf)
101+
102+
return len(buf)
103+
104+
def finish_cb(self):
105+
# print('finish: ...')
106+
self.f.close()
107+
108+
109+
class TestStreams:
110+
@classmethod
111+
def setup_class(cls):
112+
cls.tempdir = tempfile.mkdtemp()
113+
114+
@skip_if_no('jpegload')
115+
@pytest.mark.skipif(not pyvips.at_least_libvips(8, 9),
116+
reason="requires libvips >= 8.9")
117+
def test_stream(self):
118+
streami = pyvips.Streami.new_from_file(JPEG_FILE)
119+
image = pyvips.Image.new_from_stream(streami, '', access='sequential')
120+
filename = temp_filename(self.tempdir, '.png')
121+
streamo = pyvips.Streamo.new_to_file(filename)
122+
image.write_to_stream(streamo, '.png')
123+
124+
image = pyvips.Image.new_from_file(JPEG_FILE, access='sequential')
125+
image2 = pyvips.Image.new_from_file(filename, access='sequential')
126+
127+
assert abs(image - image2).abs().max() < 10
128+
129+
@skip_if_no('jpegload')
130+
@pytest.mark.skipif(not pyvips.at_least_libvips(8, 9),
131+
reason="requires libvips >= 8.9")
132+
def test_streamu(self):
133+
streamiu = Mystreami.new(JPEG_FILE)
134+
image = pyvips.Image.new_from_stream(streamiu, '', access='sequential')
135+
136+
filename = temp_filename(self.tempdir, '.jpg')
137+
streamou = Mystreamo.new(filename)
138+
image.write_to_stream(streamou, '.png')
139+
140+
image = pyvips.Image.new_from_file(JPEG_FILE, access='sequential')
141+
image2 = pyvips.Image.new_from_file(filename, access='sequential')
142+
143+
assert abs(image - image2).abs().max() < 10
144+
145+
@skip_if_no('jpegload')
146+
@pytest.mark.skipif(not pyvips.at_least_libvips(8, 9),
147+
reason="requires libvips >= 8.9")
148+
def test_streamu_pipe(self):
149+
streamiu = Mystreami.new(JPEG_FILE, True)
150+
image = pyvips.Image.new_from_stream(streamiu, '', access='sequential')
151+
152+
filename = temp_filename(self.tempdir, '.jpg')
153+
streamou = Mystreamo.new(filename)
154+
image.write_to_stream(streamou, '.png')
155+
156+
image = pyvips.Image.new_from_file(JPEG_FILE, access='sequential')
157+
image2 = pyvips.Image.new_from_file(filename, access='sequential')
158+
159+
assert abs(image - image2).abs().max() < 10

0 commit comments

Comments
 (0)
0