diff options
Diffstat (limited to 'test/test_asgi_lifespan.py')
| -rw-r--r-- | test/test_asgi_lifespan.py | 175 |
1 files changed, 89 insertions, 86 deletions
diff --git a/test/test_asgi_lifespan.py b/test/test_asgi_lifespan.py index 84e9fea4..499f523d 100644 --- a/test/test_asgi_lifespan.py +++ b/test/test_asgi_lifespan.py @@ -2,123 +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')} +} -class TestASGILifespan(TestApplicationPython): - prerequisites = { - 'modules': { - 'python': lambda v: version.parse(v) >= version.parse('3.5') - } - } - load_module = 'asgi' +client = ApplicationPython(load_module='asgi') - def setup_cookies(self, prefix): - base_dir = f'{option.test_dir}/python/lifespan/empty' - os.chmod(base_dir, 0o777) +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) - for name in ['startup', 'shutdown', 'version']: - path = f'{option.test_dir}/python/lifespan/empty/{prefix}{name}' - open(path, 'a').close() - os.chmod(path, 0o777) + assert not exists, name - 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) + path = f'{option.test_dir}/python/lifespan/empty/{prefix}version' - assert not exists, name + with open(path, 'r') as f: + version = f.read() - path = f'{option.test_dir}/python/lifespan/empty/{prefix}version' + os.remove(path) - with open(path, 'r') as f: - version = f.read() + assert version == '3.0 2.0', 'version' - os.remove(path) - assert version == '3.0 2.0', 'version' +def setup_cookies(prefix): + base_dir = f'{option.test_dir}/python/lifespan/empty' - def test_asgi_lifespan(self): - self.load('lifespan/empty') + os.chmod(base_dir, 0o777) - self.setup_cookies('') + for name in ['startup', 'shutdown', 'version']: + path = f'{option.test_dir}/python/lifespan/empty/{prefix}{name}' + open(path, 'a').close() + os.chmod(path, 0o777) - assert self.get()['status'] == 204 - unit_stop() +def test_asgi_lifespan(): + client.load('lifespan/empty') - self.assert_cookies('') + setup_cookies('') - def test_asgi_lifespan_targets(self): - path = f'{option.test_dir}/python/lifespan/empty' + assert client.get()['status'] == 204 - 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": { + unit_stop() + + 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): - 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 ( - 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(wait_for_record): + client.load('lifespan/error') - def test_asgi_lifespan_error(self): - self.load('lifespan/error') + client.get() - self.get() + assert wait_for_record(r'Exception blah') is not None, 'exception' - assert self.wait_for_record(r'Exception blah') is not None, 'exception' - def test_asgi_lifespan_error_auto(self): - self.load('lifespan/error_auto') +def test_asgi_lifespan_error_auto(wait_for_record): + client.load('lifespan/error_auto') - self.get() + client.get() - assert self.wait_for_record(r'AssertionError') is not None, 'assertion' + assert wait_for_record(r'AssertionError') is not None, 'assertion' |