"""
Tests for the thread pool.

Pyro - Python Remote Objects.  Copyright by Irmen de Jong (irmen@razorvine.net).
"""

from __future__ import print_function
import time
import random
import unittest
from Pyro4.socketserver.threadpool import Pool, PoolError, NoFreeWorkersError
from Pyro4.socketserver.threadpoolserver import SocketServer_Threadpool
import Pyro4.socketutil
import Pyro4.threadutil


JOB_TIME = 0.2


class Job(object):
    def __init__(self, name="unnamed"):
        self.name = name

    def __call__(self):
        time.sleep(JOB_TIME - random.random() / 10.0)


class SlowJob(object):
    def __init__(self, name="unnamed"):
        self.name = name

    def __call__(self):
        time.sleep(5*JOB_TIME - random.random() / 10.0)


class PoolTests(unittest.TestCase):
    def setUp(self):
        Pyro4.config.THREADPOOL_SIZE_MIN = 2
        Pyro4.config.THREADPOOL_SIZE = 4

    def tearDown(self):
        Pyro4.config.reset()

    def testCreate(self):
        with Pool() as jq:
            _ = repr(jq)
        self.assertTrue(jq.closed)

    def testSingle(self):
        with Pool() as p:
            job = Job()
            p.process(job)
            time.sleep(0.02)  # let it pick up the job
            self.assertEqual(1, len(p.busy))

    def testAllBusy(self):
        try:
            Pyro4.config.COMMTIMEOUT = 0.2
            with Pool() as p:
                for i in range(Pyro4.config.THREADPOOL_SIZE):
                    p.process(SlowJob(str(i+1)))
                # putting one more than the number of workers should raise an error:
                with self.assertRaises(NoFreeWorkersError):
                    p.process(SlowJob("toomuch"))
        finally:
            Pyro4.config.COMMTIMEOUT = 0.0

    def testClose(self):
        with Pool() as p:
            for i in range(Pyro4.config.THREADPOOL_SIZE):
                p.process(Job(str(i + 1)))
        with self.assertRaises(PoolError):
            p.process(Job(1))  # must not allow new jobs after closing
        self.assertEqual(0, len(p.busy))
        self.assertEqual(0, len(p.idle))

    def testScaling(self):
        with Pool() as p:
            for i in range(Pyro4.config.THREADPOOL_SIZE_MIN-1):
                p.process(Job("x"))
            self.assertEqual(1, len(p.idle))
            self.assertEqual(Pyro4.config.THREADPOOL_SIZE_MIN-1, len(p.busy))
            p.process(Job("x"))
            self.assertEqual(0, len(p.idle))
            self.assertEqual(Pyro4.config.THREADPOOL_SIZE_MIN, len(p.busy))
            # grow until no more free workers
            while True:
                try:
                    p.process(Job("x"))
                except NoFreeWorkersError:
                    break
            self.assertEqual(0, len(p.idle))
            self.assertEqual(Pyro4.config.THREADPOOL_SIZE, len(p.busy))
            # wait till jobs are done and check ending situation
            time.sleep(JOB_TIME*1.5)
            self.assertEqual(0, len(p.busy))
            self.assertEqual(Pyro4.config.THREADPOOL_SIZE_MIN, len(p.idle))


class ServerCallback(object):
    def __init__(self):
        self.received_denied_reason = None

    def _handshake(self, connection, denied_reason=None):
        self.received_denied_reason = denied_reason  # store the denied reason
        return True

    def handleRequest(self, connection):
        pass

    def _housekeeping(self):
        pass


class ThreadPoolServerTests(unittest.TestCase):
    def setUp(self):
        Pyro4.config.THREADPOOL_SIZE_MIN = 1
        Pyro4.config.THREADPOOL_SIZE = 1
        Pyro4.config.POLLTIMEOUT = 0.5
        Pyro4.config.COMMTIMEOUT = 0.5

    def tearDown(self):
        Pyro4.config.reset()

    def testServerPoolFull(self):
        port = Pyro4.socketutil.findProbablyUnusedPort()
        serv = SocketServer_Threadpool()
        daemon = ServerCallback()
        serv.init(daemon, "localhost", port)
        serversock = serv.sock.getsockname()
        csock1 = Pyro4.socketutil.createSocket(connect=serversock)
        csock2 = Pyro4.socketutil.createSocket(connect=serversock)
        try:
            serv.events([serv.sock])
            self.assertIsNone(daemon.received_denied_reason)
            serv.events([serv.sock])
            self.assertEqual("no free workers, increase server threadpool size", daemon.received_denied_reason)
        finally:
            csock1.close()
            csock2.close()
            serv.shutdown()


if __name__ == "__main__":
    # import sys;sys.argv = ['', 'Test.testName']
    unittest.main()

@@ -3180,22 +3180,33 @@
 allback(
-object
+Pyro4.core.Daemon
 ):%0D%0A    
@@ -3265,15 +3265,14 @@
 ason
+s
  = 
-None
+%5B%5D
 %0D%0A%0D%0A
@@ -3365,19 +3365,25 @@
 d_reason
- = 
+s.append(
 denied_r
@@ -3387,16 +3387,17 @@
 d_reason
+)
   # stor
@@ -3486,28 +3486,40 @@
 ):%0D%0A        
-pass
+time.sleep(0.05)
 %0D%0A%0D%0A    def 
@@ -4178,32 +4178,63 @@
 ct=serversock)%0D%0A
+        time.sleep(0.1) # XXX%0D%0A
         csock2 =
@@ -4277,32 +4277,63 @@
 ct=serversock)%0D%0A
+        time.sleep(0.1) # XXX%0D%0A
         try:%0D%0A  
@@ -4360,32 +4360,61 @@
 s(%5Bserv.sock%5D)%0D%0A
+            time.sleep(0.2)%0D%0A
             self
@@ -4424,15 +4424,22 @@
 sert
-Is
+Equal(%5B
 None
-(
+%5D, 
 daem
@@ -4455,32 +4455,33 @@
 ed_denied_reason
+s
 )%0D%0A            s
@@ -4517,32 +4517,128 @@
         
-self.assertEqual
+time.sleep(0.2)%0D%0A            self.assertEqual(2, len(daemon.received_denied_reasons))%0D%0A            self.assertIn
 (%22no fre
@@ -4711,16 +4711,17 @@
 d_reason
+s
 )%0D%0A     

fix threadpool unittest