Gmail open conversation test automation

build/android/gmail_test.py

+#!/usr/bin/env python
+#
+# Copyright 2017 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import os
+import sys
+import BaseHTTPServer
+import subprocess
+import argparse
+import datetime
+import signal
+import psutil
+
+import time
+
+import devil_chromium
+from devil.android import device_utils
+from devil.utils import run_tests_helper
+
+from multiprocessing import Process,Queue
+from pylib.constants import host_paths
+
+START_TRACING_CMD = "third_party/catapult/systrace/systrace/"
+START_TRACING_ARGS = ['-a', 'com.google.android.gm', 'webview']
+
+def RunTests(device, test_filter=None, title=None):
+  target = ('org.chromium.webview_ui_test.test/android.support.test.runner.'
+            'AndroidJUnitRunner')
+  extras = {}
+  if test_filter is not None:
+    extras.update({'class': test_filter})
+  if title is not None:
+    extras.update({'title': title})
+  device.StartInstrumentation(target, extras=extras, timeout=300, retries=0)
+
+def ParseArgs(argv):
+  timestring = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
+  parser = argparse.ArgumentParser()
+  parser.add_argument('-t', dest='trace_output', required=False,
+                      default='gmail_logs/%s/trace-'+timestring+'-',
+                      help='Trace output file')
+  parser.add_argument('-l', dest='logcat_output', required=False,
+                      default='gmail_logs/%s/logcat-'+timestring+'.log',
+                      help='Logcat output file')
+  parser.add_argument('-s', dest='serial', required=True, help='Device serial')
+  parser.add_argument('-f', dest='test_filter', help='Test filter')
+  parser.add_argument('-e', dest='email_title', default='Long Text',
+                      help='Email titles')
+  return parser.parse_args(argv[1:])
+
+def GetDevice(serial):
+  devices = device_utils.DeviceUtils.HealthyDevices()
+  for d in devices:
+    if d.serial == serial:
+      return d
+  raise Exception('No pixel device found')
+
+def SetUpDeviceAndGetLogcat(device):
+  device.ForceStop('com.google.android.gm')
+  device.SetProp('log.tag.GmailAnalytics', 'VERBOSE')
+
+def StartServer(port, q, trace_output, pipe, serial):
+  server_address = ('', port)
+  handler_class = GmailTracingRequestHandler
+  handler_class.protocol_version = 'HTTP/1.0'
+  httpd = GmailTracingServer(
+      server_address, handler_class, q, trace_output, pipe, serial)
+  sa = httpd.socket.getsockname()
+  print "Serving One HTTP on", sa[0], "port", sa[1], "..."
+  httpd.serve_forever()
+
+def StartServerWithProcess(port, trace_output, serial, pipe):
+  q = Queue()
+  p = Process(target=StartServer, args=(port, q, trace_output, pipe, serial))
+  p.start()
+  return p
+
+def WaitAndKill(pid):
+  time.sleep(400)
+  psutil.Process(pid).terminate()
+
+def StartTrace(pipe, output_file, serial):
+  cmd_path = os.path.abspath(os.path.join(
+      host_paths.DIR_SOURCE_ROOT, 'third_party', 'catapult', 'systrace',
+      'systrace', 'run_systrace.py'))
+  full_cmd = [cmd_path] + ['-o', output_file, '-e', serial] + START_TRACING_ARGS
+  print(full_cmd)
+  tp = subprocess.Popen(full_cmd, stdin=pipe[0], stdout=None, stderr=None)
+  return tp
+
+class GmailTracingRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+  def do_GET(self):
+    queue = self.server.queue
+    if 'start-tracing' in self.path:
+      if (queue.empty()):
+        name = self.path.split('/')[-1]
+        tracing_process = StartTrace(
+            self.server.pipe, self.server.trace_output + name + '.html',
+            self.server.serial)
+        queue.put(tracing_process)
+        self.send_response(200, 'started to trace')
+      else:
+        self.send_error(500, 'tracing process already exist')
+    elif 'end-tracing' in self.path:
+      if (queue.empty()):
+        self.send_error(
+            500, 'tracing process does not exist yet, please request '
+            '/start-tracing first')
+      else:
+        os.write(self.server.pipe[1], '\n')
+        self.send_response(200, 'ended to trace')
+        tracing_process = queue.get()
+        tracing_process.wait()
+    else:
+      self.send_error(404, 'Only allow /start-tracing or /end-tracing')
+
+def main():
+  run_tests_helper.SetLogLevel(2)
+  arguments = ParseArgs(sys.argv)
+  email_title_no_space = arguments.email_title.replace(' ', '')
+  output_dir = os.path.abspath(
+      os.path.join(
+          host_paths.DIR_SOURCE_ROOT, 'gmail_logs', email_title_no_space))
+  if not os.path.exists(os.path.abspath(output_dir)):
+    os.mkdir(output_dir)
+  device = GetDevice(arguments.serial)
+  device.adb.Reverse('tcp:7777', 'tcp:7777', allow_rebind=True)
+  # logcat_output = os.path.abspath(os.path.join(
+  # host_paths.DIR_SOURCE_ROOT,
+  # arguments.logcat_output % email_title_no_space))
+  # logcat_monitor = None
+  p = None
+  kp = None
+  pipe = os.pipe()
+  try:
+    SetUpDeviceAndGetLogcat(device)
+    # logcat_monitor.Start()
+    p = StartServerWithProcess(
+        7777, arguments.trace_output % email_title_no_space,
+        arguments.serial, pipe)
+    RunTests(device, arguments.test_filter, arguments.email_title)
+  finally:
+    # if logcat_monitor is not None:
+      # logcat_monitor.Stop()
+      # logcat_monitor.Close()
+    time.sleep(5)
+    if pipe is not None:
+      os.write(pipe[1], '\n')
+    if kp is not None:
+      kp.terminate()
+    if p is not None:
+      p.terminate()
+
+class GmailTracingServer(BaseHTTPServer.HTTPServer):
+  def __init__(self, server_address, RequestHandlerClass, queue,
+               trace_output, pipe, serial):
+    BaseHTTPServer.HTTPServer.__init__(
+        self, server_address, RequestHandlerClass)
+    self._queue = queue
+    self._trace_output = trace_output
+    self._pipe = pipe
+    self._serial = serial
+
+  @property
+  def serial(self):
+    return self._serial
+
+  @property
+  def pipe(self):
+    return self._pipe
+
+  @property
+  def queue(self):
+    return self._queue
+
+  @property
+  def trace_output(self):
+    return self._trace_output
+
+if __name__ == '__main__':
+  sys.exit(main())