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

import optparse 

import sys 

 

class TestOptParse(optparse.OptionParser): 

    """ Override the default exit and error functions so that they don't print 

    to stderr during the tests 

    """ 

    def exit(self, status=0, msg=None): 

        sys.exit(status) 

 

    def error(self, msg): 

        """error(msg : string)""" 

        self.exit(2, "%s: error: %s\n" % (self.get_prog_name(), msg)) 

 

 

class LogicalVolume(object): 

    def __init__(self, vg, name, size_mb, tag, active, zeroed): 

        self.name = name 

        self.size_mb = size_mb 

        self.volume_group = vg 

        self.tag = tag 

        self.active = active 

        self.zeroed = zeroed 

 

 

class VolumeGroup(object): 

    def __init__(self, name): 

        self.name = name 

        self.volumes = [] 

 

    def add_volume(self, name, size_mb, tag=None, active=True, zeroed=True): 

        self.volumes.append( 

            LogicalVolume(self, name, size_mb, tag, active, zeroed)) 

 

    def delete_volume(self, volume): 

        self.volumes = [vol for vol in self.volumes if vol != volume] 

 

 

class LVSubsystem(object): 

    def __init__(self, logger, executable_injector): 

        self.logger = logger 

        self.lv_calls = [] 

        self._volume_groups = [] 

        executable_injector('/usr/sbin/lvcreate', self.fake_lvcreate) 

        executable_injector('/sbin/lvcreate', self.fake_lvcreate) 

        executable_injector('/usr/sbin/lvremove', self.fake_lvremove) 

        executable_injector('/sbin/lvremove', self.fake_lvremove) 

        executable_injector('/sbin/dmsetup', self.fake_dmsetup) 

 

    def add_volume_group(self, name): 

        self._volume_groups.append(VolumeGroup(name)) 

 

    def get_logical_volumes_with_name(self, name): 

        result = [] 

        for vg in self._volume_groups: 

            for lv in vg.volumes: 

56                if name == lv.name: 

                    result.append(lv) 

        return result 

 

    def get_volume_group(self, vgname): 

        for vg in self._volume_groups: 

62            if vg.name == vgname: 

                return vg 

 

    def fake_lvcreate(self, args, stdin): 

        self.logger('lvcreate', repr(args), stdin) 

        parser = TestOptParse() 

        parser.add_option("-n", dest='name') 

        parser.add_option("-L", dest='size_mb') 

        parser.add_option("--addtag", dest='tag') 

        parser.add_option("--inactive", dest='inactive', action='store_true') 

        parser.add_option("--zero", dest='zero', default='y') 

        parser.add_option("-W", dest='wipe_sig') 

        try: 

            options, args = parser.parse_args(args[1:]) 

        except SystemExit, e: 

            self.logger("LVCREATE OPTION PARSING FAILED") 

            return (1, '', str(e)) 

 

        vgname, = args 

 

        if self.get_volume_group(vgname) is None: 

            self.logger("volume group does not exist:", vgname) 

            return (1, '', '  Volume group "%s" not found\n' % vgname) 

 

        active = not options.inactive 

        assert options.zero in ['y', 'n'] 

        zeroed = options.zero == 'y' 

 

        self.get_volume_group(vgname).add_volume( 

            options.name, 

            int(options.size_mb), 

            options.tag, 

            active, 

            zeroed) 

 

        return 0, '', '' 

 

    def fake_lvremove(self, args, stdin): 

        self.logger('lvremove', repr(args), stdin) 

        parser = TestOptParse() 

        parser.add_option( 

            "-f", "--force", dest='force', action='store_true', default=False) 

        self.logger(args, stdin) 

        try: 

            options, args = parser.parse_args(args[1:]) 

        except SystemExit, e: 

            self.logger("LVREMOVE OPTION PARSING FAILED") 

            return (1, '', str(e)) 

 

        lvpath, = args 

 

        for vg in self._volume_groups: 

            for lv in vg.volumes: 

115                if '/'.join([vg.name, lv.name]) == lvpath: 

                    vg.delete_volume(lv) 

 

        return 0, '', '' 

 

    def fake_dmsetup(self, args, stdin): 

        return 0, '', ''