build.tests.test_main

Complexity

Total Complexity

18

Size/Duplication

Total Lines	328
Duplicated Lines	0 %

Importance

Changes

0

17 Methods

Rating	Name	Duplication	Size	Complexity
A	TestMain._get_switch_mock()	0	13	1
A	TestMain.setUp()	0	9	1
A	TestMain._get_controller_mock()	0	7	1
A	TestMain.test_handle_01_features_reply()	0	23	1
A	TestMain.test_handle_echo_request()	0	13	1
A	TestMain.test_handle_features_request_sent()	0	14	1
A	TestMain.test_handle_stats_reply()	0	21	1
A	TestMain.test_handle_04_features_reply()	0	17	1
A	TestMain.test_get_event_listeners()	0	16	2
A	TestMain.test_execute()	0	11	1
A	TestMain._get_interface()	0	6	1
A	TestMain.test_handle_port_status_raw_in()	0	17	1
A	TestMain.test_handle_multipart_reply()	0	37	1
A	TestMain.test_handle_packet_in_raw_in()	0	21	1
A	TestMain.test_pack_generic_hello()	0	6	1
A	TestMain.test_handle_hello_raw_in()	0	17	1
A	TestMain.test_handle_port_desc_multipart_reply()	0	35	1

"""Test main."""

from unittest import TestCase
from unittest.mock import Mock

from pyof.utils import unpack
from pyof.v0x01.controller2switch.features_reply import \
    FeaturesReply as FReply_v0x01
from pyof.v0x01.controller2switch.stats_reply import StatsReply
from pyof.v0x04.controller2switch.features_reply import \
    FeaturesReply as FReply_v0x04
from pyof.v0x04.controller2switch.features_request import FeaturesRequest
from pyof.v0x04.controller2switch.multipart_reply import MultipartReply
from pyof.v0x04.symmetric.echo_request import EchoRequest

from kytos.core import Controller
from kytos.core.config import KytosConfig
from kytos.core.connection import Connection, ConnectionState
from kytos.core.events import KytosEvent
from kytos.core.interface import Interface
from kytos.core.switch import Switch
from napps.kytos.of_core.main import Main
from napps.kytos.of_core.utils import GenericHello


class TestMain(TestCase):
    """docstring for TestMain."""

    def setUp(self):
        """Execute steps before each tests.

        Set the server_name_url from kytos/of_core
        """
        self.server_name_url = 'http://localhost:8181/api/kytos/of_core'
        self.controller = self._get_controller_mock()
        self.napp = Main(self.controller)
        self.patched_events = []

    @staticmethod
    def _get_controller_mock():
        """Return a controller mock."""
        options = KytosConfig().options['daemon']
        controller = Controller(options)
        controller.log = Mock()
        return controller

    @staticmethod
    def _get_switch_mock(of_version, connection_state=ConnectionState.NEW,
                         dpid="00:00:00:00:00:00:00:01"):
        """Return a switch mock."""
        switch = Switch(dpid)
        address = Mock()
        port = Mock()
        socket = Mock()
        switch.connection = Connection(address, port, socket)
        switch.connection.protocol.unpack = unpack
        switch.connection.protocol.version = of_version
        switch.connection.state = connection_state
        return switch

    def _get_interface(self, interface_name, port, *args, **kwargs):
        """Return a interface mock."""
        switch1 = self._get_switch_mock(0x04)
        switch1.connection = Mock()
        iface1 = Interface(interface_name, port, switch1, *args, **kwargs)
        return iface1

    def test_get_event_listeners(self):
        """Verify all event listeners registered."""
        expected_events = [
            'kytos/of_core.v0x01.messages.in.ofpt_stats_reply',
            'kytos/of_core.v0x0[14].messages.in.ofpt_features_reply',
            'kytos/of_core.v0x04.messages.in.ofpt_multipart_reply',
            'kytos/core.openflow.raw.in',
            'kytos/of_core.v0x0[14].messages.in.ofpt_echo_request',
            'kytos/of_core.v0x0[14].messages.out.ofpt_echo_reply',
            'kytos/of_core.v0x[0-9a-f]{2}.messages.in.hello_failed',
            'kytos/of_core.v0x0[14].messages.out.hello_failed',
        ]

        actual_events = self.napp.listeners()
        for _event in expected_events:
            self.assertIn(_event, actual_events, '%s' % _event)

    def test_execute(self):
        """Test 'execute' main method."""
        dpid_01 = "00:00:00:00:00:00:00:01"
        sw_01 = self._get_switch_mock(0x01, ConnectionState.ESTABLISHED,
                                      dpid_01)
        self.napp.controller.get_switch_or_create(dpid_01, sw_01.connection)
        dpid_02 = "00:00:00:00:00:00:00:02"
        sw_04 = self._get_switch_mock(0x04, ConnectionState.ESTABLISHED,
                                      dpid_02)
        self.napp.controller.get_switch_or_create(dpid_02, sw_04.connection)
        self.napp.execute()

    def test_handle_stats_reply(self):
        """Test handling stats reply message."""
        event_name = 'kytos/of_core.v0x01.messages.in.ofpt_stats_reply'
        switch = self._get_switch_mock(0x01)
        switch.connection = Mock()

        stats_data = b'\x01\x11\x00\x0c\x00\x00\x00\x01\x00\x01\x00\x01'
        stats_reply = StatsReply()
        stats_reply.unpack(stats_data[8:])
        stats_event = KytosEvent(name=event_name,
                                 content={'source': switch.connection,
                                          'message': stats_reply})
        self.napp.handle_stats_reply(stats_event)

        desc_stats_data = b'\x01\x11\x00\x0c\x00\x00\x00\x0e\x00\x00\x00\x00'
        desc_stats_reply = StatsReply()
        desc_stats_reply.unpack(desc_stats_data[8:])
        desc_stats_event = KytosEvent(name=event_name,
                                      content={'source': switch.connection,
                                               'message': desc_stats_reply})
        self.napp.handle_stats_reply(desc_stats_event)

    def test_handle_04_features_reply(self):
        """Test handling features reply message."""
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_features_reply'
        switch = self._get_switch_mock(0x04, ConnectionState.SETUP)
        switch.connection.protocol.state = 'waiting_features_reply'

        data = b'\x04\x06\x00\x20\x00\x00\x00\x00\x00\x00\x08\x60\x6e\x7f\x74'
        data += b'\xe7\x00\x00\x00\x00\xff\x63\x00\x00\x00\x00\x00\x4f\x00\x00'
        data += b'\x00\x00'

        features_reply = FReply_v0x04()
        features_reply.unpack(data)

        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'message': features_reply})
        self.napp.handle_features_reply(event)

    def test_handle_01_features_reply(self):
        """Test handling features reply message."""
        event_name = 'kytos/of_core.v0x01.messages.in.ofpt_features_reply'
        switch = self._get_switch_mock(0x01, ConnectionState.SETUP)
        switch.connection.protocol.state = 'waiting_features_reply'

        data = b'\x01\x06\x00\x80\x00\x00\x00\x00\x00\x00\x00\xff\x12\x34\x56'
        data += b'\x78\x00\x00\x00\x00\xff\x00\x00\x00\x00\x00\x00\xa9\x00\x00'
        data += b'\x08\x43\x00\x07\xf2\x0b\xa4\xd0\x3f\x70\x50\x6f\x72\x74\x37'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x10\x00\x00\x02\x88\x00\x00\x02\x80\x00\x00\x02'
        data += b'\x88\x00\x00\x02\x88\x00\x06\xf2\x0b\xa4\x7d\xf8\xea\x50\x6f'
        data += b'\x72\x74\x36\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x00\x00\x02\x00\x00\x02\x88\x00\x00\x02\x80'
        data += b'\x00\x00\x02\x88\x00\x00\x02\x88'

        features_reply = FReply_v0x01()
        features_reply.unpack(data[8:])

        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'message': features_reply})
        self.napp.handle_features_reply(event)

    def test_handle_features_request_sent(self):
        """Test handling features request sent message."""
        event_name = 'kytos/of_core.v0x01.messages.out.ofpt_features_request'
        switch = self._get_switch_mock(0x01)
        switch.connection.protocol.state = 'sending_features'

        data = b'\x04\x05\x00\x08\x00\x00\x00\x03'
        features_request = FeaturesRequest()
        features_request.unpack(data)

        event = KytosEvent(name=event_name,
                           content={'destination': switch.connection,
                                    'message': features_request})
        self.napp.handle_features_request_sent(event)

    def test_handle_echo_request(self):
        """Test handling echo request message."""
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_echo_request'
        switch = self._get_switch_mock(0x04)

        data = b'\x04\x02\x00\x0c\x00\x00\x00\x00\x68\x6f\x67\x65'
        echo_request = EchoRequest()
        echo_request.unpack(data)

        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'message': echo_request})
        self.napp.handle_echo_request(event)

    def test_handle_hello_raw_in(self):
        """Test handling hello raw in message."""
        event_name = 'kytos/core.openflow.raw.in'
        switch = self._get_switch_mock(0x04)

        data = b'\x04\x00\x00\x10\x00\x00\x00\x3e'
        data += b'\x00\x01\x00\x08\x00\x00\x00\x10'
        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'new_data': data})
        self.napp.handle_raw_in(event)

        data_exc = b'\x00\x00\x00\x08\xb8\xd1\xb1\x0c'
        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'new_data': data_exc})
        self.napp.handle_raw_in(event)

    def test_handle_port_status_raw_in(self):
        """Test handling port_status raw in message."""
        event_name = 'kytos/core.openflow.raw.in'
        switch = self._get_switch_mock(0x04, ConnectionState.ESTABLISHED)
        switch.connection.switch = self._get_switch_mock(0x04)

        data = b'\x04\x0c\x00\x50\x00\x00\x00\x00\x02\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x01\x00\x00\x00\x00\x62\x43\xe5\xdb\x35\x0a'
        data += b'\x00\x00\x73\x31\x2d\x65\x74\x68\x31\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00\x00\x08\x40'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x98\x96'
        data += b'\x80\x00\x00\x00\x00'

        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'new_data': data})
        self.napp.handle_raw_in(event)

    def test_handle_packet_in_raw_in(self):
        """Test handling packet_in raw in message."""
        event_name = 'kytos/core.openflow.raw.in'
        switch = self._get_switch_mock(0x04, ConnectionState.ESTABLISHED)
        switch.connection.switch = self._get_switch_mock(0x04)

        data = b'\x04\x0a\x00\x94\x00\x00\x00\x00\x00\x00\x00\x02\x00\x2a\x01'
        data += b'\x01\x00\x01\x02\x03\x00\x00\x00\x00\x00\x01\x00\x50\x80\x00'
        data += b'\x00\x04\x00\x00\x00\x06\x80\x00\x0a\x02\x08\x06\x80\x00\x06'
        data += b'\x06\xff\xff\xff\xff\xff\xff\x80\x00\x08\x06\xf2\x0b\xa4\x7d'
        data += b'\xf8\xea\x80\x00\x2a\x02\x00\x01\x80\x00\x2c\x04\x0a\x00\x00'
        data += b'\x01\x80\x00\x2e\x04\x0a\x00\x00\x03\x80\x00\x30\x06\xf2\x0b'
        data += b'\xa4\x7d\xf8\xea\x80\x00\x32\x06\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00\xff\xff\xff\xff\xff\xff\xf2\x0b\xa4\x7d\xf8\xea\x08\x06'
        data += b'\x00\x01\x08\x00\x06\x04\x00\x01\xf2\x0b\xa4\x7d\xf8\xea\x0a'
        data += b'\x00\x00\x01\x00\x00\x00\x00\x00\x00\x0a\x00\x00\x03'

        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'new_data': data})
        self.napp.handle_raw_in(event)

    def test_handle_multipart_reply(self):
        """Test handling ofpt_multipart_reply."""
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_multipart_reply'
        switch = self._get_switch_mock(0x04, dpid='00:00:00:00:00:00:00:02')
        switch.connection.switch = self._get_switch_mock(0x04)
        self.napp.controller.get_switch_or_create(switch.dpid,
                                                  switch.connection)

        data = b'\x04\x13\x00\x68\xac\xc8\xdf\x58\x00\x01\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x58\x00\x00\x00\x00\x00\x38\x25\xd9\x54\xc0\x03\xe8'
        data += b'\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x12\x00\x00\x00\x00'
        data += b'\x00\x00\x02\xf4\x00\x01\x00\x10\x80\x00\x0a\x02\x88\xcc\x80'
        data += b'\x00\x0c\x02\x1e\xd7\x00\x04\x00\x18\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x10\xff\xff\xff\xfd\xff\xff\x00\x00\x00\x00\x00\x00'

        # pylint: disable=protected-access
        xid = self.napp._multipart_replies_xids[switch.dpid]
        # pylint: enable=protected-access
        multipart_reply = MultipartReply(xid=xid)
        multipart_reply.unpack(data[8:])
        stats_event = KytosEvent(name=event_name,
                                 content={'source': switch.connection,
                                          'message': multipart_reply})

        self.napp.handle_multipart_reply(stats_event)

        # test ofpmp_desc
        data = b'\x04\x12\x00\x10\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00'
        multipart_desc = MultipartReply()
        multipart_desc.unpack(data[8:])
        stats_desc_event = KytosEvent(name=event_name,
                                      content={'source': switch.connection,
                                               'message': multipart_desc})

        self.napp.handle_multipart_reply(stats_desc_event)

    def test_handle_port_desc_multipart_reply(self):
        """Test handling to ofpt_PORT_DESC."""
        event_name = 'kytos/of_core.v0x04.messages.in.ofpt_multipart_reply'
        switch = self._get_switch_mock(0x04)
        switch.connection.switch = self._get_switch_mock(0x04)
        data = b'\x04\x13\x00\x90\x00\x00\x00\x00\x00\x0d\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x07\x00\x00\x00\x00\xf2\x0b\xa4\xd0\x3f\x70'
        data += b'\x00\x00\x50\x6f\x72\x74\x37\x00\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04\x00\x00\x28\x08'
        data += b'\x00\x00\x28\x00\x00\x00\x28\x08\x00\x00\x28\x08\x00\x00\x13'
        data += b'\x88\x00\x00\x13\x88\x00\x00\x00\x06\x00\x00\x00\x00\xf2\x0b'
        data += b'\xa4\x7d\xf8\xea\x00\x00\x50\x6f\x72\x74\x36\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x04'
        data += b'\x00\x00\x28\x08\x00\x00\x28\x00\x00\x00\x28\x08\x00\x00\x28'
        data += b'\x08\x00\x00\x13\x88\x00\x00\x13\x88'

        port_desc = MultipartReply()
        port_desc.unpack(data[8:])
        interface_1 = self._get_interface("interface1", 6)
        interface_2 = self._get_interface("interface2", 7)
        switch.connection.switch.interfaces = {6: interface_1, 7: interface_2}

        stats_event = KytosEvent(name=event_name,
                                 content={'source': switch.connection,
                                          'message': port_desc})
        self.napp.handle_multipart_reply(stats_event)

        # Send port_desc pack without interface
        switch = self._get_switch_mock(0x04)
        switch.connection.switch = self._get_switch_mock(0x04)
        stats_event = KytosEvent(name=event_name,
                                 content={'source': switch.connection,
                                          'message': port_desc})

        self.napp.handle_multipart_reply(stats_event)

    def test_pack_generic_hello(self):
        """Test packing a generic hello message."""
        data = b'\x04\x00\x00\x10\x00\x00\x00\x3e'
        data += b'\x00\x01\x00\x08\x00\x00\x00\x10'
        generic_hello = GenericHello(packet=data, versions=b'\x04')
        self.assertEqual(generic_hello.pack(), data)