import re
from os.path import basename

from twisted.python import log

from buildbot.process.results import SUCCESS
from buildbot.process.results import FAILURE

from buildbot.steps.shell import Test

from buildbot.process.logobserver import LogLineObserver

class LitLogObserver(LogLineObserver):
  # Regular expressions for a regular test line.
  kTestLineRE = re.compile(r'(\w+): (.*) \(.*\)')

  # Regular expressions for verbose log start and stop markers. Verbose log
  # output always immediately follow a test.
  kTestVerboseLogStartRE = re.compile(r"""\*{4,80} TEST '(.*)' .*""")
  kTestVerboseLogStopRE = re.compile(r"""\*{10,80}""")

  # These are the codes for which we will include the log output in the buildbot
  # step results.
  failingCodes = set(['FAIL', 'XPASS', 'KPASS', 'UNRESOLVED', 'TIMEOUT', 'ERROR'])
  # Regular expressions for start of summary marker.
  kStartSummaryRE = re.compile(r'^Failing Tests \(\d*\)$')

  def __init__(self, maxLogs=None, parseSummaryOnly=False):
    log.msg(">>> LitLogObserver._init__(maxLogs=%s, parseSummaryOnly=%s)" % (maxLogs, parseSummaryOnly))
    super().__init__()
    self.resultCounts = {}
    self.maxLogs = maxLogs
    self.numLogs = 0

    # If non-null, a tuple of the last test code and name.
    self.lastTestResult = None

    # If non-null, a list of lines in the current log.
    self.activeVerboseLog = None
    # Current line will be parsed as result steps only if parserStarted is True
    self.parserStarted = not parseSummaryOnly
    self.simplifiedLog = False

  def hadFailure(self):
    for code in self.failingCodes:
      if self.resultCounts.get(code):
        return True

  def handleVerboseLogLine(self, line):
    # Append to the log.
    self.activeVerboseLog.append(line)

    # If this is a stop marker, process the test info.
    if self.kTestVerboseLogStopRE.match(line.strip()):
      self.testInfoFinished()

  def testInfoFinished(self):
    # We have finished getting information for one test, handle it.
    if self.lastTestResult:
        code, name = self.lastTestResult

        # If the test failed, add a log entry for it (unless we have reached the
        # max).
        if code in self.failingCodes and (self.maxLogs is None or
                                          self.numLogs < self.maxLogs):
          # If a verbose log was not provided, just add a one line description.
          if self.activeVerboseLog is None:
            self.activeVerboseLog = ['%s: %s' % (code, name)]

          # Add the log to the build status.
          # Make the test name short, the qualified test name is in the log anyway.
          # Otherwise, we run out of the allowed name length on some hosts.
          name_part = name.rpartition('::')
          self.step.addCompleteLog(
                      code + ': ' + name_part[0].strip() + name_part[1] + basename(name_part[2]),
                      '\n'.join(self.activeVerboseLog))
          self.numLogs += 1
    else:
        if self.activeVerboseLog:
            self.activeVerboseLog.append(
              "error: missing test status line, skipping log")

    # Reset the current state.
    self.lastTestResult = None
    self.activeVerboseLog = None

  def handleSimplifiedLogLine(self, line):
    # Check for test status line
    m = self.kTestLineRE.match(line.strip())
    if m:
      # Remember the last test result and update the result counts.
      self.lastTestResult = (code, name) = m.groups()
      self.resultCounts[code] = self.resultCounts.get(code, 0) + 1
      self.testInfoFinished()
    return

  def outLineReceived(self, line):
    # Assert - Lines after "Failing Test (\d)" will be summary line and will not contain verbose message
    if self.simplifiedLog is True:
      self.handleSimplifiedLogLine(line)
      return
    # If we are inside a verbose log, just accumulate lines until we reach the
    # stop marker.
    if self.activeVerboseLog is not None:
      self.handleVerboseLogLine(line)
      return

    # Check for the test verbose log start marker.
    m = self.kTestVerboseLogStartRE.match(line.strip())
    if m:
      self.activeVerboseLog = [line]
      if self.lastTestResult is None:
        if self.activeVerboseLog:
            self.activeVerboseLog.append(
              "error: missing test line before verbose log start.")
      elif m.group(1) != self.lastTestResult[1]:
        # This is bogus, the verbose log test name doesn't match what we
        # expect. Just note it in the log but otherwise accumulate as normal.
        self.activeVerboseLog.append(
          "error: verbose log output name didn't match expected test name")
      return

    # Otherwise, if we had any previous test consider it finished.
    #
    # FIXME: This assumes that we will always have at least one line following
    # the last test result to properly record each test; we could fix this if
    # buildbot provided us a hook for when the log is done.
    if self.lastTestResult:
      self.testInfoFinished()

    if self.kStartSummaryRE.match(line):
      self.parserStarted = True;
      self.simplifiedLog = True;

    #Assign result line only if summary marker has been matched
    #Or if all lines should be parsed
    if self.parserStarted is True:
      # Check for a new test status line.
      m = self.kTestLineRE.match(line.strip())
      if m:
        # Remember the last test result and update the result counts.
        self.lastTestResult = (code, name) = m.groups()
        self.resultCounts[code] = self.resultCounts.get(code, 0) + 1
        return

class LitTestCommand(Test):
  resultNames = {'FAIL':'unexpected failures',
                 'PASS':'expected passes',
                 'XFAIL':'expected failures',
                 'XPASS':'unexpected passes',
                 'KFAIL':'known failures',
                 'KPASS':'unknown passes',
                 'UNRESOLVED':'unresolved testcases',
                 'UNTESTED':'untested testcases',
                 'REGRESSED':'runtime performance regression',
                 'IMPROVED':'runtime performance improvement',
                 'UNSUPPORTED':'unsupported tests',
                 'TIMEOUT':'timeout waiting for results'}

  def __init__(self, ignore=[], flaky=[], max_logs=20, parseSummaryOnly=False,
               *args, **kwargs):
    log.msg(">>> LitTestCommand._init__(ignore=%s,flaky=%s,max_logs=%s,parseSummaryOnly=%s,kwargs=%s)" % (ignore, flaky, max_logs, parseSummaryOnly, kwargs))
    super().__init__(*args, **kwargs)
    self.maxLogs = int(max_logs)
    self.logObserver = LitLogObserver(self.maxLogs, parseSummaryOnly)
    self.addFactoryArguments(max_logs=max_logs)
    self.addFactoryArguments(parseSummaryOnly=parseSummaryOnly)
    self.addLogObserver('stdio', self.logObserver)

  def evaluateCommand(self, cmd):
    # Always report failure if the command itself failed.
    if cmd.rc != 0:
      return FAILURE

    # Otherwise, report failure if there were failures in the log.
    if self.logObserver.hadFailure():
      return FAILURE

    return SUCCESS

  def describe(self, done=False):
    description = Test.describe(self, done)
    for name, count in self.logObserver.resultCounts.items():
        if name in self.resultNames:
            description.append('{0} {1}'.format(count, self.resultNames[name]))
        else:
            description.append('Unexpected test result output ' + name)
    return description

## TODO: Move these unit tests to the untit tests suite.
##

import unittest

class StepProxy(object):
  def __init__(self):
    self.logs = []

  def addCompleteLog(self, name, text):
    self.logs.append((name, text))

class RemoteCommandProxy(object):
  def __init__(self, rc):
    self.rc = rc

class TestLogObserver(unittest.TestCase):
  def parse_log(self, text):
    observer = LitLogObserver()
    observer.step = StepProxy()
    for ln in text.split('\n'):
      observer.outLineReceived(ln)
    return observer

  def test_basic(self):
    obs = self.parse_log("""
PASS: test-one (1 of 4)
FAIL: test-two (2 of 4)
PASS: test-three (3 of 4)
TIMEOUT: test-four (4 of 4)
""")

    self.assertEqual(obs.resultCounts, { 'FAIL' : 1, 'TIMEOUT' : 1, 'PASS' : 2 })
    self.assertEqual(obs.step.logs, [('FAIL: test-two', 'FAIL: test-two'), ('TIMEOUT: test-four', 'TIMEOUT: test-four')])

  def test_missing_lastresult(self):
    obs = self.parse_log("""
FAIL: LLDB :: 10-breakpoint-by-address-multiple-hit-first-add-next
********** TEST 'LLDB :: 10-breakpoint-by-address-multiple-hit-first-add-next' FAIL **********
bla bla bla
**********
         TIMEOUT: LLDB :: 57-dlopen-breakpoint (51 of 57)
         ********** TEST 'LLDB :: 57-dlopen-breakpoint' TIMEOUT **********
         bla
         **********
""")

    self.assertEqual(obs.resultCounts, {'TIMEOUT': 1})

  def test_verbose_logs(self):
    obs = self.parse_log("""
FAIL: test-one (1 of 3)
FAIL: test-two (2 of 3)
**** TEST 'test-two' FAILED ****
bla bla bla
**********
FAIL: test-three (3 of 3)
""")

    self.assertEqual(obs.resultCounts, { 'FAIL' : 3 })
    self.assertEqual(obs.step.logs, [
        ('FAIL: test-one', 'FAIL: test-one'),
        ('FAIL: test-two', """\
**** TEST 'test-two' FAILED ****
bla bla bla
**********"""),
        ('FAIL: test-three', 'FAIL: test-three')])

  def test_indented_log(self):
    obs = self.parse_log("""
    FAIL: test-one (1 of 3)
    FAIL: test-two (2 of 3)
    **** TEST 'test-two' FAILED ****
    bla bla bla
    **********
    FAIL: test-three (3 of 3)
""")

    self.assertEqual(obs.resultCounts, { 'FAIL' : 3 })
    self.assertEqual(obs.step.logs, [
        ('FAIL: test-one', 'FAIL: test-one'),
        ('FAIL: test-two', """\
    **** TEST 'test-two' FAILED ****
    bla bla bla
    **********"""),
        ('FAIL: test-three', 'FAIL: test-three')])

class TestCommand(unittest.TestCase):
  def parse_log(self, text, **kwargs):
    cmd = LitTestCommand(**kwargs)
    cmd.logObserver.step = StepProxy()
    for ln in text.split('\n'):
      cmd.logObserver.outLineReceived(ln)
    return cmd

  def test_command_status(self):
    # If the command failed, the status should always be error.
    cmd = self.parse_log("")
    self.assertEqual(cmd.evaluateCommand(RemoteCommandProxy(1)), FAILURE)

    # If there were failing tests, the status should be an error (even if the
    # test command didn't report as such).
    for failing_code in set(['FAIL', 'XPASS', 'KPASS', 'UNRESOLVED', 'TIMEOUT']):
      cmd = self.parse_log("""%s: test-one (1 of 1)""" % (failing_code,))
      self.assertEqual(cmd.evaluateCommand(RemoteCommandProxy(0)), FAILURE)

  def test_max_logs(self):
    cmd = self.parse_log("""
FAIL: test-one (1 of 2)
FAIL: test-two (2 of 2)
""", max_logs=1)
    self.assertEqual(cmd.logObserver.step.logs, [('FAIL: test-one', 'FAIL: test-one')])

if __name__ == '__main__':
  unittest.main()