<commit_before># Module:   sockets
# Date:     26th June 2006
# Author:   <NAME>, prologic at shortcircuit dot net dot au

"""Sockets Test Suite

Test all functionality of the sockets module.
"""

import unittest
from time import sleep

from circuits import Component
from circuits.workers import Thread
from circuits.lib.sockets import TCPServer, TCPClient, UDPServer, UDPClient

class Manager(Thread):

    client = None
    server = None
    events = 0

    def run(self):
        while self.running:
            self.flush()
            self.server.poll()
            self.client.poll()

class Client(Component):

    channel = "client"

    connected = False
    disconnected = False
    error = None
    data = ""

    def connected(self, host, port):
        self.connected = True

    def disconnected(self):
        self.disconnected = True

    def read(self, data):
        self.data = data

    def error(self, error):
        self.error = error

class Server(Component):

    channel = "server"

    data = ""
    errors = {}
    server = None

    def connected(self, sock, host, port):
        self.server.write(sock, "Ready")

    def read(self, sock, data):
        self.data = data

    def error(self, sock, error):
        self.errors[sock] = error

class SocketsTestCase(unittest.TestCase):

    def testTCPClientServer(self):
        """Test sockets.TCPClient and sockets.TCPServer

        Test that communication between a TCPClient and
        TCPServer work correctly.
        """

        manager = Manager()
        tcpserver = TCPServer(9999, channel="server")
        tcpclient = TCPClient(channel="client")
        server = Server()
        client = Client()
        manager += tcpserver
        manager += tcpclient
        manager += server
        manager += client
        manager.server = tcpserver
        server.server = tcpserver
        manager.client = tcpclient

        manager.start()

        try:
            tcpclient.open("localhost", 9999)
            sleep(0.1)
            self.assertTrue(client.connected)

            self.assertTrue(client.data == "Ready")

            tcpclient.write("foo")
            sleep(0.1)
            self.assertTrue(server.data == "foo")

            tcpclient.close()
            sleep(0.1)
            self.assertTrue(client.disconnected)
        finally:
            manager.stop()
            manager.join()

    def testUDPClientServer(self):
        """Test sockets.UDPClient and sockets.UDPServer

        Test that communication between a UDPClient and
        UDPServer work correctly.
        """

        manager = Manager()
        udpserver = UDPServer(9999, channel="server")
        udpclient = UDPClient(10000, channel="client")
        server = Server()
        client = Client()
        manager += udpserver
        manager += udpclient
        manager += server
        manager += client
        manager.server = udpserver
        server.server = udpserver
        manager.client = udpclient

        manager.start()

        try:
            udpclient.connected = True
            #udpclient.open("localhost", 9999)
            #sleep(0.1)
            #self.assertTrue(client.connected)

            udpclient.write(("localhost", 9999), "foo")
            sleep(0.1)
            self.assertTrue(server.data == "foo")
        finally:
            manager.stop()
            manager.join()


def suite():
    return unittest.makeSuite(SocketsTestCase, "test")

if __name__ == "__main__":
    unittest.main()
<commit_msg>tests.test_sockets: Refactored these tests to not use a Manager or seprate channels. Created generic Server/Client components that accept any of the socket components as their first argument<commit_after># Module:   sockets
# Date:     26th June 2006
# Author:   <NAME>, prologic at shortcircuit dot net dot au

"""Sockets Test Suite

Test all functionality of the sockets module.
"""

import unittest

from circuits import Component
from circuits.lib.sockets import TCPServer, TCPClient, UDPServer, UDPClient

def wait():
    for x in xrange(100000):
        pass

class Client(Component):

    def __init__(self, client):
        super(Client, self).__init__()

        self.client = client
        self.manager += self.client

        self.connected = False
        self.disconnected = False
        self.data = ""

    def connected(self, host, port):
        self.connected = True

    def disconnected(self):
        self.disconnected = True

    def read(self, data):
        self.data = data

class Server(Component):

    def __init__(self, server):
        super(Server, self).__init__()

        self.server = server
        self.manager += self.server

        self.data = ""

    def connected(self, sock, host, port):
        self.server.write(sock, "Ready")

    def read(self, sock, data):
        self.data = data

class SocketsTestCase(unittest.TestCase):

    def testTCPClientServer(self):
        """Test sockets.TCPClient and sockets.TCPServer

        Test that communication between a TCPClient and
        TCPServer work correctly.
        """

        server = Server(TCPServer(9999))
        client = Client(TCPClient())

        server.start()
        client.start()

        try:
            client.client.open("localhost", 9999)
            wait()

            self.assertTrue(client.connected)
            self.assertTrue(client.data == "Ready")

            client.client.write("foo")
            wait()

            self.assertTrue(server.data == "foo")

            client.client.close()
            wait()

            self.assertTrue(client.disconnected)
        finally:
            server.stop()
            client.stop()

    def testUDPClientServer(self):
        """Test sockets.UDPClient and sockets.UDPServer

        Test that communication between a UDPClient and
        UDPServer work correctly.
        """

        server = Server(UDPServer(9999))
        client = Client(UDPClient(10000))

        server.start()
        client.start()

        try:
            client.client.write(("localhost", 9999), "foo")
            wait()

            self.assertTrue(server.data == "foo")
        finally:
            server.stop()
            client.stop()


def suite():
    return unittest.makeSuite(SocketsTestCase, "test")

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

{'fraction_non_alphanumeric': 0.10036294776708221, 'fraction_numerical': 0.011835253274420072, 'mean_word_length': 2.096238397655105, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '495416', 'n_tokens_mistral': 1787, 'n_tokens_neox': 1688, 'n_words': 459}