"""Exercise the generated single file using an empty venv and real HTTP sockets."""
import http.client
import importlib.util
import json
import os
from pathlib import Path
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
import unittest
import venv
ROOT = Path(__file__).resolve().parents[1]
APPLICATION = '''from veela import App
import importlib.metadata
import os
import sys
app = App()
@app.route('/')
async def home(request):
return {'hello': 'world'}
@app.route('/users/{name}')
async def user(request, name):
return {'name': name, 'q': request.query()}
@app.route('/echo', methods=['POST'])
async def echo(request):
return await request.json()
@app.route('/body', methods=['POST'])
async def body(request):
return await request.body()
@app.route('/invalid')
async def invalid(request):
return 'unsupported'
@app.route('/binary')
async def binary(request):
return b'\\x00\\xff', 'application/octet-stream', 202, [('X-Test', 'yes')]
@app.route('/modules')
async def modules(request):
import uvicorn, h11, click
return {'versions': {n: importlib.metadata.version(n) for n in ('uvicorn', 'h11', 'click')},
'archive': uvicorn.__file__.split('/uvicorn/')[0],
'websocket_modules': [n for n in sys.modules if n.startswith('uvicorn.protocols.websockets')],
'compiled_backends': [n for n in ('uvloop', 'httptools') if n in sys.modules]}
@app.route('/pid')
async def pid(request):
return {'pid': os.getpid()}
if os.environ.get('VEELA_DEFAULTS'):
app.run()
else:
app.run(host='0.0.0.0', port=int(os.environ['VEELA_TEST_PORT']),
workers=int(os.environ.get('VEELA_TEST_WORKERS', '1')))
'''
class StandaloneTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.temporary = tempfile.TemporaryDirectory(prefix='veela-standalone-test-')
cls.directory = Path(cls.temporary.name)
venv.EnvBuilder(with_pip=False).create(cls.directory / 'venv')
cls.python = cls.directory / 'venv' / ('Scripts/python.exe' if __import__('os').name == 'nt' else 'bin/python')
shutil.copyfile(ROOT / 'veela.py', cls.directory / 'veela.py')
(cls.directory / 'app.py').write_text(APPLICATION)
setup, startup = APPLICATION.split("if os.environ.get('VEELA_DEFAULTS'):", 1)
guarded = setup + "if __name__ == '__main__':\n" + textwrap.indent(
"if os.environ.get('VEELA_DEFAULTS'):" + startup, ' ')
(cls.directory / 'guarded.py').write_text(guarded)
package = cls.directory / 'demo'
package.mkdir()
(package / '__init__.py').write_text('')
(package / '__main__.py').write_text(APPLICATION)
@classmethod
def tearDownClass(cls):
cls.temporary.cleanup()
def start(self, default=False, workers=1, command=('app.py',), python=None):
if default:
port = 8000
else:
with socket.socket() as s:
s.bind(('127.0.0.1', 0))
port = s.getsockname()[1]
logfile = tempfile.TemporaryFile(mode='w+')
self.addCleanup(logfile.close)
environment = {**os.environ, 'VEELA_TEST_PORT': str(port), 'VEELA_TEST_WORKERS': str(workers)}
environment.pop('PYTHONPATH', None)
environment.pop('VEELA_DEFAULTS', None)
if default:
environment['VEELA_DEFAULTS'] = '1'
process = subprocess.Popen([str(python or self.python), *command], cwd=self.directory,
env=environment, stdout=logfile, stderr=logfile,
start_new_session=os.name == 'posix')
def stop():
if process.poll() is None:
process.terminate()
try:
process.wait(timeout=10)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
self.addCleanup(stop)
deadline = time.monotonic() + 15
while time.monotonic() < deadline:
if process.poll() is not None:
logfile.seek(0)
self.fail(logfile.read())
try:
connection = http.client.HTTPConnection('127.0.0.1', port, timeout=1)
connection.request('GET', '/')
response = connection.getresponse()
self.assertEqual(response.status, 200)
self.assertEqual(json.loads(response.read()), {'hello': 'world'})
connection.close()
return process, port, logfile
except OSError:
time.sleep(0.02)
logfile.seek(0)
self.fail('Server did not start within 15 seconds:\n' + logfile.read()[-8000:])
def test_import_has_no_dependencies_and_does_not_load_or_decode_server(self):
program = '''import base64, importlib.metadata, sys, time
def forbidden_decode(*args, **kwargs):
raise AssertionError('Import decoded the server archive')
base64.b64decode = forbidden_decode
assert not list(importlib.metadata.distributions())
started = time.perf_counter()
import veela
elapsed = time.perf_counter() - started
assert not any(n in sys.modules for n in ('uvicorn', 'h11', 'click'))
assert not any('server.zip' in p for p in sys.path)
print(elapsed)
'''
result = subprocess.run([str(self.python), '-c', program], cwd=self.directory,
text=True, capture_output=True, check=True)
def test_custom_binding_json_dynamic_routes_keep_alive_and_errors(self):
_, port, _ = self.start()
connection = http.client.HTTPConnection('127.0.0.1', port, timeout=5)
self.addCleanup(connection.close)
connection.request('GET', '/users/alice?q=hello+world')
response = connection.getresponse()
self.assertEqual((response.status, json.loads(response.read())),
(200, {'name': 'alice', 'q': {'q': 'hello world'}}))
first_socket = connection.sock
for path, method, body, status, expected in [
('/echo', 'POST', json.dumps({'text': 'café', 'data': 'x' * 131072}), 200,
{'text': 'café', 'data': 'x' * 131072}),
('/echo', 'POST', '{', 400, {'error': 'Invalid JSON body'}),
('/echo', 'POST', b'\xff', 400, {'error': 'Invalid JSON body'}),
('/echo', 'POST', b'', 200, {}),
('/missing', 'GET', None, 404, {'error': 'Not Found'}),
('/invalid', 'GET', None, 500,
{'error': 'Response must be a dictionary, list, bytes, or response tuple'}),
]:
with self.subTest(path=path, body=str(body)[:20]):
connection.request(method, path, body=body, headers={'Content-Type': 'application/json'})
response = connection.getresponse()
self.assertEqual((response.status, json.loads(response.read())), (status, expected))
self.assertIs(connection.sock, first_socket)
connection.request('POST', '/body', body=iter([b'first', b'\x00second']), encode_chunked=True)
response = connection.getresponse()
self.assertEqual((response.status, response.read()), (200, b'first\x00second'))
connection.request('GET', '/binary')
response = connection.getresponse()
self.assertEqual((response.status, response.getheader('X-Test'), response.read()),
(202, 'yes', b'\x00\xff'))
connection.request('GET', '/modules')
response = connection.getresponse()
data = json.loads(response.read())
self.assertEqual(data['versions'], {'uvicorn': '0.54.0', 'h11': '0.16.0', 'click': '8.5.0'})
self.assertTrue(Path(data['archive']).is_file())
self.assertEqual(data['websocket_modules'], [])
self.assertEqual(data['compiled_backends'], [])
connection.request('GET', '/', headers={'Connection': 'Upgrade', 'Upgrade': 'websocket',
'Sec-WebSocket-Version': '13', 'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ=='})
response = connection.getresponse()
self.assertNotEqual(response.status, 101)
response.read()
def test_startup_failure_restores_imports_signals_and_temporary_files(self):
program = '''from veela import App
import pathlib, signal, sys, tempfile
tempfile.tempdir = str(pathlib.Path('scratch').resolve())
pathlib.Path(tempfile.tempdir).mkdir()
original_path = sys.path.copy()
handlers = {sig: signal.getsignal(sig) for sig in (signal.SIGINT, signal.SIGTERM)}
app = App()
for workers in (1, 2, 1, 2):
try:
app.run(port=65536, workers=workers)
except OverflowError:
pass
else:
raise AssertionError('Invalid port unexpectedly succeeded')
assert not list(pathlib.Path(tempfile.tempdir).iterdir())
assert sys.path == original_path
assert not any(n in sys.modules for n in ('uvicorn', 'h11', 'click'))
assert all(signal.getsignal(sig) == handler for sig, handler in handlers.items())
'''
(self.directory / 'startup_failure.py').write_text(program)
subprocess.run([str(self.python), 'startup_failure.py'], cwd=self.directory,
text=True, capture_output=True, check=True)
def test_invalid_worker_counts_and_unbound_apps_fail_before_server_loading(self):
program = '''from veela import App
import sys
for workers in (0, -1, 1.5, None, '2', True):
try:
App().run(workers=workers)
except ValueError:
pass
else:
raise AssertionError('Invalid worker count accepted')
try:
App().run(workers=2)
except ValueError as error:
assert 'module-level' in str(error)
else:
raise AssertionError('Unbound app accepted for multiple workers')
assert not any(n in sys.modules for n in ('uvicorn', 'h11', 'click'))
'''
subprocess.run([str(self.python), '-c', program], cwd=self.directory,
text=True, capture_output=True, check=True)
def test_default_run_binding_and_sigint_cleanup(self):
process, port, logfile = self.start(default=True)
self.shutdown(process, port, logfile, signal.SIGINT)
@unittest.skipUnless(all(importlib.util.find_spec(name) for name in ('uvloop', 'orjson')),
'Install the speedups extra to exercise native accelerators')
def test_installed_accelerators_serve_json_in_single_and_multiple_workers(self):
route = '''import asyncio
import uuid
@app.route('/accelerated', methods=['POST'])
async def accelerated(request):
return {'loop': type(asyncio.get_running_loop()).__module__,
'id': uuid.UUID('12345678-1234-5678-1234-567812345678'),
'data': await request.json()}
'''
marker = "if os.environ.get('VEELA_DEFAULTS'):"
(self.directory / 'accelerated.py').write_text(APPLICATION.replace(marker, route + marker))
for workers in (1, 2):
with self.subTest(workers=workers):
process, port, logfile = self.start(workers=workers, command=('accelerated.py',),
python=sys.executable)
connection = http.client.HTTPConnection('127.0.0.1', port, timeout=5)
self.addCleanup(connection.close)
connection.request('POST', '/accelerated', body='{"text":"café"}'.encode())
response = connection.getresponse()
self.assertEqual(response.status, 200)
self.assertEqual(json.loads(response.read()), {
'loop': 'uvloop', 'id': '12345678-1234-5678-1234-567812345678',
'data': {'text': 'café'},
})
# orjson rejects non-standard JSON accepted by the stdlib parser.
connection.request('POST', '/accelerated', body=b'{"value":NaN}')
response = connection.getresponse()
self.assertEqual((response.status, json.loads(response.read())),
(400, {'error': 'Invalid JSON body'}))
connection.close()
self.shutdown(process, port, logfile, signal.SIGTERM)
def test_multiple_workers_serve_requests_and_clean_up_on_both_signals(self):
cases = [(('app.py',), signal.SIGINT), (('-m', 'app'), signal.SIGTERM),
(('-m', 'demo'), signal.SIGINT), (('guarded.py',), signal.SIGTERM)]
for command, sig in cases:
with self.subTest(command=command, signal=sig):
process, port, logfile = self.start(workers=2, command=command)
pids = set()
deadline = time.monotonic() + 10
while len(pids) < 2 and time.monotonic() < deadline:
connection = http.client.HTTPConnection('127.0.0.1', port, timeout=5)
connection.request('GET', '/pid')
response = connection.getresponse()
self.assertEqual(response.status, 200)
pids.add(json.loads(response.read())['pid'])
connection.request('POST', '/echo', body=b'{"workers":2}')
response = connection.getresponse()
self.assertEqual((response.status, json.loads(response.read())), (200, {'workers': 2}))
connection.close()
time.sleep(0.01)
self.assertEqual(len(pids), 2)
self.assertNotIn(process.pid, pids)
self.shutdown(process, port, logfile, sig)
if os.name == 'posix':
for pid in pids:
with self.assertRaises(ProcessLookupError):
os.kill(pid, 0)
def test_sigterm_shutdown_cleans_archive(self):
process, port, logfile = self.start()
self.shutdown(process, port, logfile, signal.SIGTERM)
def shutdown(self, process, port, logfile, sig):
connection = http.client.HTTPConnection('127.0.0.1', port, timeout=5)
connection.request('GET', '/modules')
response = connection.getresponse()
archive = Path(json.loads(response.read())['archive'])
connection.close()
if sig == signal.SIGINT and os.name == 'posix':
os.killpg(process.pid, sig)
else:
process.send_signal(sig)
process.wait(timeout=10)
self.assertEqual(process.returncode, 0)
self.assertFalse(archive.parent.exists())
logfile.seek(0)
log = logfile.read()
self.assertIn('Finished server process', log)
self.assertNotIn('Waiting for application startup', log)
self.assertNotIn('Traceback', log)
self.assertNotIn('Exception in thread', log)
if __name__ == '__main__':
unittest.main()