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

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

import unittest 

import mock 

 

import lvmlib 

 

 

class ExecResultMixIn(object): 

    def assertExecutionSucceeded(self, exec_result): 

        returncode, stdout, stderr = exec_result 

 

        self.assertEquals(0, returncode) 

 

    def assertExecutionFailed(self, exec_result): 

        returncode, stdout, stderr = exec_result 

 

        self.assertEquals(1, returncode) 

 

 

class TestLVSubSystem(unittest.TestCase, ExecResultMixIn): 

    def test_lvcreate_is_mocked(self): 

        executable_injector = mock.Mock() 

 

        lvsubsystem = lvmlib.LVSubsystem(None, executable_injector) 

 

        self.assertTrue( 

            mock.call('/usr/sbin/lvcreate', lvsubsystem.fake_lvcreate) 

            in executable_injector.mock_calls 

        ) 

 

    def test_lvremove_is_mocked(self): 

        executable_injector = mock.Mock() 

 

        lvsubsystem = lvmlib.LVSubsystem(None, executable_injector) 

 

        self.assertTrue( 

            mock.call('/usr/sbin/lvremove', lvsubsystem.fake_lvremove) 

            in executable_injector.mock_calls 

        ) 

 

    def test_dmsetup_is_mocked(self): 

        executable_injector = mock.Mock() 

 

        lvsubsystem = lvmlib.LVSubsystem(None, executable_injector) 

 

        self.assertTrue( 

            mock.call('/sbin/dmsetup', lvsubsystem.fake_dmsetup) 

            in executable_injector.mock_calls 

        ) 

 

    def test_add_volume_group(self): 

        lvsubsystem = lvmlib.LVSubsystem(None, mock.Mock()) 

 

        lvsubsystem.add_volume_group('vg') 

        vg = lvsubsystem.get_volume_group('vg') 

 

        self.assertEquals('vg', vg.name) 

 

    def test_fake_lvcreate_creates_volume(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        vg = lvsubsystem.add_volume_group('vg') 

 

        exec_result = lvsubsystem.fake_lvcreate( 

            "someprog -n name -L 100 vg".split(), '') 

 

        lv, = lvsubsystem.get_logical_volumes_with_name('name') 

 

        self.assertEquals('name', lv.name) 

        self.assertEquals(lvsubsystem.get_volume_group('vg'), lv.volume_group) 

        self.assertTrue(lv.active) 

        self.assertTrue(lv.zeroed) 

        self.assertEquals(None, lv.tag) 

        self.assertEquals(100, lv.size_mb) 

 

    def test_fake_lvcreate_with_tags(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

 

        exec_result = lvsubsystem.fake_lvcreate( 

            "someprog -n name --addtag tagg -L 100 vg".split(), '') 

 

        lv, = lvsubsystem.get_logical_volumes_with_name('name') 

        self.assertEquals('tagg', lv.tag) 

 

    def test_fake_lvcreate_inactive(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

 

        exec_result = lvsubsystem.fake_lvcreate( 

            "someprog -n name --inactive -L 100 vg".split(), '') 

 

        lv, = lvsubsystem.get_logical_volumes_with_name('name') 

        self.assertFalse(lv.active) 

 

    def test_fake_lvcreate_non_zeroed(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

 

        exec_result = lvsubsystem.fake_lvcreate( 

            "someprog -n name --zero n -L 100 vg".split(), '') 

 

        lv, = lvsubsystem.get_logical_volumes_with_name('name') 

 

        self.assertFalse(lv.zeroed) 

        self.assertExecutionSucceeded(exec_result) 

 

    def test_fake_lvcreate_called_with_wrong_params(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

 

        exec_result = lvsubsystem.fake_lvcreate( 

            "someprog --something-stupid -n name n -L 100 vg".split(), '') 

 

        self.assertExecutionFailed(exec_result) 

 

    def test_fake_lvcreate_fails_if_no_volume_group_found(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

 

        exec_result = lvsubsystem.fake_lvcreate( 

            "someprog -n name -L 100 nonexisting".split(), '') 

 

        self.assertExecutionFailed(exec_result) 

 

    def test_fake_lvremove(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

        lvsubsystem.get_volume_group('vg').add_volume('lv', 100) 

 

        exec_result = lvsubsystem.fake_lvremove( 

            "someprog vg/lv".split(), '') 

 

        self.assertExecutionSucceeded(exec_result) 

 

    def test_fake_lvremove_with_force(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

        lvsubsystem.get_volume_group('vg').add_volume('lv', 100) 

 

        exec_result = lvsubsystem.fake_lvremove( 

            "someprog -f vg/lv".split(), '') 

 

        self.assertExecutionSucceeded(exec_result) 

 

    def test_fake_lvremove_with_bad_params(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

        lvsubsystem.add_volume_group('vg') 

        lvsubsystem.get_volume_group('vg').add_volume('lv', 100) 

 

        exec_result = lvsubsystem.fake_lvremove( 

            "someprog -f vg/lv --stupid-parameter".split(), '') 

 

        self.assertExecutionFailed(exec_result) 

 

    def test_fake_dmsetup_status_returns_zero(self): 

        lvsubsystem = lvmlib.LVSubsystem(mock.Mock(), mock.Mock()) 

 

        exec_result = lvsubsystem.fake_dmsetup( 

            "someprog status".split(), '') 

 

        self.assertExecutionSucceeded(exec_result)