OXIESEC PANEL

Name	Size	Modified	Perms
📁 ..	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 __init__.py	112 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📁 __pycache__	-	03/31/2022 06:22:39 AM	rwxr-xr-x
📄 _awaittests.py.3only	4.37 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 _posixifaces.py	4.58 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 _win32ifaces.py	3.79 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 _yieldfromtests.py.3only	4.17 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 connectionmixins.py	19.71 KB	09/08/2017 10:38:36 AM	rw-r--r--
📁 fake_CAs	-	03/31/2022 06:22:38 AM	rwxr-xr-x
📄 fakeendpoint.py	1.52 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 modulehelpers.py	1.69 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_cli.py	664 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_connectionlost.py	119 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_gireactornocompat.py	721 bytes	09/08/2017 10:38:36 AM	rw-r--r--
📄 process_helper.py	1.06 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 reactormixins.py	13.72 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_abstract.py	1.98 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_address.py	11.45 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_base.py	10.98 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_baseprocess.py	2.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_core.py	11 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_coroutines.py	1.67 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_default.py	3.46 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_endpoints.py	143.15 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_epollreactor.py	7.16 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_fdset.py	13.18 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_filedescriptor.py	2.74 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_gireactor.py	8.83 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_glibbase.py	2.23 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_inlinecb.py	4.55 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_inotify.py	16.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_iocp.py	5.05 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_kqueuereactor.py	1.87 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_main.py	1.4 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_newtls.py	6.54 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_pollingfile.py	1.25 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_posixbase.py	9.31 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_posixprocess.py	10.9 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_process.py	31.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_protocol.py	18.29 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_resolver.py	19.58 KB	09/08/2017 10:38:35 AM	rw-r--r--
📄 test_serialport.py	1.9 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_sigchld.py	3.84 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_socket.py	9.11 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_stdio.py	6.3 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tcp.py	83 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_threads.py	7.93 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_time.py	3.63 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_tls.py	12.32 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_udp.py	16.74 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_udp_internals.py	4.79 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_unix.py	33.94 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_win32events.py	6.35 KB	09/08/2017 10:38:36 AM	rw-r--r--
📄 test_win32serialport.py	5.16 KB	09/08/2017 10:38:35 AM	rw-r--r--

Editing: test_win32serialport.py

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

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

import os
import shutil
import tempfile

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

try:
    from twisted.internet import serialport
    import serial
except ImportError as e:
    serialport = None
    serial = None

if serialport is not None:
    class RegularFileSerial(serial.Serial):
        def __init__(self, *args, **kwargs):
            super(RegularFileSerial, self).__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(RegularFileSerialPort, self).__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 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)