<h2>Editing File: /usr/lib64/python3.6/site-packages/lsm/_transport.py</h2><form method='post'><textarea name='content' class='file-edit'># Copyright (C) 2011-2013 Red Hat, Inc.
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; If not, see &lt;http://www.gnu.org/licenses/&gt;.
#
# Author: tasleson

import json
import socket
import string
import os
import unittest
import threading

from lsm._common import LsmError, ErrorNumber
from lsm._common import SocketEOF as _SocketEOF
from lsm._data import DataDecoder as _DataDecoder
from lsm._data import DataEncoder as _DataEncoder

class TransPort(object):
    &quot;&quot;&quot;
    Provides wire serialization by using json.  Loosely conforms to json-rpc,
    however a length header was added so that we would have the ability to use
    non sax like json parsers, which are more abundant.

    &lt;Zero padded 10 digit number [1..2**32] for the length followed by
    valid json.

    Notes:
    id field (json-rpc) is present but currently not being used.
    This is available to be expanded on later.
    &quot;&quot;&quot;

    HDR_LEN = 10

    def _read_all(self, l):
        &quot;&quot;&quot;
        Reads l number of bytes before returning.  Will raise a SocketEOF
        if socket returns zero bytes (i.e. socket no longer connected)
        &quot;&quot;&quot;

        if l &lt; 1:
            raise ValueError(&quot;Trying to read less than 1 byte!&quot;)

        data = bytearray()
        while len(data) &lt; l:
            r = self.s.recv(l - len(data))
            if not r:
                raise _SocketEOF()
            data += r

        return data.decode(&quot;utf-8&quot;)

    def _send_msg(self, msg):
        &quot;&quot;&quot;
        Sends the json formatted message by pre-appending the length
        first.
        &quot;&quot;&quot;

        if msg is None or len(msg) &lt; 1:
            raise ValueError(&quot;Msg argument empty&quot;)

        # Note: Don't catch io exceptions at this level!
        s = str.zfill(str(len(msg)), self.HDR_LEN) + msg
        # common.Info(&quot;SEND: &quot;, msg)
        self.s.sendall(bytes(s.encode('utf-8')))

    def _recv_msg(self):
        &quot;&quot;&quot;
        Reads header first to get the length and then the remaining
        bytes of the message.
        &quot;&quot;&quot;
        try:
            l = self._read_all(self.HDR_LEN)
            msg = self._read_all(int(l))
            # common.Info(&quot;RECV: &quot;, msg)
        except socket.error as e:
            raise LsmError(ErrorNumber.TRANSPORT_COMMUNICATION,
                           &quot;Error while reading a message from the plug-in&quot;,
                           str(e))
        return msg

    def __init__(self, socket_descriptor):
        self.s = socket_descriptor

    @staticmethod
    def get_socket(path):
        &quot;&quot;&quot;
        Returns a connected socket from the passed in path.
        &quot;&quot;&quot;
        try:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

            if os.path.exists(path):
                if os.access(path, os.R_OK | os.W_OK):
                    s.connect(path)
                else:
                    raise LsmError(ErrorNumber.PLUGIN_SOCKET_PERMISSION,
                                   &quot;Permissions are incorrect for IPC &quot;
                                   &quot;socket file&quot;)
            else:
                raise LsmError(ErrorNumber.PLUGIN_NOT_EXIST,
                               &quot;Plug-in appears to not exist&quot;)
        except socket.error:
            # self, code, message, data=None, *args, **kwargs
            raise LsmError(ErrorNumber.PLUGIN_IPC_FAIL,
                           &quot;Unable to connect to lsmd, daemon started?&quot;)
        return s

    def close(self):
        &quot;&quot;&quot;
        Closes the transport and the underlying socket
        &quot;&quot;&quot;
        self.s.close()

    def send_req(self, method, args):
        &quot;&quot;&quot;
        Sends a request given a method and arguments.
        Note: arguments must be in the form that can be automatically
        serialized to json
        &quot;&quot;&quot;
        try:
            msg = {'method': method, 'id': 100, 'params': args}
            data = json.dumps(msg, cls=_DataEncoder)
            self._send_msg(data)
        except socket.error as se:
            raise LsmError(ErrorNumber.TRANSPORT_COMMUNICATION,
                           &quot;Error while sending a message to the plug-in&quot;,
                           str(se))

    def read_req(self):
        &quot;&quot;&quot;
        Reads a message and returns the parsed version of it.
        &quot;&quot;&quot;
        data = self._recv_msg()
        if len(data):
            # common.Info(str(data))
            return json.loads(data, cls=_DataDecoder)

    def rpc(self, method, args):
        &quot;&quot;&quot;
        Sends a request and waits for a response.
        &quot;&quot;&quot;
        self.send_req(method, args)
        (reply, msg_id) = self.read_resp()
        assert msg_id == 100
        return reply

    def send_error(self, msg_id, error_code, msg, data=None):
        &quot;&quot;&quot;
        Used to transmit an error.
        &quot;&quot;&quot;
        e = {'id': msg_id, 'error': {'code': error_code, 'message': msg,
                                     'data': data}}
        self._send_msg(json.dumps(e, cls=_DataEncoder))

    def send_resp(self, result, msg_id=100):
        &quot;&quot;&quot;
        Used to transmit a response
        &quot;&quot;&quot;
        r = {'id': msg_id, 'result': result}
        self._send_msg(json.dumps(r, cls=_DataEncoder))

    def read_resp(self):
        data = self._recv_msg()
        resp = json.loads(data, cls=_DataDecoder)

        if 'result' in resp:
            return resp['result'], resp['id']
        else:
            e = resp['error']
            raise LsmError(**e)


def _server(s):
    &quot;&quot;&quot;
    Test echo server for test case.
    &quot;&quot;&quot;
    srv = TransPort(s)

    msg = srv.read_req()

    try:
        while msg['method'] != 'done':

            if msg['method'] == 'error':
                srv.send_error(
                    msg['id'],
                    msg['params']['errorcode'],
                    msg['params']['errormsg'])
            else:
                srv.send_resp(msg['params'])
            msg = srv.read_req()
        srv.send_resp(msg['params'])
    finally:
        s.close()


class _TestTransport(unittest.TestCase):
    def setUp(self):
        (self.c, self.s) = socket.socketpair(
            socket.AF_UNIX, socket.SOCK_STREAM)

        self.client = TransPort(self.c)

        self.server = threading.Thread(target=_server, args=(self.s,))
        self.server.start()

    def test_simple(self):
        tc = ['0', ' ', '   ', '{}:&quot;&quot;', &quot;Some text message&quot;, 'DEADBEEF']

        for t in tc:
            self.client.send_req('test', t)
            reply, msg_id = self.client.read_resp()
            self.assertTrue(msg_id == 100)
            self.assertTrue(reply == t)

    def test_exceptions(self):

        e_msg = 'Test error message'
        e_code = 100

        self.client.send_req('error', {'errorcode': e_code, 'errormsg': e_msg})
        self.assertRaises(LsmError, self.client.read_resp)

        try:
            self.client.send_req('error', {'errorcode': e_code,
                                           'errormsg': e_msg})
            self.client.read_resp()
        except LsmError as e:
            self.assertTrue(e.code == e_code)
            self.assertTrue(e.msg == e_msg)

    def test_slow(self):

        # Try to test the receiver getting small chunks to read
        # in a loop
        for l in range(1, 4096, 10):

            payload = &quot;x&quot; * l
            msg = {'method': 'drip', 'id': 100, 'params': payload}
            data = json.dumps(msg, cls=_DataEncoder)

            wire = string.zfill(len(data), TransPort.HDR_LEN) + data

            self.assertTrue(len(msg) &gt;= 1)

            for i in wire:
                self.c.send(i)

            reply, msg_id = self.client.read_resp()
            self.assertTrue(payload == reply)

    def tearDown(self):
        self.client.send_req(&quot;done&quot;, None)
        resp, msg_id = self.client.read_resp()
        self.assertTrue(resp is None)
        self.server.join()


if __name__ == &quot;__main__&quot;:
    unittest.main()
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fusr%2Flib64%2Fpython3.6%2Fsite-packages%2Flsm'>Back to Directory</a>