sdk/python/tests/run_test_server.py

   1 #!/usr/bin/env python
   2
   3 import argparse
   4 import atexit
   5 import httplib2
   6 import os
   7 import pipes
   8 import random
   9 import re
  10 import shutil
  11 import signal
  12 import subprocess
  13 import string
  14 import sys
  15 import tempfile
  16 import time
  17 import unittest
  18 import yaml
  19
  20 MY_DIRNAME = os.path.dirname(os.path.realpath(__file__))
  21 if __name__ == '__main__' and os.path.exists(
  22       os.path.join(MY_DIRNAME, '..', 'arvados', '__init__.py')):
  23     # We're being launched to support another test suite.
  24     # Add the Python SDK source to the library path.
  25     sys.path.insert(1, os.path.dirname(MY_DIRNAME))
  26
  27 import arvados.api
  28 import arvados.config
  29
  30 ARVADOS_DIR = os.path.realpath(os.path.join(MY_DIRNAME, '../../..'))
  31 SERVICES_SRC_DIR = os.path.join(ARVADOS_DIR, 'services')
  32 SERVER_PID_PATH = 'tmp/pids/test-server.pid'
  33 if 'GOPATH' in os.environ:
  34     gopaths = os.environ['GOPATH'].split(':')
  35     gobins = [os.path.join(path, 'bin') for path in gopaths]
  36     os.environ['PATH'] = ':'.join(gobins) + ':' + os.environ['PATH']
  37
  38 TEST_TMPDIR = os.path.join(ARVADOS_DIR, 'tmp')
  39 if not os.path.exists(TEST_TMPDIR):
  40     os.mkdir(TEST_TMPDIR)
  41
  42 my_api_host = None
  43
  44 def find_server_pid(PID_PATH, wait=10):
  45     now = time.time()
  46     timeout = now + wait
  47     good_pid = False
  48     while (not good_pid) and (now <= timeout):
  49         time.sleep(0.2)
  50         try:
  51             with open(PID_PATH, 'r') as f:
  52                 server_pid = int(f.read())
  53             good_pid = (os.kill(server_pid, 0) is None)
  54         except IOError:
  55             good_pid = False
  56         except OSError:
  57             good_pid = False
  58         now = time.time()
  59
  60     if not good_pid:
  61         return None
  62
  63     return server_pid
  64
  65 def kill_server_pid(pidfile, wait=10, passenger_root=False):
  66     # Must re-import modules in order to work during atexit
  67     import os
  68     import signal
  69     import subprocess
  70     import time
  71     try:
  72         if passenger_root:
  73             # First try to shut down nicely
  74             restore_cwd = os.getcwd()
  75             os.chdir(passenger_root)
  76             subprocess.call([
  77                 'bundle', 'exec', 'passenger', 'stop', '--pid-file', pidfile])
  78             os.chdir(restore_cwd)
  79         now = time.time()
  80         timeout = now + wait
  81         with open(pidfile, 'r') as f:
  82             server_pid = int(f.read())
  83         while now <= timeout:
  84             if not passenger_root or timeout - now < wait / 2:
  85                 # Half timeout has elapsed. Start sending SIGTERM
  86                 os.kill(server_pid, signal.SIGTERM)
  87             # Raise OSError if process has disappeared
  88             os.getpgid(server_pid)
  89             time.sleep(0.1)
  90             now = time.time()
  91     except IOError:
  92         pass
  93     except OSError:
  94         pass
  95
  96 def find_available_port():
  97     """Return a port number that is not in use right now.
  98
  99     Some opportunity for races here, but it's better than choosing
 100     something at random and not checking at all. If all of our servers
 101     (hey Passenger) knew that listening on port 0 was a thing, the OS
 102     would take care of the races, and this wouldn't be needed at all.
 103     """
 104     port = None
 105     while port is None:
 106         port = random.randint(20000, 40000)
 107         port_hex = ':%04x ' % port
 108         try:
 109             with open('/proc/net/tcp', 'r') as f:
 110                 for line in f:
 111                     if 0 <= string.find(line, port_hex):
 112                         port = None
 113                         break
 114         except OSError:
 115             # This isn't going so well. Just use the random port.
 116             pass
 117         except IOError:
 118             pass
 119     return port
 120
 121 def run(leave_running_atexit=False):
 122     """Ensure an API server is running, and ARVADOS_API_* env vars have
 123     admin credentials for it.
 124
 125     If ARVADOS_TEST_API_HOST is set, a parent process has started a
 126     test server for us to use: we just need to reset() it using the
 127     admin token fixture.
 128
 129     If a previous call to run() started a new server process, and it
 130     is still running, we just need to reset() it to fixture state and
 131     return.
 132
 133     If neither of those options work out, we'll really start a new
 134     server.
 135     """
 136     global my_api_host
 137
 138     # Delete cached discovery document.
 139     shutil.rmtree(arvados.http_cache('discovery'))
 140
 141     pid_file = os.path.join(SERVICES_SRC_DIR, 'api', SERVER_PID_PATH)
 142     pid_file_ok = find_server_pid(pid_file, 0)
 143
 144     existing_api_host = os.environ.get('ARVADOS_TEST_API_HOST', my_api_host)
 145     if existing_api_host and pid_file_ok:
 146         if existing_api_host == my_api_host:
 147             try:
 148                 return reset()
 149             except:
 150                 # Fall through to shutdown-and-start case.
 151                 pass
 152         else:
 153             # Server was provided by parent. Can't recover if it's
 154             # unresettable.
 155             return reset()
 156
 157     # Before trying to start up our own server, call stop() to avoid
 158     # "Phusion Passenger Standalone is already running on PID 12345".
 159     # (If we've gotten this far, ARVADOS_TEST_API_HOST isn't set, so
 160     # we know the server is ours to kill.)
 161     stop(force=True)
 162
 163     restore_cwd = os.getcwd()
 164     api_src_dir = os.path.join(SERVICES_SRC_DIR, 'api')
 165     os.chdir(api_src_dir)
 166
 167     # Either we haven't started a server of our own yet, or it has
 168     # died, or we have lost our credentials, or something else is
 169     # preventing us from calling reset(). Start a new one.
 170
 171     if not os.path.exists('tmp/self-signed.pem'):
 172         # We assume here that either passenger reports its listening
 173         # address as https:/0.0.0.0:port/. If it reports "127.0.0.1"
 174         # then the certificate won't match the host and reset() will
 175         # fail certificate verification. If it reports "localhost",
 176         # clients (notably Python SDK's websocket client) might
 177         # resolve localhost as ::1 and then fail to connect.
 178         subprocess.check_call([
 179             'openssl', 'req', '-new', '-x509', '-nodes',
 180             '-out', 'tmp/self-signed.pem',
 181             '-keyout', 'tmp/self-signed.key',
 182             '-days', '3650',
 183             '-subj', '/CN=0.0.0.0'])
 184
 185     port = find_available_port()
 186     env = os.environ.copy()
 187     env['RAILS_ENV'] = 'test'
 188     env['ARVADOS_WEBSOCKETS'] = 'yes'
 189     env.pop('ARVADOS_TEST_API_HOST', None)
 190     env.pop('ARVADOS_API_HOST', None)
 191     env.pop('ARVADOS_API_HOST_INSECURE', None)
 192     env.pop('ARVADOS_API_TOKEN', None)
 193     start_msg = subprocess.check_output(
 194         ['bundle', 'exec',
 195          'passenger', 'start', '-d', '-p{}'.format(port),
 196          '--pid-file', os.path.join(os.getcwd(), pid_file),
 197          '--log-file', os.path.join(os.getcwd(), 'log/test.log'),
 198          '--ssl',
 199          '--ssl-certificate', 'tmp/self-signed.pem',
 200          '--ssl-certificate-key', 'tmp/self-signed.key'],
 201         env=env)
 202
 203     if not leave_running_atexit:
 204         atexit.register(kill_server_pid, pid_file, passenger_root=api_src_dir)
 205
 206     match = re.search(r'Accessible via: https://(.*?)/', start_msg)
 207     if not match:
 208         raise Exception(
 209             "Passenger did not report endpoint: {}".format(start_msg))
 210     my_api_host = match.group(1)
 211     os.environ['ARVADOS_API_HOST'] = my_api_host
 212
 213     # Make sure the server has written its pid file before continuing
 214     find_server_pid(pid_file)
 215
 216     reset()
 217     os.chdir(restore_cwd)
 218
 219 def reset():
 220     """Reset the test server to fixture state.
 221
 222     This resets the ARVADOS_TEST_API_HOST provided by a parent process
 223     if any, otherwise the server started by run().
 224
 225     It also resets ARVADOS_* environment vars to point to the test
 226     server with admin credentials.
 227     """
 228     existing_api_host = os.environ.get('ARVADOS_TEST_API_HOST', my_api_host)
 229     token = auth_token('admin')
 230     httpclient = httplib2.Http(ca_certs=os.path.join(
 231         SERVICES_SRC_DIR, 'api', 'tmp', 'self-signed.pem'))
 232     httpclient.request(
 233         'https://{}/database/reset'.format(existing_api_host),
 234         'POST',
 235         headers={'Authorization': 'OAuth2 {}'.format(token)})
 236     os.environ['ARVADOS_API_HOST_INSECURE'] = 'true'
 237     os.environ['ARVADOS_API_HOST'] = existing_api_host
 238     os.environ['ARVADOS_API_TOKEN'] = token
 239
 240 def stop(force=False):
 241     """Stop the API server, if one is running.
 242
 243     If force==False, kill it only if we started it ourselves. (This
 244     supports the use case where a Python test suite calls run(), but
 245     run() just uses the ARVADOS_TEST_API_HOST provided by the parent
 246     process, and the test suite cleans up after itself by calling
 247     stop(). In this case the test server provided by the parent
 248     process should be left alone.)
 249
 250     If force==True, kill it even if we didn't start it
 251     ourselves. (This supports the use case in __main__, where "run"
 252     and "stop" happen in different processes.)
 253     """
 254     global my_api_host
 255     if force or my_api_host is not None:
 256         kill_server_pid(os.path.join(SERVICES_SRC_DIR, 'api', SERVER_PID_PATH))
 257         my_api_host = None
 258
 259 def _start_keep(n, keep_args):
 260     keep0 = tempfile.mkdtemp()
 261     port = find_available_port()
 262     keep_cmd = ["keepstore",
 263                 "-volumes={}".format(keep0),
 264                 "-listen=:{}".format(port),
 265                 "-pid={}".format("{}/keep{}.pid".format(TEST_TMPDIR, n))]
 266
 267     for arg, val in keep_args.iteritems():
 268         keep_cmd.append("{}={}".format(arg, val))
 269
 270     kp0 = subprocess.Popen(keep_cmd)
 271     with open("{}/keep{}.pid".format(TEST_TMPDIR, n), 'w') as f:
 272         f.write(str(kp0.pid))
 273
 274     with open("{}/keep{}.volume".format(TEST_TMPDIR, n), 'w') as f:
 275         f.write(keep0)
 276
 277     return port
 278
 279 def run_keep(blob_signing_key=None, enforce_permissions=False):
 280     stop_keep()
 281
 282     keep_args = {}
 283     if blob_signing_key:
 284         with open(os.path.join(TEST_TMPDIR, "keep.blob_signing_key"), "w") as f:
 285             keep_args['--permission-key-file'] = f.name
 286             f.write(blob_signing_key)
 287     if enforce_permissions:
 288         keep_args['--enforce-permissions'] = 'true'
 289
 290     api = arvados.api(
 291         'v1', cache=False,
 292         host=os.environ['ARVADOS_API_HOST'],
 293         token=os.environ['ARVADOS_API_TOKEN'],
 294         insecure=True)
 295     for d in api.keep_services().list().execute()['items']:
 296         api.keep_services().delete(uuid=d['uuid']).execute()
 297     for d in api.keep_disks().list().execute()['items']:
 298         api.keep_disks().delete(uuid=d['uuid']).execute()
 299
 300     for d in range(0, 2):
 301         port = _start_keep(d, keep_args)
 302         svc = api.keep_services().create(body={'keep_service': {
 303             'uuid': 'zzzzz-bi6l4-keepdisk{:07d}'.format(d),
 304             'service_host': 'localhost',
 305             'service_port': port,
 306             'service_type': 'disk',
 307             'service_ssl_flag': False,
 308         }}).execute()
 309         api.keep_disks().create(body={
 310             'keep_disk': {'keep_service_uuid': svc['uuid'] }
 311         }).execute()
 312
 313 def _stop_keep(n):
 314     kill_server_pid("{}/keep{}.pid".format(TEST_TMPDIR, n), 0)
 315     if os.path.exists("{}/keep{}.volume".format(TEST_TMPDIR, n)):
 316         with open("{}/keep{}.volume".format(TEST_TMPDIR, n), 'r') as r:
 317             shutil.rmtree(r.read(), True)
 318         os.unlink("{}/keep{}.volume".format(TEST_TMPDIR, n))
 319     if os.path.exists(os.path.join(TEST_TMPDIR, "keep.blob_signing_key")):
 320         os.remove(os.path.join(TEST_TMPDIR, "keep.blob_signing_key"))
 321
 322 def stop_keep():
 323     _stop_keep(0)
 324     _stop_keep(1)
 325
 326 def run_keep_proxy():
 327     stop_keep_proxy()
 328
 329     admin_token = auth_token('admin')
 330     port = find_available_port()
 331     env = os.environ.copy()
 332     env['ARVADOS_API_TOKEN'] = admin_token
 333     kp = subprocess.Popen(
 334         ['keepproxy',
 335          '-pid={}/keepproxy.pid'.format(TEST_TMPDIR),
 336          '-listen=:{}'.format(port)],
 337         env=env)
 338
 339     api = arvados.api(
 340         'v1', cache=False,
 341         host=os.environ['ARVADOS_API_HOST'],
 342         token=admin_token,
 343         insecure=True)
 344     for d in api.keep_services().list(
 345             filters=[['service_type','=','proxy']]).execute()['items']:
 346         api.keep_services().delete(uuid=d['uuid']).execute()
 347     api.keep_services().create(body={'keep_service': {
 348         'service_host': 'localhost',
 349         'service_port': port,
 350         'service_type': 'proxy',
 351         'service_ssl_flag': False,
 352     }}).execute()
 353     os.environ["ARVADOS_KEEP_PROXY"] = "http://localhost:{}".format(port)
 354
 355 def stop_keep_proxy():
 356     kill_server_pid(os.path.join(TEST_TMPDIR, "keepproxy.pid"), 0)
 357
 358 def fixture(fix):
 359     '''load a fixture yaml file'''
 360     with open(os.path.join(SERVICES_SRC_DIR, 'api', "test", "fixtures",
 361                            fix + ".yml")) as f:
 362         yaml_file = f.read()
 363         try:
 364           trim_index = yaml_file.index("# Test Helper trims the rest of the file")
 365           yaml_file = yaml_file[0:trim_index]
 366         except ValueError:
 367           pass
 368         return yaml.load(yaml_file)
 369
 370 def auth_token(token_name):
 371     return fixture("api_client_authorizations")[token_name]["api_token"]
 372
 373 def authorize_with(token_name):
 374     '''token_name is the symbolic name of the token from the api_client_authorizations fixture'''
 375     arvados.config.settings()["ARVADOS_API_TOKEN"] = auth_token(token_name)
 376     arvados.config.settings()["ARVADOS_API_HOST"] = os.environ.get("ARVADOS_API_HOST")
 377     arvados.config.settings()["ARVADOS_API_HOST_INSECURE"] = "true"
 378
 379 class TestCaseWithServers(unittest.TestCase):
 380     """TestCase to start and stop supporting Arvados servers.
 381
 382     Define any of MAIN_SERVER, KEEP_SERVER, and/or KEEP_PROXY_SERVER
 383     class variables as a dictionary of keyword arguments.  If you do,
 384     setUpClass will start the corresponding servers by passing these
 385     keyword arguments to the run, run_keep, and/or run_keep_server
 386     functions, respectively.  It will also set Arvados environment
 387     variables to point to these servers appropriately.  If you don't
 388     run a Keep or Keep proxy server, setUpClass will set up a
 389     temporary directory for Keep local storage, and set it as
 390     KEEP_LOCAL_STORE.
 391
 392     tearDownClass will stop any servers started, and restore the
 393     original environment.
 394     """
 395     MAIN_SERVER = None
 396     KEEP_SERVER = None
 397     KEEP_PROXY_SERVER = None
 398
 399     @staticmethod
 400     def _restore_dict(src, dest):
 401         for key in dest.keys():
 402             if key not in src:
 403                 del dest[key]
 404         dest.update(src)
 405
 406     @classmethod
 407     def setUpClass(cls):
 408         cls._orig_environ = os.environ.copy()
 409         cls._orig_config = arvados.config.settings().copy()
 410         cls._cleanup_funcs = []
 411         os.environ.pop('ARVADOS_KEEP_PROXY', None)
 412         os.environ.pop('ARVADOS_EXTERNAL_CLIENT', None)
 413         for server_kwargs, start_func, stop_func in (
 414                 (cls.MAIN_SERVER, run, reset),
 415                 (cls.KEEP_SERVER, run_keep, stop_keep),
 416                 (cls.KEEP_PROXY_SERVER, run_keep_proxy, stop_keep_proxy)):
 417             if server_kwargs is not None:
 418                 start_func(**server_kwargs)
 419                 cls._cleanup_funcs.append(stop_func)
 420         if (cls.KEEP_SERVER is None) and (cls.KEEP_PROXY_SERVER is None):
 421             cls.local_store = tempfile.mkdtemp()
 422             os.environ['KEEP_LOCAL_STORE'] = cls.local_store
 423             cls._cleanup_funcs.append(
 424                 lambda: shutil.rmtree(cls.local_store, ignore_errors=True))
 425         else:
 426             os.environ.pop('KEEP_LOCAL_STORE', None)
 427         arvados.config.initialize()
 428
 429     @classmethod
 430     def tearDownClass(cls):
 431         for clean_func in cls._cleanup_funcs:
 432             clean_func()
 433         cls._restore_dict(cls._orig_environ, os.environ)
 434         cls._restore_dict(cls._orig_config, arvados.config.settings())
 435
 436
 437 if __name__ == "__main__":
 438     actions = ['start', 'stop',
 439                'start_keep', 'stop_keep',
 440                'start_keep_proxy', 'stop_keep_proxy']
 441     parser = argparse.ArgumentParser()
 442     parser.add_argument('action', type=str, help="one of {}".format(actions))
 443     parser.add_argument('--auth', type=str, metavar='FIXTURE_NAME', help='Print authorization info for given api_client_authorizations fixture')
 444     args = parser.parse_args()
 445
 446     if args.action == 'start':
 447         stop(force=('ARVADOS_TEST_API_HOST' not in os.environ))
 448         run(leave_running_atexit=True)
 449         host = os.environ['ARVADOS_API_HOST']
 450         if args.auth is not None:
 451             token = auth_token(args.auth)
 452             print("export ARVADOS_API_TOKEN={}".format(pipes.quote(token)))
 453             print("export ARVADOS_API_HOST={}".format(pipes.quote(host)))
 454             print("export ARVADOS_API_HOST_INSECURE=true")
 455         else:
 456             print(host)
 457     elif args.action == 'stop':
 458         stop(force=('ARVADOS_TEST_API_HOST' not in os.environ))
 459     elif args.action == 'start_keep':
 460         run_keep()
 461     elif args.action == 'stop_keep':
 462         stop_keep()
 463     elif args.action == 'start_keep_proxy':
 464         run_keep_proxy()
 465     elif args.action == 'stop_keep_proxy':
 466         stop_keep_proxy()
 467     else:
 468         print("Unrecognized action '{}'. Actions are: {}.".format(args.action, actions))