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
|
"""
tests -- the actual TestCase (just one)
"""
import sys
import os
import unittest
from . import util
from .util import Global
from .iscsi import IscsiData
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_ImmediateData_off(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_InitialR2T_off_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_on_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_off_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'])
|
