How to write tests for code using twisted.web.client.Agent and its subclasses?

Question

I read the official official test development tutorial , but in my case it didn't help much. I wrote a small library that makes extensive use of twisted.web.client.Agent and its subclasses (like BrowserLikeRedirectAgent ), but I struggled to adapt the tutorial code to my own test cases.

I looked at twisted.web.test.test_web , but I don’t understand how to assemble all the shapes. For example, I still don’t know how to get the Protocol object from Agent , according to the official tutorial

Can someone show me how to write a simple test for some code that uses agent data for GET and POST ? Any further details or tips are appreciated ...

Many thanks!

Answer 1

How to simplify life (that is, we read the code) using @inlineCallbacks .

In fact, I even went so far as to suggest not using Deferred directly if it is absolutely necessary for performance or in a specific use case, but instead always stick with @inlineCallbacks - this way I will keep your code like normal code, while still getting advantage of non-blocking behavior:

 from twisted.internet import reactor from twisted.web.client import Agent from twisted.internet.defer import inlineCallbacks from twisted.trial import unittest from twisted.web.http_headers import Headers from twisted.internet.error import DNSLookupError class SomeTestCase(unittest.TestCase): @inlineCallbacks def test_smth(self): ag = Agent(reactor) response = yield ag.request('GET', 'http://example.com/', Headers({'User-Agent': ['Twisted Web Client Example']}), None) self.assertEquals(response.code, 200) @inlineCallbacks def test_exception(self): ag = Agent(reactor) try: yield ag.request('GET', 'http://exampleeee.com/', Headers({'User-Agent': ['Twisted Web Client Example']}), None) except DNSLookupError: pass else: self.fail() 

The trial should take care of the rest (i.e. waiting on Deferred returned from the test functions ( @inlineCallbacks broken calls also “magically” return Deferred - I highly recommend reading more on @inlineCallbacks if you are not already familiar with it).

PS There is also a Twisted “plug-in” for media that allows you to return Deferred from your test functions and wait for your nose to be fired before exiting: http://nose.readthedocs.org/en/latest/api/ twistedtools.html

Answer 2

This is similar to what Mike said, but is trying to check the response processing. There are other ways to do this, but I like it. I also agree that testing the things that wrap the agent is not very useful, and testing the protocol / logic in your protocol is probably better anyway, but sometimes you just want to add some green labels.

 class MockResponse(object): def __init__(self, response_string): self.response_string = response_string def deliverBody(self, protocol): protocol.dataReceived(self.response_string) protocol.connectionLost(None) class MockAgentDeliverStuff(Agent): def request(self, method, uri, headers=None, bodyProducer=None): d = Deferred() reactor.callLater(0, d.callback, MockResponse(response_body)) return d class MyWrapperTestCase(unittest.TestCase): def setUp:(self): agent = MockAgentDeliverStuff(reactor) self.wrapper_object = MyWrapper(agent) @inlineCallbacks def test_something(self): response_object = yield self.wrapper_object("example.com") self.assertEqual(response_object, expected_object) 

Answer 3

How about this? Run the trial version as follows. Basically, you simply scoff at the agent and pretend that he is doing what is advertised, and using FakeAgent (in this case) will cause all requests to fail. If you really want to enter data into the transport, it will probably take longer. But are you really checking your code? Or an agent?

 from twisted.web import client from twisted.internet import reactor, defer class BidnessLogik(object): def __init__(self, agent): self.agent = agent self.money = None def make_moneee_quik(self): d = self.agent.request('GET', 'http://no.traffic.plz') d.addCallback(self.made_the_money).addErrback(self.no_dice) return d def made_the_money(self, *args): ##print "Moneeyyyy!" self.money = True return 'money' def no_dice(self, fail): ##print "Better luck next time!!" self.money = False return 'no dice' class FailingAgent(client.Agent): expected_uri = 'http://no.traffic.plz' expected_method = 'GET' reasons = ['No Reason'] test = None def request(self, method, uri, **kw): if self.test: self.test.assertEqual(self.expected_uri, uri) self.test.assertEqual(self.expected_method, method) self.test.assertEqual([], kw.keys()) return defer.fail(client.ResponseFailed(reasons=self.reasons, response=None)) class TestRequest(unittest.TestCase): def setUp(self): self.agent = FailingAgent(reactor) self.agent.test = self @defer.inlineCallbacks def test_foo(self): bid = BidnessLogik(self.agent) resp = yield bid.make_moneee_quik() self.assertEqual(resp, 'no dice') self.assertEqual(False, bid.money)