|
|
# encoding: utf-8
|
|
|
|
|
|
"""This file contains unittests for the enginepb.py module."""
|
|
|
|
|
|
__docformat__ = "restructuredtext en"
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# Copyright (C) 2008 The IPython Development Team
|
|
|
#
|
|
|
# Distributed under the terms of the BSD License. The full license is in
|
|
|
# the file COPYING, distributed as part of this software.
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
#-------------------------------------------------------------------------------
|
|
|
# Imports
|
|
|
#-------------------------------------------------------------------------------
|
|
|
|
|
|
# Tell nose to skip this module
|
|
|
__test__ = {}
|
|
|
|
|
|
from twisted.python import components
|
|
|
from twisted.internet import reactor, defer
|
|
|
from twisted.spread import pb
|
|
|
from twisted.internet.base import DelayedCall
|
|
|
DelayedCall.debug = True
|
|
|
|
|
|
import zope.interface as zi
|
|
|
|
|
|
from IPython.kernel.fcutil import Tub, UnauthenticatedTub
|
|
|
from IPython.kernel import engineservice as es
|
|
|
from IPython.testing.util import DeferredTestCase
|
|
|
from IPython.kernel.controllerservice import IControllerBase
|
|
|
from IPython.kernel.enginefc import FCRemoteEngineRefFromService, IEngineBase
|
|
|
from IPython.kernel.engineservice import IEngineQueued
|
|
|
from IPython.kernel.engineconnector import EngineConnector
|
|
|
|
|
|
from IPython.kernel.tests.engineservicetest import \
|
|
|
IEngineCoreTestCase, \
|
|
|
IEngineSerializedTestCase, \
|
|
|
IEngineQueuedTestCase
|
|
|
|
|
|
|
|
|
class EngineFCTest(DeferredTestCase,
|
|
|
IEngineCoreTestCase,
|
|
|
IEngineSerializedTestCase,
|
|
|
IEngineQueuedTestCase
|
|
|
):
|
|
|
|
|
|
zi.implements(IControllerBase)
|
|
|
|
|
|
def setUp(self):
|
|
|
|
|
|
# Start a server and append to self.servers
|
|
|
self.controller_reference = FCRemoteEngineRefFromService(self)
|
|
|
self.controller_tub = Tub()
|
|
|
self.controller_tub.listenOn('tcp:10105:interface=127.0.0.1')
|
|
|
self.controller_tub.setLocation('127.0.0.1:10105')
|
|
|
|
|
|
furl = self.controller_tub.registerReference(self.controller_reference)
|
|
|
self.controller_tub.startService()
|
|
|
|
|
|
# Start an EngineService and append to services/client
|
|
|
self.engine_service = es.EngineService()
|
|
|
self.engine_service.startService()
|
|
|
self.engine_tub = Tub()
|
|
|
self.engine_tub.startService()
|
|
|
engine_connector = EngineConnector(self.engine_tub)
|
|
|
d = engine_connector.connect_to_controller(self.engine_service, furl)
|
|
|
# This deferred doesn't fire until after register_engine has returned and
|
|
|
# thus, self.engine has been defined and the tets can proceed.
|
|
|
return d
|
|
|
|
|
|
def tearDown(self):
|
|
|
dlist = []
|
|
|
# Shut down the engine
|
|
|
d = self.engine_tub.stopService()
|
|
|
dlist.append(d)
|
|
|
# Shut down the controller
|
|
|
d = self.controller_tub.stopService()
|
|
|
dlist.append(d)
|
|
|
return defer.DeferredList(dlist)
|
|
|
|
|
|
#---------------------------------------------------------------------------
|
|
|
# Make me look like a basic controller
|
|
|
#---------------------------------------------------------------------------
|
|
|
|
|
|
def register_engine(self, engine_ref, id=None, ip=None, port=None, pid=None):
|
|
|
self.engine = IEngineQueued(IEngineBase(engine_ref))
|
|
|
return {'id':id}
|
|
|
|
|
|
def unregister_engine(self, id):
|
|
|
pass
|
