#!/usr/bin/env python2.7
# Copyright 2020 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 os
import sys
import unittest

import test_env
import lib.command as command
from test_case import TestCaseWithFuzzer


class CommandTest(TestCaseWithFuzzer):

    # Unit tests

    def test_list_fuzzers(self):
        # No match
        args = self.parse_args('list', 'no/match')
        command.list_fuzzers(args, self.factory)
        self.assertLogged('No matching fuzzers.')

        # Multiple matches
        args = self.parse_args('list', 'fake-package1')
        command.list_fuzzers(args, self.factory)
        self.assertLogged(
            'Found 3 matching fuzzers:',
            '  fake-package1/fake-target1',
            '  fake-package1/fake-target2',
            '  fake-package1/fake-target3',
        )

        # Exact match
        args = self.parse_args('list', 'fake-package1/fake-target1')
        command.list_fuzzers(args, self.factory)
        self.assertLogged(
            'Found 1 matching fuzzers:',
            '  fake-package1/fake-target1',
        )

    def test_start_fuzzer(self):
        name = str(self.fuzzer)

        # In the foreground
        output = 'output'
        self.host.mkdir(output)
        args = self.parse_args('start', '-f', '-o', output, name)
        command.start_fuzzer(args, self.factory)
        self.assertLogged(
            'Starting {}.'.format(self.fuzzer),
            'Outputs will be written to: {}'.format(output),
        )

        # In the background
        args = self.parse_args('start', '-o', output, name)
        command.start_fuzzer(args, self.factory)
        self.assertLogged(
            'Starting {}.'.format(self.fuzzer),
            'Outputs will be written to: {}'.format(output),
            'Check status with "fx fuzz check {}".'.format(self.fuzzer),
            'Stop manually with "fx fuzz stop {}".'.format(self.fuzzer),
        )
        cmd = [
            'python',
            sys.argv[0],
            'start',
            '--monitor',
            '--output',
            output,
            name,
        ]
        self.assertRan(*cmd)

        args = self.parse_args('start', '-m', '-o', output, name)
        command.start_fuzzer(args, self.factory)
        self.assertLogged(
            '{} has stopped.'.format(self.fuzzer),
            'Output written to: {}.'.format(output),
        )

    def test_check_fuzzer(self):
        # No name, none running
        args = self.parse_args('check')
        command.check_fuzzer(args, self.factory)
        self.assertLogged(
            'No fuzzers are running.',
            'Include \'name\' to check specific fuzzers.',
        )

        # Name provided, not installed
        fuzzer = self.create_fuzzer('fake-package2/fake-target1', resolve=False)
        args = self.parse_args('check', 'fake-package2/fake-target1')
        command.check_fuzzer(args, self.factory)
        self.assertLogged(
            'fake-package2/fake-target1: NOT INSTALLED',
            '',
        )

        # No name, some running
        self.set_running('fake-package1', 'fake-target1')
        self.set_running('fake-package1', 'fake-target3')
        args = self.parse_args('check')
        command.check_fuzzer(args, self.factory)
        self.assertLogged(
            'fake-package1/fake-target1: RUNNING',
            '    Corpus size:  0 inputs / 0 bytes',
            '',
            'fake-package1/fake-target3: RUNNING',
            '    Corpus size:  0 inputs / 0 bytes',
            '',
        )

        # Name provided, running
        args = self.parse_args('check', 'fake-package1/fake-target3')
        command.check_fuzzer(args, self.factory)
        self.assertLogged(
            'fake-package1/fake-target3: RUNNING',
            '    Corpus size:  0 inputs / 0 bytes',
            '',
        )

        # Name provided, not running
        args = self.parse_args('check', 'fake-package1/fake-target2')
        command.check_fuzzer(args, self.factory)
        self.assertLogged(
            'fake-package1/fake-target2: STOPPED',
            '    Corpus size:  0 inputs / 0 bytes',
            '',
        )

        # Add some artifacts
        fuzzer = self.create_fuzzer('fake-package1/fake-target2')
        self.host.touch(os.path.join(fuzzer.output, 'crash-deadbeef'))
        self.host.touch(os.path.join(fuzzer.output, 'leak-feedface'))
        command.check_fuzzer(args, self.factory)
        self.assertLogged(
            'fake-package1/fake-target2: STOPPED',
            '    Corpus size:  0 inputs / 0 bytes',
            '    Artifacts:',
            '        {}/crash-deadbeef'.format(fuzzer.output),
            '        {}/leak-feedface'.format(fuzzer.output),
            '',
        )

    def test_stop_fuzzer(self):
        # Not running
        args = self.parse_args('stop', str(self.fuzzer))
        command.stop_fuzzer(args, self.factory)
        self.assertLogged('{} is already stopped.'.format(self.fuzzer))

        # Running
        self.set_running(self.fuzzer.package, self.fuzzer.executable)
        args = self.parse_args('stop', str(self.fuzzer))
        command.stop_fuzzer(args, self.factory)
        self.assertLogged('Stopping {}.'.format(self.fuzzer))

    def test_repro_units(self):
        unit = 'crash-deadbeef'
        self.host.touch(unit)
        args = self.parse_args('repro', str(self.fuzzer), unit)
        command.repro_units(args, self.factory)

    def test_analyze_fuzzer(self):
        args = self.parse_args('analyze', '-l', str(self.fuzzer))
        command.analyze_fuzzer(args, self.factory)

        # We shouldn't have copied anything
        self.assertFalse([cmd for cmd in self.host.processes if 'gs' in cmd])

        # Invalid corpus
        corpus1 = 'corpus1'
        corpus2 = 'corpus2'
        local_dict = 'local_dict'
        args = self.parse_args(
            'analyze', '-c', corpus1, '-c', corpus2, '-d', local_dict,
            '{}/{}'.format(self.fuzzer.package, self.fuzzer.executable))
        self.assertError(
            lambda: command.analyze_fuzzer(args, self.factory),
            'No such directory: {}'.format(corpus1))

        self.host.mkdir(corpus1)
        foo = os.path.join(corpus1, 'foo')
        bar = os.path.join(corpus1, 'bar')
        self.host.touch(foo)
        self.host.touch(bar)
        self.assertError(
            lambda: command.analyze_fuzzer(args, self.factory),
            'No such directory: {}'.format(corpus2))

        # Invalid dictionary
        self.host.mkdir(corpus2)
        baz = os.path.join(corpus2, 'baz')
        self.host.touch(baz)
        self.assertError(
            lambda: command.analyze_fuzzer(args, self.factory),
            'No such file: {}'.format(local_dict))

        self.host.touch(local_dict)
        with self.host.temp_dir() as temp_dir:
            # Make it appear as if something was retrieved from GCS.
            qux = os.path.join(temp_dir.pathname, 'qux')
            self.host.touch(qux)
        command.analyze_fuzzer(args, self.factory)
        gcs_url = 'gs://corpus.internal.clusterfuzz.com/libFuzzer/fuchsia_{}-{}'.format(
            self.fuzzer.package, self.fuzzer.executable)
        with self.host.temp_dir() as temp_dir:
            cmd = ['gsutil', '-m', 'cp', gcs_url + '/*', temp_dir.pathname]
            self.assertRan(*cmd)

        abspath = self.ns.data_abspath(self.corpus.nspaths[0])
        self.assertScpTo(bar, foo, abspath)
        self.assertScpTo(baz, abspath)
        self.assertEqual(
            self.dictionary.nspath, self.fuzzer.ns.data(local_dict))


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