[Twisted-Python] the right way of unit testing protocols

[Tommi Virtanen]

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))

-- 
:(){ :|:&};: