On Thu, Jul 31, 2003 at 12:09:00PM +1000, Andrew Bennetts wrote:

In general you'd just do some variant on:

...

reactor.listenTCP(1234, myServerFactory) reactor.connectTCP("127.0.0.1", 1234, myClientFactory): while someConditionIsn'tSet: reactor.iterate() # at this point some exchange should have finished successfully

Or you can use the loopback module -- twisted.protocols.loopback. Many of the Twisted tests do this.

I'd like to vote _heavily_ on using twisted.protocols.loopback.loopback() Opening TCP sockets in unit tests is just not _unit_ testing in my book. Please don't do it in unit tests. It just makes it harder to run your unit tests in varying environments. (Who says I allow you to bind to port 1234? Who says it's free? Who says I allow you to listen *at all*?) </rant> Also, you should not be testing just interoperability between *your* client and *your* server, but interoperability of your server with a "standard", and interoperability of your client with a "standard". For that, I prefer doing something like this: class LDAPClientTestDriver: """ A test driver that looks somewhat like a real LDAPClient. Pass in a list of lists of LDAPProtocolResponses. For each sent LDAP message, the first item of said list is iterated through, and all the items are sent as responses to the callback. The sent LDAP messages are stored in self.sent, so you can assert that the sent messages are what they are supposed to be. """ def __init__(self, *responses): self.sent=[] self.responses=list(responses) def queue(self, x, callback): self.sent.append(x) assert self.responses, 'Ran out of responses at %r' % x responses = self.responses.pop(0) while responses: r = responses.pop(0) ret = callback(r) if responses: assert ret==0 else: assert ret==1 def assertNothingSent(self): # just a bit more explicit self.assertSent() def assertSent(self, *shouldBeSent): shouldBeSent = list(shouldBeSent) assert self.sent == shouldBeSent, \ '%s expected to send %r but sent %r' % ( self.__class__.__name__, shouldBeSent, self.sent) sentStr = ''.join([str(x) for x in self.sent]) shouldBeSentStr = ''.join([str(x) for x in shouldBeSent]) assert sentStr == shouldBeSentStr, \ '%s expected to send data %r but sent %r' % ( self.__class__.__name__, shouldBeSentStr, sentStr) class LDAPSyntaxAttributesModificationOnWire(unittest.TestCase): def testAdd(self): """Modify & commit should write the right data to the server.""" client = LDAPClientTestDriver( [ pureldap.LDAPModifyResponse(resultCode=0, matchedDN='', errorMessage=''), ]) o=ldapsyntax.LDAPEntry(client=client, dn='cn=foo,dc=example,dc=com', attributes={ 'objectClass': ['a', 'b'], 'aValue': ['a'], }) o['aValue'].add('newValue') o['aValue'].add('anotherNewValue') d=o.commit() val = deferredResult(d) client.assertSent(pureldap.LDAPModifyRequest( object='cn=foo,dc=example,dc=com', modification=[ pureldap.LDAPModification_add(vals=(('aValue', ['newValue']),)), pureldap.LDAPModification_add(vals=(('aValue', ['anotherNewValue']),)), ])) And elsewhere, test the actual serialization of the on-wire protocol message known as pureldap.LDAPModifyRequest: class KnownValues(unittest.TestCase): knownValues=( # class, args, kwargs, expected_result (pureldap.LDAPModifyRequest, [], { "object": 'cn=foo, dc=example, dc=com', "modification": [pureldap.LDAPModification_delete([('bar',)])] }, [0x66, 0x2c] + [0x04, 0x1a] + l("cn=foo, dc=example, dc=com") + [0x30, 0x0e] + [0x30, 0x0c] + [0x0a, 0x01, 0x01] + [0x30, 0x07] + [0x04, 0x03] + l("bar") + [0x31, 0x00]), ... ) def testToLDAP(self): """str(LDAPClass(...)) should give known result with known input""" for klass, args, kwargs, encoded in self.knownValues: result = klass(*args, **kwargs) result = str(result) result = map(ord, result) if result!=encoded: raise AssertionError, \ "Class %s(*%s, **%s) doesn't encode properly: " \ "%s != %s" % (klass.__name__, repr(args), repr(kwargs), repr(result), repr(encoded)) def testFromLDAP(self): """LDAPClass(encoded="...") should give known result with known input""" for klass, args, kwargs, encoded in self.knownValues: m=MutableString(s(*encoded)) m.append('foo') result = klass(encoded=m, berdecoder=pureber.BERDecoderContext()) assert m=='foo' shouldBe = klass(*args, **kwargs) #TODO shouldn't use str below assert str(result)==str(shouldBe), \ "Class %s(*%s, **%s) doesn't decode properly: " \ "%s != %s" % (klass.__name__, repr(args), repr(kwargs), repr(result), repr(shouldBe)) -- :(){ :|:&};:

On Thu, Jul 31, 2003 at 11:03:07AM -0400, Itamar Shtull-Trauring wrote:

...

Opening TCP sockets in unit tests is just not _unit_ testing in my book. Please don't do it in unit tests. It just makes it harder to run your unit tests in varying environments. (Who says I allow you to bind to port 1234? Who says it's free? Who says I allow you to listen *at all*?)

Most of Twisted's unittests listen on 0 and then figure out which port was bound to.

Which is mostly the reason why I'm bitch about it on _this_ mailing list. I dislike that method. I really do. There have been _months_ during which certain twisted unit tests randomly fail; many of those have been due to interaction with local TCP stack, networking, firewall and miscellanous experimental security features. I really think unit tests should touch the underlying libraries/OS as little as possible. For the cases where you really want to test the OS integration, you want to do things like stress tests, and for e.g. mail tests where you randomly shut down the power to a machine running a stress test, and later check that no messages were lost. -- :(){ :|:&};: