<commit_before># Copyright (c) 2007, Linden Research, Inc.
# Copyright (c) 2007, IBM Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import random
from sys import stdout
import time
import re
from tests import skipped, skip_with_pyevent
from unittest import TestCase, main

from eventlet import coros, api, tpool

r = random.WichmannHill()
_g_debug = False

def prnt(msg):
    if _g_debug:
        print msg

class yadda(object):
    def __init__(self):
        pass

    def foo(self,when,n=None):
        assert(n is not None)
        prnt("foo: %s, %s" % (when,n))
        time.sleep(r.random()/20.0)
        return n

def sender_loop(pfx):
    n = 0
    obj = tpool.Proxy(yadda())
    while n < 10:
        if not (n % 5):
            stdout.write('.')
            stdout.flush()
        api.sleep(0)
        now = time.time()
        prnt("%s: send (%s,%s)" % (pfx,now,n))
        rv = obj.foo(now,n=n)
        prnt("%s: recv %s" % (pfx, rv))
        assert(n == rv)
        api.sleep(0)
        n += 1

one = 1
two = 2
three = 3

class TestTpool(TestCase):
    def setUp(self):
        # turn off exception printing, because we'll be deliberately
        # triggering exceptions in our tests
        tpool.QUIET = True
        tpool.setup()

    def tearDown(self):
        tpool.QUIET = False
        tpool.killall()

    @skip_with_pyevent
    def test_a_buncha_stuff(self):
        pool = coros.CoroutinePool(max_size=10)
        waiters = []
        for i in range(0,9):
            waiters.append(pool.execute(sender_loop,i))
        for waiter in waiters:
            waiter.wait()

    @skip_with_pyevent
    def test_wrap_tuple(self):
        my_tuple = (1, 2)
        prox = tpool.Proxy(my_tuple)
        self.assertEqual(prox[0], 1)
        self.assertEqual(prox[1], 2)
        self.assertEqual(len(my_tuple), 2)

    @skip_with_pyevent
    def test_wrap_string(self):
        my_object = "whatever"
        prox = tpool.Proxy(my_object)
        self.assertEqual(str(my_object), str(prox))
        self.assertEqual(len(my_object), len(prox))<commit_msg>Some essentially-meaningless reorg in the tpool tests, mostly for the purposes of making them more understandable and faster-executing.<commit_after># Copyright (c) 2007, Linden Research, Inc.
# Copyright (c) 2007, IBM Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import random
from sys import stdout
import time
import re
from tests import skipped, skip_with_pyevent
from unittest import TestCase, main

from eventlet import coros, api, tpool

one = 1
two = 2
three = 3

class TestTpool(TestCase):
    def setUp(self):
        # turn off exception printing, because we'll be deliberately
        # triggering exceptions in our tests
        tpool.QUIET = True
        tpool.setup()

    def tearDown(self):
        tpool.QUIET = False
        tpool.killall()

    @skip_with_pyevent
    def test_a_buncha_stuff(self):
        assert_ = self.assert_
        class Dummy(object):
            def foo(self,when,token=None):
                assert_(token is not None)
                time.sleep(random.random()/200.0)
                return token
        
        def sender_loop(loopnum):
            obj = tpool.Proxy(Dummy())
            count = 100
            for n in xrange(count):
                api.sleep(random.random()/200.0)
                now = time.time()
                token = <PASSWORD> * count + n
                rv = obj.foo(now,token=token)
                self.assertEquals(token, rv)
                api.sleep(random.random()/200.0)

        pool = coros.CoroutinePool(max_size=10)
        waiters = []
        for i in xrange(10):
            waiters.append(pool.execute(sender_loop,i))
        for waiter in waiters:
            waiter.wait()

    @skip_with_pyevent
    def test_wrap_tuple(self):
        my_tuple = (1, 2)
        prox = tpool.Proxy(my_tuple)
        self.assertEqual(prox[0], 1)
        self.assertEqual(prox[1], 2)
        self.assertEqual(len(my_tuple), 2)

    @skip_with_pyevent
    def test_wrap_string(self):
        my_object = "whatever"
        prox = tpool.Proxy(my_object)
        self.assertEqual(str(my_object), str(prox))
        self.assertEqual(len(my_object), len(prox))
{'fraction_non_alphanumeric': 0.09925129583413322, 'fraction_numerical': 0.014782107890190056, 'mean_word_length': 2.4966442953020134, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1506606', 'n_tokens_mistral': 1671, 'n_tokens_neox': 1584, 'n_words': 489}