#!/usr/bin/env python3
# Copyright (C) 2023 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import concurrent.futures
import datetime
import difflib
import os
import subprocess
import sys
import tempfile
from binascii import unhexlify
from dataclasses import dataclass
from typing import List, Tuple, Optional

from google.protobuf import text_format, message_factory, descriptor_pool
from python.generators.diff_tests.testing import TestCase, TestType, BinaryProto
from python.generators.diff_tests.utils import (
    ColorFormatter, create_message_factory, get_env, get_trace_descriptor_path,
    read_all_tests, serialize_python_trace, serialize_textproto_trace)

ROOT_DIR = os.path.dirname(
    os.path.dirname(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))


# Performance result of running the test.
@dataclass
class PerfResult:
  test: TestCase
  ingest_time_ns: int
  real_time_ns: int

  def __init__(self, test: TestCase, perf_lines: List[str]):
    self.test = test

    assert len(perf_lines) == 1
    perf_numbers = perf_lines[0].split(',')

    assert len(perf_numbers) == 2
    self.ingest_time_ns = int(perf_numbers[0])
    self.real_time_ns = int(perf_numbers[1])


# Data gathered from running the test.
@dataclass
class TestResult:
  test: TestCase
  trace: str
  cmd: List[str]
  expected: str
  actual: str
  passed: bool
  stderr: str
  exit_code: int
  perf_result: Optional[PerfResult]

  def __init__(self, test: TestCase, gen_trace_path: str, cmd: List[str],
               expected_text: str, actual_text: str, stderr: str,
               exit_code: int, perf_lines: List[str]) -> None:
    self.test = test
    self.trace = gen_trace_path
    self.cmd = cmd
    self.stderr = stderr
    self.exit_code = exit_code

    # For better string formatting we often add whitespaces, which has to now
    # be removed.
    def strip_whitespaces(text: str):
      no_front_new_line_text = text.lstrip('\n')
      return '\n'.join(s.strip() for s in no_front_new_line_text.split('\n'))

    self.expected = strip_whitespaces(expected_text)
    self.actual = strip_whitespaces(actual_text)

    expected_content = self.expected.replace('\r\n', '\n')

    actual_content = self.actual.replace('\r\n', '\n')
    self.passed = (expected_content == actual_content)

    if self.exit_code == 0:
      self.perf_result = PerfResult(self.test, perf_lines)
    else:
      self.perf_result = None

  def write_diff(self):
    expected_lines = self.expected.splitlines(True)
    actual_lines = self.actual.splitlines(True)
    diff = difflib.unified_diff(
        expected_lines, actual_lines, fromfile='expected', tofile='actual')
    return "".join(list(diff))

  def rebase(self, rebase) -> str:
    if not rebase or self.passed:
      return ""
    if not self.test.blueprint.is_out_file():
      return f"Can't rebase expected results passed as strings.\n"
    if self.exit_code != 0:
      return f"Rebase failed for {self.test.name} as query failed\n"

    with open(self.test.expected_path, 'w') as f:
      f.write(self.actual)
    return f"Rebasing {self.test.name}\n"


# Results of running the test suite. Mostly used for printing aggregated
# results.
@dataclass
class TestResults:
  test_failures: List[str]
  perf_data: List[PerfResult]
  rebased: List[str]
  test_time_ms: int

  def str(self, no_colors: bool, tests_no: int):
    c = ColorFormatter(no_colors)
    res = (
        f"[==========] {tests_no} tests ran. ({self.test_time_ms} ms total)\n"
        f"{c.green('[  PASSED  ]')} "
        f"{tests_no - len(self.test_failures)} tests.\n")
    if len(self.test_failures) > 0:
      res += (f"{c.red('[  FAILED  ]')} " f"{len(self.test_failures)} tests.\n")
      for failure in self.test_failures:
        res += f"{c.red('[  FAILED  ]')} {failure}\n"
    return res

  def rebase_str(self):
    res = f"\n[  REBASED  ] {len(self.rebased)} tests.\n"
    for name in self.rebased:
      res += f"[  REBASED  ] {name}\n"
    return res


# Responsible for executing singular diff test.
@dataclass
class TestCaseRunner:
  test: TestCase
  trace_processor_path: str
  trace_descriptor_path: str
  colors: ColorFormatter

  def __output_to_text_proto(self, actual: str, out: BinaryProto) -> str:
    """Deserializes a binary proto and returns its text representation.

  Args:
    actual: (string) HEX encoded serialized proto message
    message_type: (string) Message type

  Returns:
    Text proto
  """
    try:
      raw_data = unhexlify(actual.splitlines()[-1][1:-1])
      out_path = os.path.dirname(self.trace_processor_path)
      descriptor_paths = [
          f.path
          for f in os.scandir(
              os.path.join(ROOT_DIR, out_path, 'gen', 'protos', 'perfetto',
                           'trace_processor'))
          if f.is_file() and os.path.splitext(f.name)[1] == '.descriptor'
      ]
      descriptor_paths.append(
          os.path.join(ROOT_DIR, out_path, 'gen', 'protos', 'third_party',
                       'pprof', 'profile.descriptor'))
      proto = create_message_factory(descriptor_paths, out.message_type)()
      proto.ParseFromString(raw_data)
      try:
        return out.post_processing(proto)
      except:
        return '<Proto post processing failed>'
      return text_format.MessageToString(proto)
    except:
      return '<Invalid input for proto deserializaiton>'

  def __run_metrics_test(self, trace_path: str,
                         metrics_message_factory) -> TestResult:

    if self.test.blueprint.is_out_file():
      with open(self.test.expected_path, 'r') as expected_file:
        expected = expected_file.read()
    else:
      expected = self.test.blueprint.out.contents

    tmp_perf_file = tempfile.NamedTemporaryFile(delete=False)
    is_json_output_file = self.test.blueprint.is_out_file(
    ) and os.path.basename(self.test.expected_path).endswith('.json.out')
    is_json_output = is_json_output_file or self.test.blueprint.is_out_json()
    cmd = [
        self.trace_processor_path,
        '--analyze-trace-proto-content',
        '--crop-track-events',
        '--run-metrics',
        self.test.blueprint.query.name,
        '--metrics-output=%s' % ('json' if is_json_output else 'binary'),
        '--perf-file',
        tmp_perf_file.name,
        trace_path,
    ]
    tp = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=get_env(ROOT_DIR))
    (stdout, stderr) = tp.communicate()

    if is_json_output:
      expected_text = expected
      actual_text = stdout.decode('utf8')
    else:
      # Expected will be in text proto format and we'll need to parse it to
      # a real proto.
      expected_message = metrics_message_factory()
      text_format.Merge(expected, expected_message)

      # Actual will be the raw bytes of the proto and we'll need to parse it
      # into a message.
      actual_message = metrics_message_factory()
      actual_message.ParseFromString(stdout)

      # Convert both back to text format.
      expected_text = text_format.MessageToString(expected_message)
      actual_text = text_format.MessageToString(actual_message)

    perf_lines = [line.decode('utf8') for line in tmp_perf_file.readlines()]
    tmp_perf_file.close()
    os.remove(tmp_perf_file.name)
    return TestResult(self.test, trace_path, cmd, expected_text, actual_text,
                      stderr.decode('utf8'), tp.returncode, perf_lines)

  # Run a query based Diff Test.
  def __run_query_test(self, trace_path: str) -> TestResult:
    # Fetch expected text.
    if self.test.expected_path:
      with open(self.test.expected_path, 'r') as expected_file:
        expected = expected_file.read()
    else:
      expected = self.test.blueprint.out.contents

    # Fetch query.
    if self.test.blueprint.is_query_file():
      query = self.test.query_path
    else:
      tmp_query_file = tempfile.NamedTemporaryFile(delete=False)
      with open(tmp_query_file.name, 'w') as query_file:
        query_file.write(self.test.blueprint.query)
      query = tmp_query_file.name

    tmp_perf_file = tempfile.NamedTemporaryFile(delete=False)
    cmd = [
        self.trace_processor_path,
        '--analyze-trace-proto-content',
        '--crop-track-events',
        '-q',
        query,
        '--perf-file',
        tmp_perf_file.name,
        trace_path,
    ]
    tp = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=get_env(ROOT_DIR))
    (stdout, stderr) = tp.communicate()

    if not self.test.blueprint.is_query_file():
      tmp_query_file.close()
      os.remove(tmp_query_file.name)
    perf_lines = [line.decode('utf8') for line in tmp_perf_file.readlines()]
    tmp_perf_file.close()
    os.remove(tmp_perf_file.name)

    actual = stdout.decode('utf8')
    if self.test.blueprint.is_out_binaryproto():
      actual = self.__output_to_text_proto(actual, self.test.blueprint.out)

    return TestResult(self.test, trace_path, cmd, expected, actual,
                      stderr.decode('utf8'), tp.returncode, perf_lines)

  def __run(self, metrics_descriptor_paths: List[str],
            extension_descriptor_paths: List[str], keep_input,
            rebase) -> Tuple[TestResult, str]:
    # We can't use delete=True here. When using that on Windows, the
    # resulting file is opened in exclusive mode (in turn that's a subtle
    # side-effect of the underlying CreateFile(FILE_ATTRIBUTE_TEMPORARY))
    # and TP fails to open the passed path.
    gen_trace_file = None
    if self.test.blueprint.is_trace_file():
      if self.test.trace_path.endswith('.py'):
        gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
        serialize_python_trace(ROOT_DIR, self.trace_descriptor_path,
                               self.test.trace_path, gen_trace_file)

      elif self.test.trace_path.endswith('.textproto'):
        gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
        serialize_textproto_trace(self.trace_descriptor_path,
                                  extension_descriptor_paths,
                                  self.test.trace_path, gen_trace_file)

    elif self.test.blueprint.is_trace_textproto():
      gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
      proto = create_message_factory([self.trace_descriptor_path] +
                                     extension_descriptor_paths,
                                     'perfetto.protos.Trace')()
      text_format.Merge(self.test.blueprint.trace.contents, proto)
      gen_trace_file.write(proto.SerializeToString())
      gen_trace_file.flush()

    else:
      gen_trace_file = tempfile.NamedTemporaryFile(delete=False)
      with open(gen_trace_file.name, 'w') as trace_file:
        trace_file.write(self.test.blueprint.trace.contents)

    if gen_trace_file:
      trace_path = os.path.realpath(gen_trace_file.name)
    else:
      trace_path = self.test.trace_path

    str = f"{self.colors.yellow('[ RUN      ]')} {self.test.name}\n"

    if self.test.type == TestType.QUERY:
      result = self.__run_query_test(trace_path)
    elif self.test.type == TestType.METRIC:
      result = self.__run_metrics_test(
          trace_path,
          create_message_factory(metrics_descriptor_paths,
                                 'perfetto.protos.TraceMetrics'))
    else:
      assert False

    if gen_trace_file:
      if keep_input:
        str += f"Saving generated input trace: {trace_path}\n"
      else:
        gen_trace_file.close()
        os.remove(trace_path)

    def write_cmdlines():
      res = ""
      if not gen_trace_file:
        res += 'Command to generate trace:\n'
        res += 'tools/serialize_test_trace.py '
        res += '--descriptor {} {} > {}\n'.format(
            os.path.relpath(self.trace_descriptor_path, ROOT_DIR),
            os.path.relpath(self.test.trace_path, ROOT_DIR),
            os.path.relpath(trace_path, ROOT_DIR))
      res += f"Command line:\n{' '.join(result.cmd)}\n"
      return res

    if result.exit_code != 0 or not result.passed:
      str += result.stderr

      if result.exit_code == 0:
        str += f"Expected did not match actual for test {self.test.name}.\n"
        str += write_cmdlines()
        str += result.write_diff()
      else:
        str += write_cmdlines()

      str += (f"{self.colors.red('[  FAILED  ]')} {self.test.name}\n")
      str += result.rebase(rebase)

      return result, str
    else:
      str += (f"{self.colors.green('[       OK ]')} {self.test.name} "
              f"(ingest: {result.perf_result.ingest_time_ns / 1000000:.2f} ms "
              f"query: {result.perf_result.real_time_ns / 1000000:.2f} ms)\n")
    return result, str

  # Run a TestCase.
  def execute(self, extension_descriptor_paths: List[str],
              metrics_descriptor: str, keep_input: bool,
              rebase: bool) -> Tuple[str, str, TestResult]:
    if metrics_descriptor:
      metrics_descriptor_paths = [metrics_descriptor]
    else:
      out_path = os.path.dirname(self.trace_processor_path)
      metrics_protos_path = os.path.join(out_path, 'gen', 'protos', 'perfetto',
                                         'metrics')
      metrics_descriptor_paths = [
          os.path.join(metrics_protos_path, 'metrics.descriptor'),
          os.path.join(metrics_protos_path, 'chrome',
                       'all_chrome_metrics.descriptor'),
          os.path.join(metrics_protos_path, 'webview',
                       'all_webview_metrics.descriptor')
      ]
    result_str = ""

    result, run_str = self.__run(metrics_descriptor_paths,
                                 extension_descriptor_paths, keep_input, rebase)
    result_str += run_str
    if not result:
      return self.test.name, result_str, None

    return self.test.name, result_str, result


# Fetches and executes all diff viable tests.
@dataclass
class DiffTestsRunner:
  tests: List[TestCase]
  trace_processor_path: str
  trace_descriptor_path: str
  test_runners: List[TestCaseRunner]

  def __init__(self, name_filter: str, trace_processor_path: str,
               trace_descriptor: str, no_colors: bool):
    self.tests = read_all_tests(name_filter, ROOT_DIR)
    self.trace_processor_path = trace_processor_path

    out_path = os.path.dirname(self.trace_processor_path)
    self.trace_descriptor_path = get_trace_descriptor_path(
        out_path, trace_descriptor)
    self.test_runners = []
    color_formatter = ColorFormatter(no_colors)
    for test in self.tests:
      self.test_runners.append(
          TestCaseRunner(test, self.trace_processor_path,
                         self.trace_descriptor_path, color_formatter))

  def run_all_tests(self, metrics_descriptor: str, keep_input: bool,
                    rebase: bool) -> TestResults:
    perf_results = []
    failures = []
    rebased = []
    test_run_start = datetime.datetime.now()

    out_path = os.path.dirname(self.trace_processor_path)
    chrome_extensions = os.path.join(out_path, 'gen', 'protos', 'third_party',
                                     'chromium',
                                     'chrome_track_event.descriptor')
    test_extensions = os.path.join(out_path, 'gen', 'protos', 'perfetto',
                                   'trace', 'test_extensions.descriptor')

    with concurrent.futures.ProcessPoolExecutor() as e:
      fut = [
          e.submit(test.execute, [chrome_extensions, test_extensions],
                   metrics_descriptor, keep_input, rebase)
          for test in self.test_runners
      ]
      for res in concurrent.futures.as_completed(fut):
        test_name, res_str, result = res.result()
        sys.stderr.write(res_str)
        if not result or not result.passed:
          if rebase:
            rebased.append(test_name)
          failures.append(test_name)
        else:
          perf_results.append(result.perf_result)
    test_time_ms = int(
        (datetime.datetime.now() - test_run_start).total_seconds() * 1000)
    return TestResults(failures, perf_results, rebased, test_time_ms)