File: test_errhandler.py

package info (click to toggle)
mpi4py 2.0.0-2.1%2Bdeb9u1
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 2,680 kB
  • sloc: python: 15,291; ansic: 7,099; makefile: 719; f90: 158; sh: 156; cpp: 121
file content (77 lines) | stat: -rw-r--r-- 2,732 bytes parent folder | download | duplicates (2)
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
from mpi4py import MPI
import mpiunittest as unittest

class TestErrhandler(unittest.TestCase):

    def testPredefined(self):
        self.assertFalse(MPI.ERRHANDLER_NULL)
        self.assertTrue(MPI.ERRORS_ARE_FATAL)
        self.assertTrue(MPI.ERRORS_RETURN)

    def testCommGetSetErrhandler(self):
        for COMM in [MPI.COMM_SELF, MPI.COMM_WORLD]:
            for ERRHANDLER in [MPI.ERRORS_ARE_FATAL, MPI.ERRORS_RETURN,
                               MPI.ERRORS_ARE_FATAL, MPI.ERRORS_RETURN, ]:
                errhdl_1 = COMM.Get_errhandler()
                self.assertNotEqual(errhdl_1, MPI.ERRHANDLER_NULL)
                COMM.Set_errhandler(ERRHANDLER)
                errhdl_2 = COMM.Get_errhandler()
                self.assertEqual(errhdl_2, ERRHANDLER)
                errhdl_2.Free()
                self.assertEqual(errhdl_2, MPI.ERRHANDLER_NULL)
                COMM.Set_errhandler(errhdl_1)
                errhdl_1.Free()
                self.assertEqual(errhdl_1, MPI.ERRHANDLER_NULL)

    def testGetErrhandler(self):
        errhdls = []
        for i in range(100):
            e = MPI.COMM_WORLD.Get_errhandler()
            errhdls.append(e)
        for e in errhdls:
            e.Free()
        for e in errhdls:
            self.assertEqual(e, MPI.ERRHANDLER_NULL)

    def testCommCallErrhandler(self):
        errhdl = MPI.COMM_SELF.Get_errhandler()
        comm = MPI.COMM_SELF.Dup()
        comm.Set_errhandler(MPI.ERRORS_RETURN)
        comm.Call_errhandler(MPI.ERR_OTHER)
        comm.Free()

    def testWinCallErrhandler(self):
        try:
            win = MPI.Win.Create(MPI.BOTTOM, 1, MPI.INFO_NULL, MPI.COMM_SELF)
        except NotImplementedError:
            return
        win.Set_errhandler(MPI.ERRORS_RETURN)
        win.Call_errhandler(MPI.ERR_OTHER)
        win.Free()

    def testFileCallErrhandler(self):
        import os, tempfile
        rank = MPI.COMM_WORLD.Get_rank()
        fd, filename = tempfile.mkstemp(prefix='mpi4py-', suffix="-%d"%rank)
        os.close(fd)
        amode = MPI.MODE_WRONLY | MPI.MODE_CREATE | MPI.MODE_DELETE_ON_CLOSE
        try:
            file = MPI.File.Open(MPI.COMM_SELF, filename, amode, MPI.INFO_NULL)
        except NotImplementedError:
            return
        file.Set_errhandler(MPI.ERRORS_RETURN)
        #file.Call_errhandler(MPI.ERR_OTHER)
        file.Call_errhandler(MPI.SUCCESS)
        file.Close()

name, version = MPI.get_vendor()
if MPI.Get_version() < (2, 0):
    del TestErrhandler.testCommCallErrhandler
    del TestErrhandler.testWinCallErrhandler
    del TestErrhandler.testFileCallErrhandler
elif name == 'Microsoft MPI':
	del TestErrhandler.testFileCallErrhandler


if __name__ == '__main__':
    unittest.main()