xref: /unit/test/test_asgi_lifespan.py (revision 1771:94cf6c5fafbd)
1import os
2from distutils.version import LooseVersion
3
4import pytest
5from conftest import unit_stop
6from unit.applications.lang.python import TestApplicationPython
7from unit.option import option
8
9
10class TestASGILifespan(TestApplicationPython):
11    prerequisites = {
12        'modules': {'python': lambda v: LooseVersion(v) >= LooseVersion('3.5')}
13    }
14    load_module = 'asgi'
15
16    def test_asgi_lifespan(self):
17        self.load('lifespan/empty')
18
19        startup_path = option.test_dir + '/python/lifespan/empty/startup'
20        shutdown_path = option.test_dir + '/python/lifespan/empty/shutdown'
21        version_path = option.test_dir + '/python/lifespan/empty/version'
22
23        os.chmod(option.test_dir + '/python/lifespan/empty', 0o777)
24
25        open(startup_path, 'a').close()
26        os.chmod(startup_path, 0o777)
27
28        open(shutdown_path, 'a').close()
29        os.chmod(shutdown_path, 0o777)
30
31        open(version_path, 'a').close()
32        os.chmod(version_path, 0o777)
33
34        assert self.get()['status'] == 204
35
36        unit_stop()
37
38        is_startup = os.path.isfile(startup_path)
39        is_shutdown = os.path.isfile(shutdown_path)
40
41        if is_startup:
42            os.remove(startup_path)
43
44        if is_shutdown:
45            os.remove(shutdown_path)
46
47        with open(version_path, 'r') as f:
48            version = f.read()
49
50        os.remove(version_path)
51
52        assert not is_startup, 'startup'
53        assert not is_shutdown, 'shutdown'
54        assert version == '3.0 2.0', 'version'
55
56    def test_asgi_lifespan_failed(self):
57        self.load('lifespan/failed')
58
59        assert self.get()['status'] == 503
60
61        assert (
62            self.wait_for_record(r'\[error\].*Application startup failed')
63            is not None
64        ), 'error message'
65        assert self.wait_for_record(r'Exception blah') is not None, 'exception'
66
67    def test_asgi_lifespan_error(self):
68        self.load('lifespan/error')
69
70        self.get()
71
72        assert self.wait_for_record(r'Exception blah') is not None, 'exception'
73
74    def test_asgi_lifespan_error_auto(self):
75        self.load('lifespan/error_auto')
76
77        self.get()
78
79        assert self.wait_for_record(r'AssertionError') is not None, 'assertion'
80