charm-neutron-openvswitch/unit_tests/test_neutron_openvswitch_networking_sriov.py

# Copyright 2019 Canonical Ltd
#
# 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

import unittest
from unittest.mock import (
    call,
    mock_open,
    patch
)
import files.neutron_openvswitch_networking_sriov as net_sriov


class NeutronOVSNetworkingSriovTest(unittest.TestCase):

    def test_parse_args(self):
        results = net_sriov.parse_args(['--start'])
        self.assertEqual(results.start, True)
        results = net_sriov.parse_args(['--stop'])
        self.assertEqual(results.stop, True)
        results = net_sriov.parse_args(['--restart'])
        self.assertEqual(results.restart, True)
        args = [
            '--start',
            '--vfs',
            'eth0:8 eth1:4',
            '--vfs-blanket',
            '8'
        ]
        results = net_sriov.parse_args(args)
        self.assertEqual(results.vfs, 'eth0:8 eth1:4')
        self.assertEqual(results.vfs_blanket, '8')

    def test_parse_vfs(self):
        vfs = ''
        result = net_sriov.parse_vfs(vfs)
        expected = {}
        self.assertEqual(result, expected)
        vfs = 'eth0:8 eth1:8'
        result = net_sriov.parse_vfs(vfs)
        expected = {
            'eth0': '8',
            'eth1': '8'
        }
        self.assertEqual(result, expected)

    @patch('files.neutron_openvswitch_networking_sriov.write_sriov_numvfs')
    def test_configure_vfs_with_vfs(self, m_write_vfs):
        vfs = {
            'eth0': '8'
        }
        net_sriov.configure_vfs(vfs)
        m_write_vfs.assert_called_once_with(
            '/sys/class/net/eth0/device/',
            '8'
        )

    @patch('files.neutron_openvswitch_networking_sriov.get_totalvfs')
    @patch('files.neutron_openvswitch_networking_sriov.find_sriov_devices')
    @patch('files.neutron_openvswitch_networking_sriov.write_sriov_numvfs')
    def test_configure_vfs_with_blanket_auto(
            self,
            m_write_vfs,
            m_find,
            m_totvfs):
        m_find.return_value = [
            '/dev/one',
            '/dev/two'
        ]
        m_totvfs.return_value = '8'
        net_sriov.configure_vfs({}, 'auto')
        m_write_vfs.assert_has_calls([
            call('/dev/one', '8'),
            call('/dev/two', '8')
        ])

    @patch('files.neutron_openvswitch_networking_sriov.get_totalvfs')
    @patch('files.neutron_openvswitch_networking_sriov.find_sriov_devices')
    @patch('files.neutron_openvswitch_networking_sriov.write_sriov_numvfs')
    def test_configure_vfs_with_blanket_too_big(
            self,
            m_write_vfs,
            m_find,
            m_totvfs):
        m_find.return_value = [
            '/dev/one',
            '/dev/two'
        ]
        m_totvfs.return_value = '2'
        net_sriov.configure_vfs({}, '8')
        m_write_vfs.assert_has_calls([
            call('/dev/one', '2'),
            call('/dev/two', '2')
        ])

    @patch('files.neutron_openvswitch_networking_sriov.get_totalvfs')
    @patch('files.neutron_openvswitch_networking_sriov.find_sriov_devices')
    @patch('files.neutron_openvswitch_networking_sriov.write_sriov_numvfs')
    def test_configure_vfs_with_blanket_smaller_than_totalvfs(
            self,
            m_write_vfs,
            m_find,
            m_totvfs):
        m_find.return_value = [
            '/dev/one',
            '/dev/two'
        ]
        m_totvfs.return_value = '8'
        net_sriov.configure_vfs({}, '2')
        m_write_vfs.assert_has_calls([
            call('/dev/one', '2'),
            call('/dev/two', '2')
        ])

    @patch('time.sleep')
    @patch('os.listdir')
    def test_wait_for_vfs(self, m_listdir, m_sleep):
        dev_list = [
            'subsystem_device',
            'subsystem_vendor',
            'uevent',
            'vendor',
            'virtfn0',
            'virtfn1',
            'virtfn2',
            'virtfn3',
        ]
        m_listdir.return_value = dev_list
        net_sriov.wait_for_vfs('dev', '4')
        m_sleep.assert_called_once_with(0.05)

    @patch('os.walk')
    def test_find_sriov_devices(self, m_walk):
        m_walk.return_value = [
            ('/one', ('dir1', 'dir2'), ('file')),
            ('/one/dir1', (), ('sriov_totalvfs')),
            ('/one/dir2', (), ('sriov_totalvfs'))
        ]
        results = net_sriov.find_sriov_devices()
        expected = [
            '/one/dir1',
            '/one/dir2'
        ]
        self.assertEqual(results, expected)

    @patch('builtins.open', new_callable=mock_open)
    def test_get_totalvfs(self, m_open):
        net_sriov.get_totalvfs('/dev/some')
        m_open.assert_called_once_with(
            '/dev/some/sriov_totalvfs'
        )

    @patch('files.neutron_openvswitch_networking_sriov.wait_for_vfs')
    @patch('builtins.open', new_callable=mock_open)
    def test_write_sriov_numvfs(self, m_open, m_wait_vfs):
        net_sriov.write_sriov_numvfs('/dev/sriov', '8')
        m_open.assert_called_once_with(
            '/dev/sriov/sriov_numvfs',
            'w'
        )
        m_wait_vfs.assert_called_once_with(
            '/dev/sriov',
            '8'
        )

    @patch('files.neutron_openvswitch_networking_sriov.start')
    @patch('files.neutron_openvswitch_networking_sriov.stop')
    def test_restart(self, m_stop, m_start):
        net_sriov.restart([], 'auto')
        m_stop.assert_called_once_with([])
        m_start.assert_called_once_with([], 'auto')

    @patch('files.neutron_openvswitch_networking_sriov.configure_vfs')
    def test_start(self, m_config):
        net_sriov.start([], 'auto')
        m_config.assert_called_once_with([], 'auto')

    @patch('files.neutron_openvswitch_networking_sriov.configure_vfs')
    def test_stop(self, m_config):
        net_sriov.stop([])
        m_config.assert_called_once_with([], stop=True)

    @patch('files.neutron_openvswitch_networking_sriov.restart')
    def test_main_with_restart(self, m_restart):
        args = ['--restart', '--vfs', 'eth0:8 eth1:2', '--vfs-blanket', 'auto']
        net_sriov.main(args)
        vfs = {
            'eth0': '8',
            'eth1': '2'
        }
        vfs_blanket = 'auto'
        m_restart.assert_called_once_with(vfs, vfs_blanket)

    @patch('files.neutron_openvswitch_networking_sriov.start')
    def test_main_with_start(self, m_start):
        args = ['--start', '--vfs', 'eth0:8 eth1:2', '--vfs-blanket', 'auto']
        net_sriov.main(args)
        vfs = {
            'eth0': '8',
            'eth1': '2'
        }
        vfs_blanket = 'auto'
        m_start.assert_called_once_with(vfs, vfs_blanket)

    @patch('files.neutron_openvswitch_networking_sriov.stop')
    def test_main_with_stop(self, m_stop):
        args = ['--stop', '--vfs', 'eth0:8 eth1:2', '--vfs-blanket', 'auto']
        net_sriov.main(args)
        vfs = {
            'eth0': '8',
            'eth1': '2'
        }
        m_stop.assert_called_once_with(vfs)