xref: /unit/test/test_asgi_targets.py (revision 1873:d0cc4fd78d43)
1from distutils.version import LooseVersion
2
3import pytest
4
5from unit.applications.lang.python import TestApplicationPython
6from unit.option import option
7
8
9class TestASGITargets(TestApplicationPython):
10    prerequisites = {
11        'modules': {'python': lambda v: LooseVersion(v) >= LooseVersion('3.5')}
12    }
13    load_module = 'asgi'
14
15    @pytest.fixture(autouse=True)
16    def setup_method_fixture(self):
17        assert 'success' in self.conf(
18            {
19                "listeners": {"*:7080": {"pass": "routes"}},
20                "routes": [
21                    {
22                        "match": {"uri": "/1"},
23                        "action": {"pass": "applications/targets/1"},
24                    },
25                    {
26                        "match": {"uri": "/2"},
27                        "action": {"pass": "applications/targets/2"},
28                    },
29                ],
30                "applications": {
31                    "targets": {
32                        "type": "python",
33                        "processes": {"spare": 0},
34                        "working_directory": option.test_dir
35                        + "/python/targets/",
36                        "path": option.test_dir + '/python/targets/',
37                        "protocol": "asgi",
38                        "targets": {
39                            "1": {
40                                "module": "asgi",
41                                "callable": "application_200",
42                            },
43                            "2": {
44                                "module": "asgi",
45                                "callable": "application_201",
46                            },
47                        },
48                    }
49                },
50            }
51        )
52
53    def conf_targets(self, targets):
54        assert 'success' in self.conf(targets, 'applications/targets/targets')
55
56    def test_asgi_targets(self):
57        assert self.get(url='/1')['status'] == 200
58        assert self.get(url='/2')['status'] == 201
59
60    def test_asgi_targets_legacy(self):
61        self.conf_targets(
62            {
63                "1": {"module": "asgi", "callable": "legacy_application_200"},
64                "2": {"module": "asgi", "callable": "legacy_application_201"},
65            }
66        )
67
68        assert self.get(url='/1')['status'] == 200
69        assert self.get(url='/2')['status'] == 201
70
71    def test_asgi_targets_mix(self):
72        self.conf_targets(
73            {
74                "1": {"module": "asgi", "callable": "application_200"},
75                "2": {"module": "asgi", "callable": "legacy_application_201"},
76            }
77        )
78
79        assert self.get(url='/1')['status'] == 200
80        assert self.get(url='/2')['status'] == 201
81
82    def test_asgi_targets_broken(self, skip_alert):
83        skip_alert(r'Python failed to get "blah" from module')
84
85        self.conf_targets(
86            {
87                "1": {"module": "asgi", "callable": "application_200"},
88                "2": {"module": "asgi", "callable": "blah"},
89            }
90        )
91
92        assert self.get(url='/1')['status'] != 200
93