NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

funcDeco.py (3799B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\util\funcDeco.py
      2 import uthread, expiringdict, numbers
      3 
      4 def CallInNewThread(context = None, returnValue = None):
      5 
      6     def ContextThread(f):
      7         if context is not None:
      8             if context.startswith('^'):
      9                 fullContext = context
     10             else:
     11                 fullContext = '^%s::%s' % (context, f.__name__)
     12         else:
     13             fullContext = f.__name__
     14 
     15         def deco(*args, **kw):
     16             uthread.worker(fullContext, f, *args, **kw)
     17             return returnValue
     18 
     19         return deco
     20 
     21     return ContextThread
     22 
     23 
     24 import unittest
     25 
     26 class TestCallInNewThread(unittest.TestCase):
     27     THREAD_CONTEXT = 'testContext'
     28 
     29     def setUp(self):
     30         mock.SetUp(self, globals())
     31 
     32         def mockWorker(funcContext, func, *args, **kwargs):
     33             self.workerCalled = True
     34             self.workerContext = funcContext
     35 
     36         self.workerCalled = False
     37         uthread.worker = mockWorker
     38         self.value = 'none'
     39         self.SetUpThreadedFunctions()
     40 
     41     def tearDown(self):
     42         mock.TearDown(self)
     43 
     44     def SetUpThreadedFunctions(self):
     45 
     46         def NonThreaded():
     47             self.value = 'non'
     48 
     49         def UglyThreaded():
     50             self.value = 'ugly'
     51 
     52         UglyThreaded = CallInNewThread()(UglyThreaded)
     53 
     54         @CallInNewThread()
     55         def NiceThreaded():
     56             self.value = 'nice'
     57 
     58         @CallInNewThread(self.THREAD_CONTEXT)
     59         def ContextThread():
     60             self.value = 'contexted'
     61 
     62         @CallInNewThread()
     63         def NonContextThread():
     64             self.value = 'non-contexted'
     65 
     66         self.NonThreaded = NonThreaded
     67         self.UglyThreaded = UglyThreaded
     68         self.NiceThreaded = NiceThreaded
     69         self.ContextThread = ContextThread
     70         self.NonContextThread = NonContextThread
     71 
     72     def testNonThreaded(self):
     73         self.NonThreaded()
     74         self.assertTrue(self.value == 'non', 'NonThreaded should immediately change the value')
     75         self.assertTrue(not self.workerCalled, 'uthread.worker was called, a thread was queued that should not have been.')
     76 
     77     def testUglyThreaded(self):
     78         self.UglyThreaded()
     79         self.assertTrue(self.value == 'none', 'UglyThreaded should not immediately change the value')
     80         self.assertTrue(self.workerCalled, 'uthread.worker was not called, and thus the thread is not queued.')
     81 
     82     def testNiceThreaded(self):
     83         self.NiceThreaded()
     84         self.assertTrue(self.value == 'none', 'NiceThreaded should not immediately change the value')
     85         self.assertTrue(self.workerCalled, 'uthread.worker was not called, and thus the thread is not queued.')
     86 
     87     def testContextGeneration(self):
     88         self.ContextThread()
     89         self.assertTrue(self.value == 'none', 'PRECONDITION: ThreadContext should not immediately change the value')
     90         self.assertTrue(self.workerCalled, 'PRECONDITION: uthread.worker was not called, and thus the thread is not queued.')
     91         expectedContext = '^%s::ContextThread' % self.THREAD_CONTEXT
     92         self.assertTrue(self.workerContext == expectedContext, 'The context thread was not assembled correctly.\nExpected:\n%s\nActual:\n%s' % (expectedContext, self.workerContext))
     93 
     94     def testContextGeneration_Default(self):
     95         self.NonContextThread()
     96         self.assertTrue(self.value == 'none', 'PRECONDITION: ThreadContext should not immediately change the value')
     97         self.assertTrue(self.workerCalled, 'PRECONDITION: uthread.worker was not called, and thus the thread is not queued.')
     98         expectedContext = 'NonContextThread'
     99         self.assertTrue(self.workerContext == expectedContext, 'The default context thread is incorrect.\nExpected:\n%s\nActual:\n%s' % (expectedContext, self.workerContext))