Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
python zmq unit test mock
import zmq
import unittest
from unittest.mock import patch
from unittest.mock import MagicMock
class MockZmqSocket(MagicMock):
def __init__(self, **kwds):
super().__init__(**kwds)
self._connected = False
def connect(self, endpoint):
assert not self._connected
assert endpoint != ""
self._connected = True
def close(self):
assert self._connected
self._connected = False
class MockZmqContext(MagicMock):
def __init__(self, **kwds):
super().__init__(**kwds)
self._destroyed = False
self.IPV6 = None
self.RCVTIMEO = None
def socket(self, sock_type):
assert not self._destroyed
assert self.IPV6 == 1
assert self.RCVTIMEO is not None
assert sock_type == zmq.REQ
return MockZmqSocket()
def destroy(self, linger):
assert not self._destroyed
self._destroyed = True
@patch.object(zmq.Context, "instance", MockZmqContext)
class TestFoo(unittest.TestCase):
def test_bar(self):
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.