NSB: Fix standalone.model.Libvirt SR-IOV modeling

Fixed standalone.model.Libvirt SR-IOV XML interface modeling, acording
to [1]:
- All PCI attributes now are printed in hexadecimal format.
- The PCI address is now added in the correct section, 'interface'.

network_services.utils.PciAddress was refactored to accept both 'domain:
bus:slot:function' and 'bus:slot:function' format inputs. This class is
used as input in the previous class, Libvirt, to print in XML the PCI
address of a SR-IOV interface.

network_services.utils.PciAddress.parse_address is now deprecated. Instead
the class standard instantiation must be used:

    libvirt_obj = utils.PciAddress(text_with_address)

A deprecation decorator is implemented along with this patch. This
decorator is used for the first time in the previously mentioned function.
This decorator stores every decorated function name and deprecation message
and raises a logging warning message the first time this function is used.

[1] https://goo.gl/so2Mrp

Change-Id: I22e95c488e27d6e2a8fdf6c1a07faab275fa6bba
Signed-off-by: Dino Simeon Madarang <dinox.madarang@intel.com>
Reviewed-by: Alain Jebara <alain.jebara@intel.com>
Reviewed-by: Deepak S <deepak.s@linux.intel.com>
Reviewed-by: Ross Brattain <ross.b.brattain@intel.com>
Reviewed-by: Rodolfo Alonso Hernandez <rodolfo.alonso.hernandez@intel.com>

3 files changed, 282 insertions, 118 deletions

diff --git a/tests/unit/benchmark/contexts/standalone/test_model.py b/tests/unit/benchmark/contexts/standalone/test_model.py
index 6899a0af6..31ec2b7d1 100644
--- a/tests/unit/benchmark/contexts/standalone/test_model.py
+++ b/tests/unit/benchmark/contexts/standalone/test_model.py
@@ -16,70 +16,137 @@
 
 # Unittest for yardstick.benchmark.contexts.standalone.model
 
-
 from __future__ import absolute_import
+import copy
 import os
 import unittest
-import errno
 import mock
 
-from yardstick.common import constants as consts
-from yardstick.benchmark.contexts.standalone.model import Libvirt
-from yardstick.benchmark.contexts.standalone.model import StandaloneContextHelper
-from yardstick.benchmark.contexts.standalone import model
-from yardstick.network_services.utils import PciAddress
+from xml.etree import ElementTree
 
+from yardstick.benchmark.contexts.standalone import model
+from yardstick.network_services import utils
+from yardstick.network_services.helpers import cpu
+
+
+XML_SAMPLE = """<?xml version="1.0"?>
+<domain type="kvm">
+    <devices>
+    </devices>
+</domain>
+"""
+
+XML_SAMPLE_INTERFACE = """<?xml version="1.0"?>
+<domain type="kvm">
+    <devices>
+        <interface>
+        </interface>
+    </devices>
+</domain>
+"""
 
 class ModelLibvirtTestCase(unittest.TestCase):
 
+    def setUp(self):
+        self.xml = ElementTree.ElementTree(
+            element=ElementTree.fromstring(XML_SAMPLE))
+        self.pci_address_str = '0001:04:03.2'
+        self.pci_address = utils.PciAddress(self.pci_address_str)
+        self.mac = '00:00:00:00:00:01'
+        self._mock_write_xml = mock.patch.object(ElementTree.ElementTree,
+                                                 'write')
+        self.mock_write_xml = self._mock_write_xml.start()
+
+        self.addCleanup(self._cleanup)
+
+    def _cleanup(self):
+        self._mock_write_xml.stop()
+
     def test_check_if_vm_exists_and_delete(self):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
+            ssh_mock.execute = mock.Mock(return_value=(0, "a", ""))
             ssh.return_value = ssh_mock
-        result = Libvirt.check_if_vm_exists_and_delete("vm_0", ssh_mock)
-        self.assertIsNone(result)
+        # NOTE(ralonsoh): this test doesn't cover function execution.
+        model.Libvirt.check_if_vm_exists_and_delete("vm_0", ssh_mock)
 
     def test_virsh_create_vm(self):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
+            ssh_mock.execute = mock.Mock(return_value=(0, "a", ""))
             ssh.return_value = ssh_mock
-        result = Libvirt.virsh_create_vm(ssh_mock, "vm_0")
-        self.assertIsNone(result)
+        # NOTE(ralonsoh): this test doesn't cover function execution.
+        model.Libvirt.virsh_create_vm(ssh_mock, "vm_0")
 
     def test_virsh_destroy_vm(self):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
+            ssh_mock.execute = mock.Mock(return_value=(0, "a", ""))
             ssh.return_value = ssh_mock
-        result = Libvirt.virsh_destroy_vm("vm_0", ssh_mock)
-        self.assertIsNone(result)
-
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.ET')
-    def test_add_interface_address(self, mock_et):
-        pci_address = PciAddress.parse_address("0000:00:04.0", multi_line=True)
-        result = Libvirt.add_interface_address("<interface/>", pci_address)
-        self.assertIsNotNone(result)
-
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.Libvirt.add_interface_address')
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.ET')
-    def test_add_ovs_interfaces(self, mock_et, mock_add_interface_address):
-        pci_address = PciAddress.parse_address("0000:00:04.0", multi_line=True)
-        result = Libvirt.add_ovs_interface("/usr/local", 0, "0000:00:04.0",
-                                                "00:00:00:00:00:01", "xml")
-        self.assertIsNone(result)
-
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.Libvirt.add_interface_address')
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.ET')
-    def test_add_sriov_interfaces(self, mock_et, mock_add_interface_address):
-        pci_address = PciAddress.parse_address("0000:00:04.0", multi_line=True)
-        result = Libvirt.add_sriov_interfaces("0000:00:05.0", "0000:00:04.0",
-                                              "00:00:00:00:00:01", "xml")
-        self.assertIsNone(result)
+        # NOTE(ralonsoh): this test doesn't cover function execution.
+        model.Libvirt.virsh_destroy_vm("vm_0", ssh_mock)
+
+    def test_add_interface_address(self):
+        xml = ElementTree.ElementTree(
+            element=ElementTree.fromstring(XML_SAMPLE_INTERFACE))
+        interface = xml.find('devices').find('interface')
+        result = model.Libvirt._add_interface_address(interface, self.pci_address)
+        self.assertEqual('pci', result.get('type'))
+        self.assertEqual('0x{}'.format(self.pci_address.domain),
+                         result.get('domain'))
+        self.assertEqual('0x{}'.format(self.pci_address.bus),
+                         result.get('bus'))
+        self.assertEqual('0x{}'.format(self.pci_address.slot),
+                         result.get('slot'))
+        self.assertEqual('0x{}'.format(self.pci_address.function),
+                         result.get('function'))
+
+    def test_add_ovs_interfaces(self):
+        xml_input = mock.Mock()
+        with mock.patch.object(ElementTree, 'parse', return_value=self.xml) \
+                as mock_parse:
+            xml = copy.deepcopy(self.xml)
+            mock_parse.return_value = xml
+            model.Libvirt.add_ovs_interface(
+                '/usr/local', 0, self.pci_address_str, self.mac, xml_input)
+            mock_parse.assert_called_once_with(xml_input)
+            self.mock_write_xml.assert_called_once_with(xml_input)
+            interface = xml.find('devices').find('interface')
+            self.assertEqual('vhostuser', interface.get('type'))
+            mac = interface.find('mac')
+            self.assertEqual(self.mac, mac.get('address'))
+            source = interface.find('source')
+            self.assertEqual('unix', source.get('type'))
+            self.assertEqual('/usr/local/var/run/openvswitch/dpdkvhostuser0',
+                             source.get('path'))
+            self.assertEqual('client', source.get('mode'))
+            _model = interface.find('model')
+            self.assertEqual('virtio', _model.get('type'))
+            driver = interface.find('driver')
+            self.assertEqual('4', driver.get('queues'))
+            host = driver.find('host')
+            self.assertEqual('off', host.get('mrg_rxbuf'))
+            self.assertIsNotNone(interface.find('address'))
+
+    def test_add_sriov_interfaces(self):
+        xml_input = mock.Mock()
+        with mock.patch.object(ElementTree, 'parse', return_value=self.xml) \
+                as mock_parse:
+            xml = copy.deepcopy(self.xml)
+            mock_parse.return_value = xml
+            vf_pci = '0001:05:04.2'
+            model.Libvirt.add_sriov_interfaces(
+                self.pci_address_str, vf_pci, self.mac, xml_input)
+            mock_parse.assert_called_once_with(xml_input)
+            self.mock_write_xml.assert_called_once_with(xml_input)
+            interface = xml.find('devices').find('interface')
+            self.assertEqual('yes', interface.get('managed'))
+            self.assertEqual('hostdev', interface.get('type'))
+            mac = interface.find('mac')
+            self.assertEqual(self.mac, mac.get('address'))
+            source = interface.find('source')
+            self.assertIsNotNone(source.find('address'))
+            self.assertIsNotNone(interface.find('address'))
 
     def test_create_snapshot_qemu(self):
         result = "/var/lib/libvirt/images/0.qcow2"
@@ -88,15 +155,15 @@ class ModelLibvirtTestCase(unittest.TestCase):
             ssh_mock.execute = \
                 mock.Mock(return_value=(0, "a", ""))
             ssh.return_value = ssh_mock
-        image = Libvirt.create_snapshot_qemu(ssh_mock, "0", "ubuntu.img")
+        image = model.Libvirt.create_snapshot_qemu(ssh_mock, "0", "ubuntu.img")
         self.assertEqual(image, result)
 
-    @mock.patch("yardstick.benchmark.contexts.standalone.model.Libvirt.pin_vcpu_for_perf")
-    @mock.patch("yardstick.benchmark.contexts.standalone.model.Libvirt.create_snapshot_qemu")
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.open')
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.write_file')
-    def test_build_vm_xml(self, mock_open, mock_write_file, mock_create_snapshot_qemu,
-                          mock_pin_vcpu_for_perf):
+    @mock.patch.object(model.Libvirt, 'pin_vcpu_for_perf')
+    @mock.patch.object(model.Libvirt, 'create_snapshot_qemu')
+    def test_build_vm_xml(self, mock_create_snapshot_qemu,
+                          *args):
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
         result = [4]
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
@@ -105,7 +172,7 @@ class ModelLibvirtTestCase(unittest.TestCase):
             ssh.return_value = ssh_mock
         mock_create_snapshot_qemu.return_value = "0.img"
 
-        status = Libvirt.build_vm_xml(ssh_mock, {}, "test", "vm_0", 0)
+        status = model.Libvirt.build_vm_xml(ssh_mock, {}, "test", "vm_0", 0)
         self.assertEqual(status[0], result[0])
 
     def test_update_interrupts_hugepages_perf(self):
@@ -114,19 +181,21 @@ class ModelLibvirtTestCase(unittest.TestCase):
             ssh_mock.execute = \
                 mock.Mock(return_value=(0, "a", ""))
             ssh.return_value = ssh_mock
-        status = Libvirt.update_interrupts_hugepages_perf(ssh_mock)
-        self.assertIsNone(status)
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
+        model.Libvirt.update_interrupts_hugepages_perf(ssh_mock)
+
+    @mock.patch.object(cpu.CpuSysCores, 'get_core_socket')
+    def test_pin_vcpu_for_perf(self, mock_get_core_socket):
+        mock_get_core_socket.return_value = {
+            'cores_per_socket': 1,
+            'thread_per_core': 1,
+            '0': [1, 2]
+        }
+        # NOTE(ralonsoh): this test doesn't cover function execution. This
+        # function needs more tests.
+        model.Libvirt.pin_vcpu_for_perf(mock.Mock())
 
-    @mock.patch("yardstick.benchmark.contexts.standalone.model.CpuSysCores")
-    @mock.patch("yardstick.benchmark.contexts.standalone.model.Libvirt.update_interrupts_hugepages_perf")
-    def test_pin_vcpu_for_perf(self, mock_update_interrupts_hugepages_perf, mock_CpuSysCores):
-        with mock.patch("yardstick.ssh.SSH") as ssh:
-            ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
-            ssh.return_value = ssh_mock
-        status = Libvirt.pin_vcpu_for_perf(ssh_mock, "vm_0", 4)
-        self.assertIsNotNone(status)
 
 class StandaloneContextHelperTestCase(unittest.TestCase):
 
@@ -148,7 +217,7 @@ class StandaloneContextHelperTestCase(unittest.TestCase):
     }
 
     def setUp(self):
-        self.helper = StandaloneContextHelper()
+        self.helper = model.StandaloneContextHelper()
 
     def test___init__(self):
         self.assertIsNone(self.helper.file_path)
@@ -159,8 +228,9 @@ class StandaloneContextHelperTestCase(unittest.TestCase):
             ssh_mock.execute = \
                 mock.Mock(return_value=(1, "a", ""))
             ssh.return_value = ssh_mock
-        status = StandaloneContextHelper.install_req_libs(ssh_mock)
-        self.assertIsNone(status)
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
+        model.StandaloneContextHelper.install_req_libs(ssh_mock)
 
     def test_get_kernel_module(self):
         with mock.patch("yardstick.ssh.SSH") as ssh:
@@ -168,10 +238,12 @@ class StandaloneContextHelperTestCase(unittest.TestCase):
             ssh_mock.execute = \
                 mock.Mock(return_value=(1, "i40e", ""))
             ssh.return_value = ssh_mock
-        status = StandaloneContextHelper.get_kernel_module(ssh_mock, "05:00.0", None)
-        self.assertEqual(status, "i40e")
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
+        model.StandaloneContextHelper.get_kernel_module(
+            ssh_mock, "05:00.0", None)
 
-    @mock.patch('yardstick.benchmark.contexts.standalone.model.StandaloneContextHelper.get_kernel_module')
+    @mock.patch.object(model.StandaloneContextHelper, 'get_kernel_module')
     def test_get_nic_details(self, mock_get_kernel_module):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
@@ -179,8 +251,10 @@ class StandaloneContextHelperTestCase(unittest.TestCase):
                 mock.Mock(return_value=(1, "i40e ixgbe", ""))
             ssh.return_value = ssh_mock
         mock_get_kernel_module.return_value = "i40e"
-        status = StandaloneContextHelper.get_nic_details(ssh_mock, self.NETWORKS, "dpdk-devbind.py")
-        self.assertIsNotNone(status)
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
+        model.StandaloneContextHelper.get_nic_details(
+            ssh_mock, self.NETWORKS, 'dpdk-devbind.py')
 
     def test_get_virtual_devices(self):
         pattern = "PCI_SLOT_NAME=0000:05:00.0"
@@ -189,8 +263,10 @@ class StandaloneContextHelperTestCase(unittest.TestCase):
             ssh_mock.execute = \
                     mock.Mock(return_value=(1, pattern, ""))
             ssh.return_value = ssh_mock
-        status = StandaloneContextHelper.get_virtual_devices(ssh_mock, "0000:00:05.0")
-        self.assertIsNotNone(status)
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
+        model.StandaloneContextHelper.get_virtual_devices(
+            ssh_mock, '0000:00:05.0')
 
     def _get_file_abspath(self, filename):
         curr_path = os.path.dirname(os.path.abspath(__file__))
@@ -204,40 +280,48 @@ class StandaloneContextHelperTestCase(unittest.TestCase):
 
     def test_parse_pod_file(self):
         self.helper.file_path = self._get_file_abspath("dummy")
-        self.assertRaises(IOError, self.helper.parse_pod_file, self.helper.file_path)
+        self.assertRaises(IOError, self.helper.parse_pod_file,
+                          self.helper.file_path)
 
         self.helper.file_path = self._get_file_abspath(self.NODE_SAMPLE)
-        self.assertRaises(TypeError, self.helper.parse_pod_file, self.helper.file_path)
+        self.assertRaises(TypeError, self.helper.parse_pod_file,
+                          self.helper.file_path)
 
         self.helper.file_path = self._get_file_abspath(self.NODE_SRIOV_SAMPLE)
         self.assertIsNotNone(self.helper.parse_pod_file(self.helper.file_path))
 
     def test_get_mac_address(self):
-        status = StandaloneContextHelper.get_mac_address()
+        status = model.StandaloneContextHelper.get_mac_address()
         self.assertIsNotNone(status)
 
     @mock.patch('yardstick.ssh.SSH')
-    def test_get_mgmt_ip(self, mock_ssh):
+    def test_get_mgmt_ip(self, *args):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                    mock.Mock(return_value=(1, "1.2.3.4 00:00:00:00:00:01", ""))
+            ssh_mock.execute = mock.Mock(
+                return_value=(1, "1.2.3.4 00:00:00:00:00:01", ""))
             ssh.return_value = ssh_mock
-        status = StandaloneContextHelper.get_mgmt_ip(ssh_mock, "00:00:00:00:00:01", "1.1.1.1/24", {})
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
+        status = model.StandaloneContextHelper.get_mgmt_ip(
+            ssh_mock, "00:00:00:00:00:01", "1.1.1.1/24", {})
         self.assertIsNotNone(status)
 
     @mock.patch('yardstick.ssh.SSH')
-    def test_get_mgmt_ip_no(self, mock_ssh):
+    def test_get_mgmt_ip_no(self, *args):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
             ssh_mock.execute = \
                     mock.Mock(return_value=(1, "", ""))
             ssh.return_value = ssh_mock
-
+        # NOTE(ralonsoh): this test doesn't cover function execution. This test
+        # should also check mocked function calls.
         model.WAIT_FOR_BOOT = 0
-        status = StandaloneContextHelper.get_mgmt_ip(ssh_mock, "99", "1.1.1.1/24", {})
+        status = model.StandaloneContextHelper.get_mgmt_ip(
+            ssh_mock, "99", "1.1.1.1/24", {})
         self.assertIsNone(status)
 
+
 class ServerTestCase(unittest.TestCase):
 
     NETWORKS = {
@@ -257,6 +341,7 @@ class ServerTestCase(unittest.TestCase):
          'cidr': '152.16.40.10/24',
          'gateway_ip': '152.16.100.20'}
     }
+
     def setUp(self):
         self.server = model.Server()
 
@@ -282,7 +367,8 @@ class ServerTestCase(unittest.TestCase):
                 'xe1': ['public_0'],
             }
         }
-        status = self.server.generate_vnf_instance({}, self.NETWORKS, "1.1.1.1/24", 'vm_0', vnf, '00:00:00:00:00:01')
+        status = self.server.generate_vnf_instance(
+            {}, self.NETWORKS, '1.1.1.1/24', 'vm_0', vnf, '00:00:00:00:00:01')
         self.assertIsNotNone(status)
 
 class OvsDeployTestCase(unittest.TestCase):
@@ -312,14 +398,17 @@ class OvsDeployTestCase(unittest.TestCase):
         self.assertIsNotNone(self.ovs_deploy.connection)
 
     @mock.patch('yardstick.benchmark.contexts.standalone.model.os')
-    def test_prerequisite(self, mock_ssh):
+    def test_prerequisite(self, *args):
+        # NOTE(ralonsoh): this test should check mocked function calls.
         self.ovs_deploy.helper = mock.Mock()
         self.assertIsNone(self.ovs_deploy.prerequisite())
 
     @mock.patch('yardstick.benchmark.contexts.standalone.model.os')
-    def test_prerequisite(self, mock_ssh):
+    def test_prerequisite_2(self, *args):
+        # NOTE(ralonsoh): this test should check mocked function calls. Rename
+        # this test properly.
         self.ovs_deploy.helper = mock.Mock()
-        self.ovs_deploy.connection.execute = \
-                    mock.Mock(return_value=(1, "1.2.3.4 00:00:00:00:00:01", ""))
+        self.ovs_deploy.connection.execute = mock.Mock(
+            return_value=(1, '1.2.3.4 00:00:00:00:00:01', ''))
         self.ovs_deploy.prerequisite = mock.Mock()
         self.assertIsNone(self.ovs_deploy.ovs_deploy())
diff --git a/tests/unit/benchmark/contexts/standalone/test_sriov.py b/tests/unit/benchmark/contexts/standalone/test_sriov.py
index 50ae5fe13..3ea673abc 100644
--- a/tests/unit/benchmark/contexts/standalone/test_sriov.py
+++ b/tests/unit/benchmark/contexts/standalone/test_sriov.py
@@ -17,12 +17,10 @@
 from __future__ import absolute_import
 import os
 import unittest
-import errno
 import mock
 
-from yardstick.common import constants as consts
+from yardstick import ssh
 from yardstick.benchmark.contexts.standalone import sriov
-from yardstick.network_services.utils import PciAddress
 
 
 class SriovContextTestCase(unittest.TestCase):
@@ -66,6 +64,9 @@ class SriovContextTestCase(unittest.TestCase):
     @mock.patch('yardstick.benchmark.contexts.standalone.sriov.Libvirt')
     @mock.patch('yardstick.benchmark.contexts.standalone.model.Server')
     def test___init__(self, mock_helper, mock_libvirt, mock_server):
+        # pylint: disable=unused-argument
+        # NOTE(ralonsoh): this test doesn't cover function execution.
+        # The pylint exception should be removed.
         self.sriov.helper = mock_helper
         self.sriov.vnf_node = mock_server
         self.assertIsNone(self.sriov.file_path)
@@ -75,14 +76,11 @@ class SriovContextTestCase(unittest.TestCase):
         self.sriov.helper.parse_pod_file = mock.Mock(return_value=[{}, {}, {}])
         self.assertIsNone(self.sriov.init(self.ATTRS))
 
-    @mock.patch('yardstick.ssh.SSH')
+    @mock.patch.object(ssh, 'SSH', return_value=(0, "a", ""))
     def test_deploy(self, mock_ssh):
-        with mock.patch("yardstick.ssh.SSH") as ssh:
-            ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
-            ssh.return_value = ssh_mock
-
+        # pylint: disable=unused-argument
+        # NOTE(ralonsoh): this test doesn't cover function execution.
+        # The pylint exception should be removed.
         self.sriov.vm_deploy = False
         self.assertIsNone(self.sriov.deploy())
 
@@ -94,20 +92,16 @@ class SriovContextTestCase(unittest.TestCase):
         self.sriov.wait_for_vnfs_to_start = mock.Mock(return_value={})
         self.assertIsNone(self.sriov.deploy())
 
-    @mock.patch('yardstick.ssh.SSH')
+    @mock.patch.object(ssh, 'SSH', return_value=(0, "a", ""))
     @mock.patch('yardstick.benchmark.contexts.standalone.sriov.Libvirt')
     def test_undeploy(self, mock_libvirt, mock_ssh):
-        with mock.patch("yardstick.ssh.SSH") as ssh:
-            ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
-            ssh.return_value = ssh_mock
-
+        # pylint: disable=unused-argument
+        # NOTE(ralonsoh): the pylint exception should be removed.
         self.sriov.vm_deploy = False
         self.assertIsNone(self.sriov.undeploy())
 
         self.sriov.vm_deploy = True
-        self.sriov.connection = ssh_mock
+        self.sriov.connection = mock_ssh
         self.sriov.vm_names = ['vm_0', 'vm_1']
         self.sriov.drivers = ['vm_0', 'vm_1']
         self.assertIsNone(self.sriov.undeploy())
@@ -246,27 +240,27 @@ class SriovContextTestCase(unittest.TestCase):
         self.sriov.drivers = []
         self.sriov.networks = self.NETWORKS
         self.sriov.helper.get_mac_address = mock.Mock(return_value="")
-        self.sriov.get_vf_data = mock.Mock(return_value="")
+        self.sriov._get_vf_data = mock.Mock(return_value="")
         self.assertIsNone(self.sriov.configure_nics_for_sriov())
 
+    @mock.patch.object(ssh, 'SSH', return_value=(0, "a", ""))
     @mock.patch('yardstick.benchmark.contexts.standalone.sriov.Libvirt')
-    def test__enable_interfaces(self, mock_libvirt):
-        with mock.patch("yardstick.ssh.SSH") as ssh:
-            ssh_mock = mock.Mock(autospec=ssh.SSH)
-            ssh_mock.execute = \
-                mock.Mock(return_value=(0, "a", ""))
-            ssh.return_value = ssh_mock
+    def test__enable_interfaces(self, mock_libvirt, mock_ssh):
+        # pylint: disable=unused-argument
+        # NOTE(ralonsoh): the pylint exception should be removed.
         self.sriov.vm_deploy = True
-        self.sriov.connection = ssh_mock
+        self.sriov.connection = mock_ssh
         self.sriov.vm_names = ['vm_0', 'vm_1']
         self.sriov.drivers = []
         self.sriov.networks = self.NETWORKS
-        self.sriov.get_vf_data = mock.Mock(return_value="")
+        self.sriov._get_vf_data = mock.Mock(return_value="")
         self.assertIsNone(self.sriov._enable_interfaces(0, 0, ["private_0"], 'test'))
 
     @mock.patch('yardstick.benchmark.contexts.standalone.model.Server')
     @mock.patch('yardstick.benchmark.contexts.standalone.sriov.Libvirt')
     def test_setup_sriov_context(self, mock_libvirt, mock_server):
+        # pylint: disable=unused-argument
+        # NOTE(ralonsoh): the pylint exception should be removed.
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
             ssh_mock.execute = \
@@ -296,7 +290,7 @@ class SriovContextTestCase(unittest.TestCase):
         self.sriov.vnf_node.generate_vnf_instance = mock.Mock(return_value={})
         self.assertIsNotNone(self.sriov.setup_sriov_context())
 
-    def test_get_vf_data(self):
+    def test__get_vf_data(self):
         with mock.patch("yardstick.ssh.SSH") as ssh:
             ssh_mock = mock.Mock(autospec=ssh.SSH)
             ssh_mock.execute = \
@@ -318,5 +312,7 @@ class SriovContextTestCase(unittest.TestCase):
             }
         }
         self.sriov.networks = self.NETWORKS
-        self.sriov.helper.get_virtual_devices = mock.Mock(return_value={"0000:00:01.0": ""})
-        self.assertIsNotNone(self.sriov.get_vf_data("", "0000:00:01.0", "00:00:00:00:00:01", "if0"))
+        self.sriov.helper.get_virtual_devices = mock.Mock(
+            return_value={'0000:00:01.0': ''})
+        self.assertIsNotNone(self.sriov._get_vf_data(
+            '0000:00:01.0', '00:00:00:00:00:01', 'if0'))
diff --git a/tests/unit/network_services/test_utils.py b/tests/unit/network_services/test_utils.py
index 993adbeae..bf98a4474 100644
--- a/tests/unit/network_services/test_utils.py
+++ b/tests/unit/network_services/test_utils.py
@@ -15,8 +15,6 @@
 
 # Unittest for yardstick.network_services.utils
 
-from __future__ import absolute_import
-
 import os
 import unittest
 import mock
@@ -62,3 +60,84 @@ class UtilsTestCase(unittest.TestCase):
             ssh.return_value = ssh_mock
             tool_path = utils.provision_tool(ssh_mock, self.DPDK_PATH)
             self.assertEqual(tool_path, self.DPDK_PATH)
+
+
+class PciAddressTestCase(unittest.TestCase):
+
+    PCI_ADDRESS_DBSF = '000A:07:03.2'
+    PCI_ADDRESS_BSF = '06:02.1'
+    PCI_ADDRESS_DBSF_MULTILINE_1 = '0001:08:04.3\nother text\n'
+    PCI_ADDRESS_DBSF_MULTILINE_2 = 'first line\n   0001:08:04.3 \nother text\n'
+    # Will match and return the first address found.
+    PCI_ADDRESS_DBSF_MULTILINE_3 = '  0001:08:04.1  \n  05:03.1 \nother\n'
+    PCI_ADDRESS_BSF_MULTILINE_1 = 'first line\n   08:04.3 \n 0002:05:03.1\n'
+    BAD_INPUT_1 = 'no address found'
+    BAD_INPUT_2 = '001:08:04.1'
+    BAD_INPUT_3 = '08:4.1'
+
+    def test_pciaddress_dbsf(self):
+        pci_address = utils.PciAddress(PciAddressTestCase.PCI_ADDRESS_DBSF)
+        self.assertEqual('000a', pci_address.domain)
+        self.assertEqual('07', pci_address.bus)
+        self.assertEqual('03', pci_address.slot)
+        self.assertEqual('2', pci_address.function)
+
+    def test_pciaddress_bsf(self):
+        pci_address = utils.PciAddress(PciAddressTestCase.PCI_ADDRESS_BSF)
+        self.assertEqual('0000', pci_address.domain)
+        self.assertEqual('06', pci_address.bus)
+        self.assertEqual('02', pci_address.slot)
+        self.assertEqual('1', pci_address.function)
+
+    def test_pciaddress_dbsf_multiline_1(self):
+        pci_address = utils.PciAddress(
+            PciAddressTestCase.PCI_ADDRESS_DBSF_MULTILINE_1)
+        self.assertEqual('0001', pci_address.domain)
+        self.assertEqual('08', pci_address.bus)
+        self.assertEqual('04', pci_address.slot)
+        self.assertEqual('3', pci_address.function)
+
+    def test_pciaddress_dbsf_multiline_2(self):
+        pci_address = utils.PciAddress(
+            PciAddressTestCase.PCI_ADDRESS_DBSF_MULTILINE_2)
+        self.assertEqual('0001', pci_address.domain)
+        self.assertEqual('08', pci_address.bus)
+        self.assertEqual('04', pci_address.slot)
+        self.assertEqual('3', pci_address.function)
+
+    def test_pciaddress_dbsf_multiline_3(self):
+        pci_address = utils.PciAddress(
+            PciAddressTestCase.PCI_ADDRESS_DBSF_MULTILINE_3)
+        self.assertEqual('0001', pci_address.domain)
+        self.assertEqual('08', pci_address.bus)
+        self.assertEqual('04', pci_address.slot)
+        self.assertEqual('1', pci_address.function)
+
+    def test_pciaddress_bsf_multiline_1(self):
+        pci_address = utils.PciAddress(
+            PciAddressTestCase.PCI_ADDRESS_BSF_MULTILINE_1)
+        self.assertEqual('0000', pci_address.domain)
+        self.assertEqual('08', pci_address.bus)
+        self.assertEqual('04', pci_address.slot)
+        self.assertEqual('3', pci_address.function)
+
+    def test_pciaddress_bad_input_no_address(self):
+        with self.assertRaises(ValueError) as exception:
+            utils.PciAddress(PciAddressTestCase.BAD_INPUT_1)
+        self.assertEqual('Invalid PCI address: {}'.format(
+                PciAddressTestCase.BAD_INPUT_1), str(exception.exception))
+
+    def test_pciaddress_bad_input_dbsf_bad_formatted(self):
+        # In this test case, the domain has only 3 characters instead of 4.
+        pci_address = utils.PciAddress(
+            PciAddressTestCase.BAD_INPUT_2)
+        self.assertEqual('0000', pci_address.domain)
+        self.assertEqual('08', pci_address.bus)
+        self.assertEqual('04', pci_address.slot)
+        self.assertEqual('1', pci_address.function)
+
+    def test_pciaddress_bad_input_bsf_bad_formatted(self):
+        with self.assertRaises(ValueError) as exception:
+            utils.PciAddress(PciAddressTestCase.BAD_INPUT_3)
+        self.assertEqual('Invalid PCI address: {}'.format(
+                PciAddressTestCase.BAD_INPUT_3), str(exception.exception))