File: _debug.py

package info (click to toggle)
python-gvm 26.5.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 5,132 kB
  • sloc: python: 44,662; makefile: 18
file content (71 lines) | stat: -rw-r--r-- 1,884 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
# SPDX-FileCopyrightText: 2024 Greenbone AG
#
# SPDX-License-Identifier: GPL-3.0-or-later

import logging

from ._connection import GvmConnection

logger = logging.getLogger("gvm.connections.debug")


class DebugConnection:
    """Wrapper around a connection for debugging purposes

    Allows to debug the connection flow including send and read data. Internally
    it uses the python `logging <https://docs.python.org/3/library/logging.html>`_
    framework to create debug messages. Please take a look at
    `the logging tutorial <https://docs.python.org/3/howto/logging.html#logging-basic-tutorial>`_
    for further details.

    Example:

    .. code-block:: python

        import logging

        logging.basicConfig(level=logging.DEBUG)

        socket_connection = UnixSocketConnection(path='/var/run/gvm.sock')
        connection = DebugConnection(socket_connection)
        gmp = GMP(connection=connection)
    """

    def __init__(self, connection: GvmConnection):
        """
        Create a new DebugConnection instance.

        Args:
            connection: GvmConnection to observe
        """
        self._connection = connection

    def read(self) -> bytes:
        data = self._connection.read()

        logger.debug("Read %s characters. Data %r", len(data), data)

        self.last_read_data = data
        return data

    def send(self, data: bytes) -> None:
        self.last_send_data = data

        logger.debug("Sending %s characters. Data %r", len(data), data)

        return self._connection.send(data)

    def connect(self) -> None:
        logger.debug("Connecting")

        return self._connection.connect()

    def disconnect(self) -> None:
        logger.debug("Disconnecting")

        return self._connection.disconnect()

    def finish_send(self) -> None:
        logger.debug("Finish send")

        self._connection.finish_send()