import os
from tests.patcher_test import ProcessBase
from tests import skip_with_pyevent

class Socket(ProcessBase):
    def test_patched_thread(self):
        new_mod = """from eventlet.green import socket
socket.gethostbyname('localhost')
socket.getaddrinfo('localhost', 80)
"""
        os.environ['EVENTLET_TPOOL_DNS'] = 'yes'
        try:
            self.write_to_tempfile("newmod", new_mod)
            output, lines = self.launch_subprocess('newmod.py')
            self.assertEqual(len(lines), 1, lines)
        finally:
            del os.environ['EVENTLET_TPOOL_DNS']

class Tpool(ProcessBase):
    @skip_with_pyevent
    def test_tpool_size(self):
        new_mod = """from eventlet import tpool
import eventlet
import time
current = [0]
highwater = [0]
def count():
    current[0] += 1
    time.sleep(0.1)
    if current[0] > highwater[0]:
        highwater[0] = current[0]
    current[0] -= 1
expected = 40
p = eventlet.GreenPool()
for i in xrange(expected):
    p.spawn(tpool.execute,count)
p.waitall()
assert highwater[0] == expected, "%s != %s" % (highwater[0], expected)"""
        os.environ['EVENTLET_THREADPOOL_SIZE'] = "40"
        try:
            self.write_to_tempfile("newmod", new_mod)
            output, lines = self.launch_subprocess('newmod.py')
            self.assertEqual(len(lines), 1, lines)
        finally:
            del os.environ['EVENTLET_THREADPOOL_SIZE']

class Hub(ProcessBase):
    def test_eventlet_hub(self):
        new_mod = """from eventlet import hubs
print hubs.get_hub()
"""
        os.environ['EVENTLET_HUB'] = 'selects'
        try:
            self.write_to_tempfile("newmod", new_mod)
            output, lines = self.launch_subprocess('newmod.py')
            self.assertEqual(len(lines), 2, "\n".join(lines))
            self.assert_("selects" in lines[0])
        finally:
            del os.environ['EVENTLET_HUB']