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