#!/usr/bin/env python3.8
# Copyright 2019 The Fuchsia 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 errno
import json
import os
import re
import subprocess
from collections import defaultdict

from .fuzzer import Fuzzer
from .process import Process


class BuildEnv(object):
    """Represents a local build environment for Fuchsia.

    This class abstracts various repository, tool, and build details.

    Attributes:
      fuchsia_dir:      Path to Fuchsia source checkout.
      cli:              Associated CLI object.
      build_dir:        Path to the Fuchsia build output.
      symbolizer_exec:  Path to the Fuchsia symbolizer executable.
      build_id_dirs:    List of paths to symbolizer debug symbols.
      gsutil:           Path to the Google Cloud Storage utility.
      llvm_cov:         Path to the LLVM/Clang coverage tool.
      llvm_prodata:     Path to the LLVM/Clang profile data tool.
  """

    def __init__(self, factory):
        assert factory, 'Factory not set.'
        self._factory = factory
        fuchsia_dir = self.host.getenv('FUCHSIA_DIR')
        if not fuchsia_dir:
            self.host.error(
                'FUCHSIA_DIR not set.', 'Have you sourced "scripts/fx-env.sh"?')
        self._fuchsia_dir = fuchsia_dir
        self._build_dir = None
        self._symbolizer_exec = None
        self._build_id_dirs = None
        self._gsutil = None
        self._llvm_cov = None
        self._llvm_profdata = None
        self._fuzzers = []

    @property
    def fuchsia_dir(self):
        return self._fuchsia_dir

    @property
    def host(self):
        return self._factory.host

    @property
    def build_dir(self):
        assert self._build_dir, 'Build directory not set'
        return self._build_dir

    @property
    def symbolizer_exec(self):
        assert self._symbolizer_exec, 'Symbolizer executable not set.'
        return self._symbolizer_exec

    @symbolizer_exec.setter
    def symbolizer_exec(self, symbolizer_exec):
        symbolizer_exec = self.abspath(symbolizer_exec)
        if not self.host.isfile(symbolizer_exec):
            self.host.error(
                'Invalid symbolizer executable: {}'.format(symbolizer_exec))
        self._symbolizer_exec = symbolizer_exec

    @property
    def build_id_dirs(self):
        assert self._build_id_dirs, 'Build ID directories not set.'
        return self._build_id_dirs

    @build_id_dirs.setter
    def build_id_dirs(self, build_id_dirs):
        abspaths = []
        for build_id_dir in build_id_dirs:
            abspath = self.abspath(build_id_dir)
            srcpath = self.srcpath(build_id_dir)
            if not self.host.isdir(abspath):
                self.host.error(
                    'Invalid build ID directory: {}'.format(srcpath))
            abspaths.append(abspath)
        self._build_id_dirs = abspaths

    @property
    def gsutil(self):
        if not self._gsutil:
            try:
                self._gsutil = self.create_process(['which',
                                                    'gsutil']).check_output()
            except subprocess.CalledProcessError:
                self.host.error(
                    'Unable to find gsutil.',
                    'Try installing the Google Cloud SDK.')
        return self._gsutil

    @gsutil.setter
    def gsutil(self, gsutil):
        abspath = self.abspath(gsutil)
        if not self.host.isfile(abspath):
            self.host.error('Invalid GS utility: {}'.format(abspath))
        self._gsutil = abspath

    @property
    def llvm_cov(self):
        assert self._llvm_cov, 'LLVM cov not set.'
        return self._llvm_cov

    @llvm_cov.setter
    def llvm_cov(self, llvm_cov):
        llvm_cov = self.abspath(llvm_cov)
        if not self.host.isfile(llvm_cov):
            self.host.error('Invalid LLVM cov: {}'.format(llvm_cov))
        self._llvm_cov = llvm_cov

    @property
    def llvm_profdata(self):
        assert self._llvm_profdata, 'LLVM profdata not set.'
        return self._llvm_profdata

    @llvm_profdata.setter
    def llvm_profdata(self, llvm_profdata):
        llvm_profdata = self.abspath(llvm_profdata)
        if not self.host.isfile(llvm_profdata):
            self.host.error('Invalid LLVM profdata: {}'.format(llvm_profdata))
        self._llvm_profdata = llvm_profdata

    # Initialization routines

    def configure(self, build_dir):
        """Sets multiple properties based on the given build directory."""
        self._build_dir = self.abspath(build_dir)
        clang_dir = '//prebuilt/third_party/clang/' + self.host.platform
        self.symbolizer_exec = build_dir + '/host_x64/symbolizer'
        self.build_id_dirs = [
            clang_dir + '/lib/debug/.build-id',
            build_dir + '/.build-id',
        ]
        self.llvm_cov = clang_dir + '/bin/llvm-cov'
        self.llvm_profdata = clang_dir + '/bin/llvm-profdata'

    def read_fuzzers(self, pathname):
        """Parses the available fuzzers from an fuzzers.json pathname."""
        with self.host.open(pathname, on_error=[
                'Failed to read fuzzers from {}.'.format(pathname),
                'Have you run "fx set ... --fuzz-with <sanitizer>"?'
        ]) as opened:
            metadata = json.load(opened)

        fuzz_specs = []
        by_label = defaultdict(dict)
        for entry in metadata:
            # Try v2 metadata first.
            label = entry.get('label')
            if label:
                by_label[label].update(entry)
                continue
            # Fallback to v1 metadata.
            package = entry['fuzzers_package']
            package_url = 'fuchsia-pkg://fuchsia.com/{}'.format(package)
            for fuzzer in entry['fuzzers']:
                fuzz_specs.append(
                    {
                        'package': package,
                        'package_url': package_url,
                        'fuzzer': fuzzer,
                        'manifest': '{}.cmx'.format(fuzzer),
                        'label': '//generated/{}:{}'.format(package, fuzzer),
                    })
        fuzz_specs += list(by_label.values())
        self._fuzzers = [
            Fuzzer(self._factory, fuzz_spec) for fuzz_spec in fuzz_specs
        ]
        self._fuzzers.sort()

    def fuzzers(self, name=None, include_tests=False):
        """Returns a (possibly filtered) list of fuzzers.

        Matches the given name against the fuzzers previously instantiated by `read_fuzzers`.

        Parameters:
            name            A name to filter on. If the name is an exact match, it will return a
                            list containing the matching fuzzer. If the name is of the form 'x/y',
                            the filtered list will include all the fuzzers where 'x' is a substring
                            of `package` and y is a substring of `executable`; otherwise it includes
                            all the fuzzers where `name` is a substring of either `package` or
                            `executable`. If blank or omitted, all fuzzers are returned.
            include_tests   A boolean flag indicating whether to include fuzzer tests as fuzzers.
                            This can be useful for commands which only act on the source tree
                            without regard to a fuzzer being deployed on a device.

        Returns:
            A list of fuzzers matching the given name.

        Raises:
            ValueError: Name is malformed, e.g. of the form 'x/y/z'.
    """
        fuzzers = [
            fuzzer for fuzzer in self._fuzzers
            if fuzzer.matches(name) and (include_tests or not fuzzer.is_test)
        ]
        for fuzzer in fuzzers:
            if name == str(fuzzer):
                return [fuzzer]
        return fuzzers

    def fuzzer_tests(self, name=None):
        """Returns a (possibly filtered) list of fuzzer tests.

        Like fuzzers(), but returns uninstrumented fuzzer_tests instead of instrumented fuzzers.

        Parameters:
            name    A name to filter on. If the name is an exact match, it will return a list
                    containing the matching fuzzer test. If the name is of the form 'x/y', the
                    filtered list will include all the fuzzer tests where 'x' is a substring of
                    `package` and y is a substring of `executable` + '_test'; otherwise it includes
                    all the fuzzer tests where `name` is a substring of either `package` or
                    `executable` + '_test'. If blank or omitted, all fuzzer tests are returned.

        Returns:
            A list of fuzzer tests matching the given name.

        Raises:
            ValueError: Name is malformed, e.g. of the form 'x/y/z'.
    """
        fuzzer_tests = [
            fuzzer for fuzzer in self._fuzzers
            if fuzzer.matches(name) and fuzzer.is_test
        ]
        for fuzzer in fuzzer_tests:
            if name == str(fuzzer):
                return [fuzzer]
        return fuzzer_tests

    # Other routines

    def abspath(self, *segments):
        """Returns absolute path to a path in the build environment.

        This method can handle three types of path inputs:
        1. GN source-absolute paths, as identified by a leading '//', e.g. '//some/src/path'.
        2. Paths that are already valid absolute paths, e.g. '/an/absolute/path'.
        3. Paths interpreted as relative to the current working directory. e.g. 'a/relative/path'.
           See also Host.getcwd().

        This method normalizes the path using os.path.normpath, so forward slashes are automatically
        converted on Windows and callers should simply use POSIX-style paths.

        Parameters:
            segments    A list of path segments.

        Returns:
            A string containing the absolute path.
        """
        assert segments, 'No path segments provided.'
        if segments[0].startswith('//'):
            joined = os.path.join(
                self.fuchsia_dir, segments[0][2:], *segments[1:])
        else:
            joined = os.path.join(*segments)
        if not os.path.isabs(joined):
            joined = os.path.join(self.host.getcwd(), joined)
        return os.path.normpath(joined)

    def srcpath(self, label_or_path):
        """Returns a GN source-absolute path for a label, like GN's `get_label_info(..., "dir")`.

        This method can handle the same types of path inputs as `abspath`, plus one more:
        4. GN relative labels, i.e. labels without a leading '//'. The specific GN target is removed
           and the remainder is treated as relative to the working directory, similar to how GN
           behaves.

        As with `abspath`, callers should simply use forward slashes, as used by both POSIX-style
        and GN-style paths.

        Parameters:
            label_or_path   A string containing either a filesystem path or GN label.

        Returns:
            A string containing the source-absolute path, that is one beginning with '//'.
        """
        joined = self.abspath(label_or_path.split(':')[0])
        if not joined.startswith(self.fuchsia_dir):
            self.host.error(
                '{} is not a path in the source tree.'.format(joined))
        return '//' + os.path.relpath(joined, self.fuchsia_dir).replace(
            os.sep, '/')

    def find_device(self, device_name=None):
        """Returns the IPv6 address for a device."""
        cmd = [
            self.abspath(self.fuchsia_dir, '.jiri_root/bin/fx'), 'ffx',
            'target', 'list', '--format', 'a'
        ]
        if device_name:
            cmd += [device_name]
        try:
            addrs = self.host.create_process(cmd).check_output().strip()
        except subprocess.CalledProcessError as err:
            return self._find_device_by_list_devices(device_name)

        if not addrs:
            self.host.error('Unable to find device.', 'Try "fx set-device".')

        addrs = addrs.split('\n')
        if len(addrs) != 1:
            self.host.error('Multiple devices found.', 'Try "fx set-device".')
        return addrs[0]

    def _find_device_by_list_devices(self, device_name=None):
        """ Find a device address by using fx list-devices

        Attributes:
            device_name: The name of the device to match to. If not present,
                         will only suceed if there is exactly 1 device.

        Returns:
            The device address.
        """
        cmd = [
            self.abspath(self.fuchsia_dir, '.jiri_root/bin/fx'), 'list-devices'
        ]
        list_devices_out = self.host.create_process(cmd).check_output().strip()
        if not list_devices_out:
            self.host.error('Unable to find device.', 'Try "fx set-device".')

        # Try parsing the returned values
        devices = list_devices_out.split('\n')

        if devices:
            if not device_name:
                if len(devices) != 1:
                    self.host.error(
                        'Multiple devices found.', 'Try "fx set-device".')

                return devices[0].split(' ', 1)[0]

            # Find the matching device name
            for device in devices:
                addr, name = device.split(' ', 1)
                if name == device_name:
                    return addr

        self.host.error('Unable to find device.', 'Try "fx set-device".')

    def symbolize(self, raw, json_output=None):
        """Symbolizes backtraces in a log file using the current build.

        Attributes:
            raw: Bytes representing unsymbolized lines.
            json_output: If present, outputs trigger information to the specified file.

        Returns:
            Bytes representing symbolized lines.
        """
        cmd = [self.symbolizer_exec]
        for build_id_dir in self.build_id_dirs:
            cmd += ['--build-id-dir', build_id_dir]
        if json_output:
            cmd += ['--json-output', json_output]
        process = self.host.create_process(cmd)
        process.stdin = subprocess.PIPE
        process.stdout = subprocess.PIPE
        popened = process.popen()
        out, _ = popened.communicate(raw)
        if popened.returncode != 0:
            out = ''
        return re.sub(r'[0-9\[\]\.]*\[klog\] INFO: ', '', out)

    def testsharder(self, executable_url, out_dir, realm_label=None):
        """Shards the available tests into _one_ test shard per environment for
        use by testrunner.

        Attributes:
            executable_url: The fuchsia pkg url of the test to generate coverage for.
            out_dir: The output directory into which to write the sharded tests file.

        Returns:
            The path to a generated file containing exactly _one_ test definition
            as extracted from the tests.json file by testsharder.
        """
        sharder_out = os.path.join(out_dir, 'testsharder_out.json')
        # Generate the tests but force 1 shard maximum
        cmd = [os.path.join(self.build_dir, 'host_x64', 'testsharder')] \
            + ['-build-dir', self.build_dir] \
            + ['-max-shards-per-env', '1'] \
            + ['-output-file', sharder_out]
        if realm_label:
            cmd = cmd + ['-realm-label', realm_label]
        self.host.create_process(cmd).check_call()
        with self.host.open(sharder_out) as f:
            sharded = json.load(f)

        # Depending on the set of tests included we could get multiple shards.
        # One per environment, but we should only have 1 AEMU* shard.
        found_aemu = False
        all_tests = []
        for shard in sharded:
            if shard['name'].startswith('AEMU'):
                if found_aemu:
                    self.host.error(
                        'Expected a single AEMU shard, but got more than one.')
                found_aemu = True

                test_out = os.path.join(
                    out_dir, 'shard_{}_tests.json'.format(shard['name']))
                all_tests = [
                    t for t in shard['tests'] if t.get('name') == executable_url
                ]

        if not found_aemu:
            self.host.error('Unable to find any tests for AEMU shards.')

        if not all_tests:
            self.host.error('Found no matching tests to run.')

        self.host.echo(
            'Found {} tests to generate coverage report for.'.format(
                len(all_tests)))

        with self.host.open(test_out, 'w') as out_file:
            json.dump(all_tests, out_file)
        return test_out

    def testrunner(self, shard_file, out_dir, device):
        """Runs testrunner over a file of tests generated by testsharder which
        generates the artifacts used for SBCC.
        Additionally, dumps logs of associated pids to an output file for symbolizing.

        Attributes:
            shard_file: A fully qualified path to the sharded tests file.
            out_dir: The output directory into which to write the log dump.
            device: Reference to the device to get logs from.

        Returns:
            The path to the testrunner output.
            The path to the log dump file.
        """
        if not self.host.isfile(shard_file):
            self.host.error(
                'Unable to find sharded test file at {}.'.format(shard_file))

        runner_out_dir = os.path.join(out_dir, 'testrunner_out')
        cmd = [os.path.join(self.build_dir, 'host_x64', 'testrunner')] \
            + ['-out-dir', runner_out_dir] \
            + ['-use-runtests', '-per-test-timeout', '600s'] \
            + [shard_file]
        out = self.host.create_process(cmd).check_output()
        # fxb/64774 Catch a timeout error and check that the device addr is correct

        # Look for the marker log in the testrunner output
        pids = []
        for line in out.splitlines():
            if 'Fuzzer built as test' in line:
                parts = line.split('][')
                if len(parts) > 2:
                    pids.append(int(parts[1]))
        if len(pids) < 1:
            self.host.error('Unable to find a matching test fuzzer pid.')
        raw = device.dump_log('--pid', str(pids[0]))

        self.host.mkdir(os.path.join(out_dir, 'log_dumps'))
        # Strip the .json suffix from the end of the shard_file name.
        shard_file_basename = os.path.basename(shard_file)[:-5]
        log_dump_out = os.path.join(
            out_dir, 'log_dumps', 'dump_{}'.format(shard_file_basename))
        with self.host.open(log_dump_out, 'w') as out_file:
            out_file.write(raw)

        return runner_out_dir, log_dump_out

    def covargs(self, runner_dir, symbolize_file, out_dir):
        """Runs covargs to generate a SBCC given a testrunner output and symbolize file.

        Attributes:
            runner_dir: The path to the testrunner output which contains a summary.json file.
            symbolize_file: The path to the symbolized profile json output.
            out_dir: The output directory into which to write the coverage report

        Returns:
            The path to the generated coverage report.
        """
        summary_json_file = os.path.join(runner_dir, 'summary.json')
        if not self.host.isfile(summary_json_file):
            self.host.error(
                'Unable to find summary.json file at {}.'.format(
                    summary_json_file))
        if not self.host.isfile(symbolize_file):
            self.host.error(
                'Unable to find symbolize file at {}.'.format(symbolize_file))

        coverage_path = os.path.join(out_dir, 'covargs_out')
        cmd = [os.path.join(self.build_dir, 'host_x64', 'covargs')] \
            + ['-llvm-cov', self.llvm_cov] \
            + ['-llvm-profdata', self.llvm_profdata] \
            + ['-summary', summary_json_file] \
            + ['-symbolize-dump', symbolize_file] \
            + ['-output-dir', coverage_path]
        for build_id_dir in self.build_id_dirs:
            cmd += ['-build-id-dir', build_id_dir]

        self.host.create_process(cmd).check_call()

        return coverage_path