<commit_before>"""Tests for events.py."""

import concurrent.futures
import gc
import os
import select
import signal
import socket
try:
    import ssl
except ImportError:
    ssl = None
import sys
import threading
import time
import unittest

from . import events
from . import transports
from . import protocols
from . import selectors
from . import unix_events


class MyProto(protocols.Protocol):
    def __init__(self):
        self.state = 'INITIAL'
        self.nbytes = 0
    def connection_made(self, transport):
        self.transport = transport
        assert self.state == 'INITIAL', self.state
        self.state = 'CONNECTED'
        transport.write(b'GET / HTTP/1.0\r\nHost: xkcd.com\r\n\r\n')
    def data_received(self, data):
        assert self.state == 'CONNECTED', self.state
        self.nbytes += len(data)
    def eof_received(self):
        assert self.state == 'CONNECTED', self.state
        self.state = 'EOF'
        self.transport.close()
    def connection_lost(self, exc):
        assert self.state in ('CONNECTED', 'EOF'), self.state
        self.state = 'CLOSED'


class EventLoopTestsMixin:

    def setUp(self):
        self.selector = self.SELECTOR_CLASS()
        self.event_loop = unix_events.UnixEventLoop(self.selector)
        events.set_event_loop(self.event_loop)

    def tearDown(self):
        self.event_loop.close()
        gc.collect()

    def testRun(self):
        el = events.get_event_loop()
        el.run()  # Returns immediately.

    def testCallLater(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        el.call_later(0.1, callback, 'hello world')
        t0 = time.monotonic()
        el.run()
        t1 = time.monotonic()
        self.assertEqual(results, ['hello world'])
        self.assertTrue(t1-t0 >= 0.09)

    def testCallRepeatedly(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        el.call_repeatedly(0.03, callback, 'ho')
        el.call_later(0.1, el.stop)
        el.run()
        self.assertEqual(results, ['ho', 'ho', 'ho'])

    def testCallSoon(self):
        el = events.get_event_loop()
        results = []
        def callback(arg1, arg2):
            results.append((arg1, arg2))
        el.call_soon(callback, 'hello', 'world')
        el.run()
        self.assertEqual(results, [('hello', 'world')])

    def testCallSoonWithHandler(self):
        el = events.get_event_loop()
        results = []
        def callback():
            results.append('yeah')
        handler = events.Handler(None, callback, ())
        self.assertEqual(el.call_soon(handler), handler)
        el.run()
        self.assertEqual(results, ['yeah'])

    def testCallSoonThreadsafe(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        def run():
            el.call_soon_threadsafe(callback, 'hello')
        t = threading.Thread(target=run)
        el.call_later(0.1, callback, 'world')
        t0 = time.monotonic()
        t.start()
        el.run()
        t1 = time.monotonic()
        t.join()
        self.assertEqual(results, ['hello', 'world'])
        self.assertTrue(t1-t0 >= 0.09)

    def testCallSoonThreadsafeWithHandler(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        handler = events.Handler(None, callback, ('hello',))
        def run():
            self.assertEqual(el.call_soon_threadsafe(handler), handler)
        t = threading.Thread(target=run)
        el.call_later(0.1, callback, 'world')
        t0 = time.monotonic()
        t.start()
        el.run()
        t1 = time.monotonic()
        t.join()
        self.assertEqual(results, ['hello', 'world'])
        self.assertTrue(t1-t0 >= 0.09)

    def testCallEveryIteration(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        handler = el.call_every_iteration(callback, 'ho')
        el.run_once()
        self.assertEqual(results, ['ho'])
        el.run_once()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])
        handler.cancel()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])

    def testCallEveryIterationWithHandler(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        handler = events.Handler(None, callback, ('ho',))
        self.assertEqual(el.call_every_iteration(handler), handler)
        el.run_once()
        self.assertEqual(results, ['ho'])
        el.run_once()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])
        handler.cancel()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])

    def testWrapFuture(self):
        el = events.get_event_loop()
        def run(arg):
            time.sleep(0.1)
            return arg
        ex = concurrent.futures.ThreadPoolExecutor(1)
        f1 = ex.submit(run, 'oi')
        f2 = el.wrap_future(f1)
        res = el.run_until_complete(f2)
        self.assertEqual(res, 'oi')

    def testRunInExecutor(self):
        el = events.get_event_loop()
        def run(arg):
            time.sleep(0.1)
            return arg
        f2 = el.run_in_executor(None, run, 'yo')
        res = el.run_until_complete(f2)
        self.assertEqual(res, 'yo')

    def testRunInExecutorWithHandler(self):
        el = events.get_event_loop()
        def run(arg):
            time.sleep(0.1)
            return arg
        handler = events.Handler(None, run, ('yo',))
        f2 = el.run_in_executor(None, handler)
        res = el.run_until_complete(f2)
        self.assertEqual(res, 'yo')

    def testReaderCallback(self):
        el = events.get_event_loop()
        r, w = unix_events.socketpair()
        bytes_read = []
        def reader():
            data = r.recv(1024)
            if data:
                bytes_read.append(data)
            else:
                self.assertTrue(el.remove_reader(r.fileno()))
                r.close()
        el.add_reader(r.fileno(), reader)
        el.call_later(0.05, w.send, b'abc')
        el.call_later(0.1, w.send, b'def')
        el.call_later(0.15, w.close)
        el.run()
        self.assertEqual(b''.join(bytes_read), b'abcdef')

    def testReaderCallbackWithHandler(self):
        el = events.get_event_loop()
        r, w = unix_events.socketpair()
        bytes_read = []
        def reader():
            data = r.recv(1024)
            if data:
                bytes_read.append(data)
            else:
                self.assertTrue(el.remove_reader(r.fileno()))
                r.close()
        handler = events.Handler(None, reader, ())
        self.assertEqual(el.add_reader(r.fileno(), handler), handler)
        el.call_later(0.05, w.send, b'abc')
        el.call_later(0.1, w.send, b'def')
        el.call_later(0.15, w.close)
        el.run()
        self.assertEqual(b''.join(bytes_read), b'abcdef')

    def testReaderCallbackCancel(self):
        el = events.get_event_loop()
        r, w = unix_events.socketpair()
        bytes_read = []
        def reader():
            data = r.recv(1024)
            if data:
                bytes_read.append(data)
            if sum(len(b) for b in bytes_read) >= 6:
                handler.cancel()
            if not data:
                r.close()
        handler = el.add_reader(r.fileno(), reader)
        el.call_later(0.05, w.send, b'abc')<commit_msg>Make tests allow for spurious readiness notifications.<commit_after>"""Tests for events.py."""

import concurrent.futures
import gc
import os
import select
import signal
import socket
try:
    import ssl
except ImportError:
    ssl = None
import sys
import threading
import time
import unittest

from . import events
from . import transports
from . import protocols
from . import selectors
from . import unix_events


class MyProto(protocols.Protocol):
    def __init__(self):
        self.state = 'INITIAL'
        self.nbytes = 0
    def connection_made(self, transport):
        self.transport = transport
        assert self.state == 'INITIAL', self.state
        self.state = 'CONNECTED'
        transport.write(b'GET / HTTP/1.0\r\nHost: xkcd.com\r\n\r\n')
    def data_received(self, data):
        assert self.state == 'CONNECTED', self.state
        self.nbytes += len(data)
    def eof_received(self):
        assert self.state == 'CONNECTED', self.state
        self.state = 'EOF'
        self.transport.close()
    def connection_lost(self, exc):
        assert self.state in ('CONNECTED', 'EOF'), self.state
        self.state = 'CLOSED'


class EventLoopTestsMixin:

    def setUp(self):
        self.selector = self.SELECTOR_CLASS()
        self.event_loop = unix_events.UnixEventLoop(self.selector)
        events.set_event_loop(self.event_loop)

    def tearDown(self):
        self.event_loop.close()
        gc.collect()

    def testRun(self):
        el = events.get_event_loop()
        el.run()  # Returns immediately.

    def testCallLater(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        el.call_later(0.1, callback, 'hello world')
        t0 = time.monotonic()
        el.run()
        t1 = time.monotonic()
        self.assertEqual(results, ['hello world'])
        self.assertTrue(t1-t0 >= 0.09)

    def testCallRepeatedly(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        el.call_repeatedly(0.03, callback, 'ho')
        el.call_later(0.1, el.stop)
        el.run()
        self.assertEqual(results, ['ho', 'ho', 'ho'])

    def testCallSoon(self):
        el = events.get_event_loop()
        results = []
        def callback(arg1, arg2):
            results.append((arg1, arg2))
        el.call_soon(callback, 'hello', 'world')
        el.run()
        self.assertEqual(results, [('hello', 'world')])

    def testCallSoonWithHandler(self):
        el = events.get_event_loop()
        results = []
        def callback():
            results.append('yeah')
        handler = events.Handler(None, callback, ())
        self.assertEqual(el.call_soon(handler), handler)
        el.run()
        self.assertEqual(results, ['yeah'])

    def testCallSoonThreadsafe(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        def run():
            el.call_soon_threadsafe(callback, 'hello')
        t = threading.Thread(target=run)
        el.call_later(0.1, callback, 'world')
        t0 = time.monotonic()
        t.start()
        el.run()
        t1 = time.monotonic()
        t.join()
        self.assertEqual(results, ['hello', 'world'])
        self.assertTrue(t1-t0 >= 0.09)

    def testCallSoonThreadsafeWithHandler(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        handler = events.Handler(None, callback, ('hello',))
        def run():
            self.assertEqual(el.call_soon_threadsafe(handler), handler)
        t = threading.Thread(target=run)
        el.call_later(0.1, callback, 'world')
        t0 = time.monotonic()
        t.start()
        el.run()
        t1 = time.monotonic()
        t.join()
        self.assertEqual(results, ['hello', 'world'])
        self.assertTrue(t1-t0 >= 0.09)

    def testCallEveryIteration(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        handler = el.call_every_iteration(callback, 'ho')
        el.run_once()
        self.assertEqual(results, ['ho'])
        el.run_once()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])
        handler.cancel()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])

    def testCallEveryIterationWithHandler(self):
        el = events.get_event_loop()
        results = []
        def callback(arg):
            results.append(arg)
        handler = events.Handler(None, callback, ('ho',))
        self.assertEqual(el.call_every_iteration(handler), handler)
        el.run_once()
        self.assertEqual(results, ['ho'])
        el.run_once()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])
        handler.cancel()
        el.run_once()
        self.assertEqual(results, ['ho', 'ho', 'ho'])

    def testWrapFuture(self):
        el = events.get_event_loop()
        def run(arg):
            time.sleep(0.1)
            return arg
        ex = concurrent.futures.ThreadPoolExecutor(1)
        f1 = ex.submit(run, 'oi')
        f2 = el.wrap_future(f1)
        res = el.run_until_complete(f2)
        self.assertEqual(res, 'oi')

    def testRunInExecutor(self):
        el = events.get_event_loop()
        def run(arg):
            time.sleep(0.1)
            return arg
        f2 = el.run_in_executor(None, run, 'yo')
        res = el.run_until_complete(f2)
        self.assertEqual(res, 'yo')

    def testRunInExecutorWithHandler(self):
        el = events.get_event_loop()
        def run(arg):
            time.sleep(0.1)
            return arg
        handler = events.Handler(None, run, ('yo',))
        f2 = el.run_in_executor(None, handler)
        res = el.run_until_complete(f2)
        self.assertEqual(res, 'yo')

    def testReaderCallback(self):
        el = events.get_event_loop()
        r, w = unix_events.socketpair()
        bytes_read = []
        def reader():
            try:
                data = r.recv(1024)
            except BlockingIOError:
                # Spurious readiness notifications are possible
                # at least on Linux -- see man select.
                return
            if data:
                bytes_read.append(data)
            else:
                self.assertTrue(el.remove_reader(r.fileno()))
                r.close()
        el.add_reader(r.fileno(), reader)
        el.call_later(0.05, w.send, b'abc')
        el.call_later(0.1, w.send, b'def')
        el.call_later(0.15, w.close)
        el.run()
        self.assertEqual(b''.join(bytes_read), b'abcdef')

    def testReaderCallbackWithHandler(self):
        el = events.get_event_loop()
        r, w = unix_events.socketpair()
        bytes_read = []
        def reader():
            try:
                data = r.recv(1024)
            except BlockingIOError:
                # Spurious readiness notifications are possible
                # at least on Linux -- see man select.
                return
            if data:
                bytes_read.append(data)
            else:
                self.assertTrue(el.remove_reader(r.fileno()))
                r.close()
        handler = events.Handler(None, reader, ())
        self.assertEqual(el.add_reader(r.fileno(), handler), handler)
        el.call_later(0.05, w.send, b'abc')
        el.call_later(0.1, w.send, b'def')
        el.call_later(0.15, w.close)
        el.run()
        self.assertEqual(b''.join(bytes_read), b'abcdef')

    def testReaderCallbackCancel(self):
        el = events.get_event_loop()
        r, w = unix_events.socketpair()
        bytes_read = []
        def reader():
            data = r.recv(1024)
            if data:
                bytes_read.append(data)
            if sum(len(b) for b in bytes_read) >= 6:
                handler.cancel()
            if not data:
                r.close()
        handler = el.add_reader(r.fileno(), reader)
        el.call_later(0.05, w.send, b'abc')
{'fraction_non_alphanumeric': 0.12919388596435594, 'fraction_numerical': 0.010908860277795395, 'mean_word_length': 2.0137614678899083, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, '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': '5284988', 'n_tokens_mistral': 4994, 'n_tokens_neox': 4840, 'n_words': 961}