diff options
Diffstat (limited to 'test/test_asgi_lifespan.py')
| -rw-r--r-- | test/test_asgi_lifespan.py | 166 |
1 files changed, 85 insertions, 81 deletions
diff --git a/test/test_asgi_lifespan.py b/test/test_asgi_lifespan.py index 0d0cb162..499f523d 100644 --- a/test/test_asgi_lifespan.py +++ b/test/test_asgi_lifespan.py @@ -2,122 +2,126 @@ import os from conftest import unit_stop from packaging import version -from unit.applications.lang.python import TestApplicationPython +from unit.applications.lang.python import ApplicationPython from unit.option import option prerequisites = { 'modules': {'python': lambda v: version.parse(v) >= version.parse('3.5')} } +client = ApplicationPython(load_module='asgi') -class TestASGILifespan(TestApplicationPython): - load_module = 'asgi' - def setup_cookies(self, prefix): - base_dir = f'{option.test_dir}/python/lifespan/empty' +def assert_cookies(prefix): + for name in ['startup', 'shutdown']: + path = f'{option.test_dir}/python/lifespan/empty/{prefix}{name}' + exists = os.path.isfile(path) + if exists: + os.remove(path) - os.chmod(base_dir, 0o777) + assert not exists, name - for name in ['startup', 'shutdown', 'version']: - path = f'{option.test_dir}/python/lifespan/empty/{prefix}{name}' - open(path, 'a').close() - os.chmod(path, 0o777) + path = f'{option.test_dir}/python/lifespan/empty/{prefix}version' - def assert_cookies(self, prefix): - for name in ['startup', 'shutdown']: - path = f'{option.test_dir}/python/lifespan/empty/{prefix}{name}' - exists = os.path.isfile(path) - if exists: - os.remove(path) + with open(path, 'r') as f: + version = f.read() - assert not exists, name + os.remove(path) - path = f'{option.test_dir}/python/lifespan/empty/{prefix}version' + assert version == '3.0 2.0', 'version' - with open(path, 'r') as f: - version = f.read() - os.remove(path) +def setup_cookies(prefix): + base_dir = f'{option.test_dir}/python/lifespan/empty' - assert version == '3.0 2.0', 'version' + os.chmod(base_dir, 0o777) - def test_asgi_lifespan(self): - self.load('lifespan/empty') + for name in ['startup', 'shutdown', 'version']: + path = f'{option.test_dir}/python/lifespan/empty/{prefix}{name}' + open(path, 'a').close() + os.chmod(path, 0o777) - self.setup_cookies('') - assert self.get()['status'] == 204 +def test_asgi_lifespan(): + client.load('lifespan/empty') - unit_stop() + setup_cookies('') - self.assert_cookies('') + assert client.get()['status'] == 204 - def test_asgi_lifespan_targets(self): - path = f'{option.test_dir}/python/lifespan/empty' + unit_stop() - assert 'success' in self.conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [ - { - "match": {"uri": "/1"}, - "action": {"pass": "applications/targets/1"}, - }, - { - "match": {"uri": "/2"}, - "action": {"pass": "applications/targets/2"}, - }, - ], - "applications": { + assert_cookies('') + + +def test_asgi_lifespan_targets(): + path = f'{option.test_dir}/python/lifespan/empty' + + assert 'success' in client.conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [ + { + "match": {"uri": "/1"}, + "action": {"pass": "applications/targets/1"}, + }, + { + "match": {"uri": "/2"}, + "action": {"pass": "applications/targets/2"}, + }, + ], + "applications": { + "targets": { + "type": client.get_application_type(), + "processes": {"spare": 0}, + "working_directory": path, + "path": path, "targets": { - "type": self.get_application_type(), - "processes": {"spare": 0}, - "working_directory": path, - "path": path, - "targets": { - "1": {"module": "asgi", "callable": "application"}, - "2": { - "module": "asgi", - "callable": "application2", - }, + "1": {"module": "asgi", "callable": "application"}, + "2": { + "module": "asgi", + "callable": "application2", }, - } - }, - } - ) + }, + } + }, + } + ) + + setup_cookies('') + setup_cookies('app2_') + + assert client.get(url="/1")['status'] == 204 + assert client.get(url="/2")['status'] == 204 + + unit_stop() - self.setup_cookies('') - self.setup_cookies('app2_') + assert_cookies('') + assert_cookies('app2_') - assert self.get(url="/1")['status'] == 204 - assert self.get(url="/2")['status'] == 204 - unit_stop() +def test_asgi_lifespan_failed(wait_for_record): + client.load('lifespan/failed') - self.assert_cookies('') - self.assert_cookies('app2_') + assert client.get()['status'] == 503 - def test_asgi_lifespan_failed(self, wait_for_record): - self.load('lifespan/failed') + assert ( + wait_for_record(r'\[error\].*Application startup failed') is not None + ), 'error message' + assert wait_for_record(r'Exception blah') is not None, 'exception' - assert self.get()['status'] == 503 - assert ( - wait_for_record(r'\[error\].*Application startup failed') - is not None - ), 'error message' - assert wait_for_record(r'Exception blah') is not None, 'exception' +def test_asgi_lifespan_error(wait_for_record): + client.load('lifespan/error') - def test_asgi_lifespan_error(self, wait_for_record): - self.load('lifespan/error') + client.get() - self.get() + assert wait_for_record(r'Exception blah') is not None, 'exception' - assert wait_for_record(r'Exception blah') is not None, 'exception' - def test_asgi_lifespan_error_auto(self, wait_for_record): - self.load('lifespan/error_auto') +def test_asgi_lifespan_error_auto(wait_for_record): + client.load('lifespan/error_auto') - self.get() + client.get() - assert wait_for_record(r'AssertionError') is not None, 'assertion' + assert wait_for_record(r'AssertionError') is not None, 'assertion' |