<commit_before># -*- coding: utf-8 -*-

# This file is part of the Rocket Web Server
# Copyright (c) 2010 <NAME>
#
# See the included LICENSE.txt file for licensing details.

# Import System Modules
import time
import socket
import unittest
import threading
try:
    from queue import Queue
except ImportError:
    from Queue import Queue

# Import Custom Modules
from rocket import monitor, listener, connection

# Constants

# Define Tests
class MonitorTest(unittest.TestCase):
    def setUp(self):
        self.active_queue = Queue()
        self.monitor_queue = Queue()
        self.timeout = 1
        self.interface = ("127.0.0.1", 45451)

    def testNotActive(self):
        self.monitor = monitor.Monitor(self.monitor_queue,
                                       self.active_queue,
                                       self.timeout)

        self.assert_(not self.monitor.active)

    def testMonitor(self):
        self.testNotActive() # create self.monitor
        
        self.monitor.start()
        
        # Start the listener
        self.listener = listener.Listener(self.interface,
                                          5,
                                          self.active_queue)
        self.listener.start()
        
        # Create a socket connecting to listener's port
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        sock.connect(self.interface)

        # Verify that listener put it in the active queue
        time.sleep(0.5)
        self.assertEqual(self.active_queue.qsize(), 1)
        
        # Put it in the monitor queue
        conn = self.active_queue.get()
        conn = connection.Connection(*conn)
        self.monitor_queue.put(conn)
        self.assertEqual(self.monitor_queue.qsize(), 1)
        
        # Wait for the monitor queue to see it
        attempts = 20
        while attempts > 0 and self.monitor_queue.qsize():
            time.sleep(0.5)
            attempts -= 1
        self.assertEqual(self.monitor_queue.qsize(), 0)
        
        # Send something to the socket to see if it get put back on the active
        # queue.
        sock.send("test data")
        
        # Give monitor a chance to see it
        time.sleep(0.5)
        
        # Finally check to make sure that it's on the active queue
        self.assertEqual(self.active_queue.qsize(), 1)
        conn2 = self.active_queue.get()
        self.assert_(conn is conn2)
    
    def tearDown(self):
        try:<commit_msg>Make test_monitor.py a little more flexible and robust.<commit_after># -*- coding: utf-8 -*-

# This file is part of the Rocket Web Server
# Copyright (c) 2010 <NAME>
#
# See the included LICENSE.txt file for licensing details.

# Import System Modules
import time
import types
import socket
import unittest
import threading
try:
    from queue import Queue
except ImportError:
    from Queue import Queue

# Import Custom Modules
from rocket import monitor, listener, connection

# Constants

# Define Tests
class MonitorTest(unittest.TestCase):
    def setUp(self):
        self.active_queue = Queue()
        self.monitor_queue = Queue()
        self.timeout = 1
        self.interface = ("127.0.0.1", 45451)

    def _waitForEqual(self, a, b):
        attempts = 80
        while attempts > 0:
            if isinstance(a, types.FunctionType):
                _a = a()
            else:
                _a = a
            if isinstance(b, types.FunctionType):
                _b = b()
            else:
                _b = b
            if _a == _b:
                return True
            time.sleep(0.25)
            attempts -= 1
        return False
        
    def testNotActive(self):
        self.monitor = monitor.Monitor(self.monitor_queue,
                                       self.active_queue,
                                       self.timeout)

        self.assert_(not self.monitor.active)

    def testMonitor(self):
        self.testNotActive() # create self.monitor
        
        self.monitor.start()
        
        # Start the listener
        self.listener = listener.Listener(self.interface,
                                          5,
                                          self.active_queue)
        self.listener.start()
        
        # Create a socket connecting to listener's port
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        sock.connect(self.interface)

        # Verify that listener put it in the active queue
        self._waitForEqual(self.active_queue.qsize, 1)
        self.assertEqual(self.active_queue.qsize(), 1)
        
        # Put it in the monitor queue
        conn = self.active_queue.get()
        conn = connection.Connection(*conn)
        self.monitor_queue.put(conn)
        self.assertEqual(self.monitor_queue.qsize(), 1)
        
        # Wait for the monitor queue to see it
        self._waitForEqual(self.monitor_queue.qsize, 0)
        self.assertEqual(self.monitor_queue.qsize(), 0)
        
        # Send something to the socket to see if it get put back on the active
        # queue.
        sock.send("test data")
        
        # Give monitor a chance to see it
        self._waitForEqual(self.active_queue.qsize, 1)
        
        # Finally check to make sure that it's on the active queue
        self.assertEqual(self.active_queue.qsize(), 1)
        conn2 = self.active_queue.get()
        self.assert_(conn is conn2)
    
    def tearDown(self):
        try:
{'fraction_non_alphanumeric': 0.08259373301938054, 'fraction_numerical': 0.012678862524904909, 'mean_word_length': 1.7582417582417582, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6849076', 'n_tokens_mistral': 1529, 'n_tokens_neox': 1463, 'n_words': 431}