93 lines
5.0 KiB
Python
93 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""POSIX integration test for the actual npm launcher, including teardown."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
origin = 'http://127.0.0.1:3002'
|
|
|
|
def stop(proc):
|
|
if proc.poll() is None:
|
|
os.killpg(proc.pid, signal.SIGINT)
|
|
try:
|
|
proc.wait(timeout=12)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
proc.wait()
|
|
raise AssertionError('Launcher did not stop on Ctrl+C')
|
|
|
|
with tempfile.TemporaryDirectory(prefix='webka-launcher-') as temporary:
|
|
data = Path(temporary, 'data')
|
|
env = dict(os.environ, WEBKA_DATA=str(data), WEBKA_MDNS='0')
|
|
for key in ['WEBKA_ORIGIN', 'WEBKA_ALLOWED_ORIGINS']:
|
|
env.pop(key, None)
|
|
for iteration in range(2):
|
|
logfile = Path(temporary, f'launch-{iteration}.log')
|
|
with logfile.open('w') as output:
|
|
proc = subprocess.Popen(['npm', 'run', 'dev:all', '--', '--host', '0.0.0.0', '--port', '3002'], cwd=root, env=env, stdout=output, stderr=subprocess.STDOUT, start_new_session=True)
|
|
try:
|
|
for _ in range(1200):
|
|
if proc.poll() is not None:
|
|
raise AssertionError(logfile.read_text())
|
|
if 'Webka is ready!' in logfile.read_text():
|
|
break
|
|
time.sleep(0.5)
|
|
else:
|
|
raise AssertionError('Startup timed out: ' + logfile.read_text()[-2000:])
|
|
if iteration == 1:
|
|
assert 'Frontend is up to date.' in logfile.read_text()
|
|
assert 'Installing frontend dependencies' not in logfile.read_text()
|
|
for route in ['/login', '/host', '/watch', '/settings/accounts', '/settings/general']:
|
|
with urllib.request.urlopen(origin + route, timeout=3) as response:
|
|
assert response.status == 200
|
|
with urllib.request.urlopen(origin + '/api/bootstrap') as response:
|
|
assert json.load(response)['needs_setup'] is (iteration == 0)
|
|
body = json.dumps({'username': 'owner', 'password': 'launch-test-password-123', 'setup_token': (data / 'setup-token').read_text()}).encode()
|
|
if iteration == 0:
|
|
req = urllib.request.Request(origin + '/api/bootstrap', data=body, headers={'Origin': origin, 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(req) as response:
|
|
assert response.status == 201
|
|
req = urllib.request.Request(origin + '/api/login', data=body, headers={'Origin': origin, 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(req) as response:
|
|
cookie = response.headers['Set-Cookie'].split(';')[0]
|
|
req = urllib.request.Request(origin + '/api/status', headers={'Cookie': cookie})
|
|
with urllib.request.urlopen(req) as response:
|
|
status = json.load(response)
|
|
assert ':3002/watch' in status['local_url']
|
|
lan_origin = status['local_url'].removesuffix('/watch')
|
|
req = urllib.request.Request(origin + '/api/login', data=body, headers={'Origin': lan_origin, 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(req) as response:
|
|
assert response.status == 200
|
|
req = urllib.request.Request(origin + '/api/logout', data=b'{}', headers={'Origin': 'http://evil.example', 'Content-Type': 'application/json', 'Cookie': cookie})
|
|
try:
|
|
urllib.request.urlopen(req)
|
|
raise AssertionError('Untrusted origin accepted')
|
|
except urllib.error.HTTPError as error:
|
|
assert error.code == 403
|
|
finally:
|
|
stop(proc)
|
|
# Assert that the actual server, not just npm, released the listener.
|
|
with socket.socket() as server:
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server.bind(('0.0.0.0', 3002))
|
|
print(f'PASS: launch {iteration + 1}, routes, local/LAN auth, QR port, rejected foreign origin, Ctrl+C cleanup.', flush=True)
|
|
with socket.socket() as occupied:
|
|
occupied.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
occupied.bind(('127.0.0.1', 3002))
|
|
occupied.listen()
|
|
proc = subprocess.Popen(['npm', 'run', 'dev:all', '--', '--port', '3002'], cwd=root, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True)
|
|
try:
|
|
stdout, stderr = proc.communicate(timeout=20)
|
|
assert proc.returncode != 0 and 'already in use' in stderr, stdout + stderr
|
|
finally:
|
|
stop(proc)
|
|
print('PASS: occupied port fails before building or starting a server.', flush=True)
|