glass beadian
aa5e262f3a
Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""Tests for the Windows worker port (win/). Runs on Linux; no GPU, no Windows host.
|
|
|
|
Covers the three pieces the wave doc says must hold before first hardware contact:
|
|
1. ffmpeg-qsv.py argument rewriting matches the validated Linux adapter's
|
|
contract and fails closed on unreviewed commands (exit 78).
|
|
2. the capability-query passthrough list matches what the shipped runner
|
|
actually generates (verified in the 0.2.0 bundle).
|
|
3. pt-worker-win.py Meter/Admission semantics are byte-identical to the
|
|
Linux controller's for every decision input (states, hysteresis, floors).
|
|
"""
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
WIN = Path(__file__).resolve().parents[2] / 'win'
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def load(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
qsv = load('qsv_win', WIN / 'ffmpeg-qsv.py')
|
|
win = load('worker_win', WIN / 'pt-worker-win.py')
|
|
linux = load('worker_linux', ROOT / 'pt-worker.py')
|
|
|
|
PROBE = {'streams': [{'codec_type': 'video', 'codec_name': 'h264', 'pix_fmt': 'yuv420p',
|
|
'width': 1920, 'height': 1080}],
|
|
'format': {'duration': '10.0'}}
|
|
|
|
|
|
class RewriteTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.old_state, self.old_probe = qsv.STATE, qsv.probe
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
qsv.STATE = Path(self.tmp.name)
|
|
def stub_probe(path): # no real ffprobe.exe on this Linux host
|
|
name = str(path).lower()
|
|
if name.endswith(('.png', '.mjpeg', '.webp')):
|
|
return {'streams': [{**PROBE['streams'][0], 'codec_name': 'png',
|
|
'pix_fmt': 'rgba'}], 'format': {'duration': '0'}}
|
|
return PROBE
|
|
qsv.probe = stub_probe
|
|
|
|
def tearDown(self):
|
|
qsv.STATE, qsv.probe = self.old_state, self.old_probe
|
|
self.tmp.cleanup()
|
|
|
|
def rewritten(self, args, data=PROBE):
|
|
out, _ = qsv.rewrite(args, input_probe=data)
|
|
return out
|
|
|
|
def test_h264_web_video_becomes_qsv_pipeline(self):
|
|
out = self.rewritten(['-i', 'in.mp4', '-vf', 'scale=w=-2:h=720', '-c:v', 'libx264',
|
|
'-preset', 'veryfast', '-crf', '18', '-maxrate', '1500K',
|
|
'-bufsize', '3000K', '-g', '96', '-x264-params', 'nofast-pskip=1',
|
|
'-movflags', '+faststart', 'out.mp4'])
|
|
text = ' '.join(out)
|
|
self.assertIn('-hwaccel qsv', text)
|
|
self.assertIn('-hwaccel_output_format qsv', text)
|
|
self.assertIn('-c:v h264_qsv', text)
|
|
self.assertIn('scale_qsv=w=1280:h=720', text)
|
|
self.assertIn('-preset medium', text)
|
|
self.assertIn('-b:v 1500K', text) # runner's rate cap kept, CRF dropped
|
|
self.assertIn('-g 96', text) # runner's GOP kept
|
|
self.assertNotIn('libx264', text)
|
|
self.assertNotIn('-crf', text)
|
|
|
|
def test_hevc_source_uses_hevc_qsv_decode(self):
|
|
data = {**PROBE, 'streams': [{**PROBE['streams'][0], 'codec_name': 'hevc', 'pix_fmt': 'yuv420p10le'}]}
|
|
out = self.rewritten(['-i', 'in.mkv', '-c:v', 'libx264', 'out.mp4'], data)
|
|
self.assertIn('-c:v hevc_qsv', ' '.join(out))
|
|
|
|
def test_no_vf_gets_default_scale_qsv(self):
|
|
out = self.rewritten(['-i', 'in.mp4', '-c:v', 'libx264', 'out.mp4'])
|
|
self.assertIn('scale_qsv=w=1920:h=1080', ' '.join(out))
|
|
|
|
def test_audio_only_passthrough(self):
|
|
out, info = qsv.rewrite(['-i', 'in.mp4', '-vn', '-c:a', 'aac', 'out.m4a'])
|
|
self.assertEqual(info, None)
|
|
|
|
def test_capability_queries_passthrough(self):
|
|
for args in (['-version'], ['-encoders'], ['-hwaccels']):
|
|
out, info = qsv.rewrite(args)
|
|
self.assertEqual((out, info), (args, None))
|
|
|
|
def test_fail_closed_cpu_or_unknown(self):
|
|
# Note: -vf contents are validated after the (stubbed) probe, matching
|
|
# the Linux adapter, which needs the probe to resolve -2 dimensions.
|
|
for args in (['-i', 'in.mp4', '-c:v', 'mpeg4', 'out.mp4'],
|
|
['-i', 'in.mp4', '-c:v', 'libx264', '-filter_complex', 'x', 'out.mp4'],
|
|
['-i', 'in.mp4', '-c:v', 'libx264', '-vf', 'scale=w=-2:h=720,hue=s=0', 'out.mp4'],
|
|
['-i', 'in.png', '-c:v', 'libx264', '-vf', 'scale=w=-2:h=720', 'out.mp4'],
|
|
['-i', 'in.mp4', '-c:v', 'libx264', '-vf', 'scale=w=0:h=720', 'out.mp4']):
|
|
with self.assertRaises(ValueError):
|
|
qsv.rewrite(args)
|
|
|
|
def test_refusal_process_exit_78(self):
|
|
import os
|
|
env = {**os.environ, 'PT_STATE_DIR': str(self.tmp.name), 'PT_FFPROBE_EXE': 'ffprobe.exe'}
|
|
p = subprocess.run([sys.executable, str(WIN / 'ffmpeg-qsv.py'),
|
|
'-i', 'x.mp4', '-c:v', 'libx264', 'o.mp4'],
|
|
capture_output=True, text=True, env=env)
|
|
self.assertEqual(p.returncode, 78)
|
|
|
|
def test_events_have_fixed_reasons_only(self):
|
|
import os
|
|
env = {**os.environ, 'PT_STATE_DIR': str(self.tmp.name), 'PT_FFPROBE_EXE': 'ffprobe.exe'}
|
|
p = subprocess.run([sys.executable, str(WIN / 'ffmpeg-qsv.py'),
|
|
'-i', 'x.mp4', '-c:v', 'libx264', 'o.mp4'],
|
|
capture_output=True, text=True, env=env)
|
|
events = [json.loads(line) for line in
|
|
(Path(self.tmp.name) / 'events.jsonl').read_text().splitlines()]
|
|
self.assertEqual(events[0]['kind'], 'gpu_failure')
|
|
self.assertEqual(events[0]['reason'], 'ValueError')
|
|
self.assertNotIn('x.mp4', p.stderr)
|
|
|
|
|
|
class RunnerContractTests(unittest.TestCase):
|
|
"""Verified directly in the shipped runner 0.2.0 bundle this session."""
|
|
|
|
def test_queries_actually_issued_by_runner(self):
|
|
bundle = Path('/tmp/peertube-worker-inspect/runtime/node_modules/@peertube/peertube-runner'
|
|
'/dist/peertube-runner.js')
|
|
if not bundle.exists():
|
|
self.skipTest('pinned runner bundle not present')
|
|
text = bundle.read_text(errors='replace')
|
|
# Exactly the queries fluent-ffmpeg's capability probe issues in 0.2.0
|
|
# (-hwaccels is NOT issued; the shim passes it through anyway, harmlessly).
|
|
for query in ('-version', '-formats', '-encoders', '-filters', '-codecs'):
|
|
self.assertIn(query, text)
|
|
self.assertNotIn('-hwaccels', text)
|
|
self.assertIn('encodersToTry = this.availableEncoders.encodersToTry', text)
|
|
self.assertIn('"libx264"', text)
|
|
|
|
def test_spawn_without_shell_and_ffprobe_sibling(self):
|
|
bundle = Path('/tmp/peertube-worker-inspect/runtime/node_modules/@peertube/peertube-runner'
|
|
'/dist/peertube-runner.js')
|
|
if not bundle.exists():
|
|
self.skipTest('pinned runner bundle not present')
|
|
text = bundle.read_text(errors='replace')
|
|
self.assertNotIn('shell: true', text) # EINVAL on .cmd/.bat shims
|
|
self.assertIn('utils.isWindows ? "ffprobe.exe" : "ffprobe"', text)
|
|
self.assertIn('!utils.isWindows', text) # `nice` skipped on win32
|
|
|
|
|
|
class AdmissionParityTests(unittest.TestCase):
|
|
"""The Windows controller must decide identically to the Linux controller."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.old_win, self.old_lin = win.STATE, linux.STATE
|
|
win.STATE = linux.STATE = Path(self.tmp.name)
|
|
self.now = time.time()
|
|
|
|
def tearDown(self):
|
|
win.STATE, linux.STATE = self.old_win, self.old_lin
|
|
self.tmp.cleanup()
|
|
|
|
def meter_with_transfers(self, module, mbps):
|
|
meter = module.Meter()
|
|
for direction in ('download', 'upload'):
|
|
meter.consume({'kind': 'transfer', 'time': self.now, 'ok': True, 'direction': direction,
|
|
'bytes': int(mbps * 1e6 / 8), 'seconds': 1})
|
|
return meter
|
|
|
|
def sample(self, **overrides):
|
|
base = {'tailscale_path': 'direct', 'latency_ms': 5, 'network_ok': True,
|
|
'registered': True, 'gpu_ready': True, 'gpu_tested_at': self.now,
|
|
'gpu_accessible': True, 'on_ac': True, 'disk_free_bytes': 100 * 1024 ** 3,
|
|
'temperature_c': 50, 'throttle_counts': {}, 'active_jobs': 0}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
def parity(self, overrides, control=None, decisions=1):
|
|
control = control or {'paused': False}
|
|
outputs = []
|
|
for module in (linux, win):
|
|
policy = module.Admission({})
|
|
meter = self.meter_with_transfers(module, 50)
|
|
for _ in range(decisions):
|
|
outputs.append(policy.decide(self.sample(**overrides), meter, control, self.now))
|
|
self.assertEqual(outputs[0], outputs[1],
|
|
f'Linux and Windows controllers diverge on {overrides}: {outputs}')
|
|
return outputs[0]
|
|
|
|
def test_happy_path(self):
|
|
self.parity({}, decisions=4)
|
|
|
|
def test_power_loss_drains_identically(self):
|
|
self.assertEqual(self.parity({'on_ac': False}), 'DRAINING')
|
|
|
|
def test_unknown_power_drains_identically(self):
|
|
self.assertEqual(self.parity({'on_ac': None}), 'DRAINING')
|
|
|
|
def test_missing_gpu_proof_blocks_identically(self):
|
|
self.assertEqual(self.parity({'gpu_ready': False, 'gpu_accessible': False}), 'GPU_UNAVAILABLE')
|
|
|
|
def test_gpu_failure_event_after_proof_blocks_identically(self):
|
|
outputs = []
|
|
for module in (linux, win):
|
|
policy = module.Admission({})
|
|
meter = self.meter_with_transfers(module, 50)
|
|
meter.consume({'kind': 'gpu_failure', 'time': self.now + 1})
|
|
outputs.append(policy.decide(self.sample(gpu_tested_at=self.now), meter,
|
|
{'paused': False}, self.now + 2))
|
|
self.assertEqual(outputs[0], 'GPU_UNAVAILABLE')
|
|
self.assertEqual(outputs[0], outputs[1])
|
|
|
|
def test_low_disk_and_thermal(self):
|
|
self.assertEqual(self.parity({'disk_free_bytes': 10 * 1024 ** 3}), 'LOW_DISK')
|
|
self.assertEqual(self.parity({'temperature_c': 90}), 'THERMAL_THROTTLED')
|
|
|
|
def test_derp_disallowed_and_latency_ceiling(self):
|
|
self.assertEqual(self.parity({'tailscale_path': 'DERP'}), 'NETWORK_DEGRADED')
|
|
self.assertEqual(self.parity({'latency_ms': 500}), 'NETWORK_DEGRADED')
|
|
|
|
def test_bandwidth_floor_hysteresis(self):
|
|
# 5 Mbps < 10 Mbps floor: every sample is bad, both controllers degrade.
|
|
for module in (linux, win):
|
|
policy = module.Admission({})
|
|
meter = self.meter_with_transfers(module, 5)
|
|
outputs = [policy.decide(self.sample(), meter, {'paused': False}, self.now)
|
|
for _ in range(3)]
|
|
self.assertEqual(outputs, ['NETWORK_DEGRADED'] * 3, module.__name__)
|
|
|
|
def test_recovery_factor(self):
|
|
outputs = []
|
|
for module in (linux, win):
|
|
policy = module.Admission({})
|
|
meter = self.meter_with_transfers(module, 11)
|
|
outputs.append(policy.decide(self.sample(), meter, {'paused': False}, self.now))
|
|
outputs.append(policy.decide(self.sample(), meter, {'paused': False}, self.now))
|
|
self.assertEqual(outputs[0], 'NETWORK_DEGRADED')
|
|
self.assertEqual(outputs[0], outputs[1])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=1)
|