blob: 4255d21b40713b2540755e068a2723c679be4ac1 (
plain)
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
68
69
70
71
72
73
74
75
76
|
#!/usr/bin/python3
# Copyright 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import logging.handlers
import common
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import constants
from autotest_lib.client.cros import xmlrpc_server
from autotest_lib.client.cros.networking import apmanager_proxy
class ApmanagerXmlRpcDelegate(xmlrpc_server.XmlRpcDelegate):
"""Exposes methods called remotely during APManager autotests.
All instance methods of this object without a preceding '_' are exposed via
an XMLRPC server. This is not a stateless handler object, which means that
if you store state inside the delegate, that state will remain around for
future calls.
"""
def __init__(self):
self._apmanager_proxy = apmanager_proxy.ApmanagerProxy()
def __enter__(self):
super(ApmanagerXmlRpcDelegate, self).__enter__()
def __exit__(self, exception, value, traceback):
super(ApmanagerXmlRpcDelegate, self).__exit__(exception, value, traceback)
@xmlrpc_server.dbus_safe(None)
def start_service(self, config_params):
"""Create/start an AP service.
@param config_params dictionary of configuration parameters.
@return string object path for the AP service.
"""
return self._apmanager_proxy.start_service(config_params)
def terminate_service(self, service):
"""Remove/terminate an AP service.
@param service string object path of the AP service.
"""
self._apmanager_proxy.terminate_service(service)
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
handler = logging.handlers.SysLogHandler(address='/dev/log')
formatter = logging.Formatter(
'apmanager_xmlrpc_server: [%(levelname)s] %(message)s')
handler.setFormatter(formatter)
logging.getLogger().addHandler(handler)
logging.debug('apmanager_xmlrpc_server main...')
server = xmlrpc_server.XmlRpcServer('localhost',
constants.APMANAGER_XMLRPC_SERVER_PORT)
if server is None:
raise error.TestFail('Failed to setup xmlrpc server for apmanager')
else:
logging.debug('Server setup')
server.register_delegate(ApmanagerXmlRpcDelegate())
server.run()
|
