Hello,
New to nameko and I’m trying to figure out how I can unit test my api gateway. I have an endpoint that calls out to rabbit/nameko using the ClusterRpcProxy for a value. I’d like to write a simple pytest that mocks the proxy response with a fixed value so that we can verify the rest of the API call. can anyone point me to a basic example/sample or let me know where I can get more information on testing with CluserRpcProxy?
Thanks for any assistance.
It does help and I believe I’m at least down a better path but, no ah ha moment yet. I refactored my gateway(I’m using falcon) so that I had something like your example. however I’m still getting hung up on using ClusterRpcProxy and doing what your example is doing in test_service.py:10. here is a quick and dirty re-write of what I’m trying.
myResource.py
import json
import falcon
import uuid
from nameko.standalone.rpc import ClusterRpcProxy
class Resource(object):
my_rpc = ClusterRpcProxy({'AMQP_URI': "amqp://guest:guest@:localhost:5672"})
def __init__(self):
self.logger = Logger.ConfigLogging("config.yml").logger
self.logger.debug("ending init for resource")
def on_get(self, req, resp):
ms_id = uuid.uuid1()
with self.my_rpc as rpc:
message = rpc.TerribleForm.hello(str(ms_id))
doc = {
"id": str(ms_id),
"message": message
}
resp.body = json.dumps(doc, ensure_ascii=False)
resp.status = falcon.HTTP_200
self.logger.debug("end get function.")
my_test.py
import msgpack
import pytest
import falcon
from falcon import testing
from nameko.rpc import rpc
from app import app
@pytest.fixture
def client():
return testing.TestClient(app)
def test_get(client, monkeypatch):
fake_hello = "Hello, 284f42ee-b12e-11e9-824e-0242ac120002!"
#tried lots of stuff below. monkeypatch is my latest attempt but got No module named 'terribleForm_rpc'
monkeypatch.setattr('my_rpc.TerribleForm.hello', lambda: fake_hello)
doc = {
"id": "X",
"message": "Hello, 284f42ee-b12e-11e9-824e-0242ac120002!",
"Status": "pending"
}
response = client.simulate_get('/my_resource')
result_doc = msgpack.unpackb(response.content, raw=False)
assert result_doc['message'] == doc['message']
assert response.status == falcon.HTTP_OK
I’ve also tried similar to the example you’ve pointed out by not using monkeypatch. Please let me know if you can suggest anything
Thanks again.
I think your issue might be with mocking of ClusterRpcProxy context manager.
I’ve done this locally:
# tree
.
└── src
├── __init__.py
└── app.py
└── my_resource.py
└── my_test.pypy
# app.py
import falcon
from .my_resource import Resource
api = application = falcon.API()
resources = Resource()
api.add_route('/resources', resources)
# my_resource.py
import falcon
import msgpack
from nameko.standalone.rpc import ClusterRpcProxy
class Resource(object):
config = {'AMQP_URI': 'amqp://guest:guest@localhost:5672'}
def on_get(self, req, resp):
with ClusterRpcProxy(self.config) as cluster_rpc:
message = cluster_rpc.my_service.remote_method('hello')
doc = {
'message': message
}
resp.data = msgpack.packb(doc, use_bin_type=True)
resp.content_type = falcon.MEDIA_JSON
resp.status = falcon.HTTP_200
# my_test.py
import falcon
from falcon import testing
import msgpack
import pytest
from mock import patch, MagicMock
from .app import api
@pytest.fixture
def client():
return testing.TestClient(api)
def test_foo(client):
with patch('src.my_resource.ClusterRpcProxy') as mock_proxy:
foo = MagicMock()
foo.my_service.remote_method.return_value = 'foo'
mock_proxy.return_value.__enter__.return_value = foo
doc = {
'message': 'foo'
}
response = client.simulate_get('/resources')
result_doc = msgpack.unpackb(response.content, raw=False)
assert result_doc == doc
assert response.status == falcon.HTTP_OK
and then my test passed:
$ pytest .
collected 1 item
src/my_test.py .
Hope this helps.
Thank you! I had a few issues but the biggest was with mocking and the way I had setup my package structure. Had to restructure but it was for the better.