Enable traffic generator PID in "NSPerf" scenario setup

Now "NSPerf" scenario will inform about the traffic generator PIDs after
setup process. With this information, IterationIPC runner will be able
to receive the messages sent by those traffic generators and control
the main iteration loop.

The following example, using vFW as VNF and OpenStack as context, makes
use of this new runner implementation:
  /samples/vnf_samples/nsut/vfw/
    tc_heat_rfc2544_ipv4_1rule_1flow_64B_trex_iterationipc.yaml

JIRA: YARDSTICK-1127

Change-Id: I46b1368bc209680b88ff9fb5c3b9beadf6271ac9
Signed-off-by: Rodolfo Alonso Hernandez <rodolfo.alonso.hernandez@intel.com>

13 files changed, 205 insertions, 56 deletions

diff --git a/samples/vnf_samples/nsut/vfw/tc_heat_rfc2544_ipv4_1rule_1flow_64B_trex_iterationipc.yaml b/samples/vnf_samples/nsut/vfw/tc_heat_rfc2544_ipv4_1rule_1flow_64B_trex_iterationipc.yaml
new file mode 100644
index 000000000..184ed6881
--- /dev/null
+++ b/samples/vnf_samples/nsut/vfw/tc_heat_rfc2544_ipv4_1rule_1flow_64B_trex_iterationipc.yaml
@@ -0,0 +1,96 @@
+# Copyright (c) 2016-2017 Intel Corporation
+#
+# 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.
+
+---
+{% set provider = provider or none %}
+{% set physical_networks = physical_networks or ['physnet1', 'physnet2'] %}
+{% set segmentation_id = segmentation_id or none %}
+
+schema: yardstick:task:0.1
+scenarios:
+- type: NSPerf
+  traffic_profile: ../../traffic_profiles/ipv4_throughput.yaml
+  topology: vfw-tg-topology.yaml
+  nodes:
+    tg__0: trafficgen_1.yardstick
+    vnf__0: vnf.yardstick
+  options:
+    hugepages_gb: 8
+    framesize:
+      uplink: {64B: 100}
+      downlink: {64B: 100}
+    flow:
+      src_ip: [{'tg__0': 'xe0'}]
+      dst_ip: [{'tg__0': 'xe1'}]
+      count: 1
+    traffic_type: 4
+    rfc2544:
+      allowed_drop_rate: 0.0001 - 0.0001
+    vnf__0:
+      rules: acl_1rule.yaml
+      vnf_config: {lb_config: 'SW', lb_count: 1, worker_config: '1C/1T', worker_threads: 1}
+  runner:
+    type: IterationIPC
+    iterations: 10
+    timeout: 60
+context:
+  # put node context first, so we don't HEAT deploy if node has errors
+  name: yardstick
+  image: yardstick-samplevnfs
+  flavor:
+    vcpus: 10
+    ram: 12288
+    disk: 6
+    extra_specs:
+      hw:cpu_sockets: 1
+      hw:cpu_cores: 10
+      hw:cpu_threads: 1
+  user: ubuntu
+  placement_groups:
+    pgrp1:
+      policy: "availability"
+  servers:
+    vnf:
+      floating_ip: true
+      placement: "pgrp1"
+    trafficgen_1:
+      floating_ip: true
+      placement: "pgrp1"
+  networks:
+    mgmt:
+      cidr: '10.0.1.0/24'
+    xe0:
+      cidr: '10.0.2.0/24'
+      gateway_ip: 'null'
+      {% if provider %}
+      provider: {{ provider }}
+      physical_network: {{ physical_networks[0] }}
+        {% if segmentation_id %}
+      segmentation_id: {{ segmentation_id }}
+        {% endif %}
+      {% endif %}
+      port_security_enabled: False
+      enable_dhcp: 'false'
+    xe1:
+      cidr: '10.0.3.0/24'
+      gateway_ip: 'null'
+      {% if provider %}
+      provider: {{ provider }}
+      physical_network: {{ physical_networks[1] }}
+        {% if segmentation_id %}
+      segmentation_id: {{ segmentation_id }}
+        {% endif %}
+      {% endif %}
+      port_security_enabled: False
+      enable_dhcp: 'false'
diff --git a/yardstick/benchmark/runners/iteration_ipc.py b/yardstick/benchmark/runners/iteration_ipc.py
index 690f2660b..43aa7f489 100644
--- a/yardstick/benchmark/runners/iteration_ipc.py
+++ b/yardstick/benchmark/runners/iteration_ipc.py
@@ -44,7 +44,7 @@ class RunnerIterationIPCEndpoint(consumer.NotificationHandler):
     """Endpoint class for ``RunnerIterationIPCConsumer``"""
 
     def tg_method_started(self, ctxt, **kwargs):
-        if ctxt['id'] in self._ctx_pids:
+        if ctxt['id'] in self._ctx_ids:
             self._queue.put(
                 {'id': ctxt['id'],
                  'action': messaging.TG_METHOD_STARTED,
@@ -53,7 +53,7 @@ class RunnerIterationIPCEndpoint(consumer.NotificationHandler):
                 QUEUE_PUT_TIMEOUT)
 
     def tg_method_finished(self, ctxt, **kwargs):
-        if ctxt['id'] in self._ctx_pids:
+        if ctxt['id'] in self._ctx_ids:
             self._queue.put(
                 {'id': ctxt['id'],
                  'action': messaging.TG_METHOD_FINISHED,
@@ -61,7 +61,7 @@ class RunnerIterationIPCEndpoint(consumer.NotificationHandler):
                      kwargs)})
 
     def tg_method_iteration(self, ctxt, **kwargs):
-        if ctxt['id'] in self._ctx_pids:
+        if ctxt['id'] in self._ctx_ids:
             self._queue.put(
                 {'id': ctxt['id'],
                  'action': messaging.TG_METHOD_ITERATION,
@@ -124,7 +124,8 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
 
     if 'setup' not in run_step:
         raise exceptions.RunnerIterationIPCSetupActionNeeded()
-    producer_ctxs = benchmark.setup()
+    benchmark.setup()
+    producer_ctxs = benchmark.get_mq_ids()
     if not producer_ctxs:
         raise exceptions.RunnerIterationIPCNoCtxs()
 
diff --git a/yardstick/benchmark/scenarios/base.py b/yardstick/benchmark/scenarios/base.py
index 30ac1bea9..90a87ac29 100644
--- a/yardstick/benchmark/scenarios/base.py
+++ b/yardstick/benchmark/scenarios/base.py
@@ -119,3 +119,7 @@ class Scenario(object):
             except TypeError:
                 dic[k] = v
         return dic
+
+    def get_mq_ids(self):  # pragma: no cover
+        """Return stored MQ producer IDs, if defined"""
+        pass
diff --git a/yardstick/benchmark/scenarios/networking/vnf_generic.py b/yardstick/benchmark/scenarios/networking/vnf_generic.py
index eb62d6222..3bb168c70 100644
--- a/yardstick/benchmark/scenarios/networking/vnf_generic.py
+++ b/yardstick/benchmark/scenarios/networking/vnf_generic.py
@@ -50,7 +50,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
 
     __scenario_type__ = "NSPerf"
 
-    def __init__(self, scenario_cfg, context_cfg):  # Yardstick API
+    def __init__(self, scenario_cfg, context_cfg):  # pragma: no cover
         super(NetworkServiceTestCase, self).__init__()
         self.scenario_cfg = scenario_cfg
         self.context_cfg = context_cfg
@@ -61,6 +61,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
         self.traffic_profile = None
         self.node_netdevs = {}
         self.bin_path = get_nsb_option('bin_path', '')
+        self._mq_ids = []
 
     def _get_ip_flow_range(self, ip_start_range):
 
@@ -168,18 +169,18 @@ class NetworkServiceTestCase(scenario_base.Scenario):
         topology_yaml = vnfdgen.generate_vnfd(topology, topolgy_data)
         self.topology = topology_yaml["nsd:nsd-catalog"]["nsd"][0]
 
-    def _find_vnf_name_from_id(self, vnf_id):
+    def _find_vnf_name_from_id(self, vnf_id):  # pragma: no cover
         return next((vnfd["vnfd-id-ref"]
                      for vnfd in self.topology["constituent-vnfd"]
                      if vnf_id == vnfd["member-vnf-index"]), None)
 
-    def _find_vnfd_from_vnf_idx(self, vnf_id):
+    def _find_vnfd_from_vnf_idx(self, vnf_id):  # pragma: no cover
         return next((vnfd
                      for vnfd in self.topology["constituent-vnfd"]
                      if vnf_id == vnfd["member-vnf-index"]), None)
 
     @staticmethod
-    def find_node_if(nodes, name, if_name, vld_id):
+    def find_node_if(nodes, name, if_name, vld_id):  # pragma: no cover
         try:
             # check for xe0, xe1
             intf = nodes[name]["interfaces"][if_name]
@@ -272,14 +273,14 @@ class NetworkServiceTestCase(scenario_base.Scenario):
             node0_if["peer_intf"] = node1_copy
             node1_if["peer_intf"] = node0_copy
 
-    def _update_context_with_topology(self):
+    def _update_context_with_topology(self):  # pragma: no cover
         for vnfd in self.topology["constituent-vnfd"]:
             vnf_idx = vnfd["member-vnf-index"]
             vnf_name = self._find_vnf_name_from_id(vnf_idx)
             vnfd = self._find_vnfd_from_vnf_idx(vnf_idx)
             self.context_cfg["nodes"][vnf_name].update(vnfd)
 
-    def _generate_pod_yaml(self):
+    def _generate_pod_yaml(self):  # pragma: no cover
         context_yaml = os.path.join(LOG_DIR, "pod-{}.yaml".format(self.scenario_cfg['task_id']))
         # convert OrderedDict to a list
         # pod.yaml nodes is a list
@@ -293,7 +294,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
                            explicit_start=True)
 
     @staticmethod
-    def _serialize_node(node):
+    def _serialize_node(node):  # pragma: no cover
         new_node = copy.deepcopy(node)
         # name field is required
         # remove context suffix
@@ -315,7 +316,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
         self._update_context_with_topology()
 
     @classmethod
-    def get_vnf_impl(cls, vnf_model_id):
+    def get_vnf_impl(cls, vnf_model_id):  # pragma: no cover
         """ Find the implementing class from vnf_model["vnf"]["name"] field
 
         :param vnf_model_id: parsed vnfd model ID field
@@ -343,7 +344,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
         raise exceptions.IncorrectConfig(error_msg=message)
 
     @staticmethod
-    def create_interfaces_from_node(vnfd, node):
+    def create_interfaces_from_node(vnfd, node):  # pragma: no cover
         ext_intfs = vnfd["vdu"][0]["external-interface"] = []
         # have to sort so xe0 goes first
         for intf_name, intf in sorted(node['interfaces'].items()):
@@ -412,10 +413,7 @@ class NetworkServiceTestCase(scenario_base.Scenario):
         return vnfs
 
     def setup(self):
-        """ Setup infrastructure, provission VNFs & start traffic
-
-        :return:
-        """
+        """Setup infrastructure, provission VNFs & start traffic"""
         # 1. Verify if infrastructure mapping can meet topology
         self.map_topology_to_infrastructure()
         # 1a. Load VNF models
@@ -457,6 +455,11 @@ class NetworkServiceTestCase(scenario_base.Scenario):
         for traffic_gen in traffic_runners:
             LOG.info("Starting traffic on %s", traffic_gen.name)
             traffic_gen.run_traffic(self.traffic_profile)
+            self._mq_ids.append(traffic_gen.get_mq_producer_id())
+
+    def get_mq_ids(self):  # pragma: no cover
+        """Return stored MQ producer IDs"""
+        return self._mq_ids
 
     def run(self, result):  # yardstick API
         """ Yardstick calls run() at intervals defined in the yaml and
@@ -495,10 +498,10 @@ class NetworkServiceTestCase(scenario_base.Scenario):
             LOG.exception("")
             raise RuntimeError("Error in teardown")
 
-    def pre_run_wait_time(self, time_seconds):
+    def pre_run_wait_time(self, time_seconds):  # pragma: no cover
         """Time waited before executing the run method"""
         time.sleep(time_seconds)
 
-    def post_run_wait_time(self, time_seconds):
+    def post_run_wait_time(self, time_seconds):  # pragma: no cover
         """Time waited after executing the run method"""
         pass
diff --git a/yardstick/common/messaging/consumer.py b/yardstick/common/messaging/consumer.py
index 24ec6f184..c99d7ed27 100644
--- a/yardstick/common/messaging/consumer.py
+++ b/yardstick/common/messaging/consumer.py
@@ -29,9 +29,9 @@ LOG = logging.getLogger(__name__)
 class NotificationHandler(object):
     """Abstract class to define a endpoint object for a MessagingConsumer"""
 
-    def __init__(self, _id, ctx_pids, queue):
+    def __init__(self, _id, ctx_ids, queue):
         self._id = _id
-        self._ctx_pids = ctx_pids
+        self._ctx_ids = ctx_ids
         self._queue = queue
 
 
@@ -43,11 +43,11 @@ class MessagingConsumer(object):
     the messages published by a `MessagingNotifier`.
     """
 
-    def __init__(self, topic, pids, endpoints, fanout=True):
+    def __init__(self, topic, ids, endpoints, fanout=True):
         """Init function.
 
         :param topic: (string) MQ exchange topic
-        :param pids: (list of int) list of PIDs of the processes implementing
+        :param ids: (list of int) list of IDs of the processes implementing
                      the MQ Notifier which will be in the message context
         :param endpoints: (list of class) list of classes implementing the
                           methods (see `MessagingNotifier.send_message) used by
@@ -58,7 +58,7 @@ class MessagingConsumer(object):
         :returns: `MessagingConsumer` class object
         """
 
-        self._pids = pids
+        self._ids = ids
         self._endpoints = endpoints
         self._transport = oslo_messaging.get_rpc_transport(
             cfg.CONF, url=messaging.TRANSPORT_URL)
diff --git a/yardstick/common/messaging/producer.py b/yardstick/common/messaging/producer.py
index b6adc0c17..aadab649d 100644
--- a/yardstick/common/messaging/producer.py
+++ b/yardstick/common/messaging/producer.py
@@ -34,18 +34,18 @@ class MessagingProducer(object):
     messages in a message queue.
     """
 
-    def __init__(self, topic, pid=os.getpid(), fanout=True):
+    def __init__(self, topic, _id=os.getpid(), fanout=True):
         """Init function.
 
         :param topic: (string) MQ exchange topic
-        :param pid: (int) PID of the process implementing this MQ Notifier
+        :param id: (int) ID of the process implementing this MQ Notifier
         :param fanout: (bool) MQ clients may request that a copy of the message
                        be delivered to all servers listening on a topic by
                        setting fanout to ``True``, rather than just one of them
         :returns: `MessagingNotifier` class object
         """
         self._topic = topic
-        self._pid = pid
+        self._id = _id
         self._fanout = fanout
         self._transport = oslo_messaging.get_rpc_transport(
             cfg.CONF, url=messaging.TRANSPORT_URL)
@@ -65,6 +65,11 @@ class MessagingProducer(object):
                        consumer endpoints
         :param payload: (subclass `Payload`) payload content
         """
-        self._notifier.cast({'pid': self._pid},
+        self._notifier.cast({'id': self._id},
                             method,
                             **payload.obj_to_dict())
+
+    @property
+    def id(self):
+        """Return MQ producer ID"""
+        return self._id
diff --git a/yardstick/network_services/vnf_generic/vnf/base.py b/yardstick/network_services/vnf_generic/vnf/base.py
index c385eb6fd..fb41a4e4a 100644
--- a/yardstick/network_services/vnf_generic/vnf/base.py
+++ b/yardstick/network_services/vnf_generic/vnf/base.py
@@ -147,9 +147,9 @@ class TrafficGeneratorProducer(producer.MessagingProducer):
     This message producer must be instantiated in the process created
     "run_traffic" process.
     """
-    def __init__(self, pid):
+    def __init__(self, _id):
         super(TrafficGeneratorProducer, self).__init__(messaging.TOPIC_TG,
-                                                       pid=pid)
+                                                       _id=_id)
 
     def tg_method_started(self, version=1):
         """Send a message to inform the traffic generation has started"""
@@ -324,9 +324,15 @@ class GenericTrafficGen(GenericVNF):
         """
         pass
 
-    def _setup_mq_producer(self, pid):
+    @staticmethod
+    def _setup_mq_producer(id):
         """Setup the TG MQ producer to send messages between processes
 
         :return: (derived class from ``MessagingProducer``) MQ producer object
         """
-        return TrafficGeneratorProducer(pid)
+        return TrafficGeneratorProducer(id)
+
+    def get_mq_producer_id(self):
+        """Return the MQ producer ID if initialized"""
+        if self._mq_producer:
+            return self._mq_producer.get_id()
diff --git a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
index a37f4f72d..bc65380d3 100644
--- a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
+++ b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py
@@ -918,12 +918,12 @@ class SampleVNFTrafficGen(GenericTrafficGen):
                 LOG.info("%s TG Server is up and running.", self.APP_NAME)
                 return self._tg_process.exitcode
 
-    def _traffic_runner(self, traffic_profile, mq_pid):
+    def _traffic_runner(self, traffic_profile, mq_id):
         # always drop connections first thing in new processes
         # so we don't get paramiko errors
         self.ssh_helper.drop_connection()
         LOG.info("Starting %s client...", self.APP_NAME)
-        self._mq_producer = self._setup_mq_producer(mq_pid)
+        self._mq_producer = self._setup_mq_producer(mq_id)
         self.resource_helper.run_traffic(traffic_profile, self._mq_producer)
 
     def run_traffic(self, traffic_profile):
@@ -937,9 +937,9 @@ class SampleVNFTrafficGen(GenericTrafficGen):
         name = '{}-{}-{}-{}'.format(self.name, self.APP_NAME,
                                     traffic_profile.__class__.__name__,
                                     os.getpid())
-        mq_pid = uuid.uuid1().int
-        self._traffic_process = Process(name=name, target=self._traffic_runner,
-                                        args=(traffic_profile, mq_pid))
+        self._traffic_process = Process(
+            name=name, target=self._traffic_runner,
+            args=(traffic_profile, uuid.uuid1().int))
         self._traffic_process.start()
         # Wait for traffic process to start
         while self.resource_helper.client_started.value == 0:
@@ -948,8 +948,6 @@ class SampleVNFTrafficGen(GenericTrafficGen):
             if not self._traffic_process.is_alive():
                 break
 
-        return mq_pid
-
     def collect_kpi(self):
         # check if the tg processes have exited
         physical_node = Context.get_physical_node_from_server(
diff --git a/yardstick/tests/functional/common/messaging/test_messaging.py b/yardstick/tests/functional/common/messaging/test_messaging.py
index 99874343b..f3e31e718 100644
--- a/yardstick/tests/functional/common/messaging/test_messaging.py
+++ b/yardstick/tests/functional/common/messaging/test_messaging.py
@@ -32,25 +32,25 @@ class DummyPayload(payloads.Payload):
 class DummyEndpoint(consumer.NotificationHandler):
 
     def info(self, ctxt, **kwargs):
-        if ctxt['pid'] in self._ctx_pids:
-            self._queue.put('ID {}, data: {}, pid: {}'.format(
-                self._id, kwargs['data'], ctxt['pid']))
+        if ctxt['id'] in self._ctx_ids:
+            self._queue.put('Nr {}, data: {}, id: {}'.format(
+                self._id, kwargs['data'], ctxt['id']))
 
 
 class DummyConsumer(consumer.MessagingConsumer):
 
-    def __init__(self, _id, ctx_pids, queue):
+    def __init__(self, _id, ctx_ids, queue):
         self._id = _id
-        endpoints = [DummyEndpoint(_id, ctx_pids, queue)]
-        super(DummyConsumer, self).__init__(TOPIC, ctx_pids, endpoints)
+        endpoints = [DummyEndpoint(_id, ctx_ids, queue)]
+        super(DummyConsumer, self).__init__(TOPIC, ctx_ids, endpoints)
 
 
 class DummyProducer(producer.MessagingProducer):
     pass
 
 
-def _run_consumer(_id, ctx_pids, queue):
-    _consumer = DummyConsumer(_id, ctx_pids, queue)
+def _run_consumer(_id, ctx_ids, queue):
+    _consumer = DummyConsumer(_id, ctx_ids, queue)
     _consumer.start_rpc_server()
     _consumer.wait()
 
@@ -67,8 +67,8 @@ class MessagingTestCase(base.BaseFunctionalTestCase):
         num_consumers = 10
         ctx_1 = 100001
         ctx_2 = 100002
-        producers = [DummyProducer(TOPIC, pid=ctx_1),
-                     DummyProducer(TOPIC, pid=ctx_2)]
+        producers = [DummyProducer(TOPIC, _id=ctx_1),
+                     DummyProducer(TOPIC, _id=ctx_2)]
 
         processes = []
         for i in range(num_consumers):
@@ -91,7 +91,7 @@ class MessagingTestCase(base.BaseFunctionalTestCase):
             output.append(output_queue.get(True, 1))
 
         self.assertEqual(num_consumers * 4, len(output))
-        msg_template = 'ID {}, data: {}, pid: {}'
+        msg_template = 'Nr {}, data: {}, id: {}'
         for i in range(num_consumers):
             for ctx in [ctx_1, ctx_2]:
                 for message in ['message 0', 'message 1']:
diff --git a/yardstick/tests/unit/benchmark/scenarios/networking/test_vnf_generic.py b/yardstick/tests/unit/benchmark/scenarios/networking/test_vnf_generic.py
index bb1a7aaca..77a54c0b8 100644
--- a/yardstick/tests/unit/benchmark/scenarios/networking/test_vnf_generic.py
+++ b/yardstick/tests/unit/benchmark/scenarios/networking/test_vnf_generic.py
@@ -553,6 +553,7 @@ class TestNetworkServiceTestCase(unittest.TestCase):
             tgen.verify_traffic = lambda x: verified_dict
             tgen.terminate = mock.Mock(return_value=True)
             tgen.name = "tgen__1"
+            tgen.run_traffic.return_value = 'tg_id'
             vnf = mock.Mock(autospec=GenericVNF)
             vnf.runs_traffic = False
             vnf.terminate = mock.Mock(return_value=True)
@@ -565,7 +566,6 @@ class TestNetworkServiceTestCase(unittest.TestCase):
             self.s.load_vnf_models = mock.Mock(return_value=self.s.vnfs)
             self.s._fill_traffic_profile = \
                 mock.Mock(return_value=TRAFFIC_PROFILE)
-            self.assertIsNone(self.s.setup())
 
     def test_setup_exception(self):
         with mock.patch("yardstick.ssh.SSH") as ssh:
@@ -656,6 +656,9 @@ class TestNetworkServiceTestCase(unittest.TestCase):
         )
         self.assertEqual(self.s.topology, 'fake_nsd')
 
+    def test_get_mq_ids(self):
+        self.assertEqual(self.s._mq_ids, self.s.get_mq_ids())
+
     def test_teardown(self):
         vnf = mock.Mock(autospec=GenericVNF)
         vnf.terminate = mock.Mock(return_value=True)
diff --git a/yardstick/tests/unit/common/messaging/test_producer.py b/yardstick/tests/unit/common/messaging/test_producer.py
index 0289689dc..22286e5c3 100644
--- a/yardstick/tests/unit/common/messaging/test_producer.py
+++ b/yardstick/tests/unit/common/messaging/test_producer.py
@@ -44,3 +44,10 @@ class MessagingProducerTestCase(ut_base.BaseUnitTestCase):
                 topic='test_topic', fanout=True, server=messaging.SERVER)
             mock_RPCClient.assert_called_once_with('test_rpc_transport',
                                                    'test_Target')
+
+    def test_id(self):
+        with mock.patch.object(oslo_messaging, 'RPCClient'), \
+                mock.patch.object(oslo_messaging, 'get_rpc_transport'), \
+                mock.patch.object(oslo_messaging, 'Target'):
+            msg_producer = _MessagingProducer('topic', 'id_to_check')
+        self.assertEqual('id_to_check', msg_producer.id)
diff --git a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py
index 11cba4d79..43e5ac839 100644
--- a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py
+++ b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_base.py
@@ -145,6 +145,24 @@ VNFD = {
 }
 
 
+class _DummyGenericTrafficGen(base.GenericTrafficGen):  # pragma: no cover
+
+    def run_traffic(self, *args):
+        pass
+
+    def terminate(self):
+        pass
+
+    def collect_kpi(self):
+        pass
+
+    def instantiate(self, *args):
+        pass
+
+    def scale(self, flavor=''):
+        pass
+
+
 class FileAbsPath(object):
     def __init__(self, module_file):
         super(FileAbsPath, self).__init__()
@@ -240,6 +258,15 @@ class GenericTrafficGenTestCase(unittest.TestCase):
                "scale, terminate")
         self.assertEqual(msg, str(exc.exception))
 
+    def test_get_mq_producer_id(self):
+        vnfd = {'benchmark': {'kpi': mock.ANY},
+                'vdu': [{'external-interface': 'ext_int'}]
+                }
+        tg = _DummyGenericTrafficGen('name', vnfd)
+        tg._mq_producer = mock.Mock()
+        tg._mq_producer.get_id.return_value = 'fake_id'
+        self.assertEqual('fake_id', tg.get_mq_producer_id())
+
 
 class TrafficGeneratorProducerTestCase(unittest.TestCase):
 
@@ -250,15 +277,15 @@ class TrafficGeneratorProducerTestCase(unittest.TestCase):
     @mock.patch.object(cfg, 'CONF')
     def test__init(self, mock_config, mock_transport, mock_rpcclient,
                    mock_target):
-        pid = uuid.uuid1().int
-        tg_producer = base.TrafficGeneratorProducer(pid)
+        _id = uuid.uuid1().int
+        tg_producer = base.TrafficGeneratorProducer(_id)
         mock_transport.assert_called_once_with(
             mock_config, url='rabbit://yardstick:yardstick@localhost:5672/')
         mock_target.assert_called_once_with(topic=messaging.TOPIC_TG,
                                             fanout=True,
                                             server=messaging.SERVER)
         mock_rpcclient.assert_called_once_with('rpc_transport', 'rpc_target')
-        self.assertEqual(pid, tg_producer._pid)
+        self.assertEqual(_id, tg_producer._id)
         self.assertEqual(messaging.TOPIC_TG, tg_producer._topic)
 
     @mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target')
diff --git a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_trex.py b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_trex.py
index 08ae27422..700e910f9 100644
--- a/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_trex.py
+++ b/yardstick/tests/unit/network_services/vnf_generic/vnf/test_tg_trex.py
@@ -454,9 +454,8 @@ class TestTrexTrafficGen(unittest.TestCase):
         self.sut.ssh_helper.run = mock.Mock()
         self.sut._traffic_runner = mock.Mock(return_value=0)
         self.sut.resource_helper.client_started.value = 1
-        result = self.sut.run_traffic(mock_traffic_profile)
+        self.sut.run_traffic(mock_traffic_profile)
         self.sut._traffic_process.terminate()
-        self.assertIsNotNone(result)
 
     def test_terminate(self):
         vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]