Source code for google.appengine.ext.testbed.apiserver_util

#!/usr/bin/env python
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Utility class for testbed only used for py_test.

This file wraps imports and attributes that testbed would need for accessing
api_server and datastore_emulator.
testbed uses cloud datastore emulator if and only if emulator_util can be
import by it.
"""

import atexit
import datetime
import functools
import os
import subprocess
import sys
import tempfile
import google

from google.appengine.tools.devappserver2 import constants


def _get_cloud_sdk_platform_dir():
  """Returns the path of google-cloud-sdk/platform directory."""
  res = os.path.dirname(os.path.realpath(__file__))




  for _ in range(5):
    res = os.path.dirname(res)
  if not os.path.exists(res):
    raise OSError('Cannot locate Google Cloud SDK. Please make sure you are '
                  'using testbed from Google Cloud SDK.')
  return res


def _get_emulator_cmd_inside_cloud_sdk():
  """Try to get the path to datastore emulator in cloud sdk.

  Returns:
    A string representing the path to the Cloud Datastore Emulator shell script.

  Raises:
    OSError: cannot find Datastore Emulator.
  """
  emulator_script = (
      'cloud_datastore_emulator.cmd' if sys.platform.startswith('win')
      else 'cloud_datastore_emulator')
  res = os.path.join(
      _get_cloud_sdk_platform_dir(), 'cloud-datastore-emulator',
      emulator_script)
  if not os.path.exists(res):
    raise OSError('Cannot find Cloud Datastore Emulator. Please make sure you '
                  'installed it with Google Cloud SDK.')
  return res


def _get_api_server_cmd():
  """Get the command to invoke api_server.

  Returns:
    A list of strings representing the command.

  Raises:
    OSError: cannot find api server.
  """











  api_server_path = os.path.join(_get_cloud_sdk_platform_dir(),
                                 'google_appengine', 'api_server.py')
  if not os.path.exists(api_server_path):
    raise OSError('Cannot find api_server.py. Please make sure you have '
                  'installed gcloud app Python Extensions.')
  return [sys.executable, api_server_path,
          '--support_datastore_emulator',
          '--datastore_emulator_cmd',
          _get_emulator_cmd_inside_cloud_sdk()]
































[docs]def get_port(line):
  """Get port number out of a line like "some message: localhost:[port]."""
  stripped = line.strip()
  separator_index = stripped.rfind(':')
  return int(stripped[separator_index+1:])


[docs]def read_lines_until(in_file, stop_string, timeout=30):
  """Read lines in an input file until stop_string is found."""
  lines = []
  t1 = datetime.datetime.now()
  while True:
    line = in_file.readline()
    if line:
      lines.append(line)
      if constants.GRPC_API_SERVER_STARTING_MSG in line:
        return lines
    else:
      t2 = datetime.datetime.now()
      if (t2 - t1).total_seconds() <= timeout:
        continue
      raise AssertionError('Did not see string "%s" in input: %s' %
                           (stop_string, ''.join(lines)))





[docs]def setup_api_server():
  """Launches api_server.

  Returns:
    Two integers, first is the port number for api_server's http endpoint,
    second is the port number of the Cloud Datastore Emulator.
  """
  api_server_output = tempfile.NamedTemporaryFile()
  api_server_proc = subprocess.Popen(
      _get_api_server_cmd(),
      stdout=api_server_output,
      stderr=subprocess.STDOUT)

  with open(api_server_output.name, 'r') as in_file:


    lines = read_lines_until(in_file, constants.GRPC_API_SERVER_STARTING_MSG)
  atexit.register(api_server_proc.terminate)

  api_server_starting_line = next(
      x for x in lines if constants.API_SERVER_STARTING_MSG in x)
  datastore_emulator_starting_line = next(
      x for x in lines if constants.DATASTORE_EMULATOR_STARTING_MSG in x)
  return (get_port(api_server_starting_line),
          get_port(datastore_emulator_starting_line))