Tests: added ASGI Lifespan.
This commit is contained in:
parent
9d8d2c1758
commit
152ad526f4
5 changed files with 114 additions and 0 deletions
27
test/python/lifespan/empty/asgi.py
Normal file
27
test/python/lifespan/empty/asgi.py
Normal file
|
@ -0,0 +1,27 @@
|
|||
import os
|
||||
|
||||
|
||||
async def application(scope, receive, send):
|
||||
if scope['type'] == 'lifespan':
|
||||
with open('version', 'w+') as f:
|
||||
f.write(
|
||||
scope['asgi']['version'] + ' ' + scope['asgi']['spec_version']
|
||||
)
|
||||
while True:
|
||||
message = await receive()
|
||||
if message['type'] == 'lifespan.startup':
|
||||
os.remove('startup')
|
||||
await send({'type': 'lifespan.startup.complete'})
|
||||
elif message['type'] == 'lifespan.shutdown':
|
||||
os.remove('shutdown')
|
||||
await send({'type': 'lifespan.shutdown.complete'})
|
||||
return
|
||||
|
||||
if scope['type'] == 'http':
|
||||
await send(
|
||||
{
|
||||
'type': 'http.response.start',
|
||||
'status': 204,
|
||||
'headers': [(b'content-length', b'0'),],
|
||||
}
|
||||
)
|
3
test/python/lifespan/error/asgi.py
Normal file
3
test/python/lifespan/error/asgi.py
Normal file
|
@ -0,0 +1,3 @@
|
|||
async def application(scope, receive, send):
|
||||
if scope['type'] != 'http':
|
||||
raise Exception('Exception blah')
|
2
test/python/lifespan/error_auto/asgi.py
Normal file
2
test/python/lifespan/error_auto/asgi.py
Normal file
|
@ -0,0 +1,2 @@
|
|||
async def application(scope, receive, send):
|
||||
assert scope['type'] == 'http'
|
11
test/python/lifespan/failed/asgi.py
Normal file
11
test/python/lifespan/failed/asgi.py
Normal file
|
@ -0,0 +1,11 @@
|
|||
async def application(scope, receive, send):
|
||||
if scope['type'] == 'lifespan':
|
||||
while True:
|
||||
message = await receive()
|
||||
if message['type'] == 'lifespan.startup':
|
||||
await send({"type": "lifespan.startup.failed"})
|
||||
raise Exception('Exception blah')
|
||||
|
||||
elif message['type'] == 'lifespan.shutdown':
|
||||
await send({'type': 'lifespan.shutdown.complete'})
|
||||
return
|
71
test/test_asgi_lifespan.py
Normal file
71
test/test_asgi_lifespan.py
Normal file
|
@ -0,0 +1,71 @@
|
|||
import os
|
||||
import pytest
|
||||
from distutils.version import LooseVersion
|
||||
|
||||
from unit.applications.lang.python import TestApplicationPython
|
||||
from conftest import option
|
||||
|
||||
|
||||
class TestASGILifespan(TestApplicationPython):
|
||||
prerequisites = {
|
||||
'modules': {'python': lambda v: LooseVersion(v) >= LooseVersion('3.5')}
|
||||
}
|
||||
load_module = 'asgi'
|
||||
|
||||
def test_asgi_lifespan(self):
|
||||
self.load('lifespan/empty')
|
||||
|
||||
startup_path = option.test_dir + '/python/lifespan/empty/startup'
|
||||
shutdown_path = option.test_dir + '/python/lifespan/empty/shutdown'
|
||||
version_path = option.test_dir + '/python/lifespan/empty/version'
|
||||
|
||||
open(startup_path, 'a').close()
|
||||
open(shutdown_path, 'a').close()
|
||||
open(version_path, 'a').close()
|
||||
|
||||
assert self.get()['status'] == 204
|
||||
|
||||
self.stop()
|
||||
|
||||
is_startup = os.path.isfile(startup_path)
|
||||
is_shutdown = os.path.isfile(shutdown_path)
|
||||
|
||||
if is_startup:
|
||||
os.remove(startup_path)
|
||||
|
||||
if is_shutdown:
|
||||
os.remove(shutdown_path)
|
||||
|
||||
with open(version_path, 'r') as f:
|
||||
version = f.read()
|
||||
|
||||
os.remove(version_path)
|
||||
|
||||
assert not is_startup, 'startup'
|
||||
assert not is_shutdown, 'shutdown'
|
||||
assert version == '3.0 2.0', 'version'
|
||||
|
||||
def test_asgi_lifespan_failed(self):
|
||||
self.load('lifespan/failed')
|
||||
|
||||
assert self.get()['status'] == 503
|
||||
|
||||
assert (
|
||||
self.wait_for_record(r'\[error\].*Application startup failed')
|
||||
is not None
|
||||
), 'error message'
|
||||
assert self.wait_for_record(r'Exception blah') is not None, 'exception'
|
||||
|
||||
def test_asgi_lifespan_error(self):
|
||||
self.load('lifespan/error')
|
||||
|
||||
self.get()
|
||||
|
||||
assert self.wait_for_record(r'Exception blah') is not None, 'exception'
|
||||
|
||||
def test_asgi_lifespan_error_auto(self):
|
||||
self.load('lifespan/error_auto')
|
||||
|
||||
self.get()
|
||||
|
||||
assert self.wait_for_record(r'AssertionError') is not None, 'assertion'
|
Loading…
Reference in a new issue