Abstract out the network transport format again

Author: MetRonnie

cylc/flow/network/__init__.py

@@ -95,11 +95,19 @@ class ResponseDict(TypedDict, total=False):
         back-compat issues."""
 
 
-def load_server_response(message: str) -> 'ResponseDict':
+def serialize(data: object) -> str:
+    """Convert the structure holding a message to a JSON message string."""
+    # Abstract out the transport format in order to allow it to be changed
+    # in future.
+    return json.dumps(data)
+
+
+def deserialize(message: str) -> 'ResponseDict':
     """Convert a JSON message string to dict with an added 'user' field."""
+    # Abstract out the transport format in order to allow it to be changed
+    # in future.
     msg = json.loads(message)
-    if 'user' not in msg:
-        msg['user'] = getpass.getuser()  # assume this is the user
+    msg['user'] = getpass.getuser()  # assume this is the user
     return msg
 
 

cylc/flow/network/client.py

@@ -20,7 +20,6 @@
     abstractmethod,
 )
 import asyncio
-import json
 import os
 from shutil import which
 import socket
@@ -51,7 +50,8 @@
 from cylc.flow.network import (
     ZMQSocketBase,
     get_location,
-    load_server_response,
+    deserialize,
+    serialize,
 )
 from cylc.flow.network.client_factory import CommsMeth
 from cylc.flow.network.server import PB_METHOD_MAP
@@ -284,7 +284,7 @@ async def async_request(
         args: Optional[Dict[str, Any]] = None,
         timeout: Optional[float] = None,
         req_meta: Optional[Dict[str, Any]] = None
-    ) -> Union[bytes, object]:
+    ) -> object:
         """Send an asynchronous request using asyncio.
 
         Has the same arguments and return values as ``serial_request``.
@@ -306,7 +306,7 @@ async def async_request(
         if req_meta:
             msg['meta'].update(req_meta)
         LOG.debug('zmq:send %s', msg)
-        message = json.dumps(msg)
+        message = serialize(msg)
         self.socket.send_string(message)
 
         # receive response
@@ -326,7 +326,7 @@ async def async_request(
         if command in PB_METHOD_MAP:
             return res
 
-        response: ResponseDict = load_server_response(res.decode())
+        response: ResponseDict = deserialize(res.decode())
 
         try:
             return response['data']
@@ -338,7 +338,7 @@ async def async_request(
                     f"{response}"
                 )
                 wflow_cylc_ver = response.get('cylc_version')
-                if wflow_cylc_ver:
+                if wflow_cylc_ver and wflow_cylc_ver != CYLC_VERSION:
                     error += (
                         f"\n(Workflow is running in Cylc {wflow_cylc_ver})"
                     )

cylc/flow/network/replier.py

@@ -15,7 +15,6 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """Server for workflow runtime API."""
 
-import json
 from queue import Queue
 from typing import (
     TYPE_CHECKING,
@@ -30,7 +29,8 @@
 )
 from cylc.flow.network import (
     ZMQSocketBase,
-    load_server_response,
+    deserialize,
+    serialize,
 )
 
 
@@ -116,7 +116,7 @@ def listener(self) -> None:
             res: ResponseDict
             response: bytes
             try:
-                message = load_server_response(msg)
+                message = deserialize(msg)
             except Exception as exc:  # purposefully catch generic exception
                 # failed to decode message, possibly resulting from failed
                 # authentication
@@ -126,7 +126,7 @@ def listener(self) -> None:
                     'error': {'message': str(exc)},
                     'cylc_version': CYLC_VERSION,
                 }
-                response = json.dumps(res).encode()
+                response = serialize(res).encode()
             else:
                 # success case - serve the request
                 res = self.server.receiver(message)
@@ -135,5 +135,5 @@ def listener(self) -> None:
                 if isinstance(data, bytes):
                     response = data
                 else:
-                    response = json.dumps(res).encode()
+                    response = serialize(res).encode()
             self.socket.send(response)  # type: ignore[union-attr]

tests/integration/network/test_replier.py

@@ -21,7 +21,7 @@
 import pytest
 
 from cylc.flow import __version__ as CYLC_VERSION
-from cylc.flow.network import load_server_response
+from cylc.flow.network import deserialize
 from cylc.flow.network.client import WorkflowRuntimeClient
 from cylc.flow.scheduler import Scheduler
 
@@ -33,7 +33,7 @@ async def test_listener(one: Scheduler, start):
         # (without directly calling listener):
         client = WorkflowRuntimeClient(one.workflow)
         client.socket.send_string(r'Not JSON')
-        res = load_server_response(
+        res = deserialize(
             (await client.socket.recv()).decode()
         )
         assert res['error']