<commit_before>#!/usr/bin/env python

import unittest
import sys
import os.path
import SocketServer
import threading
import socket
import signal

from copy import copy

sys.path.append(os.path.dirname(__file__) + "/../../..")

from pox.lib.epoll_select import EpollSelect

class TCPEcho(SocketServer.StreamRequestHandler):
  def handle(self):
    data = self.rfile.readline()
    print "got data: %s" % data
    self.wfile.write(data)

class ForkingTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
  def start(self):
    self.pid = os.fork()
    if self.pid == 0:
      # child
      self.serve_forever()

  def stop(self):
    os.kill(self.pid, signal.SIGKILL)

def sort_fdlists(rl,wl,xl) :
  key = lambda(x): x.fileno() if hasattr(x, "fileno") else x

  return (
            sorted(rl, key=key),
            sorted(wl, key=key),
            sorted(xl, key=key)
        )

class EpollSelectTest(unittest.TestCase):
  def setUp(self):
    self.es = EpollSelect()
    self.server = ForkingTCPServer(("localhost", 0), TCPEcho)
    self.ip, self.port = self.server.server_address
    self.server.start()

  def tearDown(self):
    self.es.close()
    self.server.stop()

  def test_create(self):
    pass

  def test_read_one_socket(self):
    c = socket.create_connection( (self.ip, self.port))
    ret  = self.es.select([c], [], [c], 0.1)
    self.assertEqual(([],[],[]), ret)
    # socket is ready to send?
    ret  = self.es.select([c], [c], [c], 0.1)<commit_msg>tests: Skip epoll tests unless on Linux<commit_after>#!/usr/bin/env python

import unittest
import sys
import os.path
import SocketServer
import threading
import socket
import signal

from copy import copy

sys.path.append(os.path.dirname(__file__) + "/../../..")

from pox.lib.epoll_select import EpollSelect

class TCPEcho(SocketServer.StreamRequestHandler):
  def handle(self):
    data = self.rfile.readline()
    print "got data: %s" % data
    self.wfile.write(data)

class ForkingTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
  def start(self):
    self.pid = os.fork()
    if self.pid == 0:
      # child
      self.serve_forever()

  def stop(self):
    os.kill(self.pid, signal.SIGKILL)

def sort_fdlists(rl,wl,xl) :
  key = lambda(x): x.fileno() if hasattr(x, "fileno") else x

  return (
            sorted(rl, key=key),
            sorted(wl, key=key),
            sorted(xl, key=key)
        )

@unittest.skipUnless(sys.platform.startswith("linux"), "requires Linux")
class EpollSelectTest(unittest.TestCase):
  def setUp(self):
    self.es = EpollSelect()
    self.server = ForkingTCPServer(("localhost", 0), TCPEcho)
    self.ip, self.port = self.server.server_address
    self.server.start()

  def tearDown(self):
    self.es.close()
    self.server.stop()

  def test_create(self):
    pass

  def test_read_one_socket(self):
    c = socket.create_connection( (self.ip, self.port))
    ret  = self.es.select([c], [], [c], 0.1)
    self.assertEqual(([],[],[]), ret)
    # socket is ready to send?
    ret  = self.es.select([c], [c], [c], 0.1)
{'fraction_non_alphanumeric': 0.14785737651292116, 'fraction_numerical': 0.003925417075564278, 'mean_word_length': 3.3009845288326303, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '111170', 'n_tokens_mistral': 1089, 'n_tokens_neox': 1075, 'n_words': 234}