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
|
#!/usr/bin/env python3
"""
Unit tests for open-iscsi, using the unittest built-in package
"""
import sys
import unittest
import os
import time
from harness import util
from harness.util import Global
from harness.iscsi import IscsiData
__version__ = '1.0'
class TestRegression(unittest.TestCase):
"""
Regression testing
"""
@classmethod
def setUpClass(cls):
util.verify_needed_commands_exist(['parted', 'fio', 'mkfs', 'bonnie++', 'sgdisk', 'iscsiadm'])
util.vprint('*** Starting %s' % cls.__name__)
# XXX validate that target exists?
# an array of first burts, max burst, and max recv values, for testing
cls.param_values = [[4096, 4096, 4096],
[8192, 4096, 4096],
[16384, 4096, 4096],
[32768, 4096, 4096],
[65536, 4096, 4096],
[131972, 4096, 4096],
[4096, 8192, 4096],
[4096, 16384, 4096],
[4096, 32768, 4096],
[4096, 65536, 4096],
[4096, 131072, 4096],
[4096, 4096, 8192],
[4096, 4096, 16384],
[4096, 4096, 32768],
[4096, 4096, 65536],
[4096, 4096, 131072]]
def setUp(self):
if Global.debug or Global.verbosity > 1:
# this makes debug printing a little more clean
print('', file=sys.stderr)
def iscsi_logout(self):
res = util.run_cmd(['iscsiadm', '-m', 'node',
'-T', Global.target,
'-p', Global.ipnr,
'--logout'])
if res not in [0, 21]:
self.fail('logout failed')
self.assertFalse(os.path.exists(Global.device), '%s: exists after logout!' % Global.device)
def test_InitialR2T_on(self):
"""Test Initial Request to Transmit on, but Immediate Data off"""
i = 1
for v in self.param_values:
with self.subTest('Testing FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v), i=i):
util.vprint('Running subtest %d: FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v) % i)
self.iscsi_logout()
iscsi_data = IscsiData('No', 'Yes', 'None', 'None', v[0], v[1], v[2])
iscsi_data.update_cfg(Global.target, Global.ipnr)
self.run_the_rest()
i += 1
def test_ImmediateData_on(self):
"""Test Initial Request to Transmit off, Immediate Data on"""
i = 1
for v in self.param_values:
with self.subTest('Testing FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v), i=i):
self.iscsi_logout()
iscsi_data = IscsiData('Yes', 'No', 'None', 'None', v[0], v[1], v[2])
iscsi_data.update_cfg(Global.target, Global.ipnr)
self.run_the_rest()
i += 1
def test_InitialR2T_and_ImmediateData_on(self):
"""Test Initial Request to Transmit and Immediate Data on"""
i = 1
for v in self.param_values:
with self.subTest('Testing FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v), i=i):
util.vprint('Running subtest %d: FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v) % i)
self.iscsi_logout()
iscsi_data = IscsiData('Yes', 'Yes', 'None', 'None', v[0], v[1], v[2])
iscsi_data.update_cfg(Global.target, Global.ipnr)
self.run_the_rest()
i += 1
def test_InitialR2T_and_ImmediateData_off(self):
"""Test Initial Request to Transmit and Immediate Data off"""
i = 1
for v in self.param_values:
with self.subTest('Testing FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v), i=i):
util.vprint('Running subtest %d: FirstBurst={} MaxBurts={} MaxRecv={}'.format(*v) % i)
self.iscsi_logout()
iscsi_data = IscsiData('No', 'No', 'None', 'None', v[0], v[1], v[2])
iscsi_data.update_cfg(Global.target, Global.ipnr)
self.run_the_rest()
i += 1
def run_the_rest(self):
res = util.run_cmd(['iscsiadm', '-m', 'node',
'-T', Global.target,
'-p', Global.ipnr,
'--login'])
self.assertEqual(res, 0, 'cannot login to device')
# wait a few seconds for the device to show up
if not util.wait_for_path(Global.device):
self.fail('%s: does not exist after login' % Global.device)
(res, reason) = util.run_fio()
self.assertEqual(res, 0, reason)
(res, reason) = util.run_parted()
self.assertEqual(res, 0, reason)
(res, reason) = util.run_mkfs()
self.assertEqual(res, 0, reason)
(res, reason) = util.run_bonnie()
self.assertEqual(res, 0, reason)
@classmethod
def tearDownClass(cls):
# restore iscsi config
iscsi_data = IscsiData()
iscsi_data.update_cfg(Global.target, Global.ipnr)
# log out of iscsi connection
util.run_cmd(['iscsiadm', '-m', 'node',
'-T', Global.target,
'-p', Global.ipnr,
'--logout'])
if __name__ == '__main__':
# do our own hackery first, to get access to verbosity, debug, etc,
# as well as add our own command-line options
util.setup_testProgram_overrides(__version__, 'test-open-iscsi.py')
# now run the tests
unittest.main()
|
