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