#!/usr/bin/python2
# pylint: disable-msg=C0111
#
# Copyright 2019 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file
"""Test for skylab migration unittest."""

from __future__ import unicode_literals
from __future__ import print_function

import copy
import json
import os.path
import subprocess
import tempfile
import unittest
import shutil

import mock

import common
from autotest_lib.cli import skylab_migration


class ExecuteWithTempfileUnittest(unittest.TestCase):

    def test_call_with_tempfile(self):
        with mock.patch('subprocess.check_output') as check_output:
            check_output.return_value = b'\n'.join([b'x', b'y', b'z'])
            commandOutput = skylab_migration.call_with_tempfile([], [])
            self.assertEqual(commandOutput.output, ['x', 'y', 'z'])

    def test_call_with_tempfile_real(self):
        commandOutput = skylab_migration.call_with_tempfile(
            ['/bin/cat', skylab_migration._TEMPPATH], ['a', 'b', 'c'])
        self.assertEqual(commandOutput.output, ['a', 'b', 'c'])


class MigrationUnittest(unittest.TestCase):

    def setUp(self):
        super(MigrationUnittest, self).setUp()
        self._tempdir = tempfile.mkdtemp()

        def do_nothing(*args, **kwargs):
            pass

        self.__patches = {
            'call_with_tempfile':
                mock.patch.object(
                    skylab_migration, 'call_with_tempfile', new=None),
            'popen':
                mock.patch('subprocess.Popen', new=None),
            'check_call':
                mock.patch.object(subprocess, 'check_call', new=None),
            'call':
                mock.patch.object(subprocess, 'call', new=None),
            'check_output':
                mock.patch.object(subprocess, 'check_output', new=None),
            'mkdtemp':
                mock.patch.object(tempfile, 'mkdtemp', new=None),
            'mkstemp':
                mock.patch.object(tempfile, 'mkstemp', new=None),
            'NamedTemporaryFile':
                mock.patch('tempfile.NamedTemporaryFile', new=None),
            'stderr_log':
                mock.patch.object(
                    skylab_migration, 'stderr_log', new=do_nothing)
        }
        for x in self.__patches.values():
            x.start()

    def tearDown(self):
        for x in self.__patches.values():
            x.stop()
        shutil.rmtree(self._tempdir, ignore_errors=True)
        super(MigrationUnittest, self).tearDown()

    def test_find_atest(self):
        atest_exe = skylab_migration.find_atest_path()
        self.assertTrue(os.path.exists(atest_exe))

    def test_brief_info_cmd(self):
        return self.assertEqual(skylab_migration.AtestCmd.brief_info_cmd()[:-1],
                                [skylab_migration._ATEST_EXE] +
                                'host list --parse -M'.split())

    def test_brief_info(self):
        with mock.patch.object(skylab_migration, 'call_with_tempfile') as call_:
            call_.return_value = skylab_migration.CommandOutput(
                exit_code=0,
                output=[
                    'key1=a|Labels=x', 'key1=b|Labels=y', 'key1=c|Labels=z'
                ])
            items = list(
                skylab_migration.AtestCmd.brief_info(hostnames=['a', 'b', 'c']))
            self.assertEqual(items, [
                {
                    'key1': 'a'
                },
                {
                    'key1': 'b'
                },
                {
                    'key1': 'c'
                },
            ])

    def test_rename_cmd_for_migration(self):
        cmd = skylab_migration.AtestCmd.rename_cmd(for_migration=True)
        self.assertEqual(cmd, [
            skylab_migration._ATEST_EXE,
            'host',
            'rename',
            '--non-interactive',
            '--for-migration',
            '--parse',
            '-M',
            skylab_migration._TEMPPATH,
        ])

    def test_rename_cmd_for_rollback(self):
        cmd = skylab_migration.AtestCmd.rename_cmd(for_migration=False)
        self.assertEqual(cmd, [
            skylab_migration._ATEST_EXE,
            'host',
            'rename',
            '--non-interactive',
            '--for-rollback',
            '--parse',
            '-M',
            skylab_migration._TEMPPATH,
        ])

    def test_rename_filter(self):
        expected = ['10', '20']
        actual = list(
            skylab_migration.AtestCmd.rename_filter(['10 to 1', '20 to 2']))
        self.assertEqual(expected, actual)

    def test_rename(self):
        with mock.patch.object(skylab_migration, 'call_with_tempfile') as call_:
            output = skylab_migration.CommandOutput(
                exit_code=0,
                output=[
                    'a to a.suffix', 'b to b.suffix', 'c to c.suffix',
                    'd to d.suffix'
                ])
            expected = ['a', 'b', 'c', 'd']
            call_.return_value = output
            actual = list(skylab_migration.AtestCmd.rename(hostnames=[]))
            self.assertEqual(expected, actual)

    def test_statjson_cmd(self):
        self.assertEqual(
            skylab_migration.AtestCmd.statjson_cmd(hostname='H'),
            [skylab_migration._ATEST_EXE, 'host', 'statjson', '--', 'H'])

    def test_atest_lock_cmd(self):
        self.assertEqual(
            skylab_migration.AtestCmd.atest_lock_cmd(reason='R'), [
                skylab_migration._ATEST_EXE, 'host', 'mod', '--lock', '-r', 'R',
                '-M', skylab_migration._TEMPPATH
            ])

    def test_atest_lock(self):
        # just check that traversing the body of atest_lock doesn't throw an exception
        with mock.patch.object(skylab_migration, 'call_with_tempfile') as call_:
            call_.return_value = skylab_migration.CommandOutput(
                exit_code=0, output=['a', 'b'])

    def test_atest_unlock_cmd(self):
        self.assertEqual(skylab_migration.AtestCmd.atest_unlock_cmd(), [
            skylab_migration._ATEST_EXE, 'host', 'mod', '--unlock', '-M',
            skylab_migration._TEMPPATH
        ])

    def test_atest_unlock(self):
        with mock.patch.object(skylab_migration.AtestCmd,
                               'atest_unlock') as atest_unlock:
            atest_unlock.return_value = ['a', 'b']
            expected = ['a', 'b']
            actual = list(
                skylab_migration.AtestCmd.atest_unlock(hostnames=['a', 'b']))
            self.assertEqual(expected, actual)

    def test_add_one_dut_cmd(self):
        expected = [
            skylab_migration._SKYLAB_EXE, 'add-dut', '-skip-image-download',
            '-skip-install-firmware', '-skip-install-os', '-specs-file',
            skylab_migration._TEMPPATH
        ]
        actual = skylab_migration.SkylabCmd.add_one_dut_cmd()
        self.assertEqual(expected, actual)

    def test_atest_get_migration_plan_cmd(self):
        expected = [
            skylab_migration._ATEST_EXE, 'host', 'get_migration_plan',
            '--ratio', '0.1', '--mlist', skylab_migration._TEMPPATH
        ]
        actual = skylab_migration.AtestCmd.atest_get_migration_plan_cmd(
            ratio=0.1)
        self.assertEqual(expected, actual)

    def test_atest_get_migration_plan(self):
        with mock.patch.object(skylab_migration,
                               'call_with_tempfile') as call_with_tempfile:
            call_with_tempfile.return_value = skylab_migration.CommandOutput(
                exit_code=0,
                output=[json.dumps({
                    'transfer': [],
                    'retain': []
                })])
            out = skylab_migration.AtestCmd.atest_get_migration_plan(
                ratio=0.4, hostnames=[])
            self.assertEqual(out['transfer'], [])
            self.assertEqual(out['retain'], [])

    def test_lock_smoke_test(self):
        # just make sure Migration.lock doesn't throw an exception
        with mock.patch.object(skylab_migration, 'call_with_tempfile') as call_:
            call_.return_value = skylab_migration.CommandOutput(
                exit_code=0, output=['a', 'b'])
            skylab_migration.Migration.lock(
                hostnames=[], reason='reason', retries=3)

    def test_lock_single_host(self):
        pass
        # def atest_lock(hostnames=[], **kwargs):
        #     """successfully lock every hostname"""
        #     for item in hostnames:
        #         yield item

        # with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
        #     atest_cmd.atest_lock = atest_lock
        #     summary = skylab_migration.Migration.lock(
        #         hostnames=['HHH'], reason=None, retries=1)
        #     self.assertEqual(summary.locked, {'HHH'})
        #     self.assertEqual(summary.not_locked, set())
        #     self.assertEqual(list(summary.tries), ['HHH'])

    def test_lock_one_good_one_bad(self):
        # TODO(gregorynisbet): effectively just a smoke test
        def atest_lock(hostnames=[], **kwargs):
            return Nothing

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.atest_lock = atest_lock

    def test_ensure_lock_smoke_test(self):

        def brief_info(hostnames=[], **kwargs):
            if False:
                yield 42

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.brief_info = brief_info
            summary = skylab_migration.Migration.ensure_lock(hostnames=[])
            self.assertEqual(summary.locked, set())
            self.assertEqual(summary.not_locked, set())

    def test_ensure_lock_one_good_one_bad(self):

        def brief_info(**kwargs):
            yield {'Host': 'a', 'Locked': True}
            yield {'Host': 'b', 'Locked': False}

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.brief_info = brief_info
            summary = skylab_migration.Migration.ensure_lock(
                hostnames=['a', 'b'])
            self.assertEqual(summary.locked, {'a'})
            self.assertEqual(summary.not_locked, {'b'})

    def test_rename_smoke_test(self):

        def atest_cmd_rename(**kwargs):
            if False:
                yield 42

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.rename = atest_cmd_rename
            summary = skylab_migration.Migration.rename(hostnames=[])
            self.assertEqual(summary.renamed, set())
            self.assertEqual(summary.not_renamed, set())

    def test_rename_one_good_one_bad(self):

        def atest_cmd_rename(**kwargs):
            yield 'GOOD'

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.rename = atest_cmd_rename
            summary = skylab_migration.Migration.rename(
                hostnames=['GOOD', 'BAD'])
            self.assertEqual(summary.renamed, set(['GOOD']))
            self.assertEqual(summary.not_renamed, set(['BAD']))

    def test_add_to_skylab_inventory_and_drone_smoke_test(self):
        summary = skylab_migration.Migration.add_to_skylab_inventory_and_drone(
            hostnames=[],
            use_quick_add=False)
        self.assertEqual(summary.complete, set())
        self.assertEqual(summary.without_drone, set())
        self.assertEqual(summary.not_started, set())

    def test_add_to_skylab_inventory_and_drone_one_of_each(self):

        @staticmethod
        def atest_statjson(hostname=None):
            return hostname

        @staticmethod
        def add_one_dut(add_dut_content=None):
            if add_dut_content in ('GOOD', 'MEDIUM'):
                return skylab_migration.CommandOutput(output=[], exit_code=0)
            else:
                return skylab_migration.CommandOutput(output=[], exit_code=1)

        @staticmethod
        def assign_one_dut(hostname=None):
            if hostname == 'GOOD':
                return skylab_migration.CommandOutput(output=[], exit_code=0)
            else:
                return skylab_migration.CommandOutput(output=[], exit_code=1)

        with mock.patch.object(
                skylab_migration.AtestCmd, 'statjson', new=atest_statjson):
            with mock.patch.object(
                    skylab_migration.SkylabCmd, 'add_one_dut', new=add_one_dut):
                summary = skylab_migration.Migration.add_to_skylab_inventory_and_drone(
                    use_quick_add=False,
                    hostnames=['GOOD', 'MEDIUM', 'BAD'])
                self.assertEqual(summary.complete, {'GOOD', 'MEDIUM'})
                self.assertEqual(summary.not_started, {'BAD'})

    def test_migrate_known_good_duts_until_max_duration_sync_smoke_test(self):

        def brief_info(**kwargs):
            if False:
                yield 42

        def rename(**kwargs):
            if False:
                yield 42

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.brief_info = brief_info
            atest_cmd.rename = rename
            summary = skylab_migration.Migration.migrate_known_good_duts_until_max_duration_sync(
                use_quick_add=False,
                hostnames=[])
            self.assertEqual(summary.success, set())
            self.assertEqual(summary.failure, set())

    def test_migrate_known_good_duts_until_max_duration_one_good_one_bad(self):

        def brief_info(**kwargs):
            return [
                {
                    'Host': 'GOOD',
                    'Status': 'Ready'
                },
                {
                    'Host': 'BAD',
                    'Status': 'Ready'
                },
            ]

        inventory_return = skylab_migration.AddToSkylabInventoryAndDroneStatus(
            complete=['GOOD', 'BAD'],
            not_started=[],
            without_drone=[],
        )

        def atest_cmd_rename(hostname=None, **kwargs):
            yield 'GOOD'

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.brief_info = brief_info
            atest_cmd.rename = atest_cmd_rename
            with mock.patch.object(skylab_migration, 'SkylabCmd') as skylab_cmd:
                with mock.patch.object(skylab_migration.Migration,
                                       'add_to_skylab_inventory_and_drone'
                                      ) as add_to_skylab_obj:
                    add_to_skylab_obj.return_value = inventory_return
                    summary = skylab_migration.Migration.migrate_known_good_duts_until_max_duration_sync(
                        use_quick_add=False,
                        hostnames=['GOOD', 'BAD'])
                    self.assertEqual(summary.success, set(['GOOD']))
                    self.assertEqual(summary.failure, set(['BAD']))
                    self.assertEqual(summary.needs_add_to_skylab, set())
                    self.assertEqual(summary.needs_drone, set())
                    self.assertEqual(summary.needs_rename, set(['BAD']))
                    add_to_skylab_obj.assert_called()

    def test_migrate_duts_unconditionally_smoke_test(self):

        def brief_info(**kwargs):
            if False:
                yield 42

        def rename(**kwargs):
            if False:
                yield 42

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.brief_info = brief_info
            atest_cmd.rename = rename
            summary = skylab_migration.Migration.migrate_duts_unconditionally(
                use_quick_add=False,
                hostnames=[])
            self.assertEqual(summary.success, set())
            self.assertEqual(summary.failure, set())

    def test_migrate_duts_unconditionally_one_good_one_bad(self):

        def brief_info(**kwargs):
            return [
                {
                    'Host': 'GOOD',
                    'Status': 'Ready'
                },
                {
                    'Host': 'BAD',
                    'Status': 'Ready'
                },
            ]

        inventory_retval = skylab_migration.AddToSkylabInventoryAndDroneStatus(
            complete=['GOOD', 'BAD'],
            not_started=[],
            without_drone=[],
        )

        def atest_cmd_rename(hostname=None, **kwargs):
            yield 'GOOD'

        with mock.patch.object(skylab_migration, 'AtestCmd') as atest_cmd:
            atest_cmd.brief_info = brief_info
            atest_cmd.rename = atest_cmd_rename
            with mock.patch.object(skylab_migration, 'SkylabCmd') as skylab_cmd:
                with mock.patch.object(skylab_migration.Migration,
                                       'add_to_skylab_inventory_and_drone'
                                      ) as add_to_skylab_obj:
                    add_to_skylab_obj.return_value = inventory_retval
                    summary = skylab_migration.Migration.migrate_duts_unconditionally(
                        use_quick_add=False,
                        hostnames=['GOOD', 'BAD'])
                    self.assertEqual(summary.success, set(['GOOD']))
                    self.assertEqual(summary.failure, set(['BAD']))

    @mock.patch.object(skylab_migration.Migration, 'migration_plan')
    @mock.patch.object(skylab_migration.Migration,
                       'migrate_known_good_duts_until_max_duration_sync')
    @mock.patch.object(skylab_migration.Migration,
                       'migrate_duts_unconditionally')
    @mock.patch.object(skylab_migration.Migration, 'ensure_lock')
    @mock.patch.object(skylab_migration.Migration, 'lock')
    def test_migrate_smoke_test(self, lock, ensure_lock,
                                migrate_duts_unconditionally, known_good,
                                migration_plan):
        lock.return_value = skylab_migration.LockCommandStatus(
            locked=[], not_locked=[], tries=None)
        ensure_lock.return_value = skylab_migration.LockCommandStatus(
            locked=[], not_locked=[], tries=None)
        known_good.return_value = migrate_duts_unconditionally.return_value = skylab_migration.MigrateDutCommandStatus(
            success=[],
            failure=[],
            needs_add_to_skylab=[],
            needs_drone=[],
            needs_rename=[])
        migration_plan.return_value = skylab_migration.MigrationPlan(
            transfer=[], retain=[])
        skylab_migration.Migration.migrate(
            hostnames=[],
            reason='test',
            interval_len=0,
            max_duration=10,
            immediately=True)


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