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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
"""
The GPU re-encoding plan, and the one thing that must stay true across both
encoders.
`hwaccel.py` exists so that a low-power node can re-encode a browser-hostile
video at all — see its module docstring. What is tested here is the part that
holds whether or not the machine running the tests has a GPU: the shape of the
plan, what a demotion does, and the agreement between the codec string the node
announces and the arguments each encoder is actually given.
The end-to-end fallback — a hardware mode that fails, the stream playing anyway
— is in `test_stream_video_transcode.py`, because it needs a real ffmpeg and a
real file.
"""
import re
import sys
from pathlib import Path
import pytest
from meshbay_node import hwaccel
from meshbay_node.transport import webrtc_server
from conftest import needs_subprocess
VAAPI = next(c for c in hwaccel.CANDIDATES if c.name == "vaapi")
FAKE = hwaccel.Encoder(candidate=VAAPI, variant="standard", extra=(),
device="/dev/dri/renderD128")
@pytest.fixture(autouse=True)
def _clean():
hwaccel._reset_for_tests()
yield
hwaccel._reset_for_tests()
hwaccel.set_enabled(True)
async def test_no_working_encoder_means_the_software_path_and_nothing_else(monkeypatch):
"""
The guarantee this module owes every node that has no GPU, no driver, or a
driver that fails its test encode: one mode, the one that was always there.
A second mode would mean a second ffmpeg spawn and a second wait on a
machine that can never benefit from either.
"""
monkeypatch.setattr(hwaccel, "encoder", _fixed(None))
assert await hwaccel.modes_for("hevc") == [hwaccel.SW]
async def test_the_plan_always_ends_in_software(monkeypatch):
monkeypatch.setattr(hwaccel, "encoder", _fixed(FAKE))
assert await hwaccel.modes_for("hevc") == [hwaccel.HW, hwaccel.HWENC, hwaccel.SW]
async def test_a_demotion_is_remembered_for_that_source_codec_only(monkeypatch):
"""
iHD decodes HEVC and has no MPEG-4 Part 2 decoder at all, so "the GPU
cannot do this one" is a fact about the pair, not about the machine.
Demoting the machine would cost every HEVC film the acceleration that one
Xvid file proved was unavailable to *it*.
"""
monkeypatch.setattr(hwaccel, "encoder", _fixed(FAKE))
hwaccel.demote("mpeg4", hwaccel.HW, "no decoder")
assert await hwaccel.modes_for("mpeg4") == [hwaccel.HWENC, hwaccel.SW]
assert await hwaccel.modes_for("hevc") == [hwaccel.HW, hwaccel.HWENC, hwaccel.SW]
async def test_the_operator_switch_turns_off_the_probe_itself(monkeypatch):
"""`hardware_video_encode = false` must cost no subprocess, not merely go
unused: the probe is a test encode, and a node whose operator said no
should never spawn it."""
spawned = []
monkeypatch.setattr(hwaccel, "_test_encode",
lambda *a, **k: spawned.append(a) or None)
hwaccel.set_enabled(False)
assert await hwaccel.encoder() is None
assert await hwaccel.modes_for("hevc") == [hwaccel.SW]
assert spawned == []
def test_every_encoder_produces_what_the_node_announces():
"""
`stream_init` carries `avc1.640029` and the client puts it through
MediaSource.isTypeSupported before it trusts a byte of the stream. The
string is not a label chosen next to libx264 — it is a claim about the
bytes, and it has to stay true for whichever encoder produced them.
So the claim is read out of the streaming code and decoded (profile_idc
0x64 = 100 = High, level_idc 0x29 = 41 = 4.1) rather than restated here,
and every mode is checked against it. Changing the arguments of one encoder
without the other, or changing either without the announced string, fails.
"""
source = Path(webrtc_server.__file__).read_text(encoding="utf-8")
announced = set(re.findall(r"avc1\.([0-9a-f]{6})", source))
assert announced == {"640029"}, (
"the streaming path announces a codec string this test does not know "
f"how to check: {announced}")
(code,) = announced
profile_idc, constraints, level_idc = (int(code[i:i + 2], 16) for i in (0, 2, 4))
assert (profile_idc, constraints, level_idc) == (100, 0, 41), (
"the announced string no longer says High/4.1, so the arguments below "
"are no longer what it claims")
for mode in (hwaccel.SW, hwaccel.HW, hwaccel.HWENC):
args = hwaccel.codec_args(mode, FAKE)
assert args[args.index("-profile:v") + 1] == "high", \
f"{mode} does not encode the High profile the node announces"
assert args[args.index("-level") + 1] == "4.1", \
f"{mode} does not encode at the level the node announces"
def test_the_software_mode_never_touches_the_gpu():
"""The fallback has to be a real fallback: if `sw` carried a VA-API
argument, the mode that exists for machines where the GPU failed would
fail with it."""
for args in (hwaccel.input_args(hwaccel.SW, FAKE),
hwaccel.codec_args(hwaccel.SW, FAKE)):
joined = " ".join(args)
assert "vaapi" not in joined and "hwupload" not in joined, joined
assert hwaccel.input_args(hwaccel.HW, None) == []
assert "libx264" in hwaccel.codec_args(hwaccel.HW, None)
def _fixed(value):
async def _f():
return value
return _f
@needs_subprocess
async def test_the_probe_accepts_only_what_the_node_announces():
"""The acceptance test is neither always true nor always false.
Both halves are run with libx264, which is on any machine that can run
these tests, so what is being checked is the judgement itself rather than
anybody's GPU: the same encoder is accepted when it writes High at level
4.1 and rejected when it writes level 3.1. Without this, an encoder whose
arguments were spelled in a way ffmpeg accepts but misreads would pass the
probe and then announce a level it had not produced.
"""
def _software(label: str, extra: tuple[str, ...]) -> hwaccel.Candidate:
return hwaccel.Candidate(
name=label, encoder="libx264", platforms=(sys.platform,),
hwaccel="none", hw_filter="null", sw_filter="format=yuv420p",
variants=((label, extra),))
right = _software("as announced", ("-preset", "veryfast", "-crf", "30"))
assert await hwaccel._test_encode(right, None, right.variants[0][1]) is None
wrong = _software("a level lower",
("-preset", "veryfast", "-crf", "30", "-level", "3.1"))
why = await hwaccel._test_encode(wrong, None, wrong.variants[0][1])
assert why and "31" in why, (
f"an encoder that wrote another level must be rejected, and say so: {why!r}")
@needs_subprocess
async def test_a_missing_encoder_is_rejected_rather_than_raised():
"""ffmpeg exits non-zero for an encoder it does not have; nothing here may
turn that into an exception, because it happens on every machine that lacks
one of the three candidates — which is most of them."""
absent = hwaccel.Candidate(
name="not a thing", encoder="h264_definitely_not", platforms=(sys.platform,),
hwaccel="none", hw_filter="null", sw_filter="null", variants=(("x", ()),))
assert await hwaccel._test_encode(absent, None, ()) is not None
|