Merge "Updated usb_metric and tests to work with py3.4"
diff --git a/tools/lab/metrics/usb_metric.py b/tools/lab/metrics/usb_metric.py
index 2fa569a..8b8d375 100644
--- a/tools/lab/metrics/usb_metric.py
+++ b/tools/lab/metrics/usb_metric.py
@@ -14,15 +14,29 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+import io
import subprocess
+import sys
+
from metrics.metric import Metric
from utils import job
from utils import time_limit
+def _get_output(stdout):
+ if sys.version_info[0] == 2:
+ return iter(stdout.readline, '')
+ else:
+ return io.TextIOWrapper(stdout, encoding="utf-8")
+
+
class UsbMetric(Metric):
"""Class to determine all USB Device traffic over a timeframe."""
+ USB_IO_COMMAND = 'cat /sys/kernel/debug/usb/usbmon/0u | grep -v \'S Ci\''
+ USBMON_CHECK_COMMAND = 'grep usbmon /proc/modules'
+ USBMON_INSTALL_COMMAND = 'modprobe usbmon'
+ DEVICES = 'devices'
def check_usbmon(self):
"""Checks if the kernel module 'usbmon' is installed.
@@ -33,11 +47,11 @@
job.Error: When the module could not be loaded.
"""
try:
- self._shell.run('grep usbmon /proc/modules')
+ self._shell.run(self.USBMON_CHECK_COMMAND)
except job.Error:
print('Kernel module not loaded, attempting to load usbmon')
try:
- self._shell.run('modprobe usbmon')
+ self._shell.run(self.USBMON_INSTALL_COMMAND)
except job.Error as error:
raise job.Error('Cannot load usbmon: %s' % error.result.stderr)
@@ -59,12 +73,12 @@
with time_limit.TimeLimit(time):
# Lines matching 'S Ci' do not match output, and only 4 bytes/sec
process = subprocess.Popen(
- 'cat /sys/kernel/debug/usb/usbmon/0u | grep -v \'S Ci\'',
+ self.USB_IO_COMMAND,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
shell=True)
- for line in iter(process.stdout.readline, ''):
- line = line.decode('utf-8')
+
+ for line in _get_output(process.stdout):
spl_line = line.split(' ')
# Example line spl_line[3] " "[5]
# ffff88080bb00780 2452973093 C Ii:2:003:1 0:8 8 = 00000000
@@ -114,12 +128,15 @@
dev_byte_dict: A dictionary with the key as 'bus:device', leading
0's stripped from bus, and value as the number of bytes transferred.
Returns:
- List of populated Device object. Only devices that have transferred
- data in the timeframe calculated are added.
+ List of populated Device objects.
"""
devices = []
- for dev in dev_byte_dict:
- devices.append(Device(dev, dev_byte_dict[dev], dev_name_dict[dev]))
+ for dev in dev_name_dict:
+ if dev in dev_byte_dict:
+ devices.append(
+ Device(dev, dev_byte_dict[dev], dev_name_dict[dev]))
+ else:
+ devices.append(Device(dev, 0, dev_name_dict[dev]))
return devices
def gather_metric(self):
@@ -133,7 +150,7 @@
self.check_usbmon()
dev_byte_dict = self.get_bytes()
dev_name_dict = self.match_device_id()
- return {'devices': self.gen_output(dev_name_dict, dev_byte_dict)}
+ return {self.DEVICES: self.gen_output(dev_name_dict, dev_byte_dict)}
class Device:
@@ -153,6 +170,7 @@
self.name = name
def __eq__(self, other):
- return self.dev_id == other.dev_id and \
+ return isinstance(other, Device) and \
+ self.dev_id == other.dev_id and \
self.trans_bytes == other.trans_bytes and \
self.name == other.name
diff --git a/tools/lab/tests/usb_metric_test.py b/tools/lab/tests/usb_metric_test.py
index 3e2ddb5..1a7162a 100644
--- a/tools/lab/tests/usb_metric_test.py
+++ b/tools/lab/tests/usb_metric_test.py
@@ -14,10 +14,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from StringIO import StringIO
+from io import BytesIO
import unittest
-import fake
+from tests import fake
from metrics.usb_metric import Device
from metrics.usb_metric import UsbMetric
import mock
@@ -32,8 +32,8 @@
def test_check_get_bytes_2(self):
with mock.patch('subprocess.Popen') as mock_popen:
- mock_popen.return_value.stdout = StringIO(
- 'x x C Ii:2:003:1 0:8 8 = x x\nx x S Ii:2:004:1 -115:8 8 <')
+ mock_popen.return_value.stdout = BytesIO(
+ b'x x C Ii:2:003:1 0:8 8 = x x\nx x S Ii:2:004:1 -115:8 8 <')
self.assertEquals(UsbMetric().get_bytes(0),
{'2:003': 8,
@@ -41,11 +41,12 @@
def test_check_get_bytes_empty(self):
with mock.patch('subprocess.Popen') as mock_popen:
- mock_popen.return_value.stdout = StringIO('')
+ mock_popen.return_value.stdout = BytesIO(b'')
self.assertEquals(UsbMetric().get_bytes(0), {})
def test_match_device_id(self):
- mock_lsusb = 'Bus 003 Device 047: ID 18d1:d00d Device 0\nBus 003 Device 001: ID 1d6b:0002 Device 1'
+ mock_lsusb = ('Bus 003 Device 047: ID 18d1:d00d Device 0\n'
+ 'Bus 003 Device 001: ID 1d6b:0002 Device 1')
exp_res = {'3:047': 'Device 0', '3:001': 'Device 1'}
fake_result = fake.FakeResult(stdout=mock_lsusb)
fake_shell = fake.MockShellCommand(fake_result=fake_result)
@@ -61,13 +62,18 @@
self.assertEquals(m.match_device_id(), exp_res)
def test_gen_output(self):
- dev_name_dict = {'1:001': 'Device 1', '1:002': 'Device 2'}
+ dev_name_dict = {
+ '1:001': 'Device 1',
+ '1:002': 'Device 2',
+ '1:003': 'Device 3'
+ }
dev_byte_dict = {'1:001': 256, '1:002': 200}
- exp_res = [
- Device('1:002', 200, 'Device 2'),
- Device('1:001', 256, 'Device 1'),
- ]
+ dev_1 = Device('1:002', 200, 'Device 2')
+ dev_2 = Device('1:001', 256, 'Device 1')
+ dev_3 = Device('1:003', 0, 'Device 3')
act_out = UsbMetric().gen_output(dev_name_dict, dev_byte_dict)
- self.assertListEqual(exp_res, act_out)
+
+ self.assertTrue(dev_1 in act_out and dev_2 in act_out and
+ dev_3 in act_out)