FM / test

/usr/lib/python3/dist-packages/twisted/internet/test/

/usr/lib/python3/dist-packages/twisted/internet/test

Name	Size	Mode	Actions
fake_CAs/	-	0755	rm
__pycache__/	-	0755	rm
connectionmixins.py	20184	0644	edit dl rm
fakeendpoint.py	1663	0644	edit dl rm
modulehelpers.py	1672	0644	edit dl rm
process_cli.py	547	0644	edit dl rm
process_connectionlost.py	126	0644	edit dl rm
process_gireactornocompat.py	816	0644	edit dl rm
process_helper.py	1282	0644	edit dl rm
reactormixins.py	14609	0644	edit dl rm
test_abstract.py	1976	0644	edit dl rm
test_address.py	8266	0644	edit dl rm
test_asyncioreactor.py	9327	0644	edit dl rm
test_base.py	13684	0644	edit dl rm
test_baseprocess.py	2588	0644	edit dl rm
test_core.py	11125	0644	edit dl rm
test_default.py	3419	0644	edit dl rm
test_defer_await.py	6514	0644	edit dl rm
test_defer_yieldfrom.py	4848	0644	edit dl rm
test_endpoints.py	148720	0644	edit dl rm
test_epollreactor.py	7322	0644	edit dl rm
test_error.py	1090	0644	edit dl rm
test_fdset.py	13497	0644	edit dl rm
test_filedescriptor.py	2761	0644	edit dl rm
test_gireactor.py	6516	0644	edit dl rm
test_glibbase.py	2217	0644	edit dl rm
test_inlinecb.py	11482	0644	edit dl rm
test_inotify.py	18562	0644	edit dl rm
test_iocp.py	7700	0644	edit dl rm
test_kqueuereactor.py	1853	0644	edit dl rm
test_main.py	1331	0644	edit dl rm
test_newtls.py	6532	0644	edit dl rm
test_pollingfile.py	1297	0644	edit dl rm
test_posixbase.py	9718	0644	edit dl rm
test_posixprocess.py	11203	0644	edit dl rm
test_process.py	36732	0644	edit dl rm
test_protocol.py	18520	0644	edit dl rm
test_resolver.py	19610	0644	edit dl rm
test_serialport.py	1988	0644	edit dl rm
test_sigchld.py	3922	0644	edit dl rm
test_socket.py	9443	0644	edit dl rm
test_stdio.py	6597	0644	edit dl rm
test_tcp.py	106621	0644	edit dl rm
test_testing.py	16239	0644	edit dl rm
test_threads.py	8734	0644	edit dl rm
test_time.py	3758	0644	edit dl rm
test_tls.py	13349	0644	edit dl rm
test_udp.py	16881	0644	edit dl rm
test_udp_internals.py	4837	0644	edit dl rm
test_unix.py	34854	0644	edit dl rm
test_win32events.py	6473	0644	edit dl rm
test_win32serialport.py	5330	0644	edit dl rm
_posixifaces.py	4391	0644	edit dl rm
_win32ifaces.py	3975	0644	edit dl rm
__init__.py	112	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/internet/test/test_win32serialport.py (5330B)

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.internet.serialport}.
"""

import os
import shutil
import tempfile

from twisted.internet.protocol import Protocol
from twisted.internet.test.test_serialport import DoNothing
from twisted.python.failure import Failure
from twisted.python.runtime import platform
from twisted.trial import unittest

testingForced = "TWISTED_FORCE_SERIAL_TESTS" in os.environ

try:
    import serial  # type: ignore[import]

from twisted.internet import serialport
except ImportError:
    if testingForced:
        raise

serialport = None  # type: ignore[assignment]
    serial = None

if serialport is not None:

class RegularFileSerial(serial.Serial):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.captured_args = args
            self.captured_kwargs = kwargs

def _reconfigurePort(self):
            pass

def _reconfigure_port(self):
            pass

class RegularFileSerialPort(serialport.SerialPort):
        _serialFactory = RegularFileSerial

def __init__(self, *args, **kwargs):
            cbInQue = kwargs.get("cbInQue")

if "cbInQue" in kwargs:
                del kwargs["cbInQue"]

self.comstat = serial.win32.COMSTAT
            self.comstat.cbInQue = cbInQue

super().__init__(*args, **kwargs)

def _clearCommError(self):
            return True, self.comstat

class CollectReceivedProtocol(Protocol):
    def __init__(self):
        self.received_data = []

def dataReceived(self, data):
        self.received_data.append(data)

class Win32SerialPortTests(unittest.TestCase):
    """
    Minimal testing for Twisted's Win32 serial port support.
    """

if not testingForced:
        if not platform.isWindows():
            skip = "This test must run on Windows."

elif not serialport:
            skip = "Windows serial port support is not available."

def setUp(self):
        # Re-usable protocol and reactor
        self.protocol = Protocol()
        self.reactor = DoNothing()

self.directory = tempfile.mkdtemp()
        self.path = os.path.join(self.directory, "fake_serial")

data = b"1234"
        with open(self.path, "wb") as f:
            f.write(data)

def tearDown(self):
        shutil.rmtree(self.directory)

def test_serialPortDefaultArgs(self):
        """
        Test correct positional and keyword arguments have been
        passed to the C{serial.Serial} object.
        """
        port = RegularFileSerialPort(self.protocol, self.path, self.reactor)
        # Validate args
        self.assertEqual((self.path,), port._serial.captured_args)
        # Validate kwargs
        kwargs = port._serial.captured_kwargs
        self.assertEqual(9600, kwargs["baudrate"])
        self.assertEqual(serial.EIGHTBITS, kwargs["bytesize"])
        self.assertEqual(serial.PARITY_NONE, kwargs["parity"])
        self.assertEqual(serial.STOPBITS_ONE, kwargs["stopbits"])
        self.assertEqual(0, kwargs["xonxoff"])
        self.assertEqual(0, kwargs["rtscts"])
        self.assertEqual(None, kwargs["timeout"])
        port.connectionLost(Failure(Exception("Cleanup")))

def test_serialPortInitiallyConnected(self):
        """
        Test the port is connected at initialization time, and
        C{Protocol.makeConnection} has been called on the desired protocol.
        """
        self.assertEqual(0, self.protocol.connected)

port = RegularFileSerialPort(self.protocol, self.path, self.reactor)
        self.assertEqual(1, port.connected)
        self.assertEqual(1, self.protocol.connected)
        self.assertEqual(port, self.protocol.transport)
        port.connectionLost(Failure(Exception("Cleanup")))

def common_exerciseHandleAccess(self, cbInQue):
        port = RegularFileSerialPort(
            protocol=self.protocol,
            deviceNameOrPortNumber=self.path,
            reactor=self.reactor,
            cbInQue=cbInQue,
        )
        port.serialReadEvent()
        port.write(b"")
        port.write(b"abcd")
        port.write(b"ABCD")
        port.serialWriteEvent()
        port.serialWriteEvent()
        port.connectionLost(Failure(Exception("Cleanup")))

# No assertion since the point is simply to make sure that in all cases
        # the port handle resolves instead of raising an exception.

def test_exerciseHandleAccess_1(self):
        self.common_exerciseHandleAccess(cbInQue=False)

def test_exerciseHandleAccess_2(self):
        self.common_exerciseHandleAccess(cbInQue=True)

def common_serialPortReturnsBytes(self, cbInQue):
        protocol = CollectReceivedProtocol()

port = RegularFileSerialPort(
            protocol=protocol,
            deviceNameOrPortNumber=self.path,
            reactor=self.reactor,
            cbInQue=cbInQue,
        )
        port.serialReadEvent()
        self.assertTrue(all(isinstance(d, bytes) for d in protocol.received_data))
        port.connectionLost(Failure(Exception("Cleanup")))

def test_serialPortReturnsBytes_1(self):
        self.common_serialPortReturnsBytes(cbInQue=False)

def test_serialPortReturnsBytes_2(self):
        self.common_serialPortReturnsBytes(cbInQue=True)

cmd: