[Password manager tests] Write to spreadsheets.

components/test/data/password_manager/automated_tests/run_tests.py

 # -*- coding: utf-8 -*-
 # Copyright 2014 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.
 
-"""This file allows the bots to be easily configure and run the tests."""
+"""This file allows the bots to be easily configure and run the tests.
 
+Running this script requires passing --config-path with a path to a config file
+of the following structure:
+  [credentials]
+  pkey=full_path
+  client_email=email_assigned_by_google_dev_console
+  [drive]
+  key=sheet_key_from_sheet_url
+  [data_files]
+  passwords_path=full_path_to_the_file_with_passwords
+"""
+from Sheet import Sheet
+from apiclient.discovery import build
+from datetime import datetime
+from gdata.gauth import OAuth2TokenFromCredentials
+from gdata.spreadsheet.service import SpreadsheetsService
+from oauth2client.client import SignedJwtAssertionCredentials
+import ConfigParser
 import argparse
+import httplib2
+import oauth2client.tools
 import os
 import tempfile
 
 from environment import Environment
 import tests
 
+_CREDENTIAL_SCOPES = "https://spreadsheets.google.com/feeds"
+
+# TODO(melandory): Function _authenticate belongs to separate module.
+def _authenticate(pkey, client_email):
+""" Authenticates user.
+
+  Args:
+    pkey: Full path to file with private key generated by Google
+      Developer Console.
+    client_email: Email address corresponding to private key and also
+      generated by Google Developer Console.
+"""
+  http, token = None, None
+  with open(pkey) as pkey_file:
+    private_key = pkey_file.read()
+    credentials = SignedJwtAssertionCredentials(
+        client_email, private_key, _CREDENTIAL_SCOPES)
+    http = httplib2.Http()
+    http = credentials.authorize(http)
+    build('drive', 'v2', http=http)
+    token = OAuth2TokenFromCredentials(credentials).access_token
+  return http, token
+
+# TODO(melandory): Functionality of _spredsheeet_for_logging belongs
+# to websitetests, because this way we do not need to write results of run
+# in separate file and then read it here.
+def _spredsheeet_for_logging(sheet_key, access_token):
+  """ Connects to document where result of test run will be logged.
+    Args:
+      sheet_key: Key of sheet in Trix. Can be found in sheet's URL.
+      access_token: Access token of an account which should have edit rights.
+  """
+  # Connect to trix
+  service = SpreadsheetsService(additional_headers={
+            "Authorization": "Bearer " + access_token})
+  sheet = Sheet(service, sheet_key)
+  return sheet
+
+def _try_run_individual_test(test_cmd, results_path):
+  """ Runs individual test and logs results to trix.
+
+    Args:
+      test_cmd: String contains command which runs test.
+      results_path: Full path to file where results of test run will be logged.
+  """
+  failures = []
+  # The tests can be flaky. This is why we try to rerun up to 3 times.
+  for x in xrange(3):
+    # TODO(rchtara): Using "pkill" is just temporary until a better,
+    # platform-independent solution is found.
+    # Using any kind of kill and process name isn't best solution.
+    # Mainly because it kills every running instace of Chrome on the machine,
+    # which is unpleasant experience, when you're running tests locally.
+    # In my opinion proper solution is to invoke chrome using subproceses and
+    # then kill only this particular instance of it.
+    os.system("pkill chrome")
+    try:
+      os.remove(results_path)
+    except Exception:
+      pass
+    # TODO(rchtara): Using "timeout is just temporary until a better,
+    # platform-independent solution is found.
+
+    # The website test runs in two passes, each pass has an internal
+    # timeout of 200s for waiting (see |remaining_time_to_wait| and
+    # Wait() in websitetest.py). Accounting for some more time spent on
+    # the non-waiting execution, 300 seconds should be the upper bound on
+    # the runtime of one pass, thus 600 seconds for the whole test.
+    # TODO(vabr): Use subprocess.call.
+    os.system("timeout 600 %s" % test_cmd)
+    if os.path.isfile(results_path):
+      results = open(results_path, "r")
+      count = 0 # Count the number of successful tests.
+      for line in results:
+        # TODO(melandory): We do not need to send all this data to sheet.
+        failures.append(line)
+        count += line.count("successful='True'")
+      results.close()
+      # There is only two tests running for every website: the prompt and
+      # the normal test. If both of the tests were successful, the tests
+      # would be stopped for the current website.
+      if count == 2:
+        return "pass", []
+    else:
+      pass
+  return "fail", failures
+
+
 def run_tests(
-    save_path, chrome_path, chromedriver_path,
-    passwords_path, profile_path, *args, **kwargs):
+    chrome_path, chromedriver_path,
+    profile_path, config_path, *args, **kwargs):
   """ Runs automated tests.
 
   Args:
    save_path: File, where results of run will be logged.
    chrome_path: Location of Chrome binary.
    chromedriver_path: Location of Chrome Driver.
    passwords_path: Location of file with credentials.
    profile_path: Location of profile path.
    *args: Variable length argument list.
    **kwargs: Arbitrary keyword arguments.
   """
   environment = Environment('', '', '', None, False)
   tests.Tests(environment)
-
-  xml = open(save_path, "w")
-  xml.write("<xml>")
+  config = ConfigParser.ConfigParser()
+  config.read(config_path)
+  date = datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
   try:
+    _, access_token = _authenticate(config.get("credentials", "pkey"),
+                                 config.get("credentials", "client_email"))
+    sheet = _spredsheeet_for_logging(config.get("drive", "key"), access_token)
     results = tempfile.NamedTemporaryFile(
         dir=os.path.join(tempfile.gettempdir()), delete=False)
     results_path = results.name
     results.close()
 
     full_path = os.path.realpath(__file__)
     tests_dir = os.path.dirname(full_path)
     tests_path = os.path.join(tests_dir, "tests.py")
 
     for websitetest in environment.websitetests:
-      # The tests can be flaky. This is why we try to rerun up to 3 times.
-      for x in range(0, 3):
-        # TODO(rchtara): Using "pkill" is just temporary until a better,
-        # platform-independent solution is found.
-        os.system("pkill chrome")
-        try:
-          os.remove(results_path)
-        except Exception:
-          pass
-        # TODO(rchtara): Using "timeout is just temporary until a better,
-        # platform-independent solution is found.
-
-        # The website test runs in two passes, each pass has an internal
-        # timeout of 200s for waiting (see |remaining_time_to_wait| and
-        # Wait() in websitetest.py). Accounting for some more time spent on
-        # the non-waiting execution, 300 seconds should be the upper bound on
-        # the runtime of one pass, thus 600 seconds for the whole test.
-        os.system("timeout 600 python %s %s --chrome-path %s "
-            "--chromedriver-path %s --passwords-path %s --profile-path %s "
-            "--save-path %s" %
-            (tests_path, websitetest.name, chrome_path,
-             chromedriver_path, passwords_path,
-             profile_path, results_path))
-        if os.path.isfile(results_path):
-          results = open(results_path, "r")
-          count = 0 # Count the number of successful tests.
-          for line in results:
-            xml.write(line)
-            count += line.count("successful='True'")
-          results.close()
-          # There is only two tests running for every website: the prompt and
-          # the normal test. If both of the tests were successful, the tests
-          # would be stopped for the current website.
-          if count == 2:
-            break
-        else:
-          xml.write("<result><test name='%s' type='prompt' successful='false'>"
-              "</test><test name='%s' type='normal' successful='false'></test>"
-              "</result>" % (websitetest.name, websitetest.name))
+      test_cmd = ("python %s %s --chrome-path %s "
+                  "--chromedriver-path %s"
+                  "--passwords-path %s --profile-path %s "
+                  "--save-path %s" %
+                  (tests_path, websitetest.name, chrome_path,
+                   chromedriver_path,
+                   config.get("data_files", "passwords_path"),
+                   profile_path, results_path))
+      status, log = _try_run_individual_test(
+          websitetest, test_cmd, results_path)
+      try:
+        sheet.InsertRow(sheet.row_count,
+                        [websitetest.name, status, date, " | ".join(log)])
+      except Exception:
+        pass # TODO(melandory): Sometimes writing to spreadsheet fails. We need
+             # deal with it better that just ignoring.
   finally:
     try:
       os.remove(results_path)
     except Exception:
       pass
 
-  xml.write("</xml>")
-  xml.close()
 
 if __name__ == "__main__":
   parser = argparse.ArgumentParser(
       description="Password Manager automated tests runner help.")
   parser.add_argument(
       "--chrome-path", action="store", dest="chrome_path",
       help="Set the chrome path (required).", required=True)
   parser.add_argument(
       "--chromedriver-path", action="store", dest="chromedriver_path",
       help="Set the chromedriver path (required).", required=True)
   parser.add_argument(
+      "--config-path", action="store", dest="config_path",
+      help="File with configuration data: drive credentials, password path",
+      required=True)
+  parser.add_argument(
       "--profile-path", action="store", dest="profile_path",
       help="Set the profile path (required). You just need to choose a "
            "temporary empty folder. If the folder is not empty all its content "
            "is going to be removed.",
       required=True)
-  parser.add_argument(
-      "--passwords-path", action="store", dest="passwords_path",
-      help="Set the usernames/passwords path (required).", required=True)
-  parser.add_argument("--save-path", action="store",  dest="save_path",
-      help="Write the results in a file.", required=True)
   args = vars(parser.parse_args())
   run_tests(**args)