#!/usr/bin/env python import argparse import atexit import httplib2 import os import pipes import random import re import shutil import signal import socket import subprocess import string import sys import tempfile import time import unittest import yaml MY_DIRNAME = os.path.dirname(os.path.realpath(__file__)) if __name__ == '__main__' and os.path.exists( os.path.join(MY_DIRNAME, '..', 'arvados', '__init__.py')): # We're being launched to support another test suite. # Add the Python SDK source to the library path. sys.path.insert(1, os.path.dirname(MY_DIRNAME)) import arvados import arvados.config ARVADOS_DIR = os.path.realpath(os.path.join(MY_DIRNAME, '../../..')) SERVICES_SRC_DIR = os.path.join(ARVADOS_DIR, 'services') SERVER_PID_PATH = 'tmp/pids/test-server.pid' if 'GOPATH' in os.environ: gopaths = os.environ['GOPATH'].split(':') gobins = [os.path.join(path, 'bin') for path in gopaths] os.environ['PATH'] = ':'.join(gobins) + ':' + os.environ['PATH'] TEST_TMPDIR = os.path.join(ARVADOS_DIR, 'tmp') if not os.path.exists(TEST_TMPDIR): os.mkdir(TEST_TMPDIR) my_api_host = None def find_server_pid(PID_PATH, wait=10): now = time.time() timeout = now + wait good_pid = False while (not good_pid) and (now <= timeout): time.sleep(0.2) try: with open(PID_PATH, 'r') as f: server_pid = int(f.read()) good_pid = (os.kill(server_pid, 0) is None) except IOError: good_pid = False except OSError: good_pid = False now = time.time() if not good_pid: return None return server_pid def kill_server_pid(pidfile, wait=10, passenger_root=False): # Must re-import modules in order to work during atexit import os import signal import subprocess import time try: if passenger_root: # First try to shut down nicely restore_cwd = os.getcwd() os.chdir(passenger_root) subprocess.call([ 'bundle', 'exec', 'passenger', 'stop', '--pid-file', pidfile]) os.chdir(restore_cwd) now = time.time() timeout = now + wait with open(pidfile, 'r') as f: server_pid = int(f.read()) while now <= timeout: if not passenger_root or timeout - now < wait / 2: # Half timeout has elapsed. Start sending SIGTERM os.kill(server_pid, signal.SIGTERM) # Raise OSError if process has disappeared os.getpgid(server_pid) time.sleep(0.1) now = time.time() except IOError: pass except OSError: pass def find_available_port(): """Return an IPv4 port number that is not in use right now. We assume whoever needs to use the returned port is able to reuse a recently used port without waiting for TIME_WAIT (see SO_REUSEADDR / SO_REUSEPORT). Some opportunity for races here, but it's better than choosing something at random and not checking at all. If all of our servers (hey Passenger) knew that listening on port 0 was a thing, the OS would take care of the races, and this wouldn't be needed at all. """ sock = socket.socket() sock.bind(('0.0.0.0', 0)) port = sock.getsockname()[1] sock.close() return port def run(leave_running_atexit=False): """Ensure an API server is running, and ARVADOS_API_* env vars have admin credentials for it. If ARVADOS_TEST_API_HOST is set, a parent process has started a test server for us to use: we just need to reset() it using the admin token fixture. If a previous call to run() started a new server process, and it is still running, we just need to reset() it to fixture state and return. If neither of those options work out, we'll really start a new server. """ global my_api_host # Delete cached discovery document. shutil.rmtree(arvados.http_cache('discovery')) pid_file = os.path.join(SERVICES_SRC_DIR, 'api', SERVER_PID_PATH) pid_file_ok = find_server_pid(pid_file, 0) existing_api_host = os.environ.get('ARVADOS_TEST_API_HOST', my_api_host) if existing_api_host and pid_file_ok: if existing_api_host == my_api_host: try: return reset() except: # Fall through to shutdown-and-start case. pass else: # Server was provided by parent. Can't recover if it's # unresettable. return reset() # Before trying to start up our own server, call stop() to avoid # "Phusion Passenger Standalone is already running on PID 12345". # (If we've gotten this far, ARVADOS_TEST_API_HOST isn't set, so # we know the server is ours to kill.) stop(force=True) restore_cwd = os.getcwd() api_src_dir = os.path.join(SERVICES_SRC_DIR, 'api') os.chdir(api_src_dir) # Either we haven't started a server of our own yet, or it has # died, or we have lost our credentials, or something else is # preventing us from calling reset(). Start a new one. if not os.path.exists('tmp/self-signed.pem'): # We assume here that either passenger reports its listening # address as https:/0.0.0.0:port/. If it reports "127.0.0.1" # then the certificate won't match the host and reset() will # fail certificate verification. If it reports "localhost", # clients (notably Python SDK's websocket client) might # resolve localhost as ::1 and then fail to connect. subprocess.check_call([ 'openssl', 'req', '-new', '-x509', '-nodes', '-out', 'tmp/self-signed.pem', '-keyout', 'tmp/self-signed.key', '-days', '3650', '-subj', '/CN=0.0.0.0'], stdout=sys.stderr) port = find_available_port() env = os.environ.copy() env['RAILS_ENV'] = 'test' env['ARVADOS_WEBSOCKETS'] = 'yes' env.pop('ARVADOS_TEST_API_HOST', None) env.pop('ARVADOS_API_HOST', None) env.pop('ARVADOS_API_HOST_INSECURE', None) env.pop('ARVADOS_API_TOKEN', None) start_msg = subprocess.check_output( ['bundle', 'exec', 'passenger', 'start', '-d', '-p{}'.format(port), '--pid-file', os.path.join(os.getcwd(), pid_file), '--log-file', os.path.join(os.getcwd(), 'log/test.log'), '--ssl', '--ssl-certificate', 'tmp/self-signed.pem', '--ssl-certificate-key', 'tmp/self-signed.key'], env=env) if not leave_running_atexit: atexit.register(kill_server_pid, pid_file, passenger_root=api_src_dir) match = re.search(r'Accessible via: https://(.*?)/', start_msg) if not match: raise Exception( "Passenger did not report endpoint: {}".format(start_msg)) my_api_host = match.group(1) os.environ['ARVADOS_API_HOST'] = my_api_host # Make sure the server has written its pid file before continuing find_server_pid(pid_file) reset() os.chdir(restore_cwd) def reset(): """Reset the test server to fixture state. This resets the ARVADOS_TEST_API_HOST provided by a parent process if any, otherwise the server started by run(). It also resets ARVADOS_* environment vars to point to the test server with admin credentials. """ existing_api_host = os.environ.get('ARVADOS_TEST_API_HOST', my_api_host) token = auth_token('admin') httpclient = httplib2.Http(ca_certs=os.path.join( SERVICES_SRC_DIR, 'api', 'tmp', 'self-signed.pem')) httpclient.request( 'https://{}/database/reset'.format(existing_api_host), 'POST', headers={'Authorization': 'OAuth2 {}'.format(token)}) os.environ['ARVADOS_API_HOST_INSECURE'] = 'true' os.environ['ARVADOS_API_HOST'] = existing_api_host os.environ['ARVADOS_API_TOKEN'] = token def stop(force=False): """Stop the API server, if one is running. If force==False, kill it only if we started it ourselves. (This supports the use case where a Python test suite calls run(), but run() just uses the ARVADOS_TEST_API_HOST provided by the parent process, and the test suite cleans up after itself by calling stop(). In this case the test server provided by the parent process should be left alone.) If force==True, kill it even if we didn't start it ourselves. (This supports the use case in __main__, where "run" and "stop" happen in different processes.) """ global my_api_host if force or my_api_host is not None: kill_server_pid(os.path.join(SERVICES_SRC_DIR, 'api', SERVER_PID_PATH)) my_api_host = None def _start_keep(n, keep_args): keep0 = tempfile.mkdtemp() port = find_available_port() keep_cmd = ["keepstore", "-volumes={}".format(keep0), "-listen=:{}".format(port), "-pid={}".format("{}/keep{}.pid".format(TEST_TMPDIR, n))] for arg, val in keep_args.iteritems(): keep_cmd.append("{}={}".format(arg, val)) kp0 = subprocess.Popen(keep_cmd) with open("{}/keep{}.pid".format(TEST_TMPDIR, n), 'w') as f: f.write(str(kp0.pid)) with open("{}/keep{}.volume".format(TEST_TMPDIR, n), 'w') as f: f.write(keep0) return port def run_keep(blob_signing_key=None, enforce_permissions=False): stop_keep() keep_args = {} if blob_signing_key: with open(os.path.join(TEST_TMPDIR, "keep.blob_signing_key"), "w") as f: keep_args['--permission-key-file'] = f.name f.write(blob_signing_key) if enforce_permissions: keep_args['--enforce-permissions'] = 'true' api = arvados.api( version='v1', host=os.environ['ARVADOS_API_HOST'], token=os.environ['ARVADOS_API_TOKEN'], insecure=True) for d in api.keep_services().list().execute()['items']: api.keep_services().delete(uuid=d['uuid']).execute() for d in api.keep_disks().list().execute()['items']: api.keep_disks().delete(uuid=d['uuid']).execute() for d in range(0, 2): port = _start_keep(d, keep_args) svc = api.keep_services().create(body={'keep_service': { 'uuid': 'zzzzz-bi6l4-keepdisk{:07d}'.format(d), 'service_host': 'localhost', 'service_port': port, 'service_type': 'disk', 'service_ssl_flag': False, }}).execute() api.keep_disks().create(body={ 'keep_disk': {'keep_service_uuid': svc['uuid'] } }).execute() def _stop_keep(n): kill_server_pid("{}/keep{}.pid".format(TEST_TMPDIR, n), 0) if os.path.exists("{}/keep{}.volume".format(TEST_TMPDIR, n)): with open("{}/keep{}.volume".format(TEST_TMPDIR, n), 'r') as r: shutil.rmtree(r.read(), True) os.unlink("{}/keep{}.volume".format(TEST_TMPDIR, n)) if os.path.exists(os.path.join(TEST_TMPDIR, "keep.blob_signing_key")): os.remove(os.path.join(TEST_TMPDIR, "keep.blob_signing_key")) def stop_keep(): _stop_keep(0) _stop_keep(1) def run_keep_proxy(): stop_keep_proxy() admin_token = auth_token('admin') port = find_available_port() env = os.environ.copy() env['ARVADOS_API_TOKEN'] = admin_token kp = subprocess.Popen( ['keepproxy', '-pid={}/keepproxy.pid'.format(TEST_TMPDIR), '-listen=:{}'.format(port)], env=env) api = arvados.api( version='v1', host=os.environ['ARVADOS_API_HOST'], token=admin_token, insecure=True) for d in api.keep_services().list( filters=[['service_type','=','proxy']]).execute()['items']: api.keep_services().delete(uuid=d['uuid']).execute() api.keep_services().create(body={'keep_service': { 'service_host': 'localhost', 'service_port': port, 'service_type': 'proxy', 'service_ssl_flag': False, }}).execute() os.environ["ARVADOS_KEEP_PROXY"] = "http://localhost:{}".format(port) def stop_keep_proxy(): kill_server_pid(os.path.join(TEST_TMPDIR, "keepproxy.pid"), 0) def fixture(fix): '''load a fixture yaml file''' with open(os.path.join(SERVICES_SRC_DIR, 'api', "test", "fixtures", fix + ".yml")) as f: yaml_file = f.read() try: trim_index = yaml_file.index("# Test Helper trims the rest of the file") yaml_file = yaml_file[0:trim_index] except ValueError: pass return yaml.load(yaml_file) def auth_token(token_name): return fixture("api_client_authorizations")[token_name]["api_token"] def authorize_with(token_name): '''token_name is the symbolic name of the token from the api_client_authorizations fixture''' arvados.config.settings()["ARVADOS_API_TOKEN"] = auth_token(token_name) arvados.config.settings()["ARVADOS_API_HOST"] = os.environ.get("ARVADOS_API_HOST") arvados.config.settings()["ARVADOS_API_HOST_INSECURE"] = "true" class TestCaseWithServers(unittest.TestCase): """TestCase to start and stop supporting Arvados servers. Define any of MAIN_SERVER, KEEP_SERVER, and/or KEEP_PROXY_SERVER class variables as a dictionary of keyword arguments. If you do, setUpClass will start the corresponding servers by passing these keyword arguments to the run, run_keep, and/or run_keep_server functions, respectively. It will also set Arvados environment variables to point to these servers appropriately. If you don't run a Keep or Keep proxy server, setUpClass will set up a temporary directory for Keep local storage, and set it as KEEP_LOCAL_STORE. tearDownClass will stop any servers started, and restore the original environment. """ MAIN_SERVER = None KEEP_SERVER = None KEEP_PROXY_SERVER = None @staticmethod def _restore_dict(src, dest): for key in dest.keys(): if key not in src: del dest[key] dest.update(src) @classmethod def setUpClass(cls): cls._orig_environ = os.environ.copy() cls._orig_config = arvados.config.settings().copy() cls._cleanup_funcs = [] os.environ.pop('ARVADOS_KEEP_PROXY', None) os.environ.pop('ARVADOS_EXTERNAL_CLIENT', None) for server_kwargs, start_func, stop_func in ( (cls.MAIN_SERVER, run, reset), (cls.KEEP_SERVER, run_keep, stop_keep), (cls.KEEP_PROXY_SERVER, run_keep_proxy, stop_keep_proxy)): if server_kwargs is not None: start_func(**server_kwargs) cls._cleanup_funcs.append(stop_func) if (cls.KEEP_SERVER is None) and (cls.KEEP_PROXY_SERVER is None): cls.local_store = tempfile.mkdtemp() os.environ['KEEP_LOCAL_STORE'] = cls.local_store cls._cleanup_funcs.append( lambda: shutil.rmtree(cls.local_store, ignore_errors=True)) else: os.environ.pop('KEEP_LOCAL_STORE', None) arvados.config.initialize() @classmethod def tearDownClass(cls): for clean_func in cls._cleanup_funcs: clean_func() cls._restore_dict(cls._orig_environ, os.environ) cls._restore_dict(cls._orig_config, arvados.config.settings()) if __name__ == "__main__": actions = ['start', 'stop', 'start_keep', 'stop_keep', 'start_keep_proxy', 'stop_keep_proxy'] parser = argparse.ArgumentParser() parser.add_argument('action', type=str, help="one of {}".format(actions)) parser.add_argument('--auth', type=str, metavar='FIXTURE_NAME', help='Print authorization info for given api_client_authorizations fixture') args = parser.parse_args() if args.action == 'start': stop(force=('ARVADOS_TEST_API_HOST' not in os.environ)) run(leave_running_atexit=True) host = os.environ['ARVADOS_API_HOST'] if args.auth is not None: token = auth_token(args.auth) print("export ARVADOS_API_TOKEN={}".format(pipes.quote(token))) print("export ARVADOS_API_HOST={}".format(pipes.quote(host))) print("export ARVADOS_API_HOST_INSECURE=true") else: print(host) elif args.action == 'stop': stop(force=('ARVADOS_TEST_API_HOST' not in os.environ)) elif args.action == 'start_keep': run_keep() elif args.action == 'stop_keep': stop_keep() elif args.action == 'start_keep_proxy': run_keep_proxy() elif args.action == 'stop_keep_proxy': stop_keep_proxy() else: print("Unrecognized action '{}'. Actions are: {}.".format(args.action, actions))