-
-
Notifications
You must be signed in to change notification settings - Fork 82
/
test_tcset_dst_one.py
253 lines (197 loc) · 8.54 KB
/
test_tcset_dst_one.py
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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
# encoding: utf-8
"""
.. codeauthor:: Tsuyoshi Hombashi <tsuyoshi.hombashi@gmail.com>
"""
from __future__ import absolute_import, division
import itertools
import pingparsing
import pytest
import typepy
from tcconfig._const import Tc
from tcconfig._netem_param import convert_rate_to_f
from .common import ASSERT_MARGIN, DEADLINE_TIME, execute_tcdel, runner_helper
@pytest.fixture
def device_option(request):
return request.config.getoption("--device")
@pytest.fixture
def dst_host_option(request):
return request.config.getoption("--dst-host")
@pytest.fixture
def transmitter():
transmitter = pingparsing.PingTransmitter()
transmitter.ping_option = "-i 0.2 -q"
transmitter.deadline = DEADLINE_TIME
return transmitter
@pytest.fixture
def pingparser():
return pingparsing.PingParsing()
class Test_tcset_one_network(object):
"""
Tests in this class are not executable on CI services.
Execute the following command at the local environment to running tests:
python setup.py test --addopts "--device=<test device> --dst-host=<IP address>"
These tests expected to execute in the following environment:
- Linux w/ iputils-ping package
- English locale (for parsing ping output)
"""
@pytest.mark.parametrize(
["shaping_algo", "delay"],
[[params[0], params[1]] for params in itertools.product(["htb"], [100])],
)
def test_dst_net_uniform_latency(
self, device_option, dst_host_option, transmitter, pingparser, shaping_algo, delay
):
if device_option is None:
pytest.skip("device option is null")
if typepy.is_null_string(dst_host_option):
pytest.skip("destination host is null")
for tc_target in [device_option, "--device {}".format(device_option)]:
execute_tcdel(tc_target)
transmitter.destination = dst_host_option
# w/o latency tc ---
ping_result = transmitter.ping()
assert ping_result.returncode == 0
without_tc_rtt_avg = pingparser.parse(ping_result).rtt_avg
# w/ latency tc ---
runner_helper(
" ".join(
[
Tc.Command.TCSET,
tc_target,
"--delay {}ms".format(delay),
"--shaping-algo {:s}".format(shaping_algo),
]
)
)
ping_result = transmitter.ping()
assert ping_result.returncode == 0
with_tc_rtt_avg = pingparser.parse(ping_result).rtt_avg
# assertion ---
rtt_diff = with_tc_rtt_avg - without_tc_rtt_avg
assert rtt_diff > (delay * ASSERT_MARGIN)
# finalize ---
execute_tcdel(device_option)
@pytest.mark.parametrize(["delay", "delay_distro"], [[100, 50]])
def test_dst_net_latency_distro(
self, device_option, dst_host_option, transmitter, pingparser, delay, delay_distro
):
if typepy.is_null_string(dst_host_option):
pytest.skip("destination host is null")
for tc_target in [device_option, "--device {}".format(device_option)]:
execute_tcdel(tc_target)
transmitter.destination = dst_host_option
# w/o latency tc ---
ping_result = transmitter.ping()
assert ping_result.returncode == 0
ping_stats = pingparser.parse(ping_result)
without_tc_rtt_avg = ping_stats.rtt_avg
without_tc_rtt_mdev = ping_stats.rtt_mdev
# w/ latency tc ---
runner_helper(
" ".join(
[
Tc.Command.TCSET,
tc_target,
"--delay",
"{:d}ms".format(delay),
"--delay-distro",
"{:d}ms".format(delay_distro),
]
)
)
ping_result = transmitter.ping()
assert ping_result.returncode == 0
ping_stats = pingparser.parse(ping_result)
with_tc_rtt_avg = ping_stats.rtt_avg
with_tc_rtt_mdev = ping_stats.rtt_mdev
# assertion ---
rtt_diff = with_tc_rtt_avg - without_tc_rtt_avg
assert rtt_diff > (delay * ASSERT_MARGIN)
rtt_diff = with_tc_rtt_mdev - without_tc_rtt_mdev
assert rtt_diff > (delay_distro * ASSERT_MARGIN)
# finalize ---
execute_tcdel(tc_target)
@pytest.mark.parametrize(["option", "value"], [["--loss", 10], ["--corrupt", 10]])
def test_dst_net_packet_loss(
self, device_option, dst_host_option, transmitter, pingparser, option, value
):
if typepy.is_null_string(dst_host_option):
pytest.skip("destination host is null")
for tc_target in [device_option, "--device {}".format(device_option)]:
execute_tcdel(tc_target)
transmitter.destination = dst_host_option
# w/o traffic shaping ---
ping_result = transmitter.ping()
assert ping_result.returncode == 0
without_tc_loss_rate = pingparser.parse(ping_result).packet_loss_rate
# w/ traffic shaping ---
runner_helper(
" ".join([Tc.Command.TCSET, tc_target, "{:s} {:f}".format(option, value)])
)
ping_result = transmitter.ping()
assert ping_result.returncode == 0
with_tc_loss_rate = pingparser.parse(ping_result).packet_loss_rate
# check packet loss rate ---
loss_diff = with_tc_loss_rate - without_tc_loss_rate
assert loss_diff > (value * ASSERT_MARGIN)
# finalize ---
execute_tcdel(tc_target)
@pytest.mark.parametrize(["option", "value"], [["--duplicate", 50], ["--duplicate", "10%"]])
def test_dst_net_packet_duplicate(
self, device_option, dst_host_option, transmitter, pingparser, option, value
):
if typepy.is_null_string(dst_host_option):
pytest.skip("destination host is null")
for tc_target in [device_option, "--device {}".format(device_option)]:
execute_tcdel(tc_target)
transmitter.destination = dst_host_option
# w/o packet duplicate tc ---
ping_result = transmitter.ping()
assert ping_result.returncode == 0
without_tc_duplicate_rate = pingparser.parse(ping_result).packet_duplicate_rate
# w/ packet duplicate tc ---
runner_helper(" ".join([Tc.Command.TCSET, tc_target, "{:s} {}".format(option, value)]))
ping_result = transmitter.ping()
assert ping_result.returncode == 0
with_tc_duplicate_rate = pingparser.parse(ping_result).packet_duplicate_rate
# assertion ---
duplicate_rate_diff = with_tc_duplicate_rate - without_tc_duplicate_rate
assert duplicate_rate_diff > (convert_rate_to_f(value) * ASSERT_MARGIN)
# finalize ---
execute_tcdel(tc_target)
def test_dst_net_exclude_dst_network(
self, device_option, dst_host_option, transmitter, pingparser
):
if device_option is None:
pytest.skip("device option is null")
if typepy.is_null_string(dst_host_option):
pytest.skip("destination host is null")
delay = 100
for tc_target in [device_option]:
execute_tcdel(tc_target)
transmitter.destination = dst_host_option
# w/ latency tc ---
runner_helper([Tc.Command.TCSET, tc_target, "--delay", "{:d}ms".format(delay)])
ping_result = transmitter.ping()
assert ping_result.returncode == 0
with_tc_rtt_avg = pingparser.parse(ping_result).rtt_avg
# exclude certain network ---
runner_helper(
" ".join(
[
Tc.Command.TCSET,
tc_target,
"--exclude-dst-network {:s}/24".format(
".".join(dst_host_option.split(".")[:3] + ["0"])
),
"--delay {:d}ms".format(delay),
"--overwrite",
]
)
)
without_tc_rtt_avg = pingparser.parse(transmitter.ping()).rtt_avg
# assertion ---
rtt_diff = with_tc_rtt_avg - without_tc_rtt_avg
assert rtt_diff > (delay * ASSERT_MARGIN)
# finalize ---
execute_tcdel(tc_target)