#!/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 os
import unittest
import subprocess

import test_env
from lib.device import Device
from test_case import TestCaseWithFactory


class DeviceTest(TestCaseWithFactory):

    # These tests don't use a factory-created Device.
    # This to avoid having the factory add SSH options.

    def test_ssh_config(self):
        device = Device(self.factory, addr='::1')
        self.assertFalse(device.ssh_config)

        with self.assertRaises(ValueError):
            device.ssh_config = 'no_such_config'
        ssh_config = 'ssh_config'
        self.host.touch(ssh_config)
        device.ssh_config = ssh_config
        self.assertEqual(['-F', ssh_config], device.ssh_opts())

    def test_ssh_identity(self):
        device = Device(self.factory, addr='::1')
        self.assertFalse(device.ssh_identity)

        with self.assertRaises(ValueError):
            device.ssh_identity = 'no_such_identity'
        ssh_identity = 'ssh_identity'
        self.host.touch(ssh_identity)
        device.ssh_identity = ssh_identity
        self.assertEqual(['-i', ssh_identity], device.ssh_opts())

    def test_ssh_option(self):
        device = Device(self.factory, addr='::1')
        self.assertEqual(device.ssh_options, [])

        device.ssh_options = ['StrictHostKeyChecking no']
        device.ssh_options.append('UserKnownHostsFile=/dev/null')
        self.assertEqual(
            [
                '-o',
                'StrictHostKeyChecking no',
                '-o',
                'UserKnownHostsFile=/dev/null',
            ], device.ssh_opts())

    def test_ssh_verbosity(self):
        device = Device(self.factory, addr='::1')
        self.assertEqual(device.ssh_verbosity, 0)

        with self.assertRaises(ValueError):
            device.ssh_verbosity = 4

        device.ssh_verbosity = 3
        self.assertEqual(['-vvv'], device.ssh_opts())

        device.ssh_verbosity = 2
        self.assertEqual(['-vv'], device.ssh_opts())

        device.ssh_verbosity = 1
        self.assertEqual(['-v'], device.ssh_opts())

        device.ssh_verbosity = 0
        self.assertEqual([], device.ssh_opts())

        with self.assertRaises(ValueError):
            device.ssh_verbosity = -1

    def test_reachable(self):
        cmd = ['true']
        self.assertTrue(self.device.reachable)
        self.assertSsh(*cmd)

    def test_unreachable(self):
        cmd = ['true']
        process = self.get_process(cmd, ssh=True)
        process.succeeds = False
        self.assertFalse(self.device.reachable)
        self.assertSsh(*cmd)

    def test_configure(self):
        device = Device(self.factory, addr='::1')
        device.configure()
        self.assertEqual(
            device.ssh_config,
            self.buildenv.abspath(
                self.buildenv.build_dir, 'ssh-keys', 'ssh_config'))
        self.assertTrue(device.ssh_identity)
        self.assertFalse(device.ssh_options)
        self.assertFalse(device.ssh_verbosity)

    # These tests use a Device created by the TestCase

    def test_ssh(self):
        cmd = ['some-command', '--with', 'some-argument']
        self.device.ssh(cmd).check_call()
        self.assertSsh(*cmd)

    def test_has_cs_info(self):
        url1 = 'fuchsia-pkg://fuchsia.com/http#meta/http.cmx'
        url2 = 'fuchsia-pkg://fuchsia.com/fake-package1#meta/fake-target1.cmx'
        url3 = 'fuchsia-pkg://fuchsia.com/fake-package1#meta/fake-target2.cmx'
        url4 = 'fuchsia-pkg://fuchsia.com/an-extremely-verbose-target-package#meta/an-extremely-verbose-target-executable.cmx'

        self.set_running(url1)
        self.set_running(url2, duration=10)
        self.set_running(url3, duration=10)
        self.set_running(url4)

        # Can check various URLs
        self.assertTrue(self.device.has_cs_info(url1))
        self.assertTrue(self.device.has_cs_info(url2))
        self.assertTrue(self.device.has_cs_info(url3))
        self.assertTrue(self.device.has_cs_info(url4))

        # URLs are cached until refresh.
        self.host.sleep(10)
        self.assertTrue(self.device.has_cs_info(url1))
        self.assertTrue(self.device.has_cs_info(url2))
        self.assertTrue(self.device.has_cs_info(url3))
        self.assertTrue(self.device.has_cs_info(url4))

        self.assertTrue(self.device.has_cs_info(url1, refresh=True))
        self.assertFalse(self.device.has_cs_info(url2))
        self.assertFalse(self.device.has_cs_info(url3))
        self.assertTrue(self.device.has_cs_info(url4))

    def test_isfile(self):
        some_file = 'path-to-some-file'
        cmd = ['test', '-f', some_file]
        process = self.get_process(cmd, ssh=True)
        process.succeeds = False
        self.assertFalse(self.device.isfile(some_file))
        self.assertSsh(*cmd)
        process.succeeds = True
        self.assertTrue(self.device.isfile(some_file))
        self.assertSsh(*cmd)

    def test_isdir(self):
        some_dir = 'path-to-some-directory'
        cmd = ['test', '-d', some_dir]
        process = self.get_process(cmd, ssh=True)
        process.succeeds = False
        self.assertFalse(self.device.isdir(some_dir))
        self.assertSsh(*cmd)
        process.succeeds = True
        self.assertTrue(self.device.isdir(some_dir))
        self.assertSsh(*cmd)

    def test_ls(self):
        corpus_dir = 'path-to-some-corpus'
        cmd = ['ls', '-l', corpus_dir]

        self.touch_on_device(
            '{}/feac37187e77ff60222325cf2829e2273e04f2ea'.format(corpus_dir),
            size=1796)
        self.touch_on_device(
            '{}/ff415bddb30e9904bccbbd21fb5d4aa9bae9e5a5'.format(corpus_dir),
            size=124)
        files = self.device.ls(corpus_dir)
        self.assertSsh(*cmd)
        self.assertEqual(
            files['feac37187e77ff60222325cf2829e2273e04f2ea'], 1796)

    def test_mkdir(self):
        corpus_dir = 'path-to-some-corpus'
        cmd = ['mkdir', '-p', corpus_dir]
        self.device.mkdir(corpus_dir)
        self.assertSsh(*cmd)

    def test_remove(self):
        some_file = 'path-to-some-file'
        cmd = ['rm', '-f', some_file]
        self.device.remove(some_file)
        self.assertSsh(*cmd)

        some_dir = 'path-to-some-directory'
        cmd = ['rm', '-rf', some_dir]
        self.device.remove(some_dir, recursive=True)
        self.assertSsh(*cmd)

    def test_dump_log(self):
        cmd = ['log_listener', '--dump_logs', '--some', 'other-arg']
        self.device.dump_log('--some', 'other-arg')
        self.assertSsh(*cmd)

    def test_guess_pid(self):
        cmd = [
            'log_listener', '--dump_logs', '--only', 'reset,Fuzzer,Sanitizer'
        ]
        self.assertEqual(self.device.guess_pid(), -1)
        self.assertSsh(*cmd)

        # Log lines are like '[timestamp][pid][tid][name] data'
        self.set_outputs(
            cmd, [
                '[ignored][31415][ignored][ignored] ignored',
                '[ignored][92653][ignored][ignored] ignored',
                '[malformed][58979]',
            ],
            ssh=True)

        # Should find last well-formed line
        self.assertEqual(self.device.guess_pid(), 92653)

    def test_fetch(self):
        local_path = 'test_fetch'
        remote_path = 'remote-path'

        # Fails due to missing local pathname.
        self.assertError(
            lambda: self.device.fetch(local_path, remote_path),
            'No such directory: test_fetch')

        self.host.mkdir(local_path)

        # Fails due to empty source file list.
        with self.assertRaises(ValueError):
            self.device.fetch(local_path)

        self.device.fetch(local_path, remote_path)
        self.assertScpFrom(remote_path, local_path)

        corpus_dir = remote_path + '/*'
        self.device.fetch(local_path, corpus_dir)
        self.assertScpFrom(corpus_dir, local_path)

    def test_store(self):
        local_path = 'test_store'
        remote_path = 'remote-path'

        # Globs must resolve
        self.assertError(
            lambda: self.device.store(
                remote_path, os.path.join(local_path, '*')),
            'No matching files: "test_store/*".')

        # Local path must exist
        foo = os.path.join(local_path, 'foo')
        self.assertError(
            lambda: self.device.store(remote_path, foo),
            'No matching files: "test_store/foo".')

        # Valid
        self.host.touch(foo)
        self.device.store(remote_path, foo)
        self.assertScpTo(foo, remote_path)

        # Valid globs
        bar = os.path.join(local_path, 'bar')
        baz = os.path.join(local_path, 'baz')
        self.host.touch(bar)
        self.host.touch(baz)
        self.device.store(remote_path, os.path.join(local_path, '*'))
        self.assertScpTo(bar, baz, foo, remote_path)


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