Unit testing ClustorRpcProxy

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.

Hi @cricen You can find this example of
implementation:

and test:

Hope this helps

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.