xref: /unit/test/unit/control.py (revision 2616:ab2896c980ab)
1import json
2
3from unit.http import HTTP1
4from unit.option import option
5
6
7def args_handler(conf_func):
8    def args_wrapper(self, *args):
9        argcount = conf_func.__code__.co_argcount
10        url_default = '/config'
11        conf = None
12
13        if argcount == 2:
14            url = args[0] if len(args) == 1 else url_default
15
16        elif argcount == 3:
17            conf = args[0]
18
19            if isinstance(conf, (dict, list)):
20                conf = json.dumps(conf)
21
22            url = args[1] if len(args) == 2 else url_default
23
24        url = url if url.startswith('/') else f'{url_default}/{url}'
25        arguments = (self, url) if conf is None else (self, conf, url)
26
27        return json.loads(conf_func(*arguments))
28
29    return args_wrapper
30
31
32class Control(HTTP1):
33    @args_handler
34    def conf(self, conf, url):
35        return self.put(**self._get_args(url, conf))['body']
36
37    @args_handler
38    def conf_get(self, url):
39        return self.get(**self._get_args(url))['body']
40
41    @args_handler
42    def conf_delete(self, url):
43        return self.delete(**self._get_args(url))['body']
44
45    @args_handler
46    def conf_post(self, conf, url):
47        return self.post(**self._get_args(url, conf))['body']
48
49    def _get_args(self, url, conf=None):
50        args = {
51            'url': url,
52            'sock_type': 'unix',
53            'addr': f'{option.temp_dir}/control.unit.sock',
54        }
55
56        if conf is not None:
57            args['body'] = conf
58
59        return args
60