cts: Migrate cts.py to python2/3 compatible

BUG=chromium:1031705
BRANCH=master
TEST=None

Signed-off-by: kerker <[email protected]>
Change-Id: If043aa2d7d8b758571f43730635f741e3b81d8ad
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/ec/+/2431316
Reviewed-by: Jack Rosenthal <[email protected]>
Reviewed-by: Daisuke Nojiri <[email protected]>

Author: flere114

cts/common/board.py

@@ -2,13 +2,19 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+# Note: This is a py2/3 compatible file.
+
+from __future__ import print_function
+
 from abc import ABCMeta
 from abc import abstractmethod
 import os
 import shutil
 import subprocess as sp
 import serial
 
+import six
+
 
 OCD_SCRIPT_DIR = '/usr/share/openocd/scripts'
 OPENOCD_CONFIGS = {
@@ -24,7 +30,13 @@
 REBOOT_MARKER = 'UART initialized after reboot'
 
 
-class Board(object):
+def get_subprocess_args():
+  if six.PY3:
+    return {'encoding': 'utf-8'}
+  return {}
+
+
+class Board(six.with_metaclass(ABCMeta, object)):
   """Class representing a single board connected to a host machine.
 
   Attributes:
@@ -38,8 +50,6 @@ class Board(object):
     tty: String of file descriptor for tty_port
   """
 
-  __metaclass__ = ABCMeta  # This is an Abstract Base Class (ABC)
-
   def __init__(self, board, module, hla_serial=None):
     """Initializes a board object with given attributes.
 
@@ -80,7 +90,7 @@ def get_stlink_serials():
       List of serials
     """
     usb_args = ['sudo', 'lsusb', '-v', '-d', '0x0483:0x374b']
-    st_link_info = sp.check_output(usb_args)
+    st_link_info = sp.check_output(usb_args, **get_subprocess_args())
     st_serials = []
     for line in st_link_info.split('\n'):
       if 'iSerial' not in line:
@@ -123,7 +133,7 @@ def send_openocd_commands(self, commands):
 
   def dump_openocd_log(self):
     with open(self.openocd_log) as log:
-      print log.read()
+      print(log.read())
 
   def build(self, ec_dir):
     """Builds test suite module for board.
@@ -151,7 +161,7 @@ def build(self, ec_dir):
 
   def dump_build_log(self):
     with open(self.build_log) as log:
-      print log.read()
+      print(log.read())
 
   def flash(self, image_path):
     """Flashes board with most recent build ec.bin."""
@@ -212,7 +222,7 @@ def read_tty(self, max_boot_count=1):
     line = []
     boot = 0
     while True:
-      c = self.tty.read()
+      c = self.tty.read().decode('utf-8')
       if not c:
         break
       line.append(c)
@@ -240,7 +250,8 @@ def identify_tty_port(self):
     for device in com_devices:
       self.tty_port = os.path.join(dev_dir, device)
       properties = sp.check_output(
-          ['udevadm', 'info', '-a', '-n', self.tty_port, '--query=property'])
+          ['udevadm', 'info', '-a', '-n', self.tty_port, '--query=property'],
+          **get_subprocess_args())
       for line in [l.strip() for l in properties.split('\n')]:
         if line.startswith(id_prefix):
           if self.hla_serial == line[len(id_prefix):]:
@@ -311,7 +322,7 @@ def save_serial(self):
       f.write(s)
       self.hla_serial = s
 
-    print 'Your TH serial', s, 'has been saved as', self.serial_path
+    print('Your TH serial', s, 'has been saved as', self.serial_path)
     return
 
 

cts/cts.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python2
+#!/usr/bin/env python
 #
 # Copyright 2016 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
@@ -16,6 +16,9 @@
 #   $ ./cts.py
 # It'll run mock tests. The result will be stored in CTS_TEST_RESULT_DIR.
 
+# Note: This is a py2/3 compatible file.
+
+from __future__ import print_function
 
 import argparse
 import os
@@ -76,10 +79,10 @@ def __init__(self, ec_dir, th, dut, module):
 
   def build(self):
     """Build images for DUT and TH."""
-    print 'Building DUT image...'
+    print('Building DUT image...')
     if not self.dut.build(self.ec_dir):
       raise RuntimeError('Building module %s for DUT failed' % (self.module))
-    print 'Building TH image...'
+    print('Building TH image...')
     if not self.th.build(self.ec_dir):
       raise RuntimeError('Building module %s for TH failed' % (self.module))
 
@@ -88,11 +91,11 @@ def flash_boards(self):
     cts_module = 'cts_' + self.module
     image_path = os.path.join('build', self.th.board, cts_module, 'ec.bin')
     self.identify_boards()
-    print 'Flashing TH with', image_path
+    print('Flashing TH with', image_path)
     if not self.th.flash(image_path):
       raise RuntimeError('Flashing TH failed')
     image_path = os.path.join('build', self.dut.board, cts_module, 'ec.bin')
-    print 'Flashing DUT with', image_path
+    print('Flashing DUT with', image_path)
     if not self.dut.flash(image_path):
       raise RuntimeError('Flashing DUT failed')
 
@@ -212,7 +215,7 @@ def parse_output(self, output):
 
   def get_return_code_name(self, code, strip_prefix=False):
     name = ''
-    for k, v in self.return_codes.iteritems():
+    for k, v in self.return_codes.items():
       if v == code:
         if strip_prefix:
           name = k[len(CTS_RC_PREFIX):]
@@ -299,52 +302,52 @@ def evaluate_result(self, result, expected_rc, expected_string):
 
   def run(self):
     """Resets boards, records test results in results dir."""
-    print 'Reading serials...'
+    print('Reading serials...')
     self.identify_boards()
-    print 'Opening DUT tty...'
+    print('Opening DUT tty...')
     self.dut.setup_tty()
-    print 'Opening TH tty...'
+    print('Opening TH tty...')
     self.th.setup_tty()
 
     # Boards might be still writing to tty. Wait a few seconds before flashing.
     time.sleep(3)
 
     # clear buffers
-    print 'Clearing DUT tty...'
+    print('Clearing DUT tty...')
     self.dut.read_tty()
-    print 'Clearing TH tty...'
+    print('Clearing TH tty...')
     self.th.read_tty()
 
     # Resets the boards and allows them to run tests
     # Due to current (7/27/16) version of sync function,
     # both boards must be rest and halted, with the th
     # resuming first, in order for the test suite to run in sync
-    print 'Halting TH...'
+    print('Halting TH...')
     if not self.th.reset_halt():
       raise RuntimeError('Failed to halt TH')
-    print 'Halting DUT...'
+    print('Halting DUT...')
     if not self.dut.reset_halt():
       raise RuntimeError('Failed to halt DUT')
-    print 'Resuming TH...'
+    print('Resuming TH...')
     if not self.th.resume():
       raise RuntimeError('Failed to resume TH')
-    print 'Resuming DUT...'
+    print('Resuming DUT...')
     if not self.dut.resume():
       raise RuntimeError('Failed to resume DUT')
 
     time.sleep(MAX_SUITE_TIME_SEC)
 
-    print 'Reading DUT tty...'
+    print('Reading DUT tty...')
     dut_output, _ = self.dut.read_tty()
     self.dut.close_tty()
-    print 'Reading TH tty...'
+    print('Reading TH tty...')
     th_output, _ = self.th.read_tty()
     self.th.close_tty()
 
-    print 'Halting TH...'
+    print('Halting TH...')
     if not self.th.reset_halt():
       raise RuntimeError('Failed to halt TH')
-    print 'Halting DUT...'
+    print('Halting DUT...')
     if not self.dut.reset_halt():
       raise RuntimeError('Failed to halt DUT')
 
@@ -353,7 +356,7 @@ def run(self):
                        'reading ttyACMx, please kill that process and try '
                        'again.')
 
-    print 'Pursing results...'
+      print('Pursing results...')
     th_results, dut_results = self.evaluate_run(dut_output, th_output)
 
     # Print out results
@@ -372,7 +375,7 @@ def run(self):
     with open(dest, 'w') as fl:
       fl.write(dut_output)
 
-    print self.formatted_results
+    print(self.formatted_results)
 
     # TODO(chromium:735652): Should set exit code for the shell