tests-1.4/role_status.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


"""
Test the role status message
"""
import struct
import unittest
import logging

import oftest
from oftest import config
import oftest.controller as controller
import ofp
import oftest.base_tests as base_tests

from oftest.testutils import *

def simple_role_request(test, role, gen=None, con=None):
    """
    Send a role request we expect to succeed
    """
    if con == None:
        con = test.controller
    request = ofp.message.role_request(role=role, generation_id=gen)
    response, _ = con.transact(request)
    test.assertTrue(isinstance(response, ofp.message.role_reply), "Expected a role reply")
    if role != ofp.OFPCR_ROLE_NOCHANGE:
        test.assertEquals(response.role, role)
    if gen != None:
        test.assertEquals(response.generation_id, gen)
    return response.role, response.generation_id

@disabled
class RoleStatus(unittest.TestCase):
    """
    Verify that when a connection becomes a master the existing master is
    downgraded to slave and receives a role-status message.

    Requires the switch to attempt to connect in parallel to ports 6653
    and 6753 on the configured IP.
    """

    def setUp(self):
        host = config["controller_host"]
        self.controllers = [
            controller.Controller(host=host,port=6653),
            controller.Controller(host=host,port=6753)
        ]

    def runTest(self):
        # Connect and handshake with both controllers
        for con in self.controllers:
            con.start()
            if not con.connect():
                raise AssertionError("failed to connect controller %s" % str(con))
            reply, _ = con.transact(ofp.message.features_request())
            self.assertTrue(isinstance(reply, ofp.message.features_reply))

        # Assert initial role and get generation IDs
        role, gen0 = simple_role_request(self, ofp.OFPCR_ROLE_NOCHANGE, con=self.controllers[0])
        self.assertEqual(role, ofp.OFPCR_ROLE_EQUAL)
        role, gen1 = simple_role_request(self, ofp.OFPCR_ROLE_NOCHANGE, con=self.controllers[1])
        self.assertEqual(role, ofp.OFPCR_ROLE_EQUAL)

        # Initial role assignment: controller 0 is master, controller 1 is slave
        simple_role_request(self, ofp.OFPCR_ROLE_MASTER, gen0, con=self.controllers[0])
        simple_role_request(self, ofp.OFPCR_ROLE_SLAVE, gen1, con=self.controllers[1])
        self.verify_role(self.controllers[0], ofp.OFPCR_ROLE_MASTER)
        self.verify_role(self.controllers[1], ofp.OFPCR_ROLE_SLAVE)

        # Controller 1 requests master
        # Controller 0 becomes slave
        simple_role_request(self, ofp.OFPCR_ROLE_MASTER, gen1, con=self.controllers[1])
        self.verify_role(self.controllers[0], ofp.OFPCR_ROLE_SLAVE)
        self.verify_role(self.controllers[1], ofp.OFPCR_ROLE_MASTER)

        # Controller 0 should receive a role_status message
        msg, _ = self.controllers[0].poll(ofp.message.role_status)
        self.assertIsInstance(msg, ofp.message.role_status)
        self.assertEqual(msg.role, ofp.OFPCR_ROLE_SLAVE)
        self.assertEqual(msg.reason, ofp.OFPCRR_MASTER_REQUEST)
        self.assertEqual(msg.generation_id, gen1)

    def verify_role(self, con, role):
        rcv_role, _ = simple_role_request(self, ofp.OFPCR_ROLE_NOCHANGE, con=con)
        self.assertEqual(rcv_role, role)

    def tearDown(self):
        for con in self.controllers:
            con.shutdown()