1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
"""
Tests specifically for ensuring that we dont have daemon ffmpeg processes.
We should also run this test as a script, so we can confirm that ffmpeg
quits nicely (instead of being killed).
"""
import gc
import subprocess
import sys
from testutils import (
ensure_test_files,
get_ffmpeg_pids,
no_warnings_allowed,
test_file1,
test_file2,
)
import imageio_ffmpeg
N = 2 # number of times to perform each test
def setup_module():
ensure_test_files()
@no_warnings_allowed
def test_ffmpeg_version():
version = imageio_ffmpeg.get_ffmpeg_version()
print("ffmpeg version", version)
assert version > "3.0"
@no_warnings_allowed
def test_reader_done():
"""Test that the reader is done after reading all frames"""
for _ in range(N):
pids0 = get_ffmpeg_pids()
r = imageio_ffmpeg.read_frames(test_file1)
pids1 = get_ffmpeg_pids().difference(pids0) # generator has not started
r.__next__() # == meta
pids2 = get_ffmpeg_pids().difference(pids0) # now ffmpeg is running
for frame in r:
pass
pids3 = get_ffmpeg_pids().difference(pids0) # now its not
assert len(pids1) == 0
assert len(pids2) == 1
assert len(pids3) == 0
@no_warnings_allowed
def test_reader_close():
"""Test that the reader is done after closing it"""
for _ in range(N):
pids0 = get_ffmpeg_pids()
r = imageio_ffmpeg.read_frames(test_file1)
pids1 = get_ffmpeg_pids().difference(pids0) # generator has not started
r.__next__() # == meta
pids2 = get_ffmpeg_pids().difference(pids0) # now ffmpeg is running
r.close()
pids3 = get_ffmpeg_pids().difference(pids0) # now its not
assert len(pids1) == 0
assert len(pids2) == 1
assert len(pids3) == 0
@no_warnings_allowed
def test_reader_del():
"""Test that the reader is done after deleting it"""
for _ in range(N):
pids0 = get_ffmpeg_pids()
r = imageio_ffmpeg.read_frames(test_file1)
pids1 = get_ffmpeg_pids().difference(pids0) # generator has not started
r.__next__() # == meta
pids2 = get_ffmpeg_pids().difference(pids0) # now ffmpeg is running
del r
gc.collect()
pids3 = get_ffmpeg_pids().difference(pids0) # now its not
assert len(pids1) == 0
assert len(pids2) == 1
assert len(pids3) == 0
@no_warnings_allowed
def test_write_close():
"""Test that the writer is done after closing it"""
for _ in range(N):
pids0 = get_ffmpeg_pids()
w = imageio_ffmpeg.write_frames(test_file2, (64, 64))
pids1 = get_ffmpeg_pids().difference(pids0) # generator has not started
w.send(None)
w.send(b"x" * 64 * 64 * 3)
pids2 = get_ffmpeg_pids().difference(pids0) # now ffmpeg is running
w.close()
pids3 = get_ffmpeg_pids().difference(pids0) # now its not
assert len(pids1) == 0
assert len(pids2) == 1
assert len(pids3) == 0
@no_warnings_allowed
def test_write_del():
for _ in range(N):
pids0 = get_ffmpeg_pids()
w = imageio_ffmpeg.write_frames(test_file2, (64, 64))
pids1 = get_ffmpeg_pids().difference(pids0) # generator has not started
w.send(None)
w.send(b"x" * 64 * 64 * 3)
pids2 = get_ffmpeg_pids().difference(pids0) # now ffmpeg is running
del w
gc.collect()
pids3 = get_ffmpeg_pids().difference(pids0) # now its not
assert len(pids1) == 0
assert len(pids2) == 1
assert len(pids3) == 0
def test_partial_read():
# Case: https://github.com/imageio/imageio-ffmpeg/issues/69
template = (
"import sys; import imageio_ffmpeg; f=sys.argv[1]; print(f); "
"r=imageio_ffmpeg.read_frames(f);"
)
for i in range(4):
code = template + " r.__next__();" * i
cmd = [sys.executable, "-c", code, test_file1]
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
timeout=5,
)
print(result.stdout)
assert not result.returncode
if __name__ == "__main__":
setup_module()
test_ffmpeg_version()
test_reader_done()
test_reader_close()
test_reader_del()
test_write_close()
test_write_del()
test_partial_read()
|
