Repository: opensistemas-hub/osbrain
Pull request #126 (2017-06-01 09:35:05)

Add test that triggers a bug in SYNC_PUB/SYNC_SUB channels
The bug happens when specifying a handler of the client class when making a request through a SYNC_PUB/SYNC_SUB channel.

codecov[bot]: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=h1) Report
> Merging [#126](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/04b900551d1af14c73621144c118022a8e9b09d2?src=pr&el=desc) will **increase** coverage by `0.01%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126/graphs/tree.svg?width=650&height=150&src=pr&token=LfvMoNOrOc)](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #126      +/-   ##
==========================================
+ Coverage   97.64%   97.65%   +0.01%     
==========================================
  Files           8        8              
  Lines        1188     1195       +7     
  Branches      177      180       +3     
==========================================
+ Hits         1160     1167       +7     
  Misses         20       20              
  Partials        8        8
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `95.85% <100%> (+0.04%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=footer). Last update [04b9005...54f6fb6](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

codecov[bot]: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=h1) Report
> Merging [#126](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/04b900551d1af14c73621144c118022a8e9b09d2?src=pr&el=desc) will **increase** coverage by `0.01%`.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126/graphs/tree.svg?token=LfvMoNOrOc&src=pr&width=650&height=150)](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #126      +/-   ##
==========================================
+ Coverage   97.64%   97.65%   +0.01%     
==========================================
  Files           8        8              
  Lines        1188     1195       +7     
  Branches      177      180       +3     
==========================================
+ Hits         1160     1167       +7     
  Misses         20       20              
  Partials        8        8
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `95.85% <100%> (+0.04%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=footer). Last update [04b9005...54f6fb6](https://codecov.io/gh/opensistemas-hub/osbrain/pull/126?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/osbrain/agent.py b/osbrain/agent.py
index 4ea8527..b77e811 100644
--- a/osbrain/agent.py
+++ b/osbrain/agent.py
@@ -566,7 +566,7 @@ class Agent():
         method_type = (types.MethodType, types.BuiltinMethodType)
         if isinstance(handler, method_type):
             return unbound_method(handler)
-        raise TypeError('Unknow handler type "%s"' % type(handler))
+        raise TypeError('Unknown handler type "%s"' % type(handler))
 
     def registered(self, address):
         return address in self.socket
@@ -902,7 +902,12 @@ class Agent():
             self.log_warning(error)
             return
         handler = self._pending_requests.pop(uuid)
-        handler(self, response)
+
+        if isinstance(handler, str):
+            handler = getattr(self, handler)
+            handler(response)
+        else:
+            handler(self, response)
 
     def subscribe(self, alias: str, handlers: Dict[Union[bytes, str], Any]):
         """
@@ -914,7 +919,7 @@ class Agent():
             Alias of the new subscriber socket.
         handlers
             A dictionary in which the keys represent the different topics
-            and the values the actual handlers. If ,instead of a dictionary,
+            and the values the actual handlers. If, instead of a dictionary,
             a single handler is given, it will be used to subscribe the agent
             to any topic.
         """
@@ -1308,30 +1313,28 @@ class Agent():
             return self._send_channel_async_rep(channel=channel,
                                                 message=message,
                                                 wait=wait,
-                                                on_error=on_error)
+                                                on_error=on_error,
+                                                handler=handler)
         if kind == 'SYNC_PUB':
-            if topic is None:
-                topic = ''
             return self._send_channel_sync_pub(channel=channel,
                                                message=message,
                                                topic=topic)
         if kind == 'SYNC_SUB':
-            address = channel.receiver
-            address_uuid = self._async_req_uuid[address]
-            request_uuid = unique_identifier()
-            self._pending_requests[request_uuid] = handler
-            message = (address_uuid, request_uuid, message)
-            self._send_address(channel.sender, message)
-            self._wait_received(wait, uuid=request_uuid, on_error=on_error)
-            return
+            return self._send_channel_sync_sub(channel, message, topic,
+                                               handler, wait, on_error)
+
         raise NotImplementedError('Unsupported channel kind %s!' % kind)
 
-    def _send_channel_async_rep(self, channel, message, wait, on_error):
+    def _send_channel_async_rep(self, channel, message, wait, on_error,
+                                handler=None):
         address = channel.receiver
         address_uuid = self._async_req_uuid[address]
         request_uuid = unique_identifier()
-        self._pending_requests[request_uuid] = \
-            self._async_req_handler[address_uuid]
+        if handler is not None:
+            self._pending_requests[request_uuid] = handler
+        else:
+            self._pending_requests[request_uuid] = \
+                self._async_req_handler[address_uuid]
         receiver_address = self.address[address_uuid]
         message = (address_uuid, request_uuid, message, receiver_address)
         message = serialize_message(message=message,
@@ -1339,9 +1342,12 @@ class Agent():
         self.socket[channel].send(message)
         self._wait_received(wait, uuid=request_uuid, on_error=on_error)
 
-    def _send_channel_sync_pub(self, channel, message, topic, general=True):
+    def _send_channel_sync_pub(self, channel, message, topic=None,
+                               general=True):
         message = serialize_message(message=message,
                                     serializer=channel.serializer)
+        if topic is None:
+            topic = ''
         if isinstance(topic, str):
             topic = topic.encode()
         if general:
@@ -1351,6 +1357,19 @@ class Agent():
                                   serializer=channel.serializer)
         self.socket[channel].send(message)
 
+    def _send_channel_sync_sub(self, channel, message, topic, handler, wait,
+                               on_error):
+        address = channel.receiver
+        address_uuid = self._async_req_uuid[address]
+        request_uuid = unique_identifier()
+        if handler is None:
+            raise ValueError('No handler for SYNC_PUB request')
+        self._pending_requests[request_uuid] = handler
+        message = (address_uuid, request_uuid, message)
+        self._send_address(channel.sender, message)
+        self._wait_received(wait, uuid=request_uuid, on_error=on_error)
+        return
+
     def _check_received(self, uuid, wait, on_error):
         """
         Check if the requested information has been received.


Test patch:
diff --git a/osbrain/tests/test_agent_async_requests_handlers.py b/osbrain/tests/test_agent_async_requests_handlers.py
new file mode 100644
index 0000000..b7a2535
--- /dev/null
+++ b/osbrain/tests/test_agent_async_requests_handlers.py
@@ -0,0 +1,98 @@
+'''
+Test file for asynchronous requests handlers.
+'''
+import time
+
+import pytest
+
+from osbrain import Agent
+from osbrain import run_agent
+
+from common import nsproxy  # pragma: no flakes
+
+
+class Server_ASYNC_REP(Agent):
+    def on_init(self):
+        self.received = []
+        self.bind('ASYNC_REP', alias='publish', handler='reply')
+
+    def reply(self, request):
+        self.received.append(request)
+        return 'reply!'
+
+
+class ClientWithHandler(Agent):
+    def on_init(self):
+        self.received = []
+
+    def crash_handler(self, response):
+        raise Exception()
+
+    def receive_method(self, response):
+        self.received.append(response)
+
+
+def receive_function(agent, response):
+    agent.received.append(response)
+
+
+@pytest.mark.parametrize(
+    'params',
+    [('receive_method', False),
+     (receive_function, True),
+     (lambda a, x: a.received.append(x), False)])
+def test_connect_handler_types(nsproxy, params):
+    '''
+    We should be able to specify the handler in the `connect` call in
+    different ways: method, functions, lambda expressions...
+
+    Note that this handler will be used if not overriden by the `handler`
+    parameter in the `send` call. However, that is specifically checked in
+    other test.
+    '''
+    handler, check_function = params
+
+    server = run_agent('server', base=Server_ASYNC_REP)
+    client = run_agent('client', base=ClientWithHandler)
+
+    addr = server.addr('publish')
+
+    client.connect(addr, alias='sub', handler=handler)
+    client.send('sub', 'request!')
+    time.sleep(1)
+    assert len(client.get_attr('received')) == 1
+
+    if check_function:
+        # Check that the function was not stored as a method for the object
+        with pytest.raises(Exception):
+            assert client.get_attr('receive_function')
+
+
+@pytest.mark.parametrize(
+    'params',
+    [('receive_method', False),
+     (receive_function, True),
+     (lambda a, x: a.received.append(x), False)])
+def test_send_handler_types(nsproxy, params):
+    '''
+    We should be able to override the handler in the `send` call in
+    different ways: method, functions...
+    '''
+    handler, check_function = params
+
+    server = run_agent('server', base=Server_ASYNC_REP)
+    client = run_agent('client', base=ClientWithHandler)
+
+    addr = server.addr('publish')
+
+    # Default handler should not be called when specifying a handler on `send`
+    client.connect(addr, alias='sub', handler='crash_handler')
+
+    client.send('sub', 'request!', handler=handler)
+    time.sleep(1)
+    assert len(client.get_attr('received')) == 1
+
+    if check_function:
+        # Check that the function was not stored as a method for the object
+        with pytest.raises(Exception):
+            assert client.get_attr('receive_function')
diff --git a/osbrain/tests/test_agent_sync_publications_handlers.py b/osbrain/tests/test_agent_sync_publications_handlers.py
new file mode 100644
index 0000000..4056972
--- /dev/null
+++ b/osbrain/tests/test_agent_sync_publications_handlers.py
@@ -0,0 +1,105 @@
+'''
+Test file for synchronized publications handlers.
+'''
+import time
+
+import pytest
+
+from osbrain import Agent
+from osbrain import run_agent
+
+from common import nsproxy  # pragma: no flakes
+
+
+class Server_SYNC_PUB(Agent):
+    def on_init(self):
+        self.received = []
+        self.bind('SYNC_PUB', alias='publish', handler='reply')
+
+    def reply(self, request):
+        self.received.append(request)
+        return 'reply!'
+
+    def publish(self):
+        self.send('publish', 'publication!')
+
+
+class ClientWithHandler(Agent):
+    def on_init(self):
+        self.received = []
+
+    def crash_handler(self, response):
+        raise Exception()
+
+    def receive_method(self, response):
+        self.received.append(response)
+
+
+def receive_function(agent, response):
+    agent.received.append(response)
+
+
+@pytest.mark.parametrize(
+    'params',
+    [('receive_method', False),
+     (receive_function, True),
+     (lambda a, x: a.received.append(x), False)])
+def test_connect_handler_types(nsproxy, params):
+    '''
+    The handler for the normal PUB/SUB communication is specified in the
+    `connect` call.
+
+    We should be able to specify this in various ways: method, functions,
+    lambda expressions...
+    '''
+    handler, check_function = params
+
+    server = run_agent('server', base=Server_SYNC_PUB)
+    client = run_agent('client', base=ClientWithHandler)
+
+    addr = server.addr('publish')
+
+    client.connect(addr, alias='sub', handler=handler)
+    server.publish()
+    time.sleep(1)
+    assert len(client.get_attr('received')) == 1
+
+    if check_function:
+        # Check that the function was not stored as a method for the object
+        with pytest.raises(Exception):
+            assert client.get_attr('receive_function')
+
+
+@pytest.mark.parametrize(
+    'params',
+    [('receive_method', False, False),
+     (receive_function, True, False),
+     (lambda a, x: a.received.append(x), False, False),
+     (None, False, True)])
+def test_sync_pub_send_handlers(nsproxy, params):
+    '''
+    The handler for the requests MUST be specified in the `send` call.
+    It can be specified in different ways: method, functions...
+    '''
+    handler, check_function, should_crash = params
+
+    server = run_agent('server', base=Server_SYNC_PUB)
+    client = run_agent('client', base=ClientWithHandler)
+
+    addr = server.addr('publish')
+
+    # PUB/SUB handler should not be used in the requests at all
+    client.connect(addr, alias='sub', handler='crash_handler')
+
+    if should_crash:
+        with pytest.raises(ValueError):
+            client.send('sub', 'request!')
+    else:
+        client.send('sub', 'request!', handler=handler)
+        time.sleep(1)
+        assert len(client.get_attr('received')) == 1
+
+        if check_function:
+            # Check that the function was not stored as a method for the object
+            with pytest.raises(Exception):
+                assert client.get_attr('receive_function')