# SPDX-License-Identifier: LGPL-2.1-or-later
#
# Copyright (C) 2011-2023 Red Hat, Inc.
#
# Author: Tony Asleson <tasleson@redhat.com>

import json
import socket
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):
    """
    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.

    <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.
    """

    HDR_LEN = 10

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

        if l < 1:
            raise ValueError("Trying to read less than 1 byte!")

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

        return data.decode("utf-8")

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

        if msg is None or len(msg) < 1:
            raise ValueError("Msg argument empty")

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

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

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

    @staticmethod
    def get_socket(path):
        """
        Returns a connected socket from the passed in path.
        """
        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,
                        "Permissions are incorrect for IPC "
                        "socket file")
            else:
                raise LsmError(ErrorNumber.PLUGIN_NOT_EXIST,
                               "Plug-in appears to not exist")
        except socket.error:
            # self, code, message, data=None, *args, **kwargs
            raise LsmError(ErrorNumber.PLUGIN_IPC_FAIL,
                           "Unable to connect to lsmd, daemon started?")
        return s

    def close(self):
        """
        Closes the transport and the underlying socket
        """
        self.s.close()

    def send_req(self, method, args):
        """
        Sends a request given a method and arguments.
        Note: arguments must be in the form that can be automatically
        serialized to json
        """
        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,
                           "Error while sending a message to the plug-in",
                           str(se))

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

    def rpc(self, method, args):
        """
        Sends a request and waits for a response.
        """
        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):
        """
        Used to transmit an error.
        """
        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):
        """
        Used to transmit a response
        """
        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):
    """
    Test echo server for test case.
    """
    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', ' ', '   ', '{}:""', "Some text message", '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 = "x" * l
            msg = {'method': 'drip', 'id': 100, 'params': payload}
            data = json.dumps(msg, cls=_DataEncoder)
            hdr = str(len(data))

            wire = hdr.zfill(TransPort.HDR_LEN) + data

            self.assertTrue(len(msg) >= 1)

            for i in wire:
                self.c.send(bytes(i, 'utf-8'))

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

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


if __name__ == "__main__":
    unittest.main()