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