Moved chromium_status to appengine/

appengine_apps/chromium_status/tests/local_gae.py

Index: appengine_apps/chromium_status/tests/local_gae.py
diff --git a/appengine_apps/chromium_status/tests/local_gae.py b/appengine_apps/chromium_status/tests/local_gae.py
deleted file mode 100644
index 02bcf85182c0657c4da2e915b531466b3777c6ea..0000000000000000000000000000000000000000
--- a/appengine_apps/chromium_status/tests/local_gae.py
+++ /dev/null
@@ -1,227 +0,0 @@
-# Copyright (c) 2012 The Chromium Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-"""Setups a local GAE instance to test against a live server for integration
-tests.
-
-It makes sure Google AppEngine SDK is found and starts the server on a free
-inbound TCP port.
-"""
-
-import cookielib
-import errno
-import logging
-import os
-import re
-import socket
-import subprocess
-import tempfile
-import time
-import sys
-import urllib
-import urllib2
-
-BASE_DIR = os.path.dirname(os.path.abspath(__file__))
-GAE_SDK = None
-
-
-def _load_modules():
-  """Loads all the necessary modules.
-
-  Update sys.path to be able to import chromium-status and GAE SDK.
-  """
-  global GAE_SDK
-  if GAE_SDK:
-    return
-  root_dir = BASE_DIR
-  # First, verify the Google AppEngine SDK is available.
-  while True:
-    if os.path.isfile(os.path.join(root_dir, 'google_appengine', 'VERSION')):
-      break
-    next_root = os.path.dirname(root_dir)
-    if next_root == root_dir:
-      raise Failure(
-          'Install google_appengine sdk in %s' % os.path.dirname(BASE_DIR))
-    root_dir = next_root
-  GAE_SDK = os.path.realpath(os.path.join(root_dir, 'google_appengine'))
-  # Need yaml later.
-  gae_sdk_lib = os.path.realpath(os.path.join(GAE_SDK, 'lib'))
-  sys.path.insert(0, os.path.realpath(os.path.join(gae_sdk_lib, 'yaml', 'lib')))
-
-
-class Failure(Exception):
-  pass
-
-
-def test_port(port):
-  s = socket.socket()
-  try:
-    return s.connect_ex(('127.0.0.1', port)) == 0
-  finally:
-    s.close()
-
-
-def find_free_port(base_port=8080):
-  """Finds an available port starting at 8080."""
-  port = base_port
-  max_val = (2<<16)
-  while test_port(port) and port < max_val:
-    port += 1
-  if port == max_val:
-    raise Failure('Having issues finding an available port')
-  return port
-
-
-class LocalGae(object):
-  """Wraps up starting a GAE local instance for integration tests."""
-
-  def __init__(self, base_dir=None):
-    """base_dir defaults to .. from the file's directory."""
-    # Paths
-    self.base_dir = base_dir
-    if not self.base_dir:
-      self.base_dir = os.path.dirname(os.path.abspath(__file__))
-      self.base_dir = os.path.realpath(os.path.join(self.base_dir, '..'))
-    self.test_server = None
-    self.port = None
-    self.admin_port = None
-    self.app_id = None
-    self.url = None
-    self.admin_url = None
-    self.tmp_db = None
-    self._xsrf_token = None
-    self._cookie_jar = cookielib.CookieJar()
-    cookie_processor = urllib2.HTTPCookieProcessor(self._cookie_jar)
-    redirect_handler = urllib2.HTTPRedirectHandler()
-    self._opener = urllib2.build_opener(redirect_handler, cookie_processor)
-
-  def install_prerequisites(self):
-    # Load GAE SDK.
-    _load_modules()
-
-    # Now safe to import GAE SDK modules.
-    # Unable to import 'yaml'
-    # pylint: disable=F0401
-
-    import yaml
-    self.app_id = yaml.load(
-        open(os.path.join(self.base_dir, 'app.yaml'), 'r'))['application']
-    logging.debug('Instance app id: %s' % self.app_id)
-    assert self.app_id
-
-  def start_server(self, verbose=False):
-    self.install_prerequisites()
-    self.port = find_free_port()
-    self.admin_port = find_free_port(base_port=self.port + 1)
-    if verbose:
-      stdout = None
-      stderr = None
-    else:
-      stdout = subprocess.PIPE
-      stderr = subprocess.PIPE
-    # Generate a friendly environment.
-    env = os.environ.copy()
-    env['LANGUAGE'] = 'en'
-    h, self.tmp_db = tempfile.mkstemp(prefix='local_gae')
-    os.close(h)
-    cmd = [
-        sys.executable,
-        os.path.join(GAE_SDK, 'dev_appserver.py'),
-        self.base_dir,
-        '--port', str(self.port),
-        '--admin_port', str(self.admin_port),
-        '--datastore_path', self.tmp_db,
-        '--datastore_consistency_policy', 'consistent',
-        '--skip_sdk_update_check',
-    ]
-    if verbose:
-      cmd.extend([
-          '--log_level', 'debug',
-      ])
-    self.test_server = subprocess.Popen(
-        cmd, stdout=stdout, stderr=stderr, env=env)
-    # Loop until port 127.0.0.1:port opens or the process dies.
-    while not test_port(self.port):
-      while not test_port(self.admin_port):
-        self.test_server.poll()
-        if self.test_server.returncode is not None:
-          raise Failure(
-              'Test GAE instance failed early on port %s' %
-              self.port)
-        time.sleep(0.001)
-    self.url = 'http://localhost:%d/' % self.port
-    self.admin_url = 'http://localhost:%d/' % self.admin_port
-
-  def stop_server(self):
-    if self.test_server:
-      try:
-        self.test_server.terminate()
-      except OSError as e:
-        if e.errno != errno.ESRCH:
-          raise
-      self.test_server = None
-      self.port = None
-      self.url = None
-      self.admin_url = None
-      if self.tmp_db:
-        try:
-          os.remove(self.tmp_db)
-        except OSError:
-          pass
-        self.tmp_db = None
-
-  def get(self, suburl, url=None):
-    if url is None:
-      url = self.url
-    logging.debug('GET: %r', url + suburl)
-    request = urllib2.Request(url + suburl)
-    f = self._opener.open(request)
-    data = f.read()
-    return data
-
-  def post(self, suburl, data, url=None):
-    if url is None:
-      url = self.url
-    logging.debug('POST(%r): %r', url + suburl, data)
-    request = urllib2.Request(url + suburl, urllib.urlencode(data))
-    f = self._opener.open(request)
-    return f.read()
-
-  def clear_cookies(self):
-    self._cookie_jar.clear()
-
-  def login(self, username, admin):
-    try:
-      self.get('_ah/login?email=%s&admin=%r&action=login&continue=/' % (
-          urllib.quote_plus(username), admin))
-    except urllib2.HTTPError:
-      # Ignore http errors as the continue url may be inaccessible.
-      pass
-
-  def query(self, cmd):
-    """Lame way to modify the db remotely on dev server.
-
-    Using remote_api inside the unit test is a bit too invasive.
-    """
-    data = {
-        'code': 'from google.appengine.ext import db\n' + cmd,
-        'module_name': 'default',
-        'xsrf_token': self.xsrf_token,
-    }
-    return self.post('console', data, url=self.admin_url)
-
-  @property
-  def xsrf_token(self):
-    if self._xsrf_token is None:
-      self.clear_cookies()
-      interactive = self.get('console', url=self.admin_url)
-      match = re.search(r"'xsrf_token': *'(.*?)'", interactive)
-      if not match:
-        logging.debug('interactive console output:\n%s', interactive)
-        raise Failure('could not find xsrf_token')
-      self._xsrf_token = match.group(1)
-      self.clear_cookies()
-    return self._xsrf_token
-
-# vim: ts=2:sw=2:tw=80:et: