xref: /unit/test/unit/applications/lang/python.py (revision 1984:06514cd08a35) 
1import os
2import shutil
3from urllib.parse import quote
4
5from unit.applications.proto import TestApplicationProto
6from unit.option import option
7
8
9class TestApplicationPython(TestApplicationProto):
10    application_type = "python"
11    load_module = "wsgi"
12
13    def load(self, script, name=None, module=None, **kwargs):
14        if name is None:
15            name = script
16
17        if module is None:
18            module = self.load_module
19
20        if script[0] == '/':
21            script_path = script
22        else:
23            script_path = option.test_dir + '/python/' + script
24
25        if kwargs.get('isolation') and kwargs['isolation'].get('rootfs'):
26            rootfs = kwargs['isolation']['rootfs']
27
28            if not os.path.exists(rootfs + '/app/python/'):
29                os.makedirs(rootfs + '/app/python/')
30
31            if not os.path.exists(rootfs + '/app/python/' + name):
32                shutil.copytree(script_path, rootfs + '/app/python/' + name)
33
34            script_path = '/app/python/' + name
35
36        app = {
37            "type": self.get_application_type(),
38            "processes": kwargs.pop('processes', {"spare": 0}),
39            "path": script_path,
40            "working_directory": script_path,
41            "module": module,
42        }
43
44        for attr in (
45            'callable',
46            'environment',
47            'home',
48            'limits',
49            'path',
50            'protocol',
51            'targets',
52            'threads',
53        ):
54            if attr in kwargs:
55                app[attr] = kwargs.pop(attr)
56
57        self._load_conf(
58            {
59                "listeners": {
60                    "*:7080": {"pass": "applications/" + quote(name, '')}
61                },
62                "applications": {name: app},
63            },
64            **kwargs
65        )
66