from __future__ import print_function import os import subprocess import sys import time from contextlib import contextmanager if False: from typing import (Any, Iterator, Optional) # Verify the Zulip venv is available. from tools.lib import sanity_check sanity_check.check_venv(__file__) import django import requests TOOLS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if TOOLS_DIR not in sys.path: sys.path.insert(0, os.path.dirname(TOOLS_DIR)) from zerver.lib.test_fixtures import is_template_database_current def set_up_django(external_host): # type: (str) -> None os.environ['EXTERNAL_HOST'] = external_host os.environ["TORNADO_SERVER"] = "http://127.0.0.1:9983" os.environ['DJANGO_SETTINGS_MODULE'] = 'zproject.test_settings' django.setup() os.environ['PYTHONUNBUFFERED'] = 'y' def assert_server_running(server, log_file): # type: (subprocess.Popen, Optional[str]) -> None """Get the exit code of the server, or None if it is still running.""" if server.poll() is not None: message = 'Server died unexpectedly!' if log_file: message += '\nSee %s\n' % (log_file,) raise RuntimeError(message) def server_is_up(server, log_file): # type: (subprocess.Popen, Optional[str]) -> bool assert_server_running(server, log_file) try: # We could get a 501 error if the reverse proxy is up but the Django app isn't. return requests.get('http://127.0.0.1:9981/accounts/home').status_code == 200 except Exception: return False @contextmanager def test_server_running(force=False, external_host='testserver', log_file=None, dots=False, use_db=True): # type: (bool, str, str, bool, bool) -> Iterator[None] log = sys.stdout if log_file: if os.path.exists(log_file) and os.path.getsize(log_file) < 100000: log = open(log_file, 'a') log.write('\n\n') else: log = open(log_file, 'w') set_up_django(external_host) if use_db: generate_fixtures_command = ['tools/setup/generate-fixtures'] if not is_template_database_current(): generate_fixtures_command.append('--force') subprocess.check_call(generate_fixtures_command) # Run this not through the shell, so that we have the actual PID. run_dev_server_command = ['tools/run-dev.py', '--test'] if force: run_dev_server_command.append('--force') server = subprocess.Popen(run_dev_server_command, stdout=log, stderr=log) try: # Wait for the server to start up. sys.stdout.write('Waiting for test server') while not server_is_up(server, log_file): if dots: sys.stdout.write('.') sys.stdout.flush() time.sleep(0.1) sys.stdout.write('\n') # DO OUR ACTUAL TESTING HERE!!! yield finally: assert_server_running(server, log_file) server.terminate() if __name__ == '__main__': # The code below is for testing this module works with test_server_running(): print('\n\n SERVER IS UP!\n\n')