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

import unittest 

import mock 

import LVHDSR 

import journaler 

import lvhdutil 

 

 

class SMLog(object): 

    def __call__(self, *args): 

        print args 

 

 

class Stubs(object): 

    def init_stubs(self): 

        self._stubs = [] 

 

    def stubout(self, *args, **kwargs): 

        patcher = mock.patch(*args, **kwargs) 

        self._stubs.append(patcher) 

        patcher.start() 

 

    def remove_stubs(self): 

        for patcher in self._stubs: 

            patcher.stop() 

 

 

class TestLVHDSR(unittest.TestCase, Stubs): 

 

    def setUp(self): 

        self.init_stubs() 

 

    def tearDown(self): 

        self.remove_stubs() 

 

    def create_LVHDSR(self): 

        srcmd = mock.Mock() 

        srcmd.dconf = {'device': '/dev/bar'} 

        srcmd.params = {'command': 'foo', 'session_ref': 'some session ref'} 

        return LVHDSR.LVHDSR(srcmd, "some SR UUID") 

 

    @mock.patch('lvhdutil.getVDIInfo', autospec=True) 

    @mock.patch('LVHDSR.Lock', autospec=True) 

    @mock.patch('SR.XenAPI') 

    def test_loadvids(self, mock_xenapi, mock_lock, mock_getVDIInfo): 

        """sr.allVDIs populated by _loadvdis""" 

 

        vdi_uuid = 'some VDI UUID' 

        mock_getVDIInfo.return_value = {vdi_uuid: lvhdutil.VDIInfo(vdi_uuid)} 

        sr = self.create_LVHDSR() 

 

        sr._loadvdis() 

 

        self.assertEquals([vdi_uuid], sr.allVDIs.keys()) 

 

    @mock.patch('lvhdutil.lvRefreshOnAllSlaves', autospec=True) 

    @mock.patch('lvhdutil.getVDIInfo', autospec=True) 

    @mock.patch('journaler.Journaler.getAll', autospec=True) 

    @mock.patch('LVHDSR.Lock', autospec=True) 

    @mock.patch('SR.XenAPI') 

    def test_undoAllInflateJournals( 

            self, 

            mock_xenapi, 

            mock_lock, 

            mock_getAll, 

            mock_getVDIInfo, 

            mock_lvhdutil_lvRefreshOnAllSlaves): 

        """No LV refresh on slaves when Cleaning up local LVHD SR's journal""" 

 

        self.stubout('journaler.Journaler.remove') 

        self.stubout('util.zeroOut') 

        self.stubout('lvhdutil.deflate') 

        self.stubout('util.SMlog', new_callable=SMLog) 

        self.stubout('lvmcache.LVMCache') 

 

        vdi_uuid = 'some VDI UUID' 

 

        mock_getAll.return_value = {vdi_uuid: '0'} 

        mock_getVDIInfo.return_value = {vdi_uuid: lvhdutil.VDIInfo(vdi_uuid)} 

 

        sr = self.create_LVHDSR() 

 

        sr._undoAllInflateJournals() 

        self.assertEquals(0, mock_lvhdutil_lvRefreshOnAllSlaves.call_count)