blob: d403bde9fba4cd36412c0135dd8b94dc26a86939 [file] [log] [blame]
# Copyright 2014 Google Inc. All Rights Reserved.
import json
import time
import unittest
import webtest
import analytics
import apprtc
import constants
import probers
from test_util import CapturingFunction
from test_util import ReplaceFunction
from google.appengine.api import memcache
from google.appengine.ext import testbed
class MockRequest(object):
def get(self, key):
return None
class AppRtcUnitTest(unittest.TestCase):
def setUp(self):
# First, create an instance of the Testbed class.
self.testbed = testbed.Testbed()
# Then activate the testbed, which prepares the service stubs for use.
self.testbed.activate()
def tearDown(self):
self.testbed.deactivate()
def testGenerateRandomGeneratesStringOfRightLength(self):
self.assertEqual(17, len(apprtc.generate_random(17)))
self.assertEqual(23, len(apprtc.generate_random(23)))
class AppRtcPageHandlerTest(unittest.TestCase):
def setUp(self):
# First, create an instance of the Testbed class.
self.testbed = testbed.Testbed()
# Then activate the testbed, which prepares the service stubs for use.
self.testbed.activate()
# Next, declare which service stubs you want to use.
self.testbed.init_memcache_stub()
self.test_app = webtest.TestApp(apprtc.app)
# Fake out event reporting.
self.time_now = time.time()
# Fake out event reporting and capture arguments.
self.report_event_replacement = ReplaceFunction(
analytics,
'report_event',
CapturingFunction())
def tearDown(self):
self.testbed.deactivate()
del self.report_event_replacement
def makeGetRequest(self, path):
# PhantomJS uses WebKit, so Safari is closest to the thruth.
return self.test_app.get(path, headers={'User-Agent': 'Safari'})
def makePostRequest(self, path, body=''):
return self.test_app.post(path, body, headers={'User-Agent': 'Safari'})
def verifyJoinSuccessResponse(self, response, is_initiator, room_id):
self.assertEqual(response.status_int, 200)
response_json = json.loads(response.body)
self.assertEqual('SUCCESS', response_json['result'])
params = response_json['params']
caller_id = params['client_id']
self.assertTrue(len(caller_id) > 0)
self.assertEqual(json.dumps(is_initiator), params['is_initiator'])
self.assertEqual(room_id, params['room_id'])
self.assertEqual([], params['error_messages'])
self.assertEqual([], params['warning_messages'])
return caller_id
def testConnectingWithoutRoomIdServesIndex(self):
response = self.makeGetRequest('/')
self.assertEqual(response.status_int, 200)
self.assertNotRegexpMatches(response.body, 'roomId:')
def testConnectingWithRoomIdServesIndex(self):
response = self.makeGetRequest('/r/testRoom')
self.assertEqual(response.status_int, 200)
self.assertRegexpMatches(response.body, 'roomId: \'testRoom\'')
def testJoinAndLeave(self):
room_id = 'foo'
# Join the caller.
response = self.makePostRequest('/join/' + room_id)
caller_id = self.verifyJoinSuccessResponse(response, True, room_id)
# Join the callee.
response = self.makePostRequest('/join/' + room_id)
callee_id = self.verifyJoinSuccessResponse(response, False, room_id)
# The third user will get an error.
response = self.makePostRequest('/join/' + room_id)
self.assertEqual(response.status_int, 200)
response_json = json.loads(response.body)
self.assertEqual('FULL', response_json['result'])
# The caller and the callee leave.
self.makePostRequest('/leave/' + room_id + '/' + caller_id)
self.makePostRequest('/leave/' + room_id + '/' + callee_id)
# Another user becomes the new caller.
response = self.makePostRequest('/join/' + room_id)
caller_id = self.verifyJoinSuccessResponse(response, True, room_id)
self.makePostRequest('/leave/' + room_id + '/' + caller_id)
def testCallerMessagesForwardedToCallee(self):
room_id = 'foo'
# Join the caller.
response = self.makePostRequest('/join/' + room_id)
caller_id = self.verifyJoinSuccessResponse(response, True, room_id)
# Caller's messages should be saved.
messages = ['1', '2', '3']
path = '/message/' + room_id + '/' + caller_id
for msg in messages:
response = self.makePostRequest(path, msg)
response_json = json.loads(response.body)
self.assertEqual('SUCCESS', response_json['result'])
response = self.makePostRequest('/join/' + room_id)
callee_id = self.verifyJoinSuccessResponse(response, False, room_id)
received_msgs = json.loads(response.body)['params']['messages']
self.assertEqual(messages, received_msgs)
self.makePostRequest('/leave/' + room_id + '/' + caller_id)
self.makePostRequest('/leave/' + room_id + '/' + callee_id)
def setWssHostStatus(self, index1, status1, index2, status2):
probing_results = {}
probing_results[constants.WSS_HOST_PORT_PAIRS[index1]] = {
constants.WSS_HOST_IS_UP_KEY: status1
}
probing_results[constants.WSS_HOST_PORT_PAIRS[index2]] = {
constants.WSS_HOST_IS_UP_KEY: status2
}
probers.ProbeColliderPage().store_instance_state(probing_results)
def verifyRequest(self, expectedIndex):
request = MockRequest()
wss_url, wss_post_url = apprtc.get_wss_parameters(request)
self.assertIn(constants.WSS_HOST_PORT_PAIRS[expectedIndex], wss_url)
self.assertIn(constants.WSS_HOST_PORT_PAIRS[expectedIndex], wss_post_url)
def testGetWssHostParameters(self):
request = MockRequest()
# With no status set, should use fallback.
self.verifyRequest(0)
# With an invalid value in memcache, should use fallback.
memcache_client = memcache.Client()
memcache_client.set(constants.WSS_HOST_ACTIVE_HOST_KEY, 'abc')
self.verifyRequest(0)
# With an invalid value in memcache, should use fallback.
memcache_client = memcache.Client()
memcache_client.set(constants.WSS_HOST_ACTIVE_HOST_KEY, ['abc', 'def'])
self.verifyRequest(0)
# With both hosts failing, should use fallback.
self.setWssHostStatus(0, False, 1, False)
self.verifyRequest(0)
# Second host passing.
self.setWssHostStatus(0, False, 1, True)
self.verifyRequest(1)
# Both hosts passing, but second host for longer.
self.setWssHostStatus(1, True, 0, True)
self.verifyRequest(1)
if __name__ == '__main__':
unittest.main()