1 #!/usr/bin/python 2 import unittest, os, socket, time, sys, struct 3 import common 4 import utils 5 from autotest_lib.client.bin.net import net_utils, net_utils_mock 6 from autotest_lib.client.common_lib.test_utils import mock 7 from autotest_lib.client.common_lib import error 8 9 10 class TestNetUtils(unittest.TestCase): 11 class network_interface_mock(net_utils_mock.network_interface_mock): 12 def __init__(self, iface='some_name', test_init=False): 13 super(TestNetUtils.network_interface_mock, 14 self).__init__(iface=iface, test_init=test_init) 15 16 17 def setUp(self): 18 self.god = mock.mock_god() 19 self.god.stub_function(utils, "system") 20 self.god.stub_function(utils, "system_output") 21 self.god.stub_function(utils, "module_is_loaded") 22 self.god.stub_function(net_utils, "open") 23 self.god.stub_function(time, 'sleep') 24 25 self.god.stub_with(net_utils,"bond", net_utils.bonding) 26 self.god.stub_with(os, 'open', net_utils_mock.os_open) 27 self.god.stub_with(net_utils, 'netif', net_utils_mock.netutils_netif) 28 29 os.environ['AUTODIR'] = "autodir" 30 31 32 def tearDown(self): 33 self.god.unstub_all() 34 del os.environ['AUTODIR'] 35 36 37 # 38 # test network_util 39 # 40 def test_network_util_reset(self): 41 utils.system.expect_call('service network restart', ignore_status=False) 42 net_utils.network_utils().reset() 43 self.god.check_playback() 44 45 46 def test_network_util_start(self): 47 utils.system.expect_call('service network start', ignore_status=False) 48 49 net_utils.network_utils().start() 50 self.god.check_playback() 51 52 53 def test_network_util_stop(self): 54 utils.system.expect_call('service network stop', ignore_status=False) 55 56 net_utils.network_utils().stop() 57 self.god.check_playback() 58 59 60 def test_network_util_disable_ip_local_loopback(self): 61 msg = "echo '1' > /proc/sys/net/ipv4/route/no_local_loopback" 62 utils.system.expect_call(msg, ignore_status=False) 63 msg = 'echo 1 > /proc/sys/net/ipv4/route/flush' 64 utils.system.expect_call(msg, ignore_status=False) 65 66 net_utils.network_utils().disable_ip_local_loopback() 67 self.god.check_playback() 68 69 70 def test_network_util_enable_ip_local_loopback(self): 71 msg = "echo '0' > /proc/sys/net/ipv4/route/no_local_loopback" 72 utils.system.expect_call(msg, ignore_status=False) 73 msg = 'echo 1 > /proc/sys/net/ipv4/route/flush' 74 utils.system.expect_call(msg, ignore_status=False) 75 76 net_utils.network_utils().enable_ip_local_loopback() 77 self.god.check_playback() 78 79 80 # 81 # test network_interface 82 # 83 def test_network_interface_init(self): 84 self.god.stub_function(socket, 'socket') 85 s = net_utils_mock.socket_stub('eth0', socket, socket) 86 socket.socket.expect_call(socket.PF_PACKET, 87 socket.SOCK_RAW).and_return(s) 88 self.god.stub_function(s, 'bind') 89 self.god.stub_function(s, 'settimeout') 90 s.settimeout.expect_call(net_utils.TIMEOUT) 91 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 92 mock_netif = self.network_interface_mock(iface='eth0', test_init=True) 93 self.god.check_playback() 94 self.assertEquals(mock_netif.ethtool, 'ethtool') 95 self.assertEquals(mock_netif._name, 'eth0') 96 self.assertEquals(mock_netif.was_down, 'is_down') 97 self.assertEquals(mock_netif.orig_ipaddr, 'get_ipaddr') 98 self.assertEquals(mock_netif.was_loopback_enabled, 99 'is_loopback_enabled') 100 self.assertEquals(mock_netif._socket, s) 101 102 103 def test_network_interface_restore(self): 104 mock_netif = self.network_interface_mock('eth0') 105 106 mock_netif.was_loopback_enabled = False 107 mock_netif.loopback_enabled = True 108 mock_netif.was_down = False 109 110 # restore using phyint 111 cmd = 'ifconfig %s %s' % (mock_netif._name, mock_netif.orig_ipaddr) 112 utils.system.expect_call(cmd) 113 114 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 115 'phyint', 'disable') 116 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 117 mock_netif.restore() 118 self.god.check_playback() 119 120 121 # restore using mac 122 cmd = 'ifconfig %s %s' % (mock_netif._name, mock_netif.orig_ipaddr) 123 utils.system.expect_call(cmd) 124 125 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 126 'phyint', 'disable') 127 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 128 129 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, 130 mock_netif._name, 'mac', 'disable') 131 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 132 mock_netif.restore() 133 self.god.check_playback() 134 135 # check that down is restored 136 mock_netif.was_loopback_enabled = False 137 mock_netif.loopback_enabled = True 138 mock_netif.was_down = True 139 140 cmd = 'ifconfig %s %s' % (mock_netif._name, mock_netif.orig_ipaddr) 141 utils.system.expect_call(cmd) 142 143 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 144 'phyint', 'disable') 145 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 146 147 cmd = 'ifconfig %s down' % mock_netif._name 148 utils.system.expect_call(cmd) 149 150 mock_netif.restore() 151 self.god.check_playback() 152 153 # check that loopback, down are done in sequence 154 mock_netif.was_loopback_enabled = True 155 mock_netif.loopback_enabled = True 156 mock_netif.was_down = True 157 158 cmd = 'ifconfig %s %s' % (mock_netif._name, 159 mock_netif.orig_ipaddr) 160 161 utils.system.expect_call(cmd) 162 cmd = 'ifconfig %s down' % mock_netif._name 163 utils.system.expect_call(cmd) 164 165 mock_netif.restore() 166 self.god.check_playback() 167 168 # prior loopback matches current loopback 169 mock_netif.was_loopback_enabled = False 170 mock_netif.loopback_enabled = False 171 mock_netif.was_down = True 172 173 cmd = 'ifconfig %s %s' % (mock_netif._name, 174 mock_netif.orig_ipaddr) 175 utils.system.expect_call(cmd) 176 cmd = 'ifconfig %s down' % mock_netif._name 177 utils.system.expect_call(cmd) 178 179 mock_netif.restore() 180 self.god.check_playback() 181 182 183 def test_network_interface_get_name(self): 184 mock_netif = self.network_interface_mock(iface='eth0') 185 self.assertEquals(mock_netif.get_name(), 'eth0') 186 187 188 def test_network_interface_parse_ethtool(self): 189 mock_netif = self.network_interface_mock() 190 cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name) 191 192 utils.system_output.expect_call(cmd).and_return('\n field: match') 193 self.assertEquals(mock_netif.parse_ethtool('field', 'some|match'), 194 'match') 195 196 self.god.check_playback() 197 198 utils.system_output.expect_call(cmd).and_return(None) 199 self.assertEquals(mock_netif.parse_ethtool('field', 200 'some|match'), '') 201 202 utils.system_output.expect_call(cmd).and_return(' field: match') 203 self.assertEquals(mock_netif.parse_ethtool('field', 204 'some|match'), '') 205 self.god.check_playback() 206 207 208 def test_network_interface_get_stats(self): 209 mock_netif = self.network_interface_mock() 210 self.god.stub_function(os, 'listdir') 211 stat_path = '/sys/class/net/%s/statistics/' % mock_netif._name 212 213 # no stat found 214 os.listdir.expect_call(stat_path).and_return(()) 215 self.assertEquals(mock_netif.get_stats(), {}) 216 self.god.check_playback() 217 218 # can not open stat file 219 os.listdir.expect_call(stat_path).and_return(('some_stat',)) 220 f = self.god.create_mock_class(file, 'file') 221 net_utils.open.expect_call(stat_path + 'some_stat', 'r').and_return(None) 222 self.assertEquals(mock_netif.get_stats(), {}) 223 self.god.check_playback() 224 225 # found a single stat 226 os.listdir.expect_call(stat_path).and_return(('some_stat',)) 227 f = self.god.create_mock_class(file, 'file') 228 net_utils.open.expect_call(stat_path + 'some_stat', 'r').and_return(f) 229 f.read.expect_call().and_return(1234) 230 f.close.expect_call() 231 self.assertEquals(mock_netif.get_stats(), {'some_stat':1234}) 232 self.god.check_playback() 233 234 # found multiple stats 235 os.listdir.expect_call(stat_path).and_return(('stat1','stat2')) 236 f = self.god.create_mock_class(file, 'file') 237 net_utils.open.expect_call(stat_path + 'stat1', 'r').and_return(f) 238 f.read.expect_call().and_return(1234) 239 f.close.expect_call() 240 net_utils.open.expect_call(stat_path + 'stat2', 'r').and_return(f) 241 f.read.expect_call().and_return(5678) 242 f.close.expect_call() 243 244 self.assertEquals(mock_netif.get_stats(), {'stat1':1234, 'stat2':5678}) 245 self.god.check_playback() 246 247 248 def test_network_interface_get_stats_diff(self): 249 mock_netif = self.network_interface_mock() 250 self.god.stub_function(os, 'listdir') 251 stat_path = '/sys/class/net/%s/statistics/' % mock_netif._name 252 253 os.listdir.expect_call(stat_path).and_return(('stat1','stat2', 'stat4')) 254 f = self.god.create_mock_class(file, 'file') 255 net_utils.open.expect_call(stat_path + 'stat1', 'r').and_return(f) 256 f.read.expect_call().and_return(1234) 257 f.close.expect_call() 258 net_utils.open.expect_call(stat_path + 'stat2', 'r').and_return(f) 259 f.read.expect_call().and_return(0) 260 f.close.expect_call() 261 net_utils.open.expect_call(stat_path + 'stat4', 'r').and_return(f) 262 f.read.expect_call().and_return(10) 263 f.close.expect_call() 264 self.assertEquals(mock_netif.get_stats_diff({'stat1':1, 'stat2':2, 265 'stat3':0}), 266 {'stat1':1233, 'stat2':-2, 'stat4':10}) 267 self.god.check_playback() 268 269 270 def test_network_interface_get_driver(self): 271 mock_netif = self.network_interface_mock() 272 mock_netif.get_driver = net_utils.network_interface.get_driver 273 self.god.stub_function(os, 'readlink') 274 stat_path = '/sys/class/net/%s/device/driver' % mock_netif._name 275 os.readlink.expect_call(stat_path).and_return(( 276 stat_path+'/driver_name')) 277 self.assertEquals(mock_netif.get_driver(mock_netif), 'driver_name') 278 self.god.check_playback() 279 280 281 def test_network_interface_get_carrier(self): 282 mock_netif = self.network_interface_mock() 283 self.god.stub_function(os, 'readlink') 284 stat_path = '/sys/class/net/%s/carrier' % mock_netif._name 285 f = self.god.create_mock_class(file, 'file') 286 net_utils.open.expect_call(stat_path).and_return(f) 287 f.read.expect_call().and_return(' 1 ') 288 f.close.expect_call() 289 self.assertEquals(mock_netif.get_carrier(), '1') 290 self.god.check_playback() 291 292 net_utils.open.expect_call(stat_path).and_return(f) 293 f.read.expect_call().and_return(' 0 ') 294 f.close.expect_call() 295 self.assertEquals(mock_netif.get_carrier(), '0') 296 self.god.check_playback() 297 298 net_utils.open.expect_call(stat_path).and_return(f) 299 f.read.expect_call().and_return('') 300 f.close.expect_call() 301 self.assertEquals(mock_netif.get_carrier(), '') 302 self.god.check_playback() 303 304 net_utils.open.expect_call(stat_path).and_return(None) 305 self.assertEquals(mock_netif.get_carrier(), '') 306 self.god.check_playback() 307 308 309 def test_network_interface_is_autoneg_advertised(self): 310 mock_netif = self.network_interface_mock() 311 cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name) 312 313 utils.system_output.expect_call(cmd).and_return( 314 '\n Advertised auto-negotiation: Yes') 315 self.assertEquals(mock_netif.is_autoneg_advertised(), True) 316 self.god.check_playback() 317 318 utils.system_output.expect_call(cmd).and_return( 319 '\n Advertised auto-negotiation: No') 320 self.assertEquals(mock_netif.is_autoneg_advertised(), False) 321 self.god.check_playback() 322 323 utils.system_output.expect_call(cmd).and_return('') 324 self.assertEquals(mock_netif.is_autoneg_advertised(), False) 325 self.god.check_playback() 326 327 328 def test_network_interface_get_speed(self): 329 mock_netif = self.network_interface_mock() 330 cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name) 331 332 utils.system_output.expect_call(cmd).and_return( 333 '\n Speed: 1000') 334 self.assertEquals(mock_netif.get_speed(), 1000) 335 self.god.check_playback() 336 337 utils.system_output.expect_call(cmd).and_return( 338 '\n Speed: 10000') 339 self.assertEquals(mock_netif.get_speed(), 10000) 340 self.god.check_playback() 341 342 utils.system_output.expect_call(cmd).and_return('') 343 344 try: 345 mock_netif.get_speed() 346 except ValueError: 347 pass 348 else: 349 self.assertEquals(0,1) 350 self.god.check_playback() 351 352 353 def test_network_interface_is_full_duplex(self): 354 mock_netif = self.network_interface_mock() 355 cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name) 356 357 utils.system_output.expect_call(cmd).and_return( 358 '\n Duplex: Full') 359 self.assertEquals(mock_netif.is_full_duplex(), True) 360 self.god.check_playback() 361 362 utils.system_output.expect_call(cmd).and_return( 363 '\n Duplex: Half') 364 self.assertEquals(mock_netif.is_full_duplex(), False) 365 self.god.check_playback() 366 367 utils.system_output.expect_call(cmd).and_return('') 368 self.assertEquals(mock_netif.is_full_duplex(), False) 369 self.god.check_playback() 370 371 372 def test_network_interface_is_autoneg_on(self): 373 mock_netif = self.network_interface_mock() 374 cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name) 375 376 utils.system_output.expect_call(cmd).and_return( 377 '\n Auto-negotiation: on') 378 self.assertEquals(mock_netif.is_autoneg_on(), True) 379 self.god.check_playback() 380 381 utils.system_output.expect_call(cmd).and_return( 382 '\n Auto-negotiation: off') 383 self.assertEquals(mock_netif.is_autoneg_on(), False) 384 self.god.check_playback() 385 386 utils.system_output.expect_call(cmd).and_return('') 387 self.assertEquals(mock_netif.is_autoneg_on(), False) 388 self.god.check_playback() 389 390 391 def test_network_interface_get_wakeon(self): 392 mock_netif = self.network_interface_mock() 393 cmd = '%s %s %s' % (mock_netif.ethtool, '', mock_netif._name) 394 395 utils.system_output.expect_call(cmd).and_return( 396 '\n Wake-on: g') 397 self.assertEquals(mock_netif.get_wakeon(), 'g') 398 self.god.check_playback() 399 400 401 def test_network_interface_is_rx_summing_on(self): 402 mock_netif = self.network_interface_mock() 403 cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name) 404 405 utils.system_output.expect_call(cmd).and_return( 406 '\n rx-checksumming: on') 407 self.assertEquals(mock_netif.is_rx_summing_on(), True) 408 self.god.check_playback() 409 410 utils.system_output.expect_call(cmd).and_return( 411 '\n rx-checksumming: off') 412 self.assertEquals(mock_netif.is_rx_summing_on(), False) 413 self.god.check_playback() 414 415 utils.system_output.expect_call(cmd).and_return('') 416 self.assertEquals(mock_netif.is_rx_summing_on(), False) 417 self.god.check_playback() 418 419 420 def test_network_interface_is_tx_summing_on(self): 421 mock_netif = self.network_interface_mock() 422 cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name) 423 424 utils.system_output.expect_call(cmd).and_return( 425 '\n tx-checksumming: on') 426 self.assertEquals(mock_netif.is_tx_summing_on(), True) 427 self.god.check_playback() 428 429 utils.system_output.expect_call(cmd).and_return( 430 '\n tx-checksumming: off') 431 self.assertEquals(mock_netif.is_tx_summing_on(), False) 432 self.god.check_playback() 433 434 utils.system_output.expect_call(cmd).and_return('') 435 self.assertEquals(mock_netif.is_tx_summing_on(), False) 436 self.god.check_playback() 437 438 439 def test_network_interface_is_scatter_gather_on(self): 440 mock_netif = self.network_interface_mock() 441 cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name) 442 443 utils.system_output.expect_call(cmd).and_return( 444 '\n scatter-gather: on') 445 self.assertEquals(mock_netif.is_scatter_gather_on(), True) 446 self.god.check_playback() 447 448 utils.system_output.expect_call(cmd).and_return( 449 '\n scatter-gather: off') 450 self.assertEquals(mock_netif.is_scatter_gather_on(), False) 451 self.god.check_playback() 452 453 utils.system_output.expect_call(cmd).and_return('') 454 self.assertEquals(mock_netif.is_scatter_gather_on(), False) 455 self.god.check_playback() 456 457 458 def test_network_interface_is_tso_on(self): 459 mock_netif = self.network_interface_mock() 460 cmd = '%s %s %s' % (mock_netif.ethtool, '-k', mock_netif._name) 461 462 utils.system_output.expect_call(cmd).and_return( 463 '\n tcp segmentation offload: on') 464 self.assertEquals(mock_netif.is_tso_on(), True) 465 self.god.check_playback() 466 467 utils.system_output.expect_call(cmd).and_return( 468 '\n tcp segmentation offload: off') 469 self.assertEquals(mock_netif.is_tso_on(), False) 470 self.god.check_playback() 471 472 utils.system_output.expect_call(cmd).and_return('') 473 self.assertEquals(mock_netif.is_tso_on(), False) 474 self.god.check_playback() 475 476 477 def test_network_interface_is_pause_autoneg_on(self): 478 mock_netif = self.network_interface_mock() 479 cmd = '%s %s %s' % (mock_netif.ethtool, '-a', mock_netif._name) 480 481 utils.system_output.expect_call(cmd).and_return( 482 '\n Autonegotiate: on') 483 self.assertEquals(mock_netif.is_pause_autoneg_on(), True) 484 self.god.check_playback() 485 486 utils.system_output.expect_call(cmd).and_return( 487 '\n Autonegotiate: off') 488 self.assertEquals(mock_netif.is_pause_autoneg_on(), False) 489 self.god.check_playback() 490 491 utils.system_output.expect_call(cmd).and_return('') 492 self.assertEquals(mock_netif.is_pause_autoneg_on(), False) 493 self.god.check_playback() 494 495 496 def test_network_interface_is_tx_pause_on(self): 497 mock_netif = self.network_interface_mock() 498 cmd = '%s %s %s' % (mock_netif.ethtool, '-a', mock_netif._name) 499 500 utils.system_output.expect_call(cmd).and_return( 501 '\n TX: on') 502 self.assertEquals(mock_netif.is_tx_pause_on(), True) 503 self.god.check_playback() 504 505 utils.system_output.expect_call(cmd).and_return( 506 '\n TX: off') 507 self.assertEquals(mock_netif.is_tx_pause_on(), False) 508 self.god.check_playback() 509 510 utils.system_output.expect_call(cmd).and_return('') 511 self.assertEquals(mock_netif.is_tx_pause_on(), False) 512 self.god.check_playback() 513 514 515 def test_network_interface_is_rx_pause_on(self): 516 mock_netif = self.network_interface_mock() 517 cmd = '%s %s %s' % (mock_netif.ethtool, '-a', mock_netif._name) 518 519 utils.system_output.expect_call(cmd).and_return( 520 '\n RX: on') 521 self.assertEquals(mock_netif.is_rx_pause_on(), True) 522 self.god.check_playback() 523 524 utils.system_output.expect_call(cmd).and_return( 525 '\n RX: off') 526 self.assertEquals(mock_netif.is_rx_pause_on(), False) 527 self.god.check_playback() 528 529 utils.system_output.expect_call(cmd).and_return('') 530 self.assertEquals(mock_netif.is_rx_pause_on(), False) 531 self.god.check_playback() 532 533 534 def test_network_interface_enable_loopback(self): 535 mock_netif = self.network_interface_mock('eth0') 536 537 mock_netif.was_loopback_enabled = False 538 mock_netif.loopback_enabled = False 539 mock_netif.was_down = False 540 541 self.god.stub_function(net_utils.bonding, 'is_enabled') 542 543 # restore using phyint 544 net_utils.bonding.is_enabled.expect_call().and_return(False) 545 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 546 'phyint', 'enable') 547 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 548 mock_netif.enable_loopback() 549 self.god.check_playback() 550 551 # restore using mac 552 net_utils.bonding.is_enabled.expect_call().and_return(False) 553 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 554 'phyint', 'enable') 555 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 556 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 557 'mac', 'enable') 558 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 559 mock_netif.enable_loopback() 560 self.god.check_playback() 561 562 # catch exception on phyint and mac failures 563 net_utils.bonding.is_enabled.expect_call().and_return(False) 564 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 565 'phyint', 'enable') 566 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 567 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 568 'mac', 'enable') 569 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 570 try: 571 mock_netif.enable_loopback() 572 except error.TestError: 573 pass 574 else: 575 self.assertEquals(0,1) 576 self.god.check_playback() 577 578 # catch exception on bond enabled 579 net_utils.bonding.is_enabled.expect_call().and_return(True) 580 try: 581 mock_netif.enable_loopback() 582 except error.TestError: 583 pass 584 else: 585 self.assertEquals(0,1) 586 self.god.check_playback() 587 588 # check that setting tg3 and bnx2x driver have a sleep call 589 mock_netif.driver = 'tg3' 590 net_utils.bonding.is_enabled.expect_call().and_return(False) 591 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 592 'phyint', 'enable') 593 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 594 time.sleep.expect_call(1) 595 mock_netif.enable_loopback() 596 self.god.check_playback() 597 598 mock_netif.driver = 'bnx2x' 599 net_utils.bonding.is_enabled.expect_call().and_return(False) 600 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 601 'phyint', 'enable') 602 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 603 time.sleep.expect_call(1) 604 mock_netif.enable_loopback() 605 self.god.check_playback() 606 607 608 def test_network_interface_disable_loopback(self): 609 mock_netif = self.network_interface_mock('eth0') 610 611 mock_netif.was_loopback_enabled = False 612 mock_netif.loopback_enabled = True 613 mock_netif.was_down = False 614 615 # restore using phyint 616 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 617 'phyint', 'disable') 618 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 619 mock_netif.disable_loopback() 620 self.god.check_playback() 621 622 # restore using mac 623 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 624 'phyint', 'disable') 625 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 626 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 627 'mac', 'disable') 628 utils.system.expect_call(cmd, ignore_status=True).and_return(0) 629 mock_netif.disable_loopback() 630 self.god.check_playback() 631 632 # catch exception on phyint and mac failures 633 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 634 'phyint', 'disable') 635 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 636 cmd = '%s -L %s %s %s' % (mock_netif.ethtool, mock_netif._name, 637 'mac', 'disable') 638 utils.system.expect_call(cmd, ignore_status=True).and_return(1) 639 try: 640 mock_netif.disable_loopback() 641 except error.TestError: 642 pass 643 else: 644 self.assertEquals(0,1) 645 self.god.check_playback() 646 647 648 def test_network_interface_is_loopback_enabled(self): 649 mock_netif = self.network_interface_mock('eth0') 650 mock_netif.is_loopback_enabled = \ 651 net_utils.network_interface.is_loopback_enabled 652 try: 653 mock_netif.is_loopback_enabled(mock_netif) 654 except error.TestError: 655 pass 656 else: 657 self.assertEquals(0,1) 658 self.god.check_playback() 659 660 self.god.stub_function(net_utils.bonding, 'is_enabled') 661 mock_netif._name = 'eth0' 662 net_utils.bonding.is_enabled.expect_call().and_return(False) 663 cmd = '%s -l %s' % (mock_netif.ethtool, mock_netif._name) 664 utils.system_output.expect_call(cmd).and_return('') 665 self.assertEquals(mock_netif.is_loopback_enabled(mock_netif), False) 666 self.god.check_playback() 667 668 for ifname in ('eth0', 'eth1', 'eth2', 'eth3', 'eth4'): 669 mock_netif._name = ifname 670 for bond_enable in (True, False): 671 for state in (('disabled', 'disabled', 'enabled'), 672 ('disabled', 'enabled', 'disabled'), 673 ('enabled', 'disabled', 'disabled'), 674 ('disabled', 'disabled', 'disabled')): 675 net_utils.bonding.is_enabled.expect_call().and_return( 676 bond_enable) 677 if bond_enable: 678 self.assertEquals(mock_netif.is_loopback_enabled( 679 mock_netif), False) 680 else: 681 cmd = '%s -l %s' % (mock_netif.ethtool, mock_netif._name) 682 out = 'MAC loopback is %s\n'\ 683 'PHY internal loopback is %s\n'\ 684 'PHY external loopback is %s' % ( 685 state[0], state[1], state[2]) 686 utils.system_output.expect_call(cmd).and_return(out) 687 self.assertEquals(mock_netif.is_loopback_enabled( 688 mock_netif), 'enabled' in state) 689 self.god.check_playback() 690 691 692 def test_network_interface_enable_promisc(self): 693 mock_netif = self.network_interface_mock('eth0') 694 cmd = 'ifconfig %s promisc' % mock_netif._name 695 utils.system.expect_call(cmd) 696 mock_netif.enable_promisc() 697 self.god.check_playback() 698 699 700 def test_network_interface_disable_promisc(self): 701 mock_netif = self.network_interface_mock() 702 cmd = 'ifconfig %s -promisc' % mock_netif._name 703 utils.system.expect_call(cmd) 704 mock_netif.disable_promisc() 705 self.god.check_playback() 706 707 708 def test_network_interface_get_hwaddr(self): 709 mock_netif = self.network_interface_mock() 710 f = self.god.create_mock_class(file, 'file') 711 net_utils.open.expect_call('/sys/class/net/%s/address' 712 % mock_netif._name).and_return(f) 713 hw_addr = '00:0e:0c:c3:7d:a8' 714 f.read.expect_call().and_return(' ' + hw_addr + ' ') 715 f.close.expect_call() 716 self.assertEquals(mock_netif.get_hwaddr(), hw_addr) 717 self.god.check_playback() 718 719 720 def test_network_interface_set_hwaddr(self): 721 mock_netif = self.network_interface_mock() 722 hw_addr = '00:0e:0c:c3:7d:a8' 723 cmd = 'ifconfig %s hw ether %s' % (mock_netif._name, 724 hw_addr) 725 utils.system.expect_call(cmd) 726 mock_netif.set_hwaddr(hw_addr) 727 self.god.check_playback() 728 729 730 def test_network_interface_add_maddr(self): 731 mock_netif = self.network_interface_mock() 732 maddr = '01:00:5e:00:00:01' 733 cmd = 'ip maddr add %s dev %s' % (maddr, mock_netif._name) 734 utils.system.expect_call(cmd) 735 mock_netif.add_maddr(maddr) 736 self.god.check_playback() 737 738 739 def test_network_interface_del_maddr(self): 740 mock_netif = self.network_interface_mock() 741 maddr = '01:00:5e:00:00:01' 742 cmd = 'ip maddr del %s dev %s' % (maddr, mock_netif._name) 743 utils.system.expect_call(cmd) 744 mock_netif.del_maddr(maddr) 745 self.god.check_playback() 746 747 748 def test_network_interface_get_ipaddr(self): 749 mock_netif = self.network_interface_mock() 750 ip_addr = '110.211.112.213' 751 out_format = \ 752 'eth0 Link encap:Ethernet HWaddr 00:0E:0C:C3:7D:A8\n'\ 753 ' inet addr:%s Bcast:10.246.90.255'\ 754 ' Mask:255.255.255.0\n'\ 755 ' UP BROADCAST RUNNING MASTER MULTICAST MTU:1500'\ 756 ' Metric:1\n'\ 757 ' RX packets:463070 errors:0 dropped:0 overruns:0'\ 758 ' frame:0\n'\ 759 ' TX packets:32548 errors:0 dropped:0 overruns:0'\ 760 ' carrier:0\n'\ 761 ' collisions:0 txqueuelen:0' 762 out = out_format % ip_addr 763 764 cmd = 'ifconfig %s' % mock_netif._name 765 utils.system_output.expect_call(cmd).and_return(out) 766 self.assertEquals(mock_netif.get_ipaddr(), ip_addr) 767 self.god.check_playback() 768 769 cmd = 'ifconfig %s' % mock_netif._name 770 utils.system_output.expect_call(cmd).and_return('some output') 771 self.assertEquals(mock_netif.get_ipaddr(), '0.0.0.0') 772 self.god.check_playback() 773 774 cmd = 'ifconfig %s' % mock_netif._name 775 utils.system_output.expect_call(cmd).and_return(None) 776 self.assertEquals(mock_netif.get_ipaddr(), '0.0.0.0') 777 self.god.check_playback() 778 779 ip_addr = '1.2.3.4' 780 out = out_format % ip_addr 781 cmd = 'ifconfig %s' % mock_netif._name 782 utils.system_output.expect_call(cmd).and_return(out) 783 self.assertEquals(mock_netif.get_ipaddr(), ip_addr) 784 self.god.check_playback() 785 786 787 def test_network_interface_set_ipaddr(self): 788 mock_netif = self.network_interface_mock() 789 ip_addr = '1.2.3.4' 790 cmd = 'ifconfig %s %s' % (mock_netif._name, ip_addr) 791 utils.system.expect_call(cmd) 792 mock_netif.set_ipaddr(ip_addr) 793 self.god.check_playback() 794 795 796 def test_network_interface_is_down(self): 797 mock_netif = self.network_interface_mock() 798 out_format = \ 799 'eth0 Link encap:Ethernet HWaddr 00:0E:0C:C3:7D:A8\n'\ 800 ' inet addr:1.2.3.4 Bcast:10.246.90.255'\ 801 ' Mask:255.255.255.0\n'\ 802 ' %s BROADCAST RUNNING MASTER MULTICAST MTU:1500'\ 803 ' Metric:1\n'\ 804 ' RX packets:463070 errors:0 dropped:0 overruns:0'\ 805 ' frame:0\n'\ 806 ' TX packets:32548 errors:0 dropped:0 overruns:0'\ 807 ' carrier:0\n'\ 808 ' collisions:0 txqueuelen:0' 809 for state in ('UP', 'DOWN', 'NONE', ''): 810 out = out_format % state 811 cmd = 'ifconfig %s' % mock_netif._name 812 utils.system_output.expect_call(cmd).and_return(out) 813 self.assertEquals(mock_netif.is_down(), state != 'UP') 814 self.god.check_playback() 815 816 cmd = 'ifconfig %s' % mock_netif._name 817 utils.system_output.expect_call(cmd).and_return(None) 818 self.assertEquals(mock_netif.is_down(), False) 819 self.god.check_playback() 820 821 822 def test_network_interface_up(self): 823 mock_netif = self.network_interface_mock() 824 cmd = 'ifconfig %s up' % mock_netif._name 825 utils.system.expect_call(cmd) 826 mock_netif.up() 827 self.god.check_playback() 828 829 830 def test_network_interface_down(self): 831 mock_netif = self.network_interface_mock() 832 cmd = 'ifconfig %s down' % mock_netif._name 833 utils.system.expect_call(cmd) 834 mock_netif.down() 835 self.god.check_playback() 836 837 838 def test_network_interface_wait_for_carrier(self): 839 mock_netif = self.network_interface_mock() 840 mock_netif.wait_for_carrier = \ 841 net_utils.network_interface.wait_for_carrier 842 f = self.god.create_mock_class(file, 'file') 843 spath = '/sys/class/net/%s/carrier' % mock_netif._name 844 # y = 0 - test that an exception is thrown 845 # y = 1, 100 - check that carrier is checked until timeout 846 for y in (0, 1, 100): 847 max_timeout = y 848 if y: 849 for x in xrange(max_timeout - 1): 850 net_utils.open.expect_call(spath).and_return(f) 851 f.read.expect_call().and_return(' ' + '0' + ' ') 852 f.close.expect_call() 853 time.sleep.expect_call(1) 854 855 net_utils.open.expect_call(spath).and_return(f) 856 f.read.expect_call().and_return(' ' + '1' + ' ') 857 f.close.expect_call() 858 try: 859 mock_netif.wait_for_carrier(mock_netif, max_timeout) 860 except: 861 pass 862 else: 863 if not y: 864 self.assertEquals(0, 1) 865 self.god.check_playback() 866 867 868 def test_network_interface_send(self): 869 mock_netif = self.network_interface_mock() 870 mock_netif.send('test buffer') 871 self.assertEquals(mock_netif._socket.send_val, 'test buffer') 872 873 874 def test_network_interface_recv(self): 875 mock_netif = self.network_interface_mock() 876 test_str = 'test string' 877 mock_netif._socket.recv_val = test_str 878 rcv_str = mock_netif.recv(len(test_str)) 879 self.assertEquals(rcv_str, test_str) 880 881 882 def test_network_interface_flush(self): 883 mock_netif = self.network_interface_mock() 884 self.god.stub_function(mock_netif._socket, 'close') 885 mock_netif._socket.close.expect_call() 886 s = self.god.create_mock_class(socket.socket, "socket") 887 self.god.stub_function(socket, 'socket') 888 socket.socket.expect_call(socket.PF_PACKET, 889 socket.SOCK_RAW).and_return(s) 890 s.settimeout.expect_call(net_utils.TIMEOUT) 891 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 892 893 894 # 895 # bonding tests 896 # 897 def test_bonding_is_enabled(self): 898 try: 899 net_utils.bond().is_enabled() 900 except error.TestError: 901 pass 902 else: 903 self.assertEquals(1, 0) 904 905 906 def test_bonding_is_bondable(self): 907 try: 908 net_utils.bond().is_enabled() 909 except error.TestError: 910 pass 911 else: 912 self.assertEquals(1, 0) 913 914 915 def test_bonding_enable(self): 916 try: 917 net_utils.bond().is_enabled() 918 except error.TestError: 919 pass 920 else: 921 self.assertEquals(1, 0) 922 923 924 def test_bonding_disable(self): 925 try: 926 net_utils.bond().is_enabled() 927 except error.TestError: 928 pass 929 else: 930 self.assertEquals(1, 0) 931 932 933 def test_bonding_get_mii_status(self): 934 self.assertEquals(net_utils.bond().get_mii_status(), {}) 935 936 937 def test_get_mode_bonding(self): 938 self.assertEquals(net_utils.bond().get_mode(), net_utils.bonding.NO_MODE) 939 940 941 def test_bonding_wait_for_state_change(self): 942 self.god.stub_function(utils, "ping_default_gateway") 943 944 time.sleep.expect_call(10) 945 utils.ping_default_gateway.expect_call().and_return(False) 946 self.assertEquals(net_utils.bond().wait_for_state_change(), True) 947 948 for x in xrange(9): 949 time.sleep.expect_call(10) 950 utils.ping_default_gateway.expect_call().and_return(True) 951 952 time.sleep.expect_call(10) 953 utils.ping_default_gateway.expect_call().and_return(False) 954 self.assertEquals(net_utils.bond().wait_for_state_change(), True) 955 956 for x in xrange(10): 957 time.sleep.expect_call(10) 958 utils.ping_default_gateway.expect_call().and_return(True) 959 960 self.assertEquals(net_utils.bond().wait_for_state_change(), False) 961 962 self.god.check_playback() 963 964 965 def test_bonding_get_active_interfaces(self): 966 self.assertEquals(net_utils.bond().get_active_interfaces(), []) 967 self.god.check_playback() 968 969 970 def test_bonding_get_slave_interfaces(self): 971 self.assertEquals(net_utils.bond().get_slave_interfaces(), []) 972 self.god.check_playback() 973 974 975 # 976 # ethernet tests 977 # 978 979 def test_ethernet_mac_string_to_binary(self): 980 mac_bin = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05') 981 self.assertEqual(mac_bin, '\x00\x01\x02\x03\x04\x05') 982 983 984 def test_ethernet_mac_binary_to_string(self): 985 mac_str = net_utils.ethernet.mac_binary_to_string( 986 '\x00\x01\x02\x03\x04\x05') 987 self.assertEqual(mac_str, '00:01:02:03:04:05') 988 989 990 def test_ethernet_pack(self): 991 dst = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05') 992 src = net_utils.ethernet.mac_string_to_binary('16:17:18:19:1A:1B') 993 protocol = 2030 994 payload = 'some payload' 995 frame = struct.pack("!6s6sH", dst, src, protocol) + payload 996 self.assertEquals(net_utils.ethernet.pack(dst, src,protocol, payload), 997 frame) 998 999 1000 def test_ethernet_unpack(self): 1001 dst = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05') 1002 src = net_utils.ethernet.mac_string_to_binary('16:17:18:19:1A:1B') 1003 protocol = 2030 1004 payload = 'some payload' 1005 frame = net_utils.ethernet.pack(dst, src, protocol, payload) 1006 uframe = net_utils.ethernet.unpack(frame) 1007 self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_DST_MAC], dst) 1008 self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_SRC_MAC], src) 1009 self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_PROTO], protocol) 1010 self.assertEquals(uframe[net_utils.ethernet.FRAME_KEY_PAYLOAD], payload) 1011 1012 1013 # raw_socket tests 1014 # 1015 def test_raw_socket_open(self): 1016 self.god.stub_function(socket, 'setdefaulttimeout') 1017 1018 s = self.god.create_mock_class(socket.socket, "socket") 1019 self.god.stub_function(socket, 'socket') 1020 1021 # open without a protocol 1022 socket.setdefaulttimeout.expect_call(1) 1023 socket.socket.expect_call(socket.PF_PACKET, 1024 socket.SOCK_RAW).and_return(s) 1025 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1026 s.settimeout.expect_call(1) 1027 sock = net_utils.raw_socket('eth0') 1028 sock.open(protocol=None) 1029 1030 self.god.check_playback() 1031 1032 # double open should throw an exception 1033 try: 1034 sock.open() 1035 except error.TestError: 1036 pass 1037 else: 1038 self.assertEquals(1, 0) 1039 1040 self.god.check_playback() 1041 1042 # open a protocol 1043 socket.setdefaulttimeout.expect_call(1) 1044 socket.socket.expect_call(socket.PF_PACKET, 1045 socket.SOCK_RAW, 1046 socket.htons(1234)).and_return(s) 1047 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1048 s.settimeout.expect_call(1) 1049 sock = net_utils.raw_socket('eth0') 1050 sock.open(protocol=1234) 1051 1052 self.god.check_playback() 1053 1054 1055 def test_raw_socket_close(self): 1056 self.god.stub_function(socket, 'setdefaulttimeout') 1057 1058 s = self.god.create_mock_class(socket.socket, "socket") 1059 self.god.stub_function(socket, 'socket') 1060 1061 # close without open 1062 socket.setdefaulttimeout.expect_call(1) 1063 sock = net_utils.raw_socket('eth0') 1064 try: 1065 sock.close() 1066 except error.TestError: 1067 pass 1068 else: 1069 self.assertEquals(1, 0) 1070 1071 # close after open 1072 socket.setdefaulttimeout.expect_call(1) 1073 socket.socket.expect_call(socket.PF_PACKET, 1074 socket.SOCK_RAW).and_return(s) 1075 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1076 s.settimeout.expect_call(1) 1077 sock = net_utils.raw_socket('eth0') 1078 sock.open(protocol=None) 1079 1080 s.close.expect_call() 1081 sock.close() 1082 self.god.check_playback() 1083 1084 1085 def test_raw_socket_recv(self): 1086 self.god.stub_function(socket, 'setdefaulttimeout') 1087 1088 self.god.create_mock_class(socket.socket, "socket") 1089 self.god.stub_function(socket, 'socket') 1090 1091 # rcv without open 1092 socket.setdefaulttimeout.expect_call(1) 1093 sock = net_utils.raw_socket('eth0') 1094 try: 1095 sock.recv(10) 1096 except error.TestError: 1097 pass 1098 else: 1099 self.assertEquals(1, 0) 1100 1101 self.god.check_playback() 1102 1103 # open a protocol and try to get packets of varying sizes 1104 # I could not get socket.recv to get a mock expect_call. To keep 1105 # on going, added a socket stub 1106 s = net_utils_mock.socket_stub('eth0', socket, socket) 1107 socket.socket.expect_call(socket.PF_PACKET, 1108 socket.SOCK_RAW, 1109 socket.htons(1234)).and_return(s) 1110 1111 self.god.stub_function(s, 'bind') 1112 self.god.stub_function(s, 'settimeout') 1113 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1114 s.settimeout.expect_call(1) 1115 sock.open(protocol=1234) 1116 1117 s.recv_val = '' 1118 self.assertEquals(sock.recv(1), (None, 0)) 1119 1120 s.recv_val = '\xFF' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE-5) 1121 self.assertEquals(sock.recv(1), (None, 0)) 1122 1123 # when receiving a packet, make sure the timeout is not change 1124 s.recv_val = '\xEE' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE-4) 1125 self.assertEquals(sock.recv(1), (s.recv_val, 1)) 1126 1127 s.recv_val = '\xDD' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE) 1128 self.assertEquals(sock.recv(1), (s.recv_val, 1)) 1129 1130 s.recv_val = '\xCC' * (net_utils.ethernet.ETH_PACKET_MAX_SIZE) 1131 self.assertEquals(sock.recv(1), (s.recv_val, 1)) 1132 1133 s.recv_val = '\xBB' * (net_utils.ethernet.ETH_PACKET_MAX_SIZE+1) 1134 packet, time_left = sock.recv(1) 1135 self.assertEquals(len(packet), net_utils.ethernet.ETH_PACKET_MAX_SIZE) 1136 self.assertEquals(packet, 1137 s.recv_val[:net_utils.ethernet.ETH_PACKET_MAX_SIZE]) 1138 1139 1140 # test timeout 1141 s.recv_val = '' 1142 s.throw_timeout = False 1143 sock.recv(5) 1144 self.assertEquals(sock.recv(1), (None, 0)) 1145 s.throw_timeout = True 1146 sock.recv(5) 1147 self.assertEquals(sock.recv(1), (None, 0)) 1148 1149 self.god.check_playback() 1150 1151 1152 def test_raw_socket_send(self): 1153 self.god.stub_function(socket, 'setdefaulttimeout') 1154 self.god.create_mock_class(socket.socket, "socket") 1155 self.god.stub_function(socket, 'socket') 1156 self.god.stub_function(socket, 'send') 1157 1158 # send without open 1159 socket.setdefaulttimeout.expect_call(1) 1160 sock = net_utils.raw_socket('eth0') 1161 try: 1162 sock.send('test this packet') 1163 except error.TestError: 1164 pass 1165 else: 1166 self.assertEquals(1, 0) 1167 self.god.check_playback() 1168 1169 # open a protocol and try to send a packet 1170 s = net_utils_mock.socket_stub('eth0', socket, socket) 1171 self.god.stub_function(s, 'bind') 1172 self.god.stub_function(s, 'settimeout') 1173 socket.socket.expect_call(socket.PF_PACKET, 1174 socket.SOCK_RAW, 1175 socket.htons(1234)).and_return(s) 1176 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1177 s.settimeout.expect_call(1) 1178 packet = '\xFF\xAA\xBB\xCC\xDD\x11packet data\x00\x00' 1179 s.send.expect_call(packet) 1180 sock.open(protocol=1234) 1181 sock.send(packet) 1182 self.god.check_playback() 1183 1184 1185 def test_raw_socket_send_to(self): 1186 self.god.stub_function(socket, 'setdefaulttimeout') 1187 self.god.create_mock_class(socket.socket, "socket") 1188 self.god.stub_function(socket, 'socket') 1189 self.god.stub_function(socket, 'send') 1190 1191 # send without open 1192 socket.setdefaulttimeout.expect_call(1) 1193 sock = net_utils.raw_socket('eth0') 1194 try: 1195 sock.send_to('0', '1', 1, 'test this packet') 1196 except error.TestError: 1197 pass 1198 else: 1199 self.assertEquals(1, 0) 1200 self.god.check_playback() 1201 1202 # open a protocol and try to send a packet 1203 s = net_utils_mock.socket_stub('eth0', socket, socket) 1204 self.god.stub_function(s, 'bind') 1205 self.god.stub_function(s, 'settimeout') 1206 socket.socket.expect_call(socket.PF_PACKET, 1207 socket.SOCK_RAW, 1208 socket.htons(1234)).and_return(s) 1209 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1210 s.settimeout.expect_call(1) 1211 packet = '\x00\x00packet data\x00\x00' 1212 s.send.expect_call(packet) 1213 sock.open(protocol=1234) 1214 try: 1215 sock.send_to(None, None, 1, packet) 1216 except error.TestError: 1217 pass 1218 else: 1219 self.assertEquals(1, 0) 1220 self.god.check_playback() 1221 1222 dst_mac = '\x00\x01\x02\x03\x04\x05' 1223 src_mac = '\xFF\xEE\xDD\xCC\xBB\xAA' 1224 protocol = 1234 1225 s.send.expect_call(dst_mac+src_mac+'%d'%protocol+packet) 1226 sock.send_to(dst_mac, src_mac, protocol, packet) 1227 self.god.check_playback() 1228 1229 1230 def test_raw_socket_recv_from(self): 1231 1232 def __set_clock(sock): 1233 time.clock.expect_call().and_return(0.0) 1234 time.clock.expect_call().and_return(0.0) 1235 time.clock.expect_call().and_return(float(sock.socket_timeout()) + 0.5) 1236 1237 self.god.stub_function(socket, 'setdefaulttimeout') 1238 1239 self.god.create_mock_class(socket.socket, "socket") 1240 self.god.stub_function(socket, 'socket') 1241 1242 # rcv without open 1243 socket.setdefaulttimeout.expect_call(1) 1244 sock = net_utils.raw_socket('eth0') 1245 try: 1246 sock.recv_from(None, None, None) 1247 except error.TestError: 1248 pass 1249 else: 1250 self.assertEquals(1, 0) 1251 1252 self.god.check_playback() 1253 1254 # open a protocol and try to get packets of varying sizes 1255 # I could not get socket.recv to get a mock expect_call. To keep 1256 # on going, added a socket stub 1257 s = net_utils_mock.socket_stub('eth0', socket, socket) 1258 socket.socket.expect_call(socket.PF_PACKET, 1259 socket.SOCK_RAW, 1260 socket.htons(1234)).and_return(s) 1261 1262 self.god.stub_function(s, 'bind') 1263 self.god.stub_function(s, 'settimeout') 1264 s.bind.expect_call(('eth0', net_utils.raw_socket.ETH_P_ALL)) 1265 s.settimeout.expect_call(1) 1266 sock.open(protocol=1234) 1267 1268 s.recv_val = '' 1269 dst_mac = net_utils.ethernet.mac_string_to_binary('00:01:02:03:04:05') 1270 src_mac = net_utils.ethernet.mac_string_to_binary('16:17:18:19:1A:1B') 1271 t_mac = net_utils.ethernet.mac_string_to_binary('E6:E7:E8:E9:EA:EB') 1272 protocol = 2030 1273 t_protocol = 1234 1274 data = '\xEE' * (net_utils.ethernet.ETH_PACKET_MIN_SIZE) 1275 1276 # no data to receive at socket 1277 self.assertEquals(sock.recv_from(None, None, None), None) 1278 self.assertEquals(sock.recv_from(dst_mac, None, None), None) 1279 self.assertEquals(sock.recv_from(None, src_mac, None), None) 1280 self.assertEquals(sock.recv_from(None, None, protocol), None) 1281 1282 # receive packet < min size 1283 s.recv_val = (struct.pack("!6s6sH", dst_mac, src_mac, protocol) + 1284 'packet_to_short') 1285 self.assertEquals(sock.recv_from(None, None, None), None) 1286 1287 # receive packet, filtering on mac address and protocol 1288 s.recv_val = struct.pack("!6s6sH", dst_mac, t_mac, t_protocol) + data 1289 frame = net_utils.ethernet.unpack(s.recv_val) 1290 self.assertEquals(sock.recv_from(None, None, None), frame) 1291 self.assertEquals(sock.recv_from(dst_mac, None, None), frame) 1292 1293 # use time clock to speed up the timeout in send_to() 1294 self.god.stub_function(time, 'clock') 1295 __set_clock(sock) 1296 self.assertEquals(sock.recv_from(dst_mac, src_mac, None), None) 1297 __set_clock(sock) 1298 self.assertEquals(sock.recv_from(dst_mac, None, protocol), None) 1299 __set_clock(sock) 1300 self.assertEquals(sock.recv_from(dst_mac, src_mac, protocol), None) 1301 self.god.unstub(time, 'clock') 1302 1303 s.recv_val = struct.pack("!6s6sH", dst_mac, src_mac, protocol) + data 1304 frame = net_utils.ethernet.unpack(s.recv_val) 1305 self.assertEquals(sock.recv_from(dst_mac, None, None), frame) 1306 self.assertEquals(sock.recv_from(dst_mac, src_mac, None), frame) 1307 self.assertEquals(sock.recv_from(dst_mac, src_mac, protocol), frame) 1308 self.assertEquals(sock.recv_from(None, None, protocol), frame) 1309 self.assertEquals(sock.recv_from(None, src_mac, None), frame) 1310 self.god.stub_function(time, 'clock') 1311 __set_clock(sock) 1312 self.assertEquals(sock.recv_from(None, None, t_protocol), None) 1313 __set_clock(sock) 1314 self.assertEquals(sock.recv_from(None, t_mac, None), None) 1315 self.god.unstub(time, 'clock') 1316 1317 1318 self.god.check_playback() 1319 1320 1321 if __name__ == "__main__": 1322 unittest.main() 1323
