kytos /
mef_eline
| @@ 513-563 (lines=51) @@ | ||
| 510 | send_flow_mods_mock.assert_any_call(uni_z.interface.switch, |
|
| 511 | expected_flow_mod_z) |
|
| 512 | ||
| 513 | @staticmethod |
|
| 514 | @patch('napps.kytos.mef_eline.models.EVC.send_flow_mods') |
|
| 515 | def test_install_nni_flows(send_flow_mods_mock): |
|
| 516 | """Test install nni flows method. |
|
| 517 | ||
| 518 | This test will verify the flows send to the send_flows_mods method. |
|
| 519 | """ |
|
| 520 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
|
| 521 | switch_id="switch_uni_a", is_valid=True) |
|
| 522 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
|
| 523 | switch_id="switch_uni_z", is_valid=True) |
|
| 524 | ||
| 525 | attributes = { |
|
| 526 | "name": "custom_name", |
|
| 527 | "uni_a": uni_a, |
|
| 528 | "uni_z": uni_z, |
|
| 529 | "primary_links": [ |
|
| 530 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
|
| 531 | metadata={"s_vlan": 5}), |
|
| 532 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
|
| 533 | metadata={"s_vlan": 6}) |
|
| 534 | ] |
|
| 535 | } |
|
| 536 | evc = EVC(**attributes) |
|
| 537 | evc.install_nni_flows(attributes['primary_links']) |
|
| 538 | ||
| 539 | in_vlan = evc.primary_links[0].get_metadata('s_vlan').value |
|
| 540 | out_vlan = evc.primary_links[-1].get_metadata('s_vlan').value |
|
| 541 | ||
| 542 | in_port = evc.primary_links[0].endpoint_b.port_number |
|
| 543 | out_port = evc.primary_links[-1].endpoint_a.port_number |
|
| 544 | ||
| 545 | expected_flow_mods = [ |
|
| 546 | {'match': { |
|
| 547 | 'in_port': in_port, |
|
| 548 | 'dl_vlan': in_vlan}, |
|
| 549 | 'actions': [ |
|
| 550 | {'action_type': 'set_vlan', 'vlan_id': out_vlan}, |
|
| 551 | {'action_type': 'output', 'port': out_port} |
|
| 552 | ] |
|
| 553 | }, |
|
| 554 | {'match': {'in_port': out_port, 'dl_vlan': out_vlan}, |
|
| 555 | 'actions': [ |
|
| 556 | {'action_type': 'set_vlan', 'vlan_id': in_vlan}, |
|
| 557 | {'action_type': 'output', 'port': in_port} |
|
| 558 | ] |
|
| 559 | } |
|
| 560 | ] |
|
| 561 | ||
| 562 | switch = evc.primary_links[0].endpoint_b.switch |
|
| 563 | send_flow_mods_mock.assert_called_once_with(switch, expected_flow_mods) |
|
| 564 | ||
| 565 | @patch('napps.kytos.mef_eline.models.log') |
|
| 566 | @patch('napps.kytos.mef_eline.models.EVC.choose_vlans') |
|
| @@ 242-292 (lines=51) @@ | ||
| 239 | send_flow_mods_mock.assert_any_call(uni_z.interface.switch, |
|
| 240 | expected_flow_mod_z) |
|
| 241 | ||
| 242 | @staticmethod |
|
| 243 | @patch('napps.kytos.mef_eline.models.EVC.send_flow_mods') |
|
| 244 | def test_install_nni_flows(send_flow_mods_mock): |
|
| 245 | """Test install nni flows method. |
|
| 246 | ||
| 247 | This test will verify the flows send to the send_flows_mods method. |
|
| 248 | """ |
|
| 249 | uni_a = get_uni_mocked(interface_port=2, tag_value=82, |
|
| 250 | switch_id="switch_uni_a", is_valid=True) |
|
| 251 | uni_z = get_uni_mocked(interface_port=3, tag_value=83, |
|
| 252 | switch_id="switch_uni_z", is_valid=True) |
|
| 253 | ||
| 254 | attributes = { |
|
| 255 | "name": "custom_name", |
|
| 256 | "uni_a": uni_a, |
|
| 257 | "uni_z": uni_z, |
|
| 258 | "primary_links": [ |
|
| 259 | get_link_mocked(endpoint_a_port=9, endpoint_b_port=10, |
|
| 260 | metadata={"s_vlan": 5}), |
|
| 261 | get_link_mocked(endpoint_a_port=11, endpoint_b_port=12, |
|
| 262 | metadata={"s_vlan": 6}) |
|
| 263 | ] |
|
| 264 | } |
|
| 265 | evc = EVC(**attributes) |
|
| 266 | evc.install_nni_flows(attributes['primary_links']) |
|
| 267 | ||
| 268 | in_vlan = evc.primary_links[0].get_metadata('s_vlan').value |
|
| 269 | out_vlan = evc.primary_links[-1].get_metadata('s_vlan').value |
|
| 270 | ||
| 271 | in_port = evc.primary_links[0].endpoint_b.port_number |
|
| 272 | out_port = evc.primary_links[-1].endpoint_a.port_number |
|
| 273 | ||
| 274 | expected_flow_mods = [ |
|
| 275 | {'match': { |
|
| 276 | 'in_port': in_port, |
|
| 277 | 'dl_vlan': in_vlan}, |
|
| 278 | 'actions': [ |
|
| 279 | {'action_type': 'set_vlan', 'vlan_id': out_vlan}, |
|
| 280 | {'action_type': 'output', 'port': out_port} |
|
| 281 | ] |
|
| 282 | }, |
|
| 283 | {'match': {'in_port': out_port, 'dl_vlan': out_vlan}, |
|
| 284 | 'actions': [ |
|
| 285 | {'action_type': 'set_vlan', 'vlan_id': in_vlan}, |
|
| 286 | {'action_type': 'output', 'port': in_port} |
|
| 287 | ] |
|
| 288 | } |
|
| 289 | ] |
|
| 290 | ||
| 291 | switch = evc.primary_links[0].endpoint_b.switch |
|
| 292 | send_flow_mods_mock.assert_called_once_with(switch, expected_flow_mods) |
|
| 293 | ||
| 294 | @patch('napps.kytos.mef_eline.models.log') |
|
| 295 | @patch('napps.kytos.mef_eline.models.EVC.choose_vlans') |
|
