FilesExpand file tree

 master

/

TaskThreadJoinTest.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

123 lines (98 loc) · 3.59 KB

 master

/

TaskThreadJoinTest.py

Copy path

Top

File metadata and controls

• Code
• Blame

123 lines (98 loc) · 3.59 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

#!/usr/bin/env python

# ClusterShell test suite

# Written by S. Thiell 2010-01-16

"""Unit test for ClusterShell task's join feature in multithreaded

environments"""

import sys

import time

import unittest

sys.path.insert(0, '../lib')

from ClusterShell.Task import *

from ClusterShell.Event import EventHandler

class TaskThreadJoinTest(unittest.TestCase):

def tearDown(self):

task_cleanup()

def testThreadTaskWaitWhenRunning(self):

"""test task_wait() when workers are running"""

for i in range(1, 5):

task = Task()

task.shell("sleep %d" % i)

task.resume()

task_wait()

def testThreadTaskWaitWhenSomeFinished(self):

"""test task_wait() when some workers finished"""

for i in range(1, 5):

task = Task()

task.shell("sleep %d" % i)

task.resume()

time.sleep(2)

task_wait()

def testThreadTaskWaitWhenAllFinished(self):

"""test task_wait() when all workers finished"""

for i in range(1, 3):

task = Task()

task.shell("sleep %d" % i)

task.resume()

time.sleep(4)

task_wait()

def testThreadSimpleTaskSupervisor(self):

"""test task methods from another thread"""

#print "PASS 1"

task = Task()

task.shell("sleep 3")

task.shell("echo testing", key=1)

task.resume()

task.join()

self.assertEqual(task.key_buffer(1), "testing")

#print "PASS 2"

task.shell("echo ok", key=2)

task.resume()

task.join()

#print "PASS 3"

self.assertEqual(task.key_buffer(2), "ok")

task.shell("sleep 1 && echo done", key=3)

task.resume()

task.join()

#print "PASS 4"

self.assertEqual(task.key_buffer(3), "done")

task.abort()

def testThreadTaskBuffers(self):

"""test task data access methods after join()"""

task = Task()

# test data access from main thread

# test stderr separated

task.set_default("stderr", True)

task.shell("echo foobar", key="OUT")

task.shell("echo raboof 1>&2", key="ERR")

task.resume()

task.join()

self.assertEqual(task.key_buffer("OUT"), "foobar")

self.assertEqual(task.key_error("OUT"), "")

self.assertEqual(task.key_buffer("ERR"), "")

self.assertEqual(task.key_error("ERR"), "raboof")

# test stderr merged

task.set_default("stderr", False)

task.shell("echo foobar", key="OUT")

task.shell("echo raboof 1>&2", key="ERR")

task.resume()

task.join()

self.assertEqual(task.key_buffer("OUT"), "foobar")

self.assertEqual(task.key_error("OUT"), "")

self.assertEqual(task.key_buffer("ERR"), "raboof")

self.assertEqual(task.key_error("ERR"), "")

def testThreadTaskUnhandledException(self):

"""test task unhandled exception in thread"""

class TestUnhandledException(Exception):

"""test exception"""

class RaiseOnRead(EventHandler):

def ev_read(self, worker):

raise TestUnhandledException("you should see this exception")

task = Task()

# test data access from main thread

task.shell("echo raisefoobar", key=1, handler=RaiseOnRead())

task.resume()

task.join()

self.assertEqual(task.key_buffer(1), "raisefoobar")

time.sleep(1) # for pretty display, because unhandled exception

# traceback may be sent to stderr after the join()

self.assertFalse(task.running())