[zeromq-dev] Testing and debugging

andrea crotti andrea.crotti.0 at gmail.com
Thu Aug 16 11:43:40 CEST 2012

2012/8/14 Michel Pelletier <pelletier.michel at gmail.com>:
> On Tue, Aug 14, 2012 at 7:34 AM, andrea crotti
> <andrea.crotti.0 at gmail.com> wrote:
>> 2012/8/13 Michel Pelletier <pelletier.michel at gmail.com>:
>> Thanks for the help, any example of how to mock the zmq messages maybe?
>> I was trying to find a way and I'm ending up mocking the whole module,
>> but it doesn't really work this way:
> <snip>
>> And I'm not sure it's worth actually to mock in this way, instead I
>> could just send message with "inproc" instead of the TCP connections..
> You should look at a mocking library to do the actual mocked objects
> for you.  flexmock, for example, is one I like, although there are
> several.  Here's an off the cuff example:
> from flexmock import flexmock
> def do(pub):
>     pub.send('hi there!')
> def test_do():
>     mock_pub = flexmock()
>     (flexmock(mock_pub).
>      should_receive('send').
>      with_args('hi there!').
>      once())
>     do(mock_pub)
> Put it in test.py and run it with "nosetests test.py".  Note that no
> real zmq objects are actually created, the zmq module isn't even
> imported.  The point is to isolate an test your logic as much as
> possible.  You don't need to test that zmq works to test your logic.
> Also flexmock will take care of asserting that your method was called
> once with the correct args, there's no need to reimplement that.

Yes sure but I still don't see how to actually use it for my case..
What I would like is a way to create a mock that understands if two
sockets are on the same channel, and if one sends something the other
is able to read it.

Because in this way it's really too close to the implementation and
way too long for my taste, and the moment I change something the tests
will fail..

I might try to replace a socket with something smart that just writes
and reads from the same temporary file whenever the socket called bind
or connect with the same port..

The alternative is to not bother at all and just test one method a
time (as below), passing something to and checking that we are sending
the right back, but I'm not so sure it's worth still.

    def _handle_job(self, job):
        self.logger.info("Got job %s" % job)
        # send the result back
        self.result_socket.send(str(int(job) * 2))

More information about the zeromq-dev mailing list