##// END OF EJS Templates
Merge branch 'unicode-issues'
Merge branch 'unicode-issues'

File last commit:

r2517:8185e15c
r3459:a6a9d580 merge
Show More
test_enginefc.py
91 lines | 3.2 KiB | text/x-python | PythonLexer
# 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:10111:interface=127.0.0.1')
self.controller_tub.setLocation('127.0.0.1:10111')
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