Mocking rpc call for unit testing with pytest

1

Hi fellow developers, I’m facing an issue with unit testing a service which calls other service via rpc. What I’m trying to do is basically, mock the rpc response from other service, as they have done in the nameko example for testing gateway.
I’m not sure if that’ s right approach, if anyone could point in my right direction, that would really be great. I’m following nameko-example from offical nameko repository, I use same conftest file, to provide mocked rpc and then retuen a custom response.

2

More details about the problem, since I’m unable to edit the post I’m writing it here.

Hi fellow developers, I’m facing an issue with unit testing a service which calls other service via rpc. What I’m trying to do is basically, mock the rpc response from other service, as they have done in the nameko example for testing gateway.
I’m not sure if that’ s right approach, if anyone could point in my right direction, that would really be great. I’m following nameko-example from offical nameko repository, I use same conftest file, to provide mocked rpc and then return a custom response.

I think I didn’t provide sufficient info, the thing is I can’t paste the code here. But the gist of it is

Service ABC

    class ABCService: 
         xyx_rpc = RpcProxy('XYZ')
        @rpc
        def handle_task(self, *task):
         ....
         rpc_response = xyz_rpc.perform_task(token, details)
        ....
        return MySchema().dump(data_map)

Service XYZ

 class XYZService: 
             
            @rpc
            def perform_task(self, token, details):
             ....
            some business logic
            ....
            return ResponseSchema().dump(data_map)

Test class ABC

class TestABCservice:

        def test_handle_task(self, abc_service):
         abcService = ABCService()    
     #I want to mock response from xyzservice.perform_task method
         abc_service.xyz_rpc.perform_task.return_value = {**'some':'dummy response'**}
         response = abcService.handle_task(*dummy_task)
        
       assert response = { 'status : 'success', **'some':'dummy response'**, ....}

here, abc_service is a fixture with replaced dependencies using conftest.py from gateway service in nameko example.

The issue is I get this error, I’m guessing it didn’t get mocked.

AttributeError: ‘ServiceRpc’ object has no attribute ‘perform_task’

Any help would really appreciated

3

@mattbennett any help will really appreciated. Thanks

4

Could you show the definition of abc_service fixture? Do you use worker_factory there?

What works for me is usually something along the lines of:

@pytest.fixture
def abc_service():
    return worker_factory(AbcService)

And then in the test case:

def test_xyz_method(abc_service):
    # abc_service is already an instance of service class, there's no need to instantiate it explicitly
    abc_service.xyz_rpc.perform_task.return_value = "whatever"
    abc_service.handle_task() # this calls the mocked proxy method
1 Like
5

The error you’ve posted looks like no mock has been applied. I can’t say for sure what the problem is without seeing your fixture definition, but if you just want to mock the return value of the RPC call, the worker_factory pattern @zsiciarz suggests is best.

The replace_dependencies fixture is for cases where you want to inject an alternative DependencyProvider object, whereas worker_factory skips dependency injection completely and replaces the dependency attributes on the service class directly.

1 Like
7

Thank you @zsiciarz and @mattbennett for the reply and sorry, for delayed reply, weekend and then different timezones.
Anyway, I tried what @zsiciarz suggested using worker_factory and yes, @mattbennett, I was using replace_dependencies in my fixture which I copied and edited from nameko example.

Thank you so much, it worked. I had feeling that I’m doing something wrong because this seems like a pretty common case.

Also, any suggestions on how to test db layer or database dependency class in case of mongodb. I’ve heard [mongomock] is a good library.

8

When writing and testing DependencyProviders I try to keep the interface with the service class as narrow as possible. Often that means injecting a wrapper class rather than the bare dependency itself.

A silly example:

You’re writing a service that manages a zoo. State about the animals is stored in Mongo. Rather than having a DependencyProvider that returned a bare mongo client, I’d return something with meaningful methods for the service:

DO THIS

class ZooService:
    name = "zoo"

    self.animals = Animals()  # DependencyProvider

    @entrypoint
    def method(self):
        all_mammals = self.animals.list_mammals()

NOT THIS

class ZooService:
    name = "zoo"

    self.mongo = MongoClient()  # DependencyProvider

    @entrypoint
    def method(self):
        all_mammals = self.mongo.animals.filter(type=="mammal")  # or whatever

If you follow this pattern it’s trivial to test your service method with mock responses using the worker_factory.

It’s much simpler to test a specific DependencyProvider in isolation too. If your equivalent of the Animals DP used Mongo it’d probably be quite easy to use mongomock in place of the real mongo client.

1 Like
9

Thank you again, for jolting an example again.
I see your point, and I’ve noticed same pattern in nameko example. I’m afraid I’m doing what you described not to do. I use this nameko-mongo dependency and import it directly into my service class, which is anti-pattern. I’ll refactor this as per suggested example. Thanks again.

and one last question, I can’t seem to find any docs or examples on mongomock, it’s just this [repo] and this stackoverflow answer. Am I overlooking something ?

10

I’ve not used mongomock before I’m afraid

1 Like