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