Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

124

125

126

127

128

129

130

131

132

133

import errno 

import mock 

import nfs 

import sys 

import unittest 

import util 

 

class Test_nfs(unittest.TestCase): 

 

    @mock.patch('util.pread', autospec=True) 

    def test_check_server_tcp(self, pread): 

        nfs.check_server_tcp('aServer') 

 

        pread.assert_called_once_with(['/usr/sbin/rpcinfo', '-p', 'aServer'], quiet=False) 

 

    @mock.patch('util.pread', autospec=True) 

    def test_check_server_tcp_nfsversion(self, pread): 

        nfs.check_server_tcp('aServer', 'aNfsversion') 

 

        pread.assert_called_once_with(['/usr/sbin/rpcinfo', '-p', 'aServer'], quiet=False) 

 

    @mock.patch('time.sleep', autospec=True) 

    # Can't use autospec due to http://bugs.python.org/issue17826 

    @mock.patch('util.pread') 

    def test_check_server_service(self, pread, sleep): 

        pread.side_effect=["    100003  4,3,2     udp6,tcp6,udp,tcp                nfs         superuser"] 

        service_found = nfs.check_server_service('aServer') 

 

        self.assertTrue(service_found) 

        self.assertEqual(len(pread.mock_calls), 1) 

        pread.assert_called_with(['/usr/sbin/rpcinfo', '-s', 'aServer']) 

        # Mock==1.0.1 (all that is available in Epel) doesn't suport this 

        #sleep.assert_not_called() 

 

    @mock.patch('time.sleep', autospec=True) 

    # Can't use autospec due to http://bugs.python.org/issue17826 

    @mock.patch('util.pread') 

    def test_check_server_service_with_retries(self, pread, sleep): 

        pread.side_effect=["", 

                           "", 

                           "    100003  4,3,2     udp6,tcp6,udp,tcp                nfs         superuser"] 

        service_found = nfs.check_server_service('aServer') 

 

        self.assertTrue(service_found) 

        self.assertEqual(len(pread.mock_calls), 3) 

        pread.assert_called_with(['/usr/sbin/rpcinfo', '-s', 'aServer']) 

 

    @mock.patch('time.sleep', autospec=True) 

    @mock.patch('util.pread', autospec=True) 

    def test_check_server_service_not_available(self, pread, sleep): 

        pread.return_value="" 

 

        service_found = nfs.check_server_service('aServer') 

 

        self.assertFalse(service_found) 

 

    @mock.patch('time.sleep', autospec=True) 

    # Can't use autospec due to http://bugs.python.org/issue17826 

    @mock.patch('util.pread') 

    def test_check_server_service_exception(self, pread, sleep): 

        pread.side_effect=[util.CommandException(errno.ENOMEM)] 

        with self.assertRaises(util.CommandException): 

            nfs.check_server_service('aServer') 

 

 

    @mock.patch('time.sleep', autospec=True) 

    # Can't use autospec due to http://bugs.python.org/issue17826 

    @mock.patch('util.pread') 

    def test_check_server_service_first_call_exception(self, pread, sleep): 

        pread.side_effect=[util.CommandException(errno.EPIPE), 

                            "    100003  4,3,2     udp6,tcp6,udp,tcp                nfs         superuser"] 

        service_found = nfs.check_server_service('aServer') 

 

        self.assertTrue(service_found) 

        self.assertEqual(len(pread.mock_calls), 2) 

 

    def get_soft_mount_pread(self, binary, vers): 

        return ([binary, 'remoteserver:remotepath', 'mountpoint', '-o', 

                 'soft,proto=transport,vers=%s,acdirmin=0,acdirmax=0' % vers]) 

 

    @mock.patch('util.makedirs', autospec=True) 

    @mock.patch('nfs.check_server_service', autospec=True) 

    @mock.patch('util.pread', autospec=True) 

    def test_soft_mount(self, pread, check_server_service, makedirs): 

        nfs.soft_mount('mountpoint', 'remoteserver', 'remotepath', 'transport', 

                       timeout=None) 

 

        check_server_service.assert_called_once_with('remoteserver') 

        pread.assert_called_once_with(self.get_soft_mount_pread('mount.nfs', 

                                                                '3')) 

 

    @mock.patch('util.makedirs', autospec=True) 

    @mock.patch('nfs.check_server_service', autospec=True) 

    @mock.patch('util.pread', autospec=True) 

    def test_soft_mount_nfsversion_3(self, pread, 

                                     check_server_service, makedirs): 

        nfs.soft_mount('mountpoint', 'remoteserver', 'remotepath', 'transport', 

                       timeout=None, nfsversion='3') 

 

        check_server_service.assert_called_once_with('remoteserver') 

        pread.assert_called_with(self.get_soft_mount_pread('mount.nfs', 

                                                                '3')) 

 

    @mock.patch('util.makedirs', autospec=True) 

    @mock.patch('nfs.check_server_service', autospec=True) 

    @mock.patch('util.pread', autospec=True) 

    def test_soft_mount_nfsversion_4(self, pread, 

                                     check_server_service, makedirs): 

        nfs.soft_mount('mountpoint', 'remoteserver', 'remotepath', 'transport', 

                       timeout=None, nfsversion='4') 

 

        check_server_service.assert_called_once_with('remoteserver') 

        pread.assert_called_with(self.get_soft_mount_pread('mount.nfs4', 

                                                                '4')) 

 

    def test_validate_nfsversion_invalid(self): 

        for thenfsversion in ['0', '5']: 

            self.assertRaises(nfs.NfsException, nfs.validate_nfsversion, 

                              thenfsversion) 

 

    def test_validate_nfsversion_unsupported(self): 

        for thenfsversion in ['2']: 

            self.assertRaises(nfs.NfsException, nfs.validate_nfsversion, 

                              thenfsversion) 

 

    def test_validate_nfsversion_default(self): 

        for thenfsversion in ['', None]: 

            self.assertEquals(nfs.validate_nfsversion(thenfsversion), '3') 

 

    def test_validate_nfsversion_valid(self): 

        for thenfsversion in ['3', '4', '4.1']: 

            self.assertEquals(nfs.validate_nfsversion(thenfsversion), 

                              thenfsversion)