testing/legion/rpc_methods.py

   1 # Copyright 2015 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 """Defines the task RPC methods."""
   6
   7 import logging
   8 import os
   9 import sys
  10 import threading
  11
  12 #pylint: disable=relative-import
  13 import process
  14
  15
  16 class RPCMethods(object):
  17   """Class exposing RPC methods."""
  18
  19   _dotted_whitelist = ['subprocess']
  20
  21   def __init__(self, server):
  22     self._server = server
  23     self.subprocess = process.Process
  24
  25   def _dispatch(self, method, params):
  26     obj = self
  27     if '.' in method:
  28       # Allow only white listed dotted names
  29       name, method = method.split('.')
  30       assert name in self._dotted_whitelist
  31       obj = getattr(self, name)
  32     return getattr(obj, method)(*params)
  33
  34   def Echo(self, message):
  35     """Simple RPC method to print and return a message."""
  36     logging.info('Echoing %s', message)
  37     return 'echo %s' % str(message)
  38
  39   def AbsPath(self, path):
  40     """Returns the absolute path."""
  41     return os.path.abspath(path)
  42
  43   def Quit(self):
  44     """Call _server.shutdown in another thread.
  45
  46     This is needed because server.shutdown waits for the server to actually
  47     quit. However the server cannot shutdown until it completes handling this
  48     call. Calling this in the same thread results in a deadlock.
  49     """
  50     t = threading.Thread(target=self._server.shutdown)
  51     t.start()