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

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

import unittest 

import BaseISCSI 

import mock 

import xs_errors 

import os 

 

 

class TestBase(unittest.TestCase): 

    """ Provides errorcodes.xml, so exceptions are sensible """ 

 

    def setUp(self): 

        self._xmldefs = xs_errors.XML_DEFS 

        xs_errors.XML_DEFS = os.path.join( 

            os.path.dirname(__file__), 'XE_SR_ERRORCODES.xml') 

 

    def tearDown(self): 

        xs_errors.XML_DEFS = self._xmldefs 

 

 

class NonLoadingISCSISR(BaseISCSI.BaseISCSISR): 

    def load(self, sr_uuid): 

        pass 

 

 

class TestForceTapDiskConfig(TestBase): 

 

    def _get_iscsi_sr(self, dconf=None): 

        srcmd = mock.Mock() 

        srcmd.dconf = dconf or {} 

        srcmd.params = { 

            'command': 'some_command' 

        } 

 

        iscsisr = NonLoadingISCSISR(srcmd, None) 

        return iscsisr 

 

    def test_default_value(self): 

        iscsi_sr = self._get_iscsi_sr() 

 

        self.assertEquals(False, iscsi_sr.force_tapdisk) 

 

    def test_set_to_true(self): 

        iscsi_sr = self._get_iscsi_sr({ 

            'force_tapdisk': 'true' 

        }) 

 

        self.assertEquals(True, iscsi_sr.force_tapdisk) 

 

 

class NonInitingISCSISR(BaseISCSI.BaseISCSISR): 

    def __init__(self, extra_dconf=None): 

        self.mpath = "false" 

        self.dconf = { 

            'target': 'target', 

            'localIQN': 'localIQN', 

            'targetIQN': 'targetIQN' 

        } 

        self.localIQN = self.dconf['localIQN'] 

        self.original_srcmd = mock.Mock() 

        self.original_srcmd.cmd = 'some_command' 

 

        self.dconf.update(extra_dconf or {}) 

 

 

class NonInitingMultiLUNISCSISR(BaseISCSI.BaseISCSISR): 

    def __init__(self, node1, node2): 

        self.mpath = "false" 

        self.dconf = { 

            'target': node1['ip'], 

            'localIQN': 'localIQN', 

            'targetIQN': node1['iqn'] 

        } 

 

        self.dconf.update({}) 

        self.target = node1['ip'] 

        self.port = node1['port'] 

        self.targetIQN = node1['iqn'] 

        self.attached = True 

        self.multihomed = True 

        extra_adapter = "%s:%d" % (node2['ip'], node2['port']) 

        self.adapter = { 

            extra_adapter: None 

        } 

 

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

        pass 

 

    def _init_adapters(self): 

        pass 

 

 

class TestVdiTypeSetting(TestBase): 

 

    @mock.patch('BaseISCSI.iscsilib.discovery', autospec=True) 

    @mock.patch('BaseISCSI.iscsilib.ensure_daemon_running_ok', autospec=True) 

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

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

    def load_iscsi_sr(self, convertDNS, testHost, ensure_daemon_running_ok, 

                      discovery, iscsi_sr): 

        iscsi_sr.load(None) 

 

    def test_default_vdi_type(self): 

        iscsi_sr = NonInitingISCSISR() 

 

        self.load_iscsi_sr(iscsi_sr=iscsi_sr) 

 

        self.assertEquals('phy', iscsi_sr.sr_vditype) 

 

    def test_vdi_type_modified_by_force_tapdisk(self): 

        iscsi_sr = NonInitingISCSISR(extra_dconf=dict(force_tapdisk='true')) 

 

        self.load_iscsi_sr(iscsi_sr=iscsi_sr) 

 

        self.assertEquals('aio', iscsi_sr.sr_vditype) 

 

 

class TestMultiLUNISCSISR(unittest.TestCase): 

 

    def setUp(self): 

        self.node1 = { 

            'ip': '127.0.0.1', 

            'port': 3260, 

            'iqn': 'IQN' 

        } 

        self.node2 = { 

            'ip': '127.0.0.2', 

            'port': 8080, 

            'iqn': 'IQN', 

            'tpgt': 'TPGT' 

        } 

        self.node_records = [( 

            "%s:%d" % (self.node2['ip'], self.node2['port']), 

            self.node2['tpgt'], 

            self.node2['iqn'] 

        )] 

 

    def assertActiveNodeEquals(self, node, iscsi_sr): 

        node_ip_port = "%s:%d" % (node['ip'], node['port']) 

        node_path = '/dev/iscsi/%s/%s' % (node['iqn'], node_ip_port) 

 

        self.assertEquals(node_path, iscsi_sr.path) 

        self.assertEquals(node_ip_port, iscsi_sr.tgtidx) 

        self.assertEquals(node_ip_port, iscsi_sr.address) 

 

    @mock.patch('BaseISCSI.os.path.exists', autospec=True) 

    @mock.patch('BaseISCSI.iscsilib.get_node_records', autospec=True) 

    def test_initPaths_actual_path_is_active( 

            self, 

            mock_get_node_records, 

            mock_exists): 

        mock_get_node_records.return_value = self.node_records 

        mock_exists.return_value = True 

 

        iscsi_sr = NonInitingMultiLUNISCSISR(self.node1, self.node2) 

 

        iscsi_sr._initPaths() 

 

        self.assertActiveNodeEquals(self.node1, iscsi_sr) 

 

    @mock.patch('BaseISCSI.os.path.exists', autospec=True) 

    @mock.patch('BaseISCSI.iscsilib.get_node_records', autospec=True) 

    def test_initPaths_active_path_detection( 

            self, 

            mock_get_node_records, 

            mock_exists): 

        mock_get_node_records.return_value = self.node_records 

 

        def fake_exists(path): 

            if self.node1['ip'] in path: 

                return False 

            return True 

 

        mock_exists.side_effect = fake_exists 

 

        iscsi_sr = NonInitingMultiLUNISCSISR(self.node1, self.node2) 

 

        iscsi_sr._initPaths() 

 

        self.assertActiveNodeEquals(self.node2, iscsi_sr) 

 

 

class TestISCSISR(TestBase): 

 

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

    def test_load_assert_utf_8_chap_credencials( 

            self, 

            mock__convertDNS): 

 

        """ Asserts that CHAP credentials are always encoded in UTF-8. 

 

            Xapi passes CHAP credentials to ISCSISR as strings of type 'str', 

            if they strictly contain ASCII characters, or as strings of type 

            'unicode' if they contain at least one non-ASCII character. 

        """ 

 

        s1 =  'ascii' 

        s2 = u'\u03bc\u03b9x\u03b5d'  # == 'mixed' in Greek and Latin chars 

        s3 = u'\u03c4\u03bf\u03c0\u03b9\u03ba\u03cc'  # == 'local' in Greek 

        s4 = u'\u6c5fw\u6708\u03c2\xfc\xe4\xd6'  # gibberish var char len str 

 

        # These are the sizes of the 4 strings 

        # in bytes when encoded in UTF-8 

        s1_size = 5 

        s2_size = 8 

        s3_size = 12 

        s4_size = 15 

 

        iscsi_sr = NonInitingISCSISR({ 

                       'chapuser': s1, 

                       'chappassword': s2, 

                       'incoming_chapuser': s3, 

                       'incoming_chappassword': s4 

                   }) 

 

        iscsi_sr.load(None) 

 

        self.assertEqual(iscsi_sr.chapuser, s1.encode('utf-8')) 

        self.assertEqual(iscsi_sr.chappassword, s2.encode('utf-8')) 

        self.assertEqual(iscsi_sr.incoming_chapuser, s3.encode('utf-8')) 

        self.assertEqual(iscsi_sr.incoming_chappassword, s4.encode('utf-8')) 

        self.assertEqual(len(iscsi_sr.chapuser), s1_size) 

        self.assertEqual(len(iscsi_sr.chappassword), s2_size) 

        self.assertEqual(len(iscsi_sr.incoming_chapuser), s3_size) 

        self.assertEqual(len(iscsi_sr.incoming_chappassword), s4_size)