[Twisted-Python] Conch Testing server and client
Hi, I have to write tests for a custom conch server. My approach is to make a test client and then connect it to my server in my testcase. I have been able to create a client that can interface with my server, but i'm not able to do it from within the test case. What I'm trying is : class TestAccountStatus(unittest.TestCase): def setUp(self): ssh_server = Server() self.server = reactor.listenTCP(ssh_server.port, ssh_server.application(), interface=ssh_server.interface) def tearDown(self): server, self.server = self.server,None return server.stopListening() def test_1(self): d = protocol.ClientCreator(reactor, SimpleTransport).connectTCP('localhost', self.server.getHost().port) return d The transport SimpleTransport is my custom transport.SSHClientTransport instance. The class Server() is a wrapper around an SSHFactory insatance. The problem I'm facing is that after this line > d = protocol.ClientCreator(reactor, SimpleTransport).connectTCP('localhost', self.server.getHost().port) I have to write reactor.run(), otherwise the client is not starting. If I write reactor.run() the test doesn't stop (understandable since the reactor isn't being stopped). Basically I want to know how I can start the client without running reactor.run() Thanks in Advance!! Anshul
I understand that. The problem I'm facing is that unless I write "reactor.run()" my client isn't starting up. From what I understood in the trial documentation, I should simply call the connectTCP method and then return a deferred. However, if I do that, the client isn't starting. I'll post some of the client code for you info : class SimpleTransport(transport.SSHClientTransport): def verifyHostKey(self, hostKey, fingerprint): print 'host key fingerprint: %s' % fingerprint return defer.succeed(1) def connectionSecure(self): self.requestService( SimpleUserAuth('webchick', SimpleConnection())) class SimpleUserAuth(userauth.SSHUserAuthClient): def getPassword(self): return defer.succeed("Thai1mil3ahb") def getPublicKey(self): print "trying public key" path = os.path.expanduser('~/.ssh/id_rsa') # this works with rsa too # just change the name here and in getPrivateKey if not os.path.exists(path) or self.lastPublicKey: # the file doesn't exist, or we've tried a public key return return keys.Key.fromFile(filename=path+'.pub',passphrase='pikachu').blob() def getPrivateKey(self): path = os.path.expanduser('~/.ssh/id_rsa') return defer.succeed(keys.Key.fromFile(path,passphrase='pikachu').keyObject) Now , in my test case : def test_1(self): def got_data(data): self.assertEquals(data,"a") d = protocol.ClientCreator(reactor, SimpleTransport).connectTCP('localhost', self.server.getHost().port) d.addCallback(got_data) return d The callback is firing, but the client is not started and *data* is a SimpleTransport Instance. Any thoughts on why this is happening and what I am doing wrong?
On 12:59 pm, anshul.singhle@gmail.com wrote:
Complicated, incomplete code snippets aren't very useful. Always follow http://sscce.org/ when sharing code. For example: from twisted.trial.unittest import TestCase from twisted.internet.defer import Deferred from twisted.internet import reactor class ReactorIsRunningTests(TestCase): def test_isTheReactorRunning(self): d = Deferred() reactor.callLater(1, d.callback, None) return d The output I expect from this, when run with trial, is one test method which takes 1 second to complete, and then succeeds. The output I get is: reactortests ReactorIsRunningTests test_isTheReactorRunning ... [OK] ----------------------------------------------------------------------- Ran 1 tests in 1.003s PASSED (successes=1) In other words, it behaves as I expect. Jean-Paul
Ok, here goes : The test code : from twisted.trial import unittest from twisted.internet import defer, protocol, reactor class TestAccountStatus(unittest.TestCase): def setUp(self): ssh_server = Server() self.server = reactor.listenTCP(ssh_server.port, ssh_server.application(), interface=ssh_server.interface) return self.server def tearDown(self): server, self.server = self.server,None return server.stopListening() def test_1(self): def got_data(data): self.assertEquals(data,"a") d = protocol.ClientCreator(reactor, SimpleTransport).connectTCP('localhost', self.server.getHost().port) d.addCallback(got_data) return d The client code : from twisted.conch.ssh import transport, userauth, connection, common, keys, channel from twisted.internet import defer, protocol, reactor class SimpleTransport(transport.SSHClientTransport): def verifyHostKey(self, hostKey, fingerprint): print 'host key fingerprint: %s' % fingerprint return defer.succeed(1) def connectionSecure(self): self.requestService( SimpleUserAuth('webchick', SimpleConnection())) class SimpleUserAuth(userauth.SSHUserAuthClient): def getPassword(self): print "Get Password called" return defer.succeed("Thai1mil3ahb") When I run the test I expect : tests TestAccountStatus test_1 ... Get Password called followed by some errors , but atleast this much should be printed if the client code is being called properly. Instead I get: [FAIL]: tests.TestAccountStatus.test_1 Traceback (most recent call last): File "tests.py", line 92, in got_data self.assertEquals(data,"a") twisted.trial.unittest.FailTest: not equal: a = <tests.SimpleTransport instance at 0x1024aa128> b = 'a' Instead of running the client the deferred is just returning an instance of the Client Transport used by me (SimpleTransport). Any ideas why this is happening? Is this becuse I'm not using a ClientFactory or am I using the deferred all wrong?
On Fri, Jul 01, 2011 at 05:44:31PM +0530, Anshul Singhle wrote:
What I'm trying is : class TestAccountStatus(unittest.TestCase):
Is that the Python standard library's unittest.TestCase, or twisted.trial.unittest.TestCase? As Jean-Paul says, Twisted's TestCase should automatically handle the reactor for you; that's why it exists. :)
I understand that. The problem I'm facing is that unless I write "reactor.run()" my client isn't starting up. From what I understood in the trial documentation, I should simply call the connectTCP method and then return a deferred. However, if I do that, the client isn't starting. I'll post some of the client code for you info : class SimpleTransport(transport.SSHClientTransport): def verifyHostKey(self, hostKey, fingerprint): print 'host key fingerprint: %s' % fingerprint return defer.succeed(1) def connectionSecure(self): self.requestService( SimpleUserAuth('webchick', SimpleConnection())) class SimpleUserAuth(userauth.SSHUserAuthClient): def getPassword(self): return defer.succeed("Thai1mil3ahb") def getPublicKey(self): print "trying public key" path = os.path.expanduser('~/.ssh/id_rsa') # this works with rsa too # just change the name here and in getPrivateKey if not os.path.exists(path) or self.lastPublicKey: # the file doesn't exist, or we've tried a public key return return keys.Key.fromFile(filename=path+'.pub',passphrase='pikachu').blob() def getPrivateKey(self): path = os.path.expanduser('~/.ssh/id_rsa') return defer.succeed(keys.Key.fromFile(path,passphrase='pikachu').keyObject) Now , in my test case : def test_1(self): def got_data(data): self.assertEquals(data,"a") d = protocol.ClientCreator(reactor, SimpleTransport).connectTCP('localhost', self.server.getHost().port) d.addCallback(got_data) return d The callback is firing, but the client is not started and *data* is a SimpleTransport Instance. Any thoughts on why this is happening and what I am doing wrong?
On 12:59 pm, anshul.singhle@gmail.com wrote:
Complicated, incomplete code snippets aren't very useful. Always follow http://sscce.org/ when sharing code. For example: from twisted.trial.unittest import TestCase from twisted.internet.defer import Deferred from twisted.internet import reactor class ReactorIsRunningTests(TestCase): def test_isTheReactorRunning(self): d = Deferred() reactor.callLater(1, d.callback, None) return d The output I expect from this, when run with trial, is one test method which takes 1 second to complete, and then succeeds. The output I get is: reactortests ReactorIsRunningTests test_isTheReactorRunning ... [OK] ----------------------------------------------------------------------- Ran 1 tests in 1.003s PASSED (successes=1) In other words, it behaves as I expect. Jean-Paul
Ok, here goes : The test code : from twisted.trial import unittest from twisted.internet import defer, protocol, reactor class TestAccountStatus(unittest.TestCase): def setUp(self): ssh_server = Server() self.server = reactor.listenTCP(ssh_server.port, ssh_server.application(), interface=ssh_server.interface) return self.server def tearDown(self): server, self.server = self.server,None return server.stopListening() def test_1(self): def got_data(data): self.assertEquals(data,"a") d = protocol.ClientCreator(reactor, SimpleTransport).connectTCP('localhost', self.server.getHost().port) d.addCallback(got_data) return d The client code : from twisted.conch.ssh import transport, userauth, connection, common, keys, channel from twisted.internet import defer, protocol, reactor class SimpleTransport(transport.SSHClientTransport): def verifyHostKey(self, hostKey, fingerprint): print 'host key fingerprint: %s' % fingerprint return defer.succeed(1) def connectionSecure(self): self.requestService( SimpleUserAuth('webchick', SimpleConnection())) class SimpleUserAuth(userauth.SSHUserAuthClient): def getPassword(self): print "Get Password called" return defer.succeed("Thai1mil3ahb") When I run the test I expect : tests TestAccountStatus test_1 ... Get Password called followed by some errors , but atleast this much should be printed if the client code is being called properly. Instead I get: [FAIL]: tests.TestAccountStatus.test_1 Traceback (most recent call last): File "tests.py", line 92, in got_data self.assertEquals(data,"a") twisted.trial.unittest.FailTest: not equal: a = <tests.SimpleTransport instance at 0x1024aa128> b = 'a' Instead of running the client the deferred is just returning an instance of the Client Transport used by me (SimpleTransport). Any ideas why this is happening? Is this becuse I'm not using a ClientFactory or am I using the deferred all wrong?
On Fri, Jul 01, 2011 at 05:44:31PM +0530, Anshul Singhle wrote:
What I'm trying is : class TestAccountStatus(unittest.TestCase):
Is that the Python standard library's unittest.TestCase, or twisted.trial.unittest.TestCase? As Jean-Paul says, Twisted's TestCase should automatically handle the reactor for you; that's why it exists. :)
participants (4)
-
Anshul Singhle -
exarkun＠twistedmatrix.com -
Itamar Turner-Trauring -
Tim Allen