Repository: opensistemas-hub/osbrain
Pull request #42 (2017-02-03 12:12:55)

Serialization - in reference to issue #36

coveralls: 
[![Coverage Status](https://coveralls.io/builds/9979140/badge)](https://coveralls.io/builds/9979140)

Coverage increased (+0.08%) to 96.536% when pulling **d470ce5a8959b6d84c6da1341fb030392b022a1a on Flood1993:serialization** into **92a7c1d4d0364585ae864148291523665dc076ad on opensistemas-hub:master**.

codecov-io: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?src=pr&el=h1) Report
> Merging [#42](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/92a7c1d4d0364585ae864148291523665dc076ad?src=pr&el=desc) will **increase** coverage by `-0.41%`.


```diff
@@            Coverage Diff             @@
##           master      #42      +/-   ##
==========================================
- Coverage   94.74%   94.34%   -0.41%     
==========================================
  Files           7        7              
  Lines         819      866      +47     
  Branches      118      135      +17     
==========================================
+ Hits          776      817      +41     
- Misses         29       30       +1     
- Partials       14       19       +5
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/__init__.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/92a7c1d4d0364585ae864148291523665dc076ad...d470ce5a8959b6d84c6da1341fb030392b022a1a?src=pr&el=tree#diff-b3NicmFpbi9fX2luaXRfXy5weQ==) | `100% <100%> (ø)` | :white_check_mark: |
| [osbrain/address.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/92a7c1d4d0364585ae864148291523665dc076ad...d470ce5a8959b6d84c6da1341fb030392b022a1a?src=pr&el=tree#diff-b3NicmFpbi9hZGRyZXNzLnB5) | `95.83% <75%> (-1.95%)` | :x: |
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/92a7c1d4d0364585ae864148291523665dc076ad...d470ce5a8959b6d84c6da1341fb030392b022a1a?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `90.35% <88%> (-0.52%)` | :x: |
| [osbrain/nameserver.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/92a7c1d4d0364585ae864148291523665dc076ad...d470ce5a8959b6d84c6da1341fb030392b022a1a?src=pr&el=tree#diff-b3NicmFpbi9uYW1lc2VydmVyLnB5) | `99.09% <ø> (+1.8%)` | :white_check_mark: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?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/42?src=pr&el=footer). Last update [92a7c1d...33cfa23](https://codecov.io/gh/opensistemas-hub/osbrain/compare/92a7c1d4d0364585ae864148291523665dc076ad...33cfa238054267c5bbb1b1b5a40f3be8eb6824dc?el=footer&src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
coveralls: 
[![Coverage Status](https://coveralls.io/builds/10010653/badge)](https://coveralls.io/builds/10010653)

Coverage decreased (-0.3%) to 96.11% when pulling **33cfa238054267c5bbb1b1b5a40f3be8eb6824dc on Flood1993:serialization** into **92a7c1d4d0364585ae864148291523665dc076ad on opensistemas-hub:master**.

Flood1993: Lowered the complexity of the functions. Also, the tests for checking message passing without serialization now use raw ZMQ sockets.

However, there is an issue which I have not been able to track down. In order for the tests to run OK, I've had to add explicit calls to close the ZMQ socket and destroy the ZMQ context at the end of each of the following tests (located in osbrain/tests/test_agent.py):
- test_reqrep_without_serialization
- test_pushpull_without_serialization
- test_pubsub_without_serialization

The code that manages the freeing of the resources are the lines:
`socket.close()`
`context.destroy()`

In order to reproduce the issue, first, checkout this commit 33cfa23. Then, run `pytest -k _without_serialization`. The tests should run successfully. If you remove the resource freeing calls from each function, the tests won't pass (I'm getting stuck after passing successfully test_pushpull_without_serialization and test_reqrep_without_serialization, but when running the last one, test_pubsub_without_serialization, it hangs up).

Note that that without the resource freeing calls, each test runs OK individually.
coveralls: 
[![Coverage Status](https://coveralls.io/builds/10048703/badge)](https://coveralls.io/builds/10048703)

Coverage decreased (-0.5%) to 95.991% when pulling **ab3b212a702a55a7f111de8482fd81cbe21cc550 on Flood1993:serialization** into **92a7c1d4d0364585ae864148291523665dc076ad on opensistemas-hub:master**.

coveralls: 
[![Coverage Status](https://coveralls.io/builds/10048725/badge)](https://coveralls.io/builds/10048725)

Coverage decreased (-0.6%) to 95.991% when pulling **bd439c83ad9eb792c26be213796d537347a7464a on Flood1993:serialization** into **3d00404223035db38514b22f76af43ca4809e0b8 on opensistemas-hub:master**.

Peque: Thanks for the changes!

Would you mind...?

- Removing the `pytest` file.
- Moving agent tests to a new file (i.e.: `test_agent_serialization.py`).
- Having a look at why lines 847, 848, 864, 865 are not being covered.

Thanks a lot! :blush:
coveralls: 
[![Coverage Status](https://coveralls.io/builds/10051600/badge)](https://coveralls.io/builds/10051600)

Coverage decreased (-0.5%) to 96.105% when pulling **e7f8c599ec0ec293b12aa25375372f1fd5a6c569 on Flood1993:serialization** into **437c303ecf846a04a1d25fab2da1eb050ae73328 on opensistemas-hub:master**.

codecov-io: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?src=pr&el=h1) Report
> Merging [#42](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/948411356f25922ed7b00a4ff2754ec5b18f9948?src=pr&el=desc) will **not impact** coverage by `-0.83%`.


```diff
@@            Coverage Diff             @@
##           master      #42      +/-   ##
==========================================
- Coverage   94.74%   93.92%   -0.83%     
==========================================
  Files           7        7              
  Lines         819      873      +54     
  Branches      118      134      +16     
==========================================
+ Hits          776      820      +44     
- Misses         29       34       +5     
- Partials       14       19       +5
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/__init__.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/948411356f25922ed7b00a4ff2754ec5b18f9948...e7f8c599ec0ec293b12aa25375372f1fd5a6c569?src=pr&el=tree#diff-b3NicmFpbi9fX2luaXRfXy5weQ==) | `100% <100%> (ø)` | :white_check_mark: |
| [osbrain/address.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/948411356f25922ed7b00a4ff2754ec5b18f9948...e7f8c599ec0ec293b12aa25375372f1fd5a6c569?src=pr&el=tree#diff-b3NicmFpbi9hZGRyZXNzLnB5) | `95.83% <75%> (-1.95%)` | :x: |
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/948411356f25922ed7b00a4ff2754ec5b18f9948...e7f8c599ec0ec293b12aa25375372f1fd5a6c569?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `89.63% <83.05%> (-1.24%)` | :x: |
| [osbrain/nameserver.py](https://codecov.io/gh/opensistemas-hub/osbrain/compare/948411356f25922ed7b00a4ff2754ec5b18f9948...e7f8c599ec0ec293b12aa25375372f1fd5a6c569?src=pr&el=tree#diff-b3NicmFpbi9uYW1lc2VydmVyLnB5) | `99.09% <ø> (+1.8%)` | :white_check_mark: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/42?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/42?src=pr&el=footer). Last update [9484113...fb100f8](https://codecov.io/gh/opensistemas-hub/osbrain/compare/948411356f25922ed7b00a4ff2754ec5b18f9948...fb100f88e666f88cfd8fb3aa37137ab023e037ed?src=pr&el=footer&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
coveralls: 
[![Coverage Status](https://coveralls.io/builds/10055608/badge)](https://coveralls.io/builds/10055608)

Coverage increased (+0.1%) to 96.564% when pulling **1b209f26f1b2321a2729968a4e3c34c54a524cf5 on Flood1993:serialization** into **948411356f25922ed7b00a4ff2754ec5b18f9948 on opensistemas-hub:master**.


Patch:
diff --git a/docs/source/index.rst b/docs/source/index.rst
index d0f0605..1ca91ed 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -38,6 +38,7 @@ Contents
    tutorial-2.rst
    security.rst
    developers.rst
+   serialization.rst
    api.rst
    changelog.rst
    license.rst
diff --git a/docs/source/serialization.rst b/docs/source/serialization.rst
new file mode 100644
index 0000000..10cef7e
--- /dev/null
+++ b/docs/source/serialization.rst
@@ -0,0 +1,27 @@
+.. index:: serialization
+
+.. _serialization:
+
+********
+Serialization
+********
+
+Serialization in osBrain
+========================
+Data needs to be serialized in order to be sent from one osBrain agent to
+another.
+
+osBrain has two options for serializing the data. One is through the use
+of :py:mod:`pickle` and the other one is the `'raw'` option, in which bytes
+are directly sent.
+
+For internal communication, `pickle` is the default option. The programmer
+might specify which option to use when binding a socket.
+
+osBrain uses :py:mod:`pickle` module for serialization when passing messages
+between agents internally and can use :py:mod:`pickle` and :py:mod:`dill` for
+serialization when configuring and deploying the multi-agent architectures.
+It is well known that using pickle or dill for this purpose is a security risk.
+The main problem is that allowing a program to unpickle or undill arbitrary
+data can cause arbitrary code execution and this may wreck or compromise your
+system.
\ No newline at end of file
diff --git a/osbrain/__init__.py b/osbrain/__init__.py
index 849106b..4970fd3 100644
--- a/osbrain/__init__.py
+++ b/osbrain/__init__.py
@@ -10,6 +10,7 @@ Pyro4.config.COMMTIMEOUT = 0.
 Pyro4.config.DETAILED_TRACEBACK = True
 os.environ['OSBRAIN_DEFAULT_TRANSPORT'] = 'ipc'
 os.environ['OSBRAIN_DEFAULT_SAFE'] = 'true'
+os.environ['OSBRAIN_DEFAULT_SERIALIZER'] = 'pickle'
 
 __version__ = '0.3.1'
 
diff --git a/osbrain/address.py b/osbrain/address.py
index 7be8d9d..b59da36 100644
--- a/osbrain/address.py
+++ b/osbrain/address.py
@@ -171,6 +171,26 @@ class AgentAddressKind(int):
         return self.__class__(self.ZMQ_KIND_TWIN[self])
 
 
+class AgentAddressSerializer(str):
+    """
+    Agent's address serializer class.
+
+    Each communication channel will have a serializer.
+
+    Note that for `raw` message passing, everything must be on bytes, and the
+    programmer is the one responsible for converting data to bytes.
+
+    Parameters
+    ----------
+    serializer_type : str
+        Serializer type (i.e.: 'raw', 'pickle'...).
+    """
+    def __new__(cls, value):
+        if value not in ('raw', 'pickle'):
+            raise ValueError('Invalid serializer type %s!' % value)
+        return super().__new__(cls, value)
+
+
 class SocketAddress(object):
     """
     Socket address information consisting on the host and port.
@@ -229,6 +249,8 @@ class AgentAddress():
         Agent kind.
     role : str, AgentAddressRole
         Agent role.
+    serializer : str
+        Agent serializer type.
 
     Attributes
     ----------
@@ -238,14 +260,17 @@ class AgentAddress():
         Agent kind.
     role : AgentAddressRole
         Agent role.
+    serializer : AgentAddressSerializer
+        Agent serializer.
     """
-    def __init__(self, transport, address, kind, role):
+    def __init__(self, transport, address, kind, role, serializer):
         if transport == 'tcp':
             address = SocketAddress(*address_to_host_port(address))
         self.transport = AgentAddressTransport(transport)
         self.address = address
         self.kind = AgentAddressKind(kind)
         self.role = AgentAddressRole(role)
+        self.serializer = AgentAddressSerializer(serializer)
 
     def __repr__(self):
         """
@@ -255,8 +280,9 @@ class AgentAddress():
         -------
         representation : str
         """
-        return 'AgentAddress(%s, %s, %s, %s)' % \
-            (self.transport, self.address, self.kind, self.role)
+        return 'AgentAddress(%s, %s, %s, %s, %s)' % \
+            (self.transport, self.address, self.kind, self.role,
+             self.serializer)
 
     def __hash__(self):
         return hash(self.transport) ^ hash(self.address) ^ \
@@ -268,7 +294,8 @@ class AgentAddress():
         return self.transport == other.transport \
             and self.address == other.address \
             and self.kind == other.kind \
-            and self.role == other.role
+            and self.role == other.role \
+            and self.serializer == other.serializer
 
     def twin(self):
         """
@@ -282,4 +309,5 @@ class AgentAddress():
         """
         kind = self.kind.twin()
         role = self.role.twin()
-        return self.__class__(self.transport, self.address, kind, role)
+        return self.__class__(self.transport, self.address, kind, role,
+                              self.serializer)
diff --git a/osbrain/agent.py b/osbrain/agent.py
index 0d6fa31..31d70b6 100644
--- a/osbrain/agent.py
+++ b/osbrain/agent.py
@@ -53,8 +53,8 @@ class Agent():
         A dictionary in which the key is the address or the alias and the
         value is the actual socket.
     adddress : dict
-        A dictionary in which the key is the alias and the value is the
-        actual address.
+        A dictionary in which the key is the address or the alias and the
+        value is the actual address.
     handler : dict
         A dictionary in which the key is the socket and the values are the
         handlers for each socket.
@@ -87,20 +87,16 @@ class Agent():
         self.context = zmq.Context()
         self.poller = zmq.Poller()
 
-        # This in-process socket could, eventually, handle safe access to
-        # memory from other threads (i.e. when using Pyro proxies).
-        socket = self.context.socket(zmq.REP)
-        address = 'inproc://loopback'
-        socket.bind(address)
-        self.register(socket, address, 'loopback', self._handle_loopback)
+        # A loopback socket where, for example, timers are processed
+        self.bind('REP', alias='loopback', addr='loopback',
+                  handler=self._handle_loopback, transport='inproc',
+                  serializer='pickle')
 
         # This in-process socket handles safe access to
         # memory from other threads (i.e. when using Pyro proxies).
-        socket = self.context.socket(zmq.REP)
-        address = 'inproc://_loopback_safe'
-        socket.bind(address)
-        self.register(socket, address, '_loopback_safe',
-                      self._handle_loopback_safe)
+        self.bind('REP', alias='_loopback_safe', addr='_loopback_safe',
+                  handler=self._handle_loopback_safe, transport='inproc',
+                  serializer='pickle')
 
         self.on_init()
 
@@ -389,6 +385,7 @@ class Agent():
         self.socket[alias] = socket
         self.socket[address] = socket
         self.address[alias] = address
+        self.address[address] = address
         if handler is not None:
             self.poller.register(socket, zmq.POLLIN)
             self._set_handler(socket, handler)
@@ -429,7 +426,8 @@ class Agent():
     def registered(self, address):
         return address in self.socket
 
-    def bind(self, kind, alias=None, handler=None, addr=None, transport=None):
+    def bind(self, kind, alias=None, handler=None, addr=None, transport=None,
+             serializer=None):
         """
         Bind to an agent address.
 
@@ -457,8 +455,10 @@ class Agent():
             'This socket requires a handler!'
         socket = self.context.socket(kind)
         transport = transport or os.environ.get('OSBRAIN_DEFAULT_TRANSPORT')
+        serializer = serializer or os.getenv('OSBRAIN_DEFAULT_SERIALIZER')
         addr = self._bind_socket(socket, addr=addr, transport=transport)
-        server_address = AgentAddress(transport, addr, kind, 'server')
+        server_address = AgentAddress(transport, addr, kind, 'server',
+                                      serializer)
         self.register(socket, server_address, alias, handler)
         # SUB sockets are a special case
         if kind == 'SUB':
@@ -701,6 +701,18 @@ class Agent():
         else:
             self._process_nonsub_event(socket_kind, socket, serialized)
 
+    def _deserialize_message(self, serializer, message):
+        if serializer == 'pickle':
+            return pickle.loads(message)
+        return message
+
+    def _process_rep_event(self, socket_kind, socket, handler_return,
+                           serializer):
+        if socket_kind == 'REP' and handler_return is not None:
+            if serializer == 'pickle':
+                handler_return = pickle.dumps(handler_return, -1)
+            socket.send(handler_return)
+
     def _process_nonsub_event(self, socket_kind, socket, serialized):
         """
         Process a non-SUB socket's event.
@@ -714,15 +726,27 @@ class Agent():
         serialized : bytes
             Data received on the socket.
         """
-        message = pickle.loads(serialized)
+        serializer = self._get_serialization_from_socket(socket)
+
+        message = self._deserialize_message(serializer, serialized)
+
         handlers = self.handler[socket]
         if not isinstance(handlers, list):
             handlers = [handlers]
         for handler in handlers:
             handler_return = handler(self, message)
-        if socket_kind == 'REP':
-            if handler_return is not None:
-                socket.send_pyobj(handler_return)
+
+        self._process_rep_event(socket_kind, socket, handler_return,
+                                serializer)
+
+    def _process_sub_message(self, serializer, message):
+        if serializer == 'pickle':
+            # Since SUB messages might have a `topic`, we need to be careful to
+            # only deserialize the non-topic part. Pickle objects always start
+            # with b'\x80', so the pure message part will start from that char.
+            sepp = message.index(b'\x80')
+            message = memoryview(message)[sepp:]
+        return self._deserialize_message(serializer, message)
 
     def _process_sub_event(self, socket, serialized):
         """
@@ -736,9 +760,11 @@ class Agent():
             Data received on the socket.
         """
         handlers = self.handler[socket]
-        sepp = serialized.index(b'\x80')
-        data = memoryview(serialized)[sepp:]
-        message = pickle.loads(data)
+
+        serializer = self._get_serialization_from_socket(socket)
+
+        message = self._process_sub_message(serializer, serialized)
+
         for str_topic in handlers:
             btopic = self.str2bytes(str_topic)
             if not serialized.startswith(btopic):
@@ -751,19 +777,56 @@ class Agent():
             elif nparams == 3:
                 handler(self, message, str_topic)
 
+    def _get_serialization_from_socket(self, socket):
+        """
+        Parameters
+        ----------
+        socket : zmq.Socket
+            Socket which we want to know the type of serialization from.
+
+        Return
+        ------
+        AgentAddressSerializer
+            The serializer type for that socket.
+        """
+        agent_address = None
+        for k, v in self.socket.items():
+            if v is socket:
+                agent_address = k
+
+        serializer = None
+        if isinstance(agent_address, str):
+            serializer = self.address[agent_address].serializer
+        else:
+            serializer = agent_address.serializer
+
+        return serializer
+
     def str2bytes(self, message):
         return message.encode('ascii')
 
     def send(self, address, message, topic=''):
+        """
+        TODO
+        """
         assert isinstance(topic, str), 'Topic must be of `str` type!'
-        serialized = pickle.dumps(message, -1)
+        serializer = self.address[address].serializer
+        if serializer == 'pickle':
+            serialized = pickle.dumps(message, -1)
+        elif serializer == 'raw':
+            serialized = message
         topic = self.str2bytes(topic)
         self.socket[address].send(topic + serialized)
 
     def recv(self, address):
-        serialized = self.socket[address].recv()
-        deserialized = pickle.loads(serialized)
-        return deserialized
+        """
+        TODO
+
+        This method is only used in REQREP communication patterns.
+        """
+        message = self.socket[address].recv()
+        serializer = self.address[address].serializer
+        return self._deserialize_message(serializer, message)
 
     def send_recv(self, address, message):
         self.send(address, message)
@@ -803,8 +866,8 @@ class Agent():
 
     def close_sockets(self):
         for address in self.socket:
-            if address in ('loopback', '_loopback_safe',
-                           'inproc://loopback', 'inproc://_loopback_safe'):
+            if address in ('loopback', '_loopback_safe', 'inproc://loopback',
+                           'inproc://_loopback_safe'):
                 continue
             self.socket[address].close()
 


Test patch:
diff --git a/osbrain/tests/test_address.py b/osbrain/tests/test_address.py
index 6f4b6e8..09eff51 100644
--- a/osbrain/tests/test_address.py
+++ b/osbrain/tests/test_address.py
@@ -12,6 +12,7 @@ from osbrain.address import AgentAddress
 from osbrain.address import AgentAddressKind
 from osbrain.address import AgentAddressRole
 from osbrain.address import AgentAddressTransport
+from osbrain.address import AgentAddressSerializer
 
 
 def twin_list(elements):
@@ -32,7 +33,8 @@ def test_invalid_address_to_host_port():
 
 @pytest.mark.parametrize(('address', 'host', 'port'), [
     (None, None, None),
-    (AgentAddress('tcp', '127.0.0.1:123', 'PUSH', 'server'), '127.0.0.1', 123),
+    (AgentAddress('tcp', '127.0.0.1:123', 'PUSH', 'server', 'pickle'),
+     '127.0.0.1', 123),
     (SocketAddress('127.0.0.1', 123), '127.0.0.1', 123),
     ('127.0.0.1:123', '127.0.0.1', 123),
     ('127.0.0.1', '127.0.0.1', None),
@@ -109,6 +111,17 @@ def test_role():
         AgentAddressRole(1)
 
 
+def test_serializer():
+    """
+    This test aims to cover basic AgentAddressSerializer initialization and
+    equivalence.
+    """
+    assert AgentAddressSerializer('raw') == 'raw'
+    assert AgentAddressSerializer('pickle') == 'pickle'
+    with pytest.raises(ValueError):
+        AgentAddressSerializer('foo')
+
+
 def test_socket_address():
     """
     Test basic SocketAddress operations: initialization and equivalence.
@@ -127,18 +140,36 @@ def test_agent_address():
     Test basic AgentAddress operations: initialization, equivalence and
     basic methods.
     """
-    address = AgentAddress('ipc', 'addr', 'PUSH', 'server')
+    address = AgentAddress('ipc', 'addr', 'PUSH', 'server', 'pickle')
+    # Equivalence
+    assert address == AgentAddress('ipc', 'addr', 'PUSH', 'server', 'pickle')
+    assert not address == 'foo'
+    assert address != 'foo'
+    # Basic methods
+    assert address.twin() == AgentAddress('ipc', 'addr', 'PULL', 'client',
+                                          'pickle')
+
+
+def test_agent_address_explicit_serializer():
+    """
+    Test basic AgentAddress operations: initialization, equivalence and
+    basic methods when an explicit serializer is used.
+    """
+    address = AgentAddress('ipc', 'addr', 'PUSH', 'server', 'raw')
     # Equivalence
-    assert address == AgentAddress('ipc', 'addr', 'PUSH', 'server')
+    assert address == AgentAddress('ipc', 'addr', 'PUSH', 'server', 'raw')
     assert not address == 'foo'
     assert address != 'foo'
     # Basic methods
-    assert address.twin() == AgentAddress('ipc', 'addr', 'PULL', 'client')
+    assert address.twin() == AgentAddress('ipc', 'addr', 'PULL', 'client',
+                                          'raw')
+    assert address.twin() != AgentAddress('ipc', 'addr', 'PULL', 'client',
+                                          'pickle')
 
 
 def test_agent_address_to_host_port():
     """
     An agent address should be convertible to host+port if TCP is used.
     """
-    address = AgentAddress('tcp', '127.0.0.1:1234', 'PUSH', 'server')
+    address = AgentAddress('tcp', '127.0.0.1:1234', 'PUSH', 'server', 'pickle')
     assert address_to_host_port(address) == ('127.0.0.1', 1234)
diff --git a/osbrain/tests/test_agent.py b/osbrain/tests/test_agent.py
index dd63f3e..62cac1f 100644
--- a/osbrain/tests/test_agent.py
+++ b/osbrain/tests/test_agent.py
@@ -12,6 +12,7 @@ import pytest
 from osbrain import run_logger
 from osbrain import run_agent
 from osbrain import Agent
+from osbrain import AgentAddress
 from osbrain import AgentProcess
 from osbrain import Proxy
 from osbrain import NSProxy
@@ -66,7 +67,9 @@ def test_agent_loopback(nsaddr):
     An agent should always have a _loopback_safe inproc socket.
     """
     a0 = run_agent('a0')
-    assert a0.addr('_loopback_safe') == 'inproc://_loopback_safe'
+    assert a0.addr('_loopback_safe') == \
+        AgentAddress(transport='inproc', address='_loopback_safe',
+                     kind='REP', role='server', serializer='pickle')
 
 
 def test_ping(nsaddr):
diff --git a/osbrain/tests/test_agent_serialization.py b/osbrain/tests/test_agent_serialization.py
new file mode 100644
index 0000000..74349d7
--- /dev/null
+++ b/osbrain/tests/test_agent_serialization.py
@@ -0,0 +1,189 @@
+import time
+
+import zmq
+
+from osbrain import run_agent
+
+from common import nsaddr  # pragma: no flakes
+from common import nsproxy  # pragma: no flakes
+
+
+def set_received(agent, message, topic=None):
+    agent.received = message
+
+
+def test_reqrep_raw(nsaddr):
+    """
+    Simple request-reply pattern between two agents with raw serialization.
+    """
+    def rep_handler(agent, message):
+        return b'OK'
+
+    a0 = run_agent('a0')
+    a1 = run_agent('a1')
+    addr = a0.bind('REP', 'reply', rep_handler, serializer='raw')
+    a1.connect(addr, 'request')
+    response = a1.send_recv('request', b'Hello world')
+    assert response == b'OK'
+
+
+def test_reqrep_pickle(nsaddr):
+    """
+    Simple request-reply pattern between two agents with pickle serialization.
+    """
+    def rep_handler(agent, message):
+        return 'OK'
+
+    a0 = run_agent('a0')
+    a1 = run_agent('a1')
+    addr = a0.bind('REP', 'reply', rep_handler, serializer='pickle')
+    a1.connect(addr, 'request')
+    response = a1.send_recv('request', 'Hello world')
+    assert response == 'OK'
+
+
+def test_reqrep_raw_zmq_outside(nsaddr):
+    """
+    Simple request-reply pattern between an agent and a direct ZMQ connection.
+    """
+    def rep_handler(agent, message):
+        return message
+
+    # Create an osBrain agent that will receive the message
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a1.bind('REP', transport='tcp', handler=rep_handler,
+                   serializer='raw')
+
+    # Create a raw ZeroMQ REQ socket
+    context = zmq.Context()
+    socket = context.socket(zmq.REQ)
+    socket.connect('tcp://%s:%s' % (addr.address.host, addr.address.port))
+
+    # Send the message
+    message = b'Hello world'
+    socket.send(message)
+    assert socket.recv() == message
+
+    socket.close()
+    context.destroy()
+
+
+def test_pushpull_raw(nsaddr):
+    """
+    Simple push-pull pattern test, using raw serialization between agents.
+    """
+    a0 = run_agent('a0')
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a1.bind('PULL', handler=set_received, serializer='raw')
+    a0.connect(addr, 'push')
+    message = b'Hello world'
+    a0.send('push', message)
+    while not a1.get_attr('received'):
+        time.sleep(0.01)
+    assert a1.get_attr('received') == message
+
+
+def test_pushpull_pickle(nsaddr):
+    """
+    Simple push-pull pattern test with pickle serialization.
+    """
+    a0 = run_agent('a0')
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a1.bind('PULL', handler=set_received, serializer='pickle')
+    a0.connect(addr, 'push')
+    message = 'Hello world'
+    a0.send('push', message)
+    while not a1.get_attr('received'):
+        time.sleep(0.01)
+    assert a1.get_attr('received') == message
+
+
+def test_pushpull_raw_zmq_outside(nsaddr):
+    """
+    Simple push-pull pattern test. Channel without serialization.
+
+    The message is sent from outside osBrain, through a ZMQ PUSH socket.
+    """
+    # Create an osBrain agent that will receive the message
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a1.bind('PULL', transport='tcp', handler=set_received,
+                   serializer='raw')
+
+    # Create a raw ZeroMQ PUSH socket
+    context = zmq.Context()
+    socket = context.socket(zmq.PUSH)
+    socket.connect('tcp://%s:%s' % (addr.address.host, addr.address.port))
+
+    # Send the message
+    message = b'Hello world'
+    socket.send(message)
+    while not a1.get_attr('received'):
+        time.sleep(0.01)
+    assert a1.get_attr('received') == message
+
+    socket.close()
+    context.destroy()
+
+
+def test_pubsub_raw(nsaddr):
+    """
+    Simple publisher-subscriber pattern test.
+    """
+    a0 = run_agent('a0')
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a0.bind('PUB', alias='pub', serializer='raw')
+    a1.connect(addr, handler=set_received)
+    message = b'Hello world'
+    while not a1.get_attr('received'):
+        a0.send('pub', message)
+        time.sleep(0.1)
+    assert a1.get_attr('received') == message
+
+
+def test_pubsub_pickle(nsaddr):
+    """
+    Simple publisher-subscriber pattern test with pickle serialization.
+    """
+    a0 = run_agent('a0')
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a0.bind('PUB', alias='pub', serializer='pickle')
+    a1.connect(addr, handler=set_received)
+    message = 'Hello world'
+    while not a1.get_attr('received'):
+        a0.send('pub', message)
+        time.sleep(0.1)
+    assert a1.get_attr('received') == message
+
+
+def test_pubsub_raw_zmq_outside(nsaddr):
+    """
+    Simple publisher-subscriber pattern test. Channel without serialization.
+
+    The message is sent from outside osBrain, through a ZMQ PUB socket.
+    """
+    # Create an osBrain agent that will receive the message
+    a1 = run_agent('a1')
+    a1.set_attr(received=None)
+    addr = a1.bind('SUB', transport='tcp', handler=set_received,
+                   serializer='raw')
+
+    # Create a raw ZeroMQ PUB socket
+    context = zmq.Context()
+    socket = context.socket(zmq.PUB)
+    socket.connect('tcp://%s:%s' % (addr.address.host, addr.address.port))
+
+    # Send the message
+    message = b'Hello world'
+    while not a1.get_attr('received'):
+        socket.send(message)
+        time.sleep(0.01)
+    assert a1.get_attr('received') == message
+
+    socket.close()
+    context.destroy()