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

134

135

136

137

138

139

140

141

import unittest 

import mock 

 

import cleanup 

 

import util 

 

 

class FakeXapi(object): 

    def __init__(self): 

        self.srRecord = { 

            'name_label': 'dummy' 

        } 

 

    def isPluggedHere(self): 

        return True 

 

    def isMaster(self): 

        return True 

 

 

class AlwaysLockedLock(object): 

    def acquireNoblock(self): 

        return False 

 

 

class AlwaysFreeLock(object): 

    def acquireNoblock(self): 

        return True 

 

 

class IrrelevantLock(object): 

    pass 

 

 

def create_cleanup_sr(): 

    xapi = FakeXapi() 

    return cleanup.SR(uuid=None, xapi=xapi, createLock=False, force=False) 

 

 

class TestSR(unittest.TestCase): 

    def setUp(self): 

        self.sleep_patcher = mock.patch('cleanup.time.sleep') 

        self.sleep_patcher.start() 

 

    def tearDown(self): 

        self.sleep_patcher.stop() 

 

    def setup_abort_flag(self, ipc_mock, should_abort=False): 

        flag = mock.Mock() 

        flag.test = mock.Mock(return_value=should_abort) 

 

        ipc_mock.return_value = flag 

 

    def test_lock_if_already_locked(self): 

        """ 

        Given an already locked SR, a lock call increments the lock counter 

        """ 

 

        sr = create_cleanup_sr() 

        sr._srLock = IrrelevantLock() 

        sr._locked = 1 

 

        sr.lock() 

 

        self.assertEquals(2, sr._locked) 

 

    def test_lock_if_no_locking_is_used(self): 

        """ 

        Given no srLock present, the lock operations don't touch the counter 

        """ 

 

        sr = create_cleanup_sr() 

        sr._srLock = None 

 

        sr.lock() 

 

        self.assertEquals(0, sr._locked) 

 

    @mock.patch('cleanup.IPCFlag', autospec=True) 

    def test_lock_succeeds_if_lock_is_acquired( 

            self, 

            mock_ipc_flag): 

        """ 

        After performing a lock, the counter equals to 1 

        """ 

 

        self.setup_abort_flag(mock_ipc_flag) 

        sr = create_cleanup_sr() 

        sr._srLock = AlwaysFreeLock() 

 

        sr.lock() 

 

        self.assertEquals(1, sr._locked) 

 

    @mock.patch('cleanup.IPCFlag', autospec=True) 

    def test_lock_raises_exception_if_abort_requested( 

            self, 

            mock_ipc_flag): 

        """ 

        If IPC abort was requested, lock raises AbortException 

        """ 

 

        self.setup_abort_flag(mock_ipc_flag, should_abort=True) 

        sr = create_cleanup_sr() 

        sr._srLock = AlwaysLockedLock() 

 

        self.assertRaises(cleanup.AbortException, sr.lock) 

 

    @mock.patch('cleanup.IPCFlag', autospec=True) 

    def test_lock_raises_exception_if_unable_to_acquire_lock( 

            self, 

            mock_ipc_flag): 

        """ 

        If the lock is busy, SMException is raised 

        """ 

 

        self.setup_abort_flag(mock_ipc_flag) 

        sr = create_cleanup_sr() 

        sr._srLock = AlwaysLockedLock() 

 

        self.assertRaises(util.SMException, sr.lock) 

 

    @mock.patch('cleanup.IPCFlag', autospec=True) 

    def test_lock_leaves_sr_consistent_if_unable_to_acquire_lock( 

            self, 

            mock_ipc_flag): 

        """ 

        If the lock is busy, the lock counter is not incremented 

        """ 

 

        self.setup_abort_flag(mock_ipc_flag) 

        sr = create_cleanup_sr() 

        sr._srLock = AlwaysLockedLock() 

 

        try: 

            sr.lock() 

        except: 

            pass 

 

        self.assertEquals(0, sr._locked)