import pytest from packaging import version from unit.applications.lang.python import TestApplicationPython from unit.option import option class TestASGITargets(TestApplicationPython): prerequisites = { 'modules': { 'python': lambda v: version.parse(v) >= version.parse('3.5') } } load_module = 'asgi' @pytest.fixture(autouse=True) def setup_method_fixture(self): path = f'{option.test_dir}/python/targets/' 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": { "targets": { "type": self.get_application_type(), "processes": {"spare": 0}, "working_directory": path, "path": path, "protocol": "asgi", "targets": { "1": { "module": "asgi", "callable": "application_200", }, "2": { "module": "asgi", "callable": "application_201", }, }, } }, } ) def conf_targets(self, targets): assert 'success' in self.conf(targets, 'applications/targets/targets') def test_asgi_targets(self): assert self.get(url='/1')['status'] == 200 assert self.get(url='/2')['status'] == 201 def test_asgi_targets_legacy(self): self.conf_targets( { "1": {"module": "asgi", "callable": "legacy_application_200"}, "2": {"module": "asgi", "callable": "legacy_application_201"}, } ) assert self.get(url='/1')['status'] == 200 assert self.get(url='/2')['status'] == 201 def test_asgi_targets_mix(self): self.conf_targets( { "1": {"module": "asgi", "callable": "application_200"}, "2": {"module": "asgi", "callable": "legacy_application_201"}, } ) assert self.get(url='/1')['status'] == 200 assert self.get(url='/2')['status'] == 201 def test_asgi_targets_broken(self, skip_alert): skip_alert(r'Python failed to get "blah" from module') self.conf_targets( { "1": {"module": "asgi", "callable": "application_200"}, "2": {"module": "asgi", "callable": "blah"}, } ) assert self.get(url='/1')['status'] != 200 def test_asgi_targets_prefix(self): self.conf_targets( { "1": { "module": "asgi", "callable": "application_prefix", "prefix": "/1/", }, "2": { "module": "asgi", "callable": "application_prefix", "prefix": "/api", }, } ) self.conf( [ { "match": {"uri": "/1*"}, "action": {"pass": "applications/targets/1"}, }, { "match": {"uri": "*"}, "action": {"pass": "applications/targets/2"}, }, ], "routes", ) def check_prefix(url, prefix): resp = self.get(url=url) assert resp['status'] == 200 assert resp['headers']['prefix'] == prefix check_prefix('/1', '/1') check_prefix('/11', 'NULL') check_prefix('/1/', '/1') check_prefix('/', 'NULL') check_prefix('/ap', 'NULL') check_prefix('/api', '/api') check_prefix('/api/', '/api') check_prefix('/api/test/', '/api') check_prefix('/apis', 'NULL') check_prefix('/apis/', 'NULL')