<commit_before>"""
Tests for the crochet APIs.
"""

from __future__ import absolute_import

import threading
import subprocess
import time
import gc
import sys
import weakref
import tempfile
import os
import imp

from twisted.trial.unittest import TestCase
from twisted.internet.defer import succeed, Deferred, fail, CancelledError
from twisted.python.failure import Failure
from twisted.python import threadable
from twisted.python.runtime import platform
if platform.type == "posix":
    try:
        from twisted.internet.process import reapAllProcesses
    except (SyntaxError, ImportError):
        if sys.version_info < (3, 3, 0):
            raise
        else:
            # Process support is still not ported to Python 3 on some versions
            # of Twisted.
            reapAllProcesses = None
else:
    # waitpid() is only necessary on POSIX:
    reapAllProcesses = None

from .._eventloop import (EventLoop, EventualResult, TimeoutError,
                          ResultRegistry, ReactorStopped)
from .test_setup import FakeReactor
from .. import (_main, setup, in_reactor, retrieve_result, _store, no_setup,
                run_in_reactor, wait_for_reactor, wait_for)
from ..tests import crochet_directory


class ResultRegistryTests(TestCase):
    """
    Tests for ResultRegistry.
    """
    def test_stopped_registered(self):
        """
        ResultRegistery.stop() fires registered EventualResult with
        ReactorStopped.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(None, None)
        registry.register(er)
        registry.stop()
        self.assertRaises(ReactorStopped, er.wait, timeout=0)

    def test_stopped_new_registration(self):
        """
        After ResultRegistery.stop() is called subsequent register() calls
        raise ReactorStopped.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(None, None)
        registry.stop()
        self.assertRaises(ReactorStopped, registry.register, er)

    def test_stopped_already_have_result(self):
        """
        ResultRegistery.stop() has no impact on registered EventualResult
        which already have a result.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(succeed(123), None)
        registry.register(er)
        registry.stop()
        self.assertEqual(er.wait(), 123)
        self.assertEqual(er.wait(), 123)
        self.assertEqual(er.wait(), 123)

    def test_weakref(self):
        """
        Registering an EventualResult with a ResultRegistry does not prevent
        it from being garbage collected.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(None, None)
        registry.register(er)
        ref = weakref.ref(er)
        del er
        gc.collect()
        self.assertIdentical(ref(), None)

    def test_runs_with_lock(self):
        """
        All code in ResultRegistry.stop() and register() is protected by a
        lock.
        """
        self.assertTrue(ResultRegistry.stop.synchronized)
        self.assertTrue(ResultRegistry.register.synchronized)


class EventualResultTests(TestCase):
    """
    Tests for EventualResult.
    """

    def setUp(self):
        self.patch(threadable, "isInIOThread", lambda: False)

    def test_success_result(self):
        """
        wait() returns the value the Deferred fired with.
        """
        dr = EventualResult(succeed(123), None)
        self.assertEqual(dr.wait(), 123)

    def test_later_success_result(self):
        """
        wait() returns the value the Deferred fired with, in the case where
        the Deferred is fired after wait() is called.
        """
        d = Deferred()
        def fireSoon():
            import time; time.sleep(0.01)
            d.callback(345)
        threading.Thread(target=fireSoon).start()
        dr = EventualResult(d, None)
        self.assertEqual(dr.wait(), 345)

    def test_success_result_twice(self):
        """
        A second call to wait() returns same value as the first call.
        """
        dr = EventualResult(succeed(123), None)
        self.assertEqual(dr.wait(), 123)
        self.assertEqual(dr.wait(), 123)

    def test_failure_result(self):
        """
        wait() raises the exception the Deferred fired with.
        """
        dr = EventualResult(fail(RuntimeError()), None)
        self.assertRaises(RuntimeError, dr.wait)

    def test_later_failure_result(self):
        """
        wait() raises the exception the Deferred fired with, in the case
        where the Deferred is fired after wait() is called.
        """
        d = Deferred()
        def fireSoon():
            time.sleep(0.01)
            d.errback(RuntimeError())
        threading.Thread(target=fireSoon).start()
        dr = EventualResult(d, None)
        self.assertRaises(RuntimeError, dr.wait)<commit_msg>Don't operate in non-threadsafe manner that doesn't match real usage.<commit_after>"""
Tests for the crochet APIs.
"""

from __future__ import absolute_import

import threading
import subprocess
import time
import gc
import sys
import weakref
import tempfile
import os
import imp

from twisted.trial.unittest import TestCase
from twisted.internet.defer import succeed, Deferred, fail, CancelledError
from twisted.python.failure import Failure
from twisted.python import threadable
from twisted.python.runtime import platform
if platform.type == "posix":
    try:
        from twisted.internet.process import reapAllProcesses
    except (SyntaxError, ImportError):
        if sys.version_info < (3, 3, 0):
            raise
        else:
            # Process support is still not ported to Python 3 on some versions
            # of Twisted.
            reapAllProcesses = None
else:
    # waitpid() is only necessary on POSIX:
    reapAllProcesses = None

from .._eventloop import (EventLoop, EventualResult, TimeoutError,
                          ResultRegistry, ReactorStopped)
from .test_setup import FakeReactor
from .. import (_main, setup, in_reactor, retrieve_result, _store, no_setup,
                run_in_reactor, wait_for_reactor, wait_for)
from ..tests import crochet_directory


class ResultRegistryTests(TestCase):
    """
    Tests for ResultRegistry.
    """
    def test_stopped_registered(self):
        """
        ResultRegistery.stop() fires registered EventualResult with
        ReactorStopped.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(None, None)
        registry.register(er)
        registry.stop()
        self.assertRaises(ReactorStopped, er.wait, timeout=0)

    def test_stopped_new_registration(self):
        """
        After ResultRegistery.stop() is called subsequent register() calls
        raise ReactorStopped.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(None, None)
        registry.stop()
        self.assertRaises(ReactorStopped, registry.register, er)

    def test_stopped_already_have_result(self):
        """
        ResultRegistery.stop() has no impact on registered EventualResult
        which already have a result.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(succeed(123), None)
        registry.register(er)
        registry.stop()
        self.assertEqual(er.wait(), 123)
        self.assertEqual(er.wait(), 123)
        self.assertEqual(er.wait(), 123)

    def test_weakref(self):
        """
        Registering an EventualResult with a ResultRegistry does not prevent
        it from being garbage collected.
        """
        registry = ResultRegistry(FakeReactor())
        er = EventualResult(None, None)
        registry.register(er)
        ref = weakref.ref(er)
        del er
        gc.collect()
        self.assertIdentical(ref(), None)

    def test_runs_with_lock(self):
        """
        All code in ResultRegistry.stop() and register() is protected by a
        lock.
        """
        self.assertTrue(ResultRegistry.stop.synchronized)
        self.assertTrue(ResultRegistry.register.synchronized)


def append_in_thread(l, f, *args, **kwargs):
    """
    Call a function in a thread, append its result to the given list.

    Only return once the thread has actually started.

    Will return a threading.Event that will be set when the action is done.
    """
    started = threading.Event()
    done = threading.Event()
    def go():
        started.set()
        try:
            result = f(*args, **kwargs)
        except Exception as e:
            l.extend([False, e])
        else:
            l.extend([True, result])
        done.set()
    threading.Thread(target=go).start()
    started.wait()
    return done


class EventualResultTests(TestCase):
    """
    Tests for EventualResult.
    """

    def setUp(self):
        self.patch(threadable, "isInIOThread", lambda: False)

    def test_success_result(self):
        """
        wait() returns the value the Deferred fired with.
        """
        dr = EventualResult(succeed(123), None)
        self.assertEqual(dr.wait(), 123)

    def test_later_success_result(self):
        """
        wait() returns the value the Deferred fired with, in the case where
        the Deferred is fired after wait() is called.
        """
        d = Deferred()
        dr = EventualResult(d, None)
        l = []
        done = append_in_thread(l, dr.wait)
        time.sleep(0.1)
        # At this point dr.wait() should have started:
        d.callback(345)
        done.wait()
        self.assertEqual(l, [True, 345])

    def test_success_result_twice(self):
        """
        A second call to wait() returns same value as the first call.
        """
        dr = EventualResult(succeed(123), None)
        self.assertEqual(dr.wait(), 123)
        self.assertEqual(dr.wait(), 123)

    def test_failure_result(self):
        """
        wait() raises the exception the Deferred fired with.
        """
        dr = EventualResult(fail(RuntimeError()), None)
        self.assertRaises(RuntimeError, dr.wait)

    def test_later_failure_result(self):
        """
        wait() raises the exception the Deferred fired with, in the case
        where the Deferred is fired after wait() is called.
        """
        d = Deferred()
        dr = EventualResult(d, None)
        l = []
        done = append_in_thread(l, dr.wait)
        time.sleep(0.1)
        d.errback(RuntimeError())
        done.wait()
        self.assertEqual((l[0], l[1].__class__), (False, RuntimeError))
{'fraction_non_alphanumeric': 0.0956390123923943, 'fraction_numerical': 0.008324661810613945, 'mean_word_length': 2.460556464811784, 'pattern_counts': {'":': 2, '<': 5, '<?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': '6405262', 'n_tokens_mistral': 3009, 'n_tokens_neox': 2884, 'n_words': 818}