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
|
from mpi4py import MPI
import mpiunittest as unittest
class TestRequest(unittest.TestCase):
def setUp(self):
self.REQUEST = MPI.Request()
self.STATUS = MPI.Status()
def testWait(self):
self.REQUEST.Wait()
self.REQUEST.Wait(None)
self.REQUEST.Wait(self.STATUS)
def testTest(self):
self.REQUEST.Wait()
self.REQUEST.Wait(None)
self.REQUEST.Test(self.STATUS)
def testGetStatus(self):
try: flag = self.REQUEST.Get_status()
except NotImplementedError: return
self.assertTrue(flag)
flag = self.REQUEST.Get_status(self.STATUS)
self.assertTrue(flag)
self.assertEqual(self.STATUS.Get_source(), MPI.ANY_SOURCE)
self.assertEqual(self.STATUS.Get_tag(), MPI.ANY_TAG)
self.assertEqual(self.STATUS.Get_error(), MPI.SUCCESS)
self.assertEqual(self.STATUS.Get_count(MPI.BYTE), 0)
self.assertEqual(self.STATUS.Get_elements(MPI.BYTE), 0)
try: self.assertFalse(self.STATUS.Is_cancelled())
except NotImplementedError: pass
class TestRequestArray(unittest.TestCase):
def setUp(self):
self.REQUESTS = [MPI.Request() for i in range(5)]
self.STATUSES = [MPI.Status() for i in range(5)]
def testWaitany(self):
MPI.Request.Waitany(self.REQUESTS)
MPI.Request.Waitany(self.REQUESTS, None)
MPI.Request.Waitany(self.REQUESTS, self.STATUSES[0])
def testTestany(self):
MPI.Request.Testany(self.REQUESTS)
MPI.Request.Testany(self.REQUESTS, None)
MPI.Request.Testany(self.REQUESTS, self.STATUSES[0])
def testWaitall(self):
MPI.Request.Waitall(self.REQUESTS)
MPI.Request.Waitall(self.REQUESTS, None)
for statuses in (self.STATUSES, []):
MPI.Request.Waitall(self.REQUESTS, statuses)
self.assertEqual(len(statuses), len(self.REQUESTS))
def testTestall(self):
MPI.Request.Testall(self.REQUESTS)
MPI.Request.Testall(self.REQUESTS, None)
for statuses in (self.STATUSES, []):
MPI.Request.Testall(self.REQUESTS, statuses)
self.assertEqual(len(statuses), len(self.REQUESTS))
def testWaitsome(self):
out = (MPI.UNDEFINED, [])
ret = MPI.Request.Waitsome(self.REQUESTS)
self.assertEqual((ret[0],list(ret[1])), out)
ret = MPI.Request.Waitsome(self.REQUESTS, None)
self.assertEqual((ret[0],list(ret[1])), out)
for statuses in (self.STATUSES, []):
ret = MPI.Request.Waitsome(self.REQUESTS, statuses)
self.assertEqual((ret[0],list(ret[1])), out)
self.assertEqual(len(statuses), len(self.REQUESTS))
def testTestsome(self):
out = (MPI.UNDEFINED, [])
ret = MPI.Request.Testsome(self.REQUESTS)
self.assertEqual((ret[0],list(ret[1])), out)
ret = MPI.Request.Testsome(self.REQUESTS, None)
self.assertEqual((ret[0],list(ret[1])), out)
for statuses in (self.STATUSES, []):
ret = MPI.Request.Testsome(self.REQUESTS, statuses)
self.assertEqual((ret[0],list(ret[1])), out)
self.assertEqual(len(statuses), len(self.REQUESTS))
_name, _version = MPI.get_vendor()
if (_name == 'MPICH1' or
_name == 'LAM/MPI'):
del TestRequest.testGetStatus
if __name__ == '__main__':
unittest.main()
|
