#!/usr/bin/python3 """Integration test with published runner against a loopback API; no production calls.""" import argparse import http.server import json import os import shutil import subprocess import tempfile import threading import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] UUID = '22222222-2222-4222-8222-222222222222' def main(): parser = argparse.ArgumentParser() parser.add_argument('--node', required=True) parser.add_argument('--runtime', required=True, help='directory containing installed runtime node_modules') args = parser.parse_args() with tempfile.TemporaryDirectory(prefix='pt-runner-smoke-') as directory: tmp = Path(directory) home, state = tmp / 'home', tmp / 'state' home.mkdir(); state.mkdir() (tmp / 'runtime').symlink_to(Path(args.runtime).resolve(), target_is_directory=True) shutil.copy(ROOT / 'ipc.cjs', tmp / 'ipc.cjs') fixture = tmp / 'source.mp4' subprocess.run(['/usr/bin/ffmpeg', '-v', 'error', '-f', 'lavfi', '-i', 'testsrc2=size=64x64:rate=2', '-t', '1', '-c:v', 'libx264', '-threads', '1', str(fixture)], check=True) requests, accepted, failures = [], threading.Event(), threading.Event() registered_count = 0 offered = False class API(http.server.BaseHTTPRequestHandler): def log_message(self, *_): pass def do_POST(self): nonlocal registered_count, offered length = int(self.headers.get('Content-Length', 0)) self.rfile.read(length) # discarded; never log credentials requests.append(self.path) if self.path == '/api/v1/runners/register': registered_count += 1 return self.reply({'runnerToken': 'ptrt-local-test-only'}) if self.path == '/api/v1/runners/jobs/request': jobs = [] if offered or not registered_count else [{'uuid': UUID, 'type': 'vod-web-video-transcoding', 'payload': {'input': {'videoFileUrl': f'https://canonical.invalid/api/v1/runners/jobs/{UUID}/files/videos/source/max-quality', 'separatedAudioFileUrl': []}, 'output': {'resolution': 64, 'fps': 2}}}] return self.reply({'availableJobs': jobs}) if self.path.endswith('/accept'): offered = True accepted.set() return self.reply({'job': {'uuid': UUID, 'type': 'vod-web-video-transcoding', 'jobToken': 'local-job-secret', 'payload': {'input': {'videoFileUrl': f'https://canonical.invalid/api/v1/runners/jobs/{UUID}/files/videos/source/max-quality', 'separatedAudioFileUrl': []}, 'output': {'resolution': 64, 'fps': 2}}}}) if '/files/' in self.path: blob = fixture.read_bytes() self.send_response(200); self.send_header('Content-Length', str(len(blob))); self.end_headers() self.wfile.write(blob) return if self.path.endswith('/error'): failures.set() self.send_response(204); self.end_headers() def reply(self, data): blob = json.dumps(data).encode() self.send_response(200); self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(blob))); self.end_headers(); self.wfile.write(blob) server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), API) threading.Thread(target=server.serve_forever, daemon=True).start() url = f'http://127.0.0.1:{server.server_port}' cfg = tmp / 'worker.json' cfg.write_text(json.dumps({'name': 'smoke-worker', 'peertube_url': url, 'canonical_origin': 'https://canonical.invalid'})) env = {**os.environ, 'HOME': str(home), 'XDG_CONFIG_HOME': str(home / '.config'), 'XDG_CACHE_HOME': str(home / '.cache'), 'XDG_DATA_HOME': str(home / '.local/share'), 'PT_WORKER_CONFIG': str(cfg), 'PT_STATE_DIR': str(state), 'PT_RENDER_DEVICE': str(tmp / 'missing-render'), 'FFMPEG_PATH': str(ROOT / 'ffmpeg-qsv.py'), 'FFPROBE_PATH': '/usr/bin/ffprobe', 'NODE_OPTIONS': '--require=' + str(ROOT / 'observe.cjs')} runner = Path(args.runtime) / 'node_modules/@peertube/peertube-runner/dist/peertube-runner.js' runner_config = home / '.config/peertube-runner-nodejs/default/config.toml' runner_config.parent.mkdir(parents=True) runner_config.write_text('[jobs]\nconcurrency=0\n[ffmpeg]\nthreads=2\nnice=20\n') socket = home / '.local/share/peertube-runner-nodejs/default/peertube-runner.sock' children = [] def start(): child = subprocess.Popen([args.node, str(runner), 'server', '--enable-job', 'vod-web-video-transcoding'], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) children.append(child) for _ in range(100): if socket.exists(): p = subprocess.run([args.node, str(tmp / 'ipc.cjs'), 'list-jobs'], env=env, capture_output=True, timeout=3) if p.returncode == 0: return child if child.poll() is not None: raise AssertionError('runner exited') time.sleep(.1) raise AssertionError('IPC socket unavailable') def call(command, data=None): return subprocess.run([args.node, str(tmp / 'ipc.cjs'), command], input=data, env=env, capture_output=True, text=True, timeout=20, check=True) try: child = start() call('register', 'ptrrt-local-test-only') assert registered_count == 1 assert json.loads(call('list-jobs').stdout)['active_jobs'] == 0 call('graceful-shutdown'); child.wait(timeout=10) assert not accepted.is_set(), 'enrollment accepted a job at concurrency=0' runner_config.write_text(runner_config.read_text().replace('concurrency = 0', 'concurrency = 1')) socket.unlink(missing_ok=True) child = start() assert accepted.wait(15), 'job not accepted' assert failures.wait(15), 'missing QSV did not report job failure' call('graceful-shutdown'); child.wait(timeout=10) assert registered_count == 1, 'restart duplicated registration' transcoding = home / '.cache/peertube-runner-nodejs/default/transcoding' assert not list(transcoding.iterdir()), 'temporary files not cleaned' events = (state / 'events.jsonl').read_text() assert 'gpu_failure' in events assert 'local-job-secret' not in events and 'ptrt-local-test-only' not in events assert not any(p.endswith('/success') for p in requests), 'failed QSV uploaded output' assert any('/files/' in p for p in requests), 'canonical source was not routed to loopback' print('PASS: official runner 0.2.0 registration via secret stdin, concurrency=0 enrollment, restart without re-registration, source routing, QSV failure reporting, graceful shutdown, tmp cleanup') finally: for child in children: if child.poll() is None: child.kill(); child.wait() server.shutdown(); server.server_close() if __name__ == '__main__': main()