build.tests.test_main

Complexity

Total Complexity

11

Size/Duplication

Total Lines	184
Duplicated Lines	0 %

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	TestMain.test_handle_features_reply()	0	17	1
A	TestMain.test_handle_port_status_raw_in()	0	19	1
A	TestMain.test_handle_multipart_reply()	0	21	1
A	TestMain.test_handle_features_request_sent()	0	14	1
A	TestMain.test_handle_stats_reply()	0	22	1
A	TestMain.test_handle_hello_raw_in()	0	14	1
A	TestMain._get_switch_mock()	0	10	1
A	TestMain.test_get_event_listeners()	0	16	2
A	TestMain.setUp()	0	9	1
A	TestMain._get_controller_mock()	0	7	1

"""Test main."""

from unittest import TestCase
from unittest.mock import Mock

from pyof.foundation.basic_types import DPID
from pyof.utils import unpack
from pyof.v0x01.controller2switch.stats_reply import StatsReply
from pyof.v0x04.controller2switch.features_reply import FeaturesReply
from pyof.v0x04.controller2switch.features_request import FeaturesRequest
from pyof.v0x04.controller2switch.multipart_reply import MultipartReply

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.switch import Switch
from napps.kytos.of_core.main import Main


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():
        """Return a switch mock."""
        switch = Switch('dpid')
        address = Mock()
        port = Mock()
        socket = Mock()
        switch.connection = Connection(address, port, socket)
        switch.connection.protocol.unpack = unpack
        return switch

    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_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()
        switch.connection = Mock()
        switch.connection.protocol.version = 0x01

        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_features_reply(self):
        """Test handling features reply message."""
        event_name = 'kytos/of_core.v0x0[14].messages.in.ofpt_features_reply'
        switch = self._get_switch_mock()
        switch.connection.protocol.version = 0x04
        switch.connection.protocol.state = 'waiting_features_reply'
        switch.connection.state = ConnectionState.SETUP

        message = FeaturesReply(xid=3,
                                datapath_id=DPID('00:00:00:00:00:00:00:01'),
                                n_buffers=0, n_tables=254, auxiliary_id=0,
                                capabilities=0x0000004f, reserved=0x00000000)

        event = KytosEvent(name=event_name,
                           content={'source': switch.connection,
                                    'message': message})
        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()
        switch.connection.protocol.version = 0x01
        switch.connection.protocol.state = 'sending_features'

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

        stats_event = KytosEvent(name=event_name,
                                 content={'destination': switch.connection})
        self.napp.handle_features_request_sent(stats_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()
        switch.connection.protocol.version = 0x04
        switch.connection.state = ConnectionState.NEW

        data = b'\x04\x00\x00\x10\x00\x00\x00\x3e'
        data += b'\x00\x01\x00\x08\x00\x00\x00\x10'

        stats_event = KytosEvent(name=event_name,
                                 content={'source': switch.connection,
                                          'new_data': data})
        self.napp.handle_raw_in(stats_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()
        switch.connection.protocol.version = 0x04
        switch.connection.switch = self._get_switch_mock()
        switch.connection.state = ConnectionState.ESTABLISHED

        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'

        stats_event = KytosEvent(name=event_name,
                                 content={'source': switch.connection,
                                          'new_data': data})
        self.napp.handle_raw_in(stats_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()
        switch.connection.switch = self._get_switch_mock()
        data = b'\x04\x13\x00\x68\xac\xc8\xdf\x58\x00\x01\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x58\x00\x00\x00\x00\x00\x38\x25\xd9\x54\xc0\x03\xe8\x00\x00'
        data += b'\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
        data += b'\x00\x00\x00\x00\x00\x00\x00\x12\x00\x00\x00\x00\x00\x00\x02\xf4'
        data += b'\x00\x01\x00\x10\x80\x00\x0a\x02\x88\xcc\x80\x00\x0c\x02\x1e\xd7'
        data += b'\x00\x04\x00\x18\x00\x00\x00\x00\x00\x00\x00\x10\xff\xff\xff\xfd'
        data += b'\xff\xff\x00\x00\x00\x00\x00\x00'

        multipart_reply = MultipartReply()
        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)