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