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
|
from mpi4py import MPI
import mpiunittest as unittest
class TestEnviron(unittest.TestCase):
def testIsInitialized(self):
flag = MPI.Is_initialized()
self.assertTrue(type(flag) is bool)
self.assertTrue(flag)
def testIsFinalized(self):
flag = MPI.Is_finalized()
self.assertTrue(type(flag) is bool)
self.assertFalse(flag)
def testGetVersion(self):
version = MPI.Get_version()
self.assertEqual(len(version), 2)
major, minor = version
self.assertTrue(type(major) is int)
self.assertTrue(major >= 1)
self.assertTrue(type(minor) is int)
self.assertTrue(minor >= 0)
def testGetProcessorName(self):
procname = MPI.Get_processor_name()
self.assertTrue(isinstance(procname, str))
def testWTime(self):
time1 = MPI.Wtime()
self.assertTrue(type(time1) is float)
time2 = MPI.Wtime()
self.assertTrue(type(time2) is float)
self.assertTrue(time2 >= time1)
def testWTick(self):
tick = MPI.Wtick()
self.assertTrue(type(tick) is float)
self.assertTrue(tick > 0.0)
class TestWorldAttrs(unittest.TestCase):
def testWTimeIsGlobal(self):
wtg = MPI.COMM_WORLD.Get_attr(MPI.WTIME_IS_GLOBAL)
if wtg is not None:
self.assertTrue(wtg in (True, False))
def testWTimeIsGlobal(self):
wtg = MPI.COMM_WORLD.Get_attr(MPI.WTIME_IS_GLOBAL)
if wtg is not None:
self.assertTrue(wtg in (True, False))
def testHostPorcessor(self):
size = MPI.COMM_WORLD.Get_size()
vals = list(range(size)) + [MPI.PROC_NULL]
hostproc = MPI.COMM_WORLD.Get_attr(MPI.HOST)
if hostproc is not None:
self.assertTrue(hostproc in vals)
def testIOProcessor(self):
size = MPI.COMM_WORLD.Get_size()
vals = list(range(size)) + [MPI.UNDEFINED,
MPI.ANY_SOURCE,
MPI.PROC_NULL]
ioproc = MPI.COMM_WORLD.Get_attr(MPI.IO)
if ioproc is not None:
self.assertTrue(ioproc in vals)
def testAppNum(self):
if MPI.APPNUM == MPI.KEYVAL_INVALID: return
appnum = MPI.COMM_WORLD.Get_attr(MPI.APPNUM)
if appnum is not None:
self.assertTrue(appnum == MPI.UNDEFINED or appnum >= 0)
def testUniverseSize(self):
if MPI.UNIVERSE_SIZE == MPI.KEYVAL_INVALID: return
univsz = MPI.COMM_WORLD.Get_attr(MPI.UNIVERSE_SIZE)
if univsz is not None:
self.assertTrue(univsz == MPI.UNDEFINED or univsz >= 0)
def testLastUsedCode(self):
if MPI.LASTUSEDCODE == MPI.KEYVAL_INVALID: return
lastuc = MPI.COMM_WORLD.Get_attr(MPI.LASTUSEDCODE)
self.assertTrue(lastuc >= 0)
_name, _version = MPI.get_vendor()
if (_name == 'MPICH2' and
_version > (1, 2)):
# Up to mpich2-1.3.1 when running under Hydra process manager,
# getting the universe size fails for the singleton init case
if MPI.COMM_WORLD.Get_attr(MPI.APPNUM) is None:
del TestWorldAttrs.testUniverseSize
if __name__ == '__main__':
unittest.main()
|
