# Copyright (c) 2010 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.
'''Utilities to automate running tests under app verifier.'''

import os
import os.path
import re
import subprocess
import sys
import tempfile
import win32com.client.gencache as gencache
import xml.sax
import xml.sax.handler


TRACE_RE_ = re.compile('(?P<module>\w+)'  # Module
    '(?:'
      '!'  # Separator
      '(?P<symbol>[^\+]+)'  # Symbol
      '(?:\+(?P<offset>[0-9a-f]+))?'  # Optional hex offset
    ')?'  # Sometimes we only get a module
    '(?:\s+\('  # Optional file and line
      '(?P<file>[^@]*)'  # file
      '\s+@\s*'
      '(?P<line>\d+)'  # line
    '\))?'
    , re.I)


# This is application verifier's automation library id.
LIBID_AppVerifier = '{DAB52BCB-6990-464A-AC61-F60C8EF60E24}'
try:
  VerifierLib = gencache.EnsureModule(LIBID_AppVerifier, 0, 1, 0)
except:
  VerifierLib = None


class LogTrace:
  def __init__(self, text):
    d = TRACE_RE_.match(text).groupdict()
    self.module = d['module']
    self.symbol = d['symbol']
    self.offset = d['offset']
    self.file = d['file']
    self.line = d['line']

  def __str__(self):
    ret = ''
    if self.file and self.line:
      ret = '%s (%s): AppVerifier warning C1: ' % (self.file, self.line)

    ret = '%s%s!%s' % (ret, self.module, self.symbol)

    if self.offset:
      ret = "%s+%s" % (ret, self.offset)

    return ret


class LogEntry:
  def __init__(self, stopcode, layer, severity):
    self.stopcode = stopcode
    self.layer = layer
    self.severity = severity
    self.message = ''
    self.trace = []

  def __str__(self):
    return "%s: %s\n\t%s" % (self.severity,
                             self.message,
                             '\n\t'.join(map(str,self.trace)))


class VerifierSaxHandler(xml.sax.handler.ContentHandler):
  '''Sax event handler to parse the verifier log file.

  The XML generated by verifier appears not to be standards compliant, so
  e.g. xml.dom.minidom falls down while parsing it, hence this tedious
  implementation.
  '''
  def __init__(self):
    self.log_entry = None
    self.stack = []
    self.element = None
    self.errors = 0
    self.text = ''
    self.log_entries = []

  def startElement(self, name, attrs):
    self.stack.append((self.element, self.text))
    self.element = name
    self.text = []

    if name == 'avrf:logEntry':
      self.log_entry = LogEntry(attrs.getValue('StopCode'),
                               attrs.getValue('LayerName'),
                               attrs.getValue('Severity'))
      self.errors += 1

  def endElement(self, name):
    if name == 'avrf:logEntry':
      self.log_entries.append(self.log_entry)
      self.log_entry = None
    elif name == 'avrf:message':
      self.log_entry.message = ''.join(self.text)
    elif name == 'avrf:trace':
      self.log_entry.trace.append(LogTrace(''.join(self.text)))

    (self.element, self.text) = self.stack.pop()

  def characters(self, content):
    self.text.append(content)


class AppverifierTestRunner:
  '''Encapsulates logic to run an executable under app verifier.

  Interacts with application verifiers automation library to set default
  verifier settings for a given target, process logs etc.
  '''
  # Checks we want enabled.
  default_checks_ = [
      # Basics group
      "Core",
      "Exceptions",
      "Handles",
      "Heaps",
      "InputOutput",
      "Locks",
      "Memory",
      "TLS",

      # Misc group
      "DangerousAPIs",
      "DirtyStacks",
      "TimeRollOver",
      ]

  def __init__(self, break_on_stop):
    self.manager = VerifierLib.AppVerifierManager()
    self.break_on_stop = break_on_stop

  def SetStopBreaks(self, check):
    '''Configures all the stops for check to log only.

      Arguments:
        check: an app verifier check
    '''
    error_reporting = VerifierLib.constants.VerifierErrorReportingNoBreak
    error_flags = (VerifierLib.constants.VerifierErrorFlagLogToFile |
                   VerifierLib.constants.VerifierErrorFlagLogStackTrace)
    if self.break_on_stop:
      error_reporting = VerifierLib.constants.VerifierErrorReportingBreakpoint

    try:
      for stop in check.Stops:
        stop.ErrorReporting = error_reporting
        stop.ErrorFlags = error_flags
    except:
      # Accessing or enumerating Stops fails for some checks.
      print 'Exception setting options for check', check.Name

  def ResetImage(self, image_name):
    '''Removes all verifier settings for image_name.
    '''
    # Reset the settings for our image
    try:
      self.manager.Images.Remove(image_name)
    except:
      # this fails if verifier had no settings for the image
      pass

  def SetImageDefaults(self, image_name):
    '''Configures a default set of tests for image_name

      Arguments:
        image_name: the basename of a test, e.g. 'common_unittest.exe'
    '''
    self.ResetImage(image_name)

    image = self.manager.Images.Add(image_name)
    for check in image.Checks:
      if check.Name in self.default_checks_:
        check.Enabled = True
        self.SetStopBreaks(check)

  def ClearImageLogs(self, image_name):
    '''Deletes all app verifier logs for image_name.'''
    logs = self.manager.Logs(image_name)
    if logs:
      while logs.Count:
        logs.Remove(0)

  def ProcessLog(self, log_path):
    '''Process the verifier log at log_path.

      Returns: a list of the log entries in the log.
    '''
    handler = VerifierSaxHandler()
    doc = xml.sax.parse(open(log_path, 'rb'), handler)

    return handler.log_entries

  def SaveLog(self, log):
    '''Saves log to an XML file and returns the path to the file.
    '''
    (fd, path) = tempfile.mkstemp('.xml', 'verifier_log')
    os.close(fd)
    try:
      log.SaveAsXML(path, '')
    except:
      os.remove(path)
      return None

    return path

  def ProcessLogs(self, test_name):
    '''Processes all logs for image test_name.

      Arguments:
        test_name: the base name of the test executable.

      Returns: A list of LogEntry instances for each error logged.
    '''
    logs = self.manager.Logs(test_name)
    if not logs or not logs.Count:
      return 0

    errors = []
    for log in logs:
      path = self.SaveLog(log)
      if path:
        errors.extend(self.ProcessLog(path))
        os.remove(path)

    return errors

  def RunTestWithVerifier(self, test_path):
    '''Run a single test under verifier.

      Arguments:
        test_path: full or relative path to the test to run.

      Returns:
        A tuple with the test exit code and a list of verifier errors,
        example:
          (0, [error1, error2, ...])
    '''
    test_name = os.path.basename(test_path)

    # Set up the verifier configuration
    self.SetImageDefaults(test_name)
    self.ClearImageLogs(test_name)

    # run the test.
    exit = subprocess.call(test_path)

    # Clear the verifier settings for the image.
    self.ResetImage(test_name)

    # And process the logs.
    errors = self.ProcessLogs(test_name)
    return (exit, errors)

  def DumpImages_(self):
    for image in self.manager.Images:
      print image.Name
      for check in image.Checks:
        print '\t', check.Name, check.Enabled

  def DumpLogs_(self, image_name):
    for l in self.manager.Logs(image_name):
      print l


def RunTestWithVerifier(test_path, break_on_stop = False):
  '''Runs test_path under app verifier.

    Returns: a tuple of the exit code and list of errors
  '''
  runner = AppverifierTestRunner(break_on_stop)
  return runner.RunTestWithVerifier(test_path)


def HasAppVerifier():
  '''Returns true iff application verifier is installed.'''
  if VerifierLib:
    return True

  return False


def Main():
  '''Runs all tests on command line under verifier with stops disabled.'''
  for test in sys.argv[1:]:
    (exit_code, errors) = RunTestWithVerifier(test)
    for error in errors:
      print error


if __name__ == '__main__':
  Main()