Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes
test.test_thread.ThreadRunningTests Class Reference
Inheritance diagram for test.test_thread.ThreadRunningTests:
Inheritance graph
[legend]
Collaboration diagram for test.test_thread.ThreadRunningTests:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def newtask
def task
def test_starting_threads
def test_stack_size
def test__count
def setUp

Public Attributes

 created
 running
 next_ident
 done_mutex
 running_mutex
 random_mutex

Detailed Description

Definition at line 35 of file test_thread.py.


Member Function Documentation

Definition at line 37 of file test_thread.py.

00037 
00038     def newtask(self):
00039         with self.running_mutex:
00040             self.next_ident += 1
00041             verbose_print("creating task %s" % self.next_ident)
00042             thread.start_new_thread(self.task, (self.next_ident,))
00043             self.created += 1
00044             self.running += 1

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_thread.BasicThreadTest.setUp (   self) [inherited]

Definition at line 25 of file test_thread.py.

00025 
00026     def setUp(self):
00027         self.done_mutex = thread.allocate_lock()
00028         self.done_mutex.acquire()
00029         self.running_mutex = thread.allocate_lock()
00030         self.random_mutex = thread.allocate_lock()
00031         self.created = 0
00032         self.running = 0
00033         self.next_ident = 0
00034 

Here is the caller graph for this function:

def test.test_thread.ThreadRunningTests.task (   self,
  ident 
)

Definition at line 45 of file test_thread.py.

00045 
00046     def task(self, ident):
00047         with self.random_mutex:
00048             delay = random.random() / 10000.0
00049         verbose_print("task %s will run for %sus" % (ident, round(delay*1e6)))
00050         time.sleep(delay)
00051         verbose_print("task %s done" % ident)
00052         with self.running_mutex:
00053             self.running -= 1
00054             if self.created == NUMTASKS and self.running == 0:
00055                 self.done_mutex.release()

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 105 of file test_thread.py.

00105 
00106     def test__count(self):
00107         # Test the _count() function.
00108         orig = thread._count()
00109         mut = thread.allocate_lock()
00110         mut.acquire()
00111         started = []
00112         def task():
00113             started.append(None)
00114             mut.acquire()
00115             mut.release()
00116         thread.start_new_thread(task, ())
00117         while not started:
00118             time.sleep(0.01)
00119         self.assertEqual(thread._count(), orig + 1)
00120         # Allow the task to finish.
00121         mut.release()
00122         # The only reliable way to be sure that the thread ended from the
00123         # interpreter's point of view is to wait for the function object to be
00124         # destroyed.
00125         done = []
00126         wr = weakref.ref(task, lambda _: done.append(None))
00127         del task
00128         while not done:
00129             time.sleep(0.01)
00130         self.assertEqual(thread._count(), orig)
00131 

Here is the call graph for this function:

Definition at line 64 of file test_thread.py.

00064 
00065     def test_stack_size(self):
00066         # Various stack size tests.
00067         self.assertEqual(thread.stack_size(), 0, "initial stack size is not 0")
00068 
00069         thread.stack_size(0)
00070         self.assertEqual(thread.stack_size(), 0, "stack_size not reset to default")
00071 
00072         if os.name not in ("nt", "os2", "posix"):
00073             return
00074 
00075         tss_supported = True
00076         try:
00077             thread.stack_size(4096)
00078         except ValueError:
00079             verbose_print("caught expected ValueError setting "
00080                             "stack_size(4096)")
00081         except thread.error:
00082             tss_supported = False
00083             verbose_print("platform does not support changing thread stack "
00084                             "size")
00085 
00086         if tss_supported:
00087             fail_msg = "stack_size(%d) failed - should succeed"
00088             for tss in (262144, 0x100000, 0):
00089                 thread.stack_size(tss)
00090                 self.assertEqual(thread.stack_size(), tss, fail_msg % tss)
00091                 verbose_print("successfully set stack_size(%d)" % tss)
00092 
00093             for tss in (262144, 0x100000):
00094                 verbose_print("trying stack_size = (%d)" % tss)
00095                 self.next_ident = 0
00096                 self.created = 0
00097                 for i in range(NUMTASKS):
00098                     self.newtask()
00099 
00100                 verbose_print("waiting for all tasks to complete")
00101                 self.done_mutex.acquire()
00102                 verbose_print("all tasks done")
00103 
00104             thread.stack_size(0)

Here is the call graph for this function:

Definition at line 56 of file test_thread.py.

00056 
00057     def test_starting_threads(self):
00058         # Basic test for thread creation.
00059         for i in range(NUMTASKS):
00060             self.newtask()
00061         verbose_print("waiting for tasks to complete...")
00062         self.done_mutex.acquire()
00063         verbose_print("all tasks done")

Here is the call graph for this function:


Member Data Documentation

Reimplemented from test.test_thread.BasicThreadTest.

Definition at line 53 of file test_thread.py.

Definition at line 26 of file test_thread.py.

Reimplemented from test.test_thread.BasicThreadTest.

Definition at line 94 of file test_thread.py.

Definition at line 29 of file test_thread.py.

Reimplemented from test.test_thread.BasicThreadTest.

Definition at line 53 of file test_thread.py.

Definition at line 28 of file test_thread.py.


The documentation for this class was generated from the following file: