google/appengine/tools/devappserver2/channel_test.py

#!/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.
#
"""Tests for google.appengine.tools.devappserver2.channel."""




import StringIO
import unittest
import urllib

from google.appengine.api.channel import channel_service_stub
from google.appengine.tools.devappserver2 import channel
from google.appengine.tools.devappserver2 import wsgi_test_utils


class MockChannelServiceStub(object):
  """Mock channel service to return known messages for testing."""

  def __init__(self):
    self._messages = {}
    self._connected_channel_tokens = []
    self._disconnected_channel_client_ids = []

  def set_messages(self, messages):
    self._messages = messages

  def has_channel_messages(self, token):
    return (self._messages.has_key(token) and
            self._messages[token])

  def get_channel_messages(self, token):
    return self._messages[token]

  def pop_first_message(self, token):
    if self.has_channel_messages(token):
      return self._messages[token].pop(0)
    else:
      return None

  def clear_channel_messages(self, token):
    del self._messages[token]

  def connect_channel(self, token):
    self.validate_token_and_extract_client_id(token)
    self._connected_channel_tokens.append(token)

  def disconnect_channel(self, client_id):
    self._disconnected_channel_client_ids.append(client_id)

  def get_connected_tokens(self):
    return self._connected_channel_tokens

  def set_connected_tokens(self, tokens):
    self._connected_tokens = tokens

  def validate_token_and_extract_client_id(self, token):
    if token == 'bad':
      raise channel_service_stub.InvalidTokenError()
    if token == 'expired':
      raise channel_service_stub.TokenTimedOutError()
    return 'dummy-client-id'

  def connect_and_pop_first_message(self, token):
    self.connect_channel(token)
    return self.pop_first_message(token)


def _build_environ(path, query=None):
  environ = {
      'REQUEST_METHOD': 'GET',
      'PATH_INFO': path,
  }
  if query:
    environ['QUERY_STRING'] = urllib.urlencode(query)
  return environ


def _build_poll_environ(token):
  """Build an environ for a WSGI request that performs a channel poll.

  Args:
    token: The channel to poll.
  """
  return _build_environ('/_ah/channel/dev',
                        {'command': 'poll', 'channel': token})


class DevAppserverChannelTest(wsgi_test_utils.WSGITestCase):
  """Test the dev appserver channel module."""

  def setUp(self):
    """Set up the mocks and output objects for tests."""
    self._mock_channel_service_stub = MockChannelServiceStub()
    self._old_get_channel_stub = channel._get_channel_stub
    channel._get_channel_stub = lambda: self._mock_channel_service_stub
    self._channel_app = channel.application
    self._output = StringIO.StringIO()

  def tearDown(self):
    channel._get_channel_stub = self._old_get_channel_stub

  def test_channel_request_no_messages(self):
    """Test a channel request with no pending messages."""
    environ = _build_poll_environ('id')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('200 OK', expected_headers, '', channel.application,
                        environ)

  def test_channel_request_missing_parameters(self):
    """Test a channel request with missing query string parameters."""
    environ = _build_environ('/_ah/channel/dev', {'command': 'poll'})
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('400 Bad Request', expected_headers, '',
                        channel.application, environ)

    environ = _build_environ('/_ah/channel/dev', {'channel': 'id'})
    self.assertResponse('400 Bad Request', expected_headers, '',
                        channel.application, environ)

  def test_channel_request_bad_command(self):
    """Test a channel request with an invalid command."""
    environ = _build_environ('/_ah/channel/dev', {'command': 'bad',
                                                  'channel': 'id'})
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('400 Bad Request', expected_headers, '',
                        channel.application, environ)

  def test_channel_request_bad_token(self):
    """Test a channel request with an invalid token."""
    environ = _build_poll_environ('bad')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('401 Invalid+token.', expected_headers, '',
                        channel.application, environ)

  def test_channel_request_expired_token(self):
    """Test a channel request with an expired token."""
    environ = _build_poll_environ('expired')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('401 Token+timed+out.', expected_headers, '',
                        channel.application, environ)

  def test_channel_request_other_channel_has_messages(self):
    """Test a channel request with another channel with messages."""
    self._mock_channel_service_stub.set_messages({'a': ['hello']})
    environ = _build_poll_environ('b')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('200 OK', expected_headers, '', channel.application,
                        environ)

  def test_channel_request_with_messages(self):
    """Test a channel request with a channel that has a message."""
    self._mock_channel_service_stub.set_messages({'a': ['hello']})
    environ = _build_poll_environ('a')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '5',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, 'hello',
                        channel.application, environ)

  def test_channel_request_empty_message(self):
    """Test a channel request with a channel that has a 0-length message."""
    self._mock_channel_service_stub.set_messages({'a': ['']})
    environ = _build_poll_environ('a')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, '', channel.application,
                        environ)

  def test_channel_request_with_multiple_messages(self):
    """Test a channel request with a channel that has multiple messages."""
    self._mock_channel_service_stub.set_messages({'a': ['hello', 'goodbye']})
    environ = _build_poll_environ('a')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '5',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, 'hello',
                        channel.application, environ)
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '7',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, 'goodbye',
                        channel.application, environ)

  def test_channel_request_clears_messages(self):
    """Test that request a channel's messages clears those messages."""
    self._mock_channel_service_stub.set_messages({'a': ['hello']})
    environ = _build_poll_environ('a')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '5',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, 'hello',
                        channel.application, environ)
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('200 OK', expected_headers, '', channel.application,
                        environ)

  def test_channel_request_clears_correct_messages(self):
    """Test that request a channel's messages clears only those messages."""
    self._mock_channel_service_stub.set_messages({'a': ['hello'],
                                                  'b': ['goodbye']})
    environ = _build_poll_environ('a')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '5',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, 'hello',
                        channel.application, environ)
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('200 OK', expected_headers, '', channel.application,
                        environ)
    environ = _build_poll_environ('b')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '7',
        'Content-Type': 'application/json',
    }
    self.assertResponse('200 OK', expected_headers, 'goodbye',
                        channel.application, environ)

  def test_channel_request_connects_channel(self):
    """Ensure that a channel request causes the channel to be connected."""
    mock_stub = self._mock_channel_service_stub
    mock_stub.set_connected_tokens([])
    environ = _build_poll_environ('id')
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': '0',
    }
    self.assertResponse('200 OK', expected_headers, '', channel.application,
                        environ)
    self.assertListEqual(['id'], mock_stub.get_connected_tokens())

  def test_channel_request_jsapi(self):
    """Test that requesting the jsapi script returns expected result."""
    environ = _build_environ('/_ah/channel/jsapi')
    # Read the JS API file to test that it matches.
    js_text = open(channel._JSAPI_PATH).read()
    expected_headers = {
        'Cache-Control': 'no-cache',
        'Content-Length': str(len(js_text)),
        'Content-Type': 'text/javascript',
    }
    self.assertResponse('200 OK', expected_headers, js_text,
                        channel.application, environ)


if __name__ == '__main__':
  unittest.main()