"""The tests for the ASUSWRT device tracker platform."""
import os
from datetime import timedelta
import unittest
from unittest import mock
import socket

import voluptuous as vol

from homeassistant.setup import setup_component
from homeassistant.components import device_tracker
from homeassistant.components.device_tracker import (
    CONF_CONSIDER_HOME, CONF_TRACK_NEW, CONF_NEW_DEVICE_DEFAULTS,
    CONF_AWAY_HIDE)
from homeassistant.components.device_tracker.asuswrt import (
    CONF_PROTOCOL, CONF_MODE, CONF_PUB_KEY, DOMAIN, _ARP_REGEX,
    CONF_PORT, PLATFORM_SCHEMA, Device, get_scanner, AsusWrtDeviceScanner,
    _parse_lines, SshConnection, TelnetConnection, CONF_REQUIRE_IP)
from homeassistant.const import (CONF_PLATFORM, CONF_PASSWORD, CONF_USERNAME,
                                 CONF_HOST)

import pytest
from tests.common import (
    get_test_home_assistant, get_test_config_dir, assert_setup_component,
    mock_component)

FAKEFILE = None

VALID_CONFIG_ROUTER_SSH = {DOMAIN: {
    CONF_PLATFORM: 'asuswrt',
    CONF_HOST: 'fake_host',
    CONF_USERNAME: 'fake_user',
    CONF_PROTOCOL: 'ssh',
    CONF_MODE: 'router',
    CONF_PORT: '22'
}}

WL_DATA = [
    'assoclist 01:02:03:04:06:08\r',
    'assoclist 08:09:10:11:12:14\r',
    'assoclist 08:09:10:11:12:15\r'
]

WL_DEVICES = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip=None, name=None),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip=None, name=None),
    '08:09:10:11:12:15': Device(
        mac='08:09:10:11:12:15', ip=None, name=None)
}

ARP_DATA = [
    '? (123.123.123.125) at 01:02:03:04:06:08 [ether]  on eth0\r',
    '? (123.123.123.126) at 08:09:10:11:12:14 [ether]  on br0\r',
    '? (123.123.123.127) at <incomplete>  on br0\r',
]

ARP_DEVICES = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip='123.123.123.125', name=None),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip='123.123.123.126', name=None)
}

NEIGH_DATA = [
    '123.123.123.125 dev eth0 lladdr 01:02:03:04:06:08 REACHABLE\r',
    '123.123.123.126 dev br0 lladdr 08:09:10:11:12:14 REACHABLE\r',
    '123.123.123.127 dev br0  FAILED\r',
    '123.123.123.128 dev br0 lladdr 08:09:15:15:15:15 DELAY\r',
    'fe80::feff:a6ff:feff:12ff dev br0 lladdr fc:ff:a6:ff:12:ff STALE\r',
]

NEIGH_DEVICES = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip='123.123.123.125', name=None),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip='123.123.123.126', name=None)
}

LEASES_DATA = [
    '51910 01:02:03:04:06:08 123.123.123.125 TV 01:02:03:04:06:08\r',
    '79986 01:02:03:04:06:10 123.123.123.127 android 01:02:03:04:06:15\r',
    '23523 08:09:10:11:12:14 123.123.123.126 * 08:09:10:11:12:14\r',
]

LEASES_DEVICES = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip='123.123.123.125', name='TV'),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip='123.123.123.126', name='')
}

WAKE_DEVICES = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip='123.123.123.125', name='TV'),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip='123.123.123.126', name='')
}

WAKE_DEVICES_AP = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip='123.123.123.125', name=None),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip='123.123.123.126', name=None)
}

WAKE_DEVICES_NO_IP = {
    '01:02:03:04:06:08': Device(
        mac='01:02:03:04:06:08', ip='123.123.123.125', name=None),
    '08:09:10:11:12:14': Device(
        mac='08:09:10:11:12:14', ip='123.123.123.126', name=None),
    '08:09:10:11:12:15': Device(
        mac='08:09:10:11:12:15', ip=None, name=None)
}


def setup_module():
    """Set up the test module."""
    global FAKEFILE
    FAKEFILE = get_test_config_dir('fake_file')
    with open(FAKEFILE, 'w') as out:
        out.write(' ')


def teardown_module():
    """Tear down the module."""
    os.remove(FAKEFILE)


@pytest.mark.skip(
    reason="These tests are performing actual failing network calls. They "
    "need to be cleaned up before they are re-enabled. They're frequently "
    "failing in Travis.")
class TestComponentsDeviceTrackerASUSWRT(unittest.TestCase):
    """Tests for the ASUSWRT device tracker platform."""

    hass = None

    def setup_method(self, _):
        """Set up things to be run when tests are started."""
        self.hass = get_test_home_assistant()
        mock_component(self.hass, 'zone')

    def teardown_method(self, _):
        """Stop everything that was started."""
        self.hass.stop()
        try:
            os.remove(self.hass.config.path(device_tracker.YAML_DEVICES))
        except FileNotFoundError:
            pass

    def test_parse_lines_wrong_input(self):
        """Testing parse lines."""
        output = _parse_lines("asdf asdfdfsafad", _ARP_REGEX)
        self.assertEqual(output, [])

    def test_get_device_name(self):
        """Test for getting name."""
        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.last_results = WAKE_DEVICES
        self.assertEqual('TV', scanner.get_device_name('01:02:03:04:06:08'))
        self.assertEqual(None, scanner.get_device_name('01:02:03:04:08:08'))

    def test_scan_devices(self):
        """Test for scan devices."""
        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.last_results = WAKE_DEVICES
        self.assertEqual(list(WAKE_DEVICES), scanner.scan_devices())

    def test_password_or_pub_key_required(self):
        """Test creating an AsusWRT scanner without a pass or pubkey."""
        with assert_setup_component(0, DOMAIN):
            assert setup_component(
                self.hass, DOMAIN, {DOMAIN: {
                    CONF_PLATFORM: 'asuswrt',
                    CONF_HOST: 'fake_host',
                    CONF_USERNAME: 'fake_user',
                    CONF_PROTOCOL: 'ssh'
                }})

    @mock.patch(
        'homeassistant.components.device_tracker.asuswrt.AsusWrtDeviceScanner',
        return_value=mock.MagicMock())
    def test_get_scanner_with_password_no_pubkey(self, asuswrt_mock):
        """Test creating an AsusWRT scanner with a password and no pubkey."""
        conf_dict = {
            DOMAIN: {
                CONF_PLATFORM: 'asuswrt',
                CONF_HOST: 'fake_host',
                CONF_USERNAME: 'fake_user',
                CONF_PASSWORD: 'fake_pass',
                CONF_TRACK_NEW: True,
                CONF_CONSIDER_HOME: timedelta(seconds=180),
                CONF_NEW_DEVICE_DEFAULTS: {
                    CONF_TRACK_NEW: True,
                    CONF_AWAY_HIDE: False
                }
            }
        }

        with assert_setup_component(1, DOMAIN):
            assert setup_component(self.hass, DOMAIN, conf_dict)

        conf_dict[DOMAIN][CONF_MODE] = 'router'
        conf_dict[DOMAIN][CONF_PROTOCOL] = 'ssh'
        conf_dict[DOMAIN][CONF_PORT] = 22
        self.assertEqual(asuswrt_mock.call_count, 1)
        self.assertEqual(asuswrt_mock.call_args, mock.call(conf_dict[DOMAIN]))

    @mock.patch(
        'homeassistant.components.device_tracker.asuswrt.AsusWrtDeviceScanner',
        return_value=mock.MagicMock())
    def test_get_scanner_with_pubkey_no_password(self, asuswrt_mock):
        """Test creating an AsusWRT scanner with a pubkey and no password."""
        conf_dict = {
            device_tracker.DOMAIN: {
                CONF_PLATFORM: 'asuswrt',
                CONF_HOST: 'fake_host',
                CONF_USERNAME: 'fake_user',
                CONF_PUB_KEY: FAKEFILE,
                CONF_TRACK_NEW: True,
                CONF_CONSIDER_HOME: timedelta(seconds=180),
                CONF_NEW_DEVICE_DEFAULTS: {
                    CONF_TRACK_NEW: True,
                    CONF_AWAY_HIDE: False
                }
            }
        }

        with assert_setup_component(1, DOMAIN):
            assert setup_component(self.hass, DOMAIN, conf_dict)

        conf_dict[DOMAIN][CONF_MODE] = 'router'
        conf_dict[DOMAIN][CONF_PROTOCOL] = 'ssh'
        conf_dict[DOMAIN][CONF_PORT] = 22
        self.assertEqual(asuswrt_mock.call_count, 1)
        self.assertEqual(asuswrt_mock.call_args, mock.call(conf_dict[DOMAIN]))

    def test_ssh_login_with_pub_key(self):
        """Test that login is done with pub_key when configured to."""
        ssh = mock.MagicMock()
        ssh_mock = mock.patch('pexpect.pxssh.pxssh', return_value=ssh)
        ssh_mock.start()
        self.addCleanup(ssh_mock.stop)
        conf_dict = PLATFORM_SCHEMA({
            CONF_PLATFORM: 'asuswrt',
            CONF_HOST: 'fake_host',
            CONF_USERNAME: 'fake_user',
            CONF_PUB_KEY: FAKEFILE
        })
        update_mock = mock.patch(
            'homeassistant.components.device_tracker.asuswrt.'
            'AsusWrtDeviceScanner.get_asuswrt_data')
        update_mock.start()
        self.addCleanup(update_mock.stop)
        asuswrt = device_tracker.asuswrt.AsusWrtDeviceScanner(conf_dict)
        asuswrt.connection.run_command('ls')
        self.assertEqual(ssh.login.call_count, 1)
        self.assertEqual(
            ssh.login.call_args,
            mock.call('fake_host', 'fake_user', quiet=False,
                      ssh_key=FAKEFILE, port=22)
        )

    def test_ssh_login_with_password(self):
        """Test that login is done with password when configured to."""
        ssh = mock.MagicMock()
        ssh_mock = mock.patch('pexpect.pxssh.pxssh', return_value=ssh)
        ssh_mock.start()
        self.addCleanup(ssh_mock.stop)
        conf_dict = PLATFORM_SCHEMA({
            CONF_PLATFORM: 'asuswrt',
            CONF_HOST: 'fake_host',
            CONF_USERNAME: 'fake_user',
            CONF_PASSWORD: 'fake_pass'
        })
        update_mock = mock.patch(
            'homeassistant.components.device_tracker.asuswrt.'
            'AsusWrtDeviceScanner.get_asuswrt_data')
        update_mock.start()
        self.addCleanup(update_mock.stop)
        asuswrt = device_tracker.asuswrt.AsusWrtDeviceScanner(conf_dict)
        asuswrt.connection.run_command('ls')
        self.assertEqual(ssh.login.call_count, 1)
        self.assertEqual(
            ssh.login.call_args,
            mock.call('fake_host', 'fake_user', quiet=False,
                      password='fake_pass', port=22)
        )

    def test_ssh_login_without_password_or_pubkey(self):
        """Test that login is not called without password or pub_key."""
        ssh = mock.MagicMock()
        ssh_mock = mock.patch('pexpect.pxssh.pxssh', return_value=ssh)
        ssh_mock.start()
        self.addCleanup(ssh_mock.stop)

        conf_dict = {
            CONF_PLATFORM: 'asuswrt',
            CONF_HOST: 'fake_host',
            CONF_USERNAME: 'fake_user',
        }

        with self.assertRaises(vol.Invalid):
            conf_dict = PLATFORM_SCHEMA(conf_dict)

        update_mock = mock.patch(
            'homeassistant.components.device_tracker.asuswrt.'
            'AsusWrtDeviceScanner.get_asuswrt_data')
        update_mock.start()
        self.addCleanup(update_mock.stop)

        with assert_setup_component(0, DOMAIN):
            assert setup_component(self.hass, DOMAIN,
                                   {DOMAIN: conf_dict})
        ssh.login.assert_not_called()

    def test_telnet_login_with_password(self):
        """Test that login is done with password when configured to."""
        telnet = mock.MagicMock()
        telnet_mock = mock.patch('telnetlib.Telnet', return_value=telnet)
        telnet_mock.start()
        self.addCleanup(telnet_mock.stop)
        conf_dict = PLATFORM_SCHEMA({
            CONF_PLATFORM: 'asuswrt',
            CONF_PROTOCOL: 'telnet',
            CONF_HOST: 'fake_host',
            CONF_USERNAME: 'fake_user',
            CONF_PASSWORD: 'fake_pass'
        })
        update_mock = mock.patch(
            'homeassistant.components.device_tracker.asuswrt.'
            'AsusWrtDeviceScanner.get_asuswrt_data')
        update_mock.start()
        self.addCleanup(update_mock.stop)
        asuswrt = device_tracker.asuswrt.AsusWrtDeviceScanner(conf_dict)
        asuswrt.connection.run_command('ls')
        self.assertEqual(telnet.read_until.call_count, 4)
        self.assertEqual(telnet.write.call_count, 3)
        self.assertEqual(
            telnet.read_until.call_args_list[0],
            mock.call(b'login: ')
        )
        self.assertEqual(
            telnet.write.call_args_list[0],
            mock.call(b'fake_user\n')
        )
        self.assertEqual(
            telnet.read_until.call_args_list[1],
            mock.call(b'Password: ')
        )
        self.assertEqual(
            telnet.write.call_args_list[1],
            mock.call(b'fake_pass\n')
        )
        self.assertEqual(
            telnet.read_until.call_args_list[2],
            mock.call(b'#')
        )

    def test_telnet_login_without_password(self):
        """Test that login is not called without password or pub_key."""
        telnet = mock.MagicMock()
        telnet_mock = mock.patch('telnetlib.Telnet', return_value=telnet)
        telnet_mock.start()
        self.addCleanup(telnet_mock.stop)

        conf_dict = {
            CONF_PLATFORM: 'asuswrt',
            CONF_PROTOCOL: 'telnet',
            CONF_HOST: 'fake_host',
            CONF_USERNAME: 'fake_user',
        }

        with self.assertRaises(vol.Invalid):
            conf_dict = PLATFORM_SCHEMA(conf_dict)

        update_mock = mock.patch(
            'homeassistant.components.device_tracker.asuswrt.'
            'AsusWrtDeviceScanner.get_asuswrt_data')
        update_mock.start()
        self.addCleanup(update_mock.stop)

        with assert_setup_component(0, DOMAIN):
            assert setup_component(self.hass, DOMAIN,
                                   {DOMAIN: conf_dict})
        telnet.login.assert_not_called()

    def test_get_asuswrt_data(self):
        """Test asuswrt data fetch."""
        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner._get_wl = mock.Mock()
        scanner._get_arp = mock.Mock()
        scanner._get_neigh = mock.Mock()
        scanner._get_leases = mock.Mock()
        scanner._get_wl.return_value = WL_DEVICES
        scanner._get_arp.return_value = ARP_DEVICES
        scanner._get_neigh.return_value = NEIGH_DEVICES
        scanner._get_leases.return_value = LEASES_DEVICES
        self.assertEqual(WAKE_DEVICES, scanner.get_asuswrt_data())

    def test_get_asuswrt_data_ap(self):
        """Test for get asuswrt_data in ap mode."""
        conf = VALID_CONFIG_ROUTER_SSH.copy()[DOMAIN]
        conf[CONF_MODE] = 'ap'
        scanner = AsusWrtDeviceScanner(conf)
        scanner._get_wl = mock.Mock()
        scanner._get_arp = mock.Mock()
        scanner._get_neigh = mock.Mock()
        scanner._get_leases = mock.Mock()
        scanner._get_wl.return_value = WL_DEVICES
        scanner._get_arp.return_value = ARP_DEVICES
        scanner._get_neigh.return_value = NEIGH_DEVICES
        scanner._get_leases.return_value = LEASES_DEVICES
        self.assertEqual(WAKE_DEVICES_AP, scanner.get_asuswrt_data())

    def test_get_asuswrt_data_no_ip(self):
        """Test for get asuswrt_data and not requiring ip."""
        conf = VALID_CONFIG_ROUTER_SSH.copy()[DOMAIN]
        conf[CONF_REQUIRE_IP] = False
        scanner = AsusWrtDeviceScanner(conf)
        scanner._get_wl = mock.Mock()
        scanner._get_arp = mock.Mock()
        scanner._get_neigh = mock.Mock()
        scanner._get_leases = mock.Mock()
        scanner._get_wl.return_value = WL_DEVICES
        scanner._get_arp.return_value = ARP_DEVICES
        scanner._get_neigh.return_value = NEIGH_DEVICES
        scanner._get_leases.return_value = LEASES_DEVICES
        self.assertEqual(WAKE_DEVICES_NO_IP, scanner.get_asuswrt_data())

    def test_update_info(self):
        """Test for update info."""
        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.get_asuswrt_data = mock.Mock()
        scanner.get_asuswrt_data.return_value = WAKE_DEVICES
        self.assertTrue(scanner._update_info())
        self.assertTrue(scanner.last_results, WAKE_DEVICES)
        scanner.success_init = False
        self.assertFalse(scanner._update_info())

    @mock.patch(
        'homeassistant.components.device_tracker.asuswrt.SshConnection')
    def test_get_wl(self, mocked_ssh):
        """Testing wl."""
        mocked_ssh.run_command.return_value = WL_DATA
        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.connection = mocked_ssh
        self.assertEqual(WL_DEVICES, scanner._get_wl())
        mocked_ssh.run_command.return_value = ''
        self.assertEqual({}, scanner._get_wl())

    @mock.patch(
        'homeassistant.components.device_tracker.asuswrt.SshConnection')
    def test_get_arp(self, mocked_ssh):
        """Testing arp."""
        mocked_ssh.run_command.return_value = ARP_DATA

        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.connection = mocked_ssh
        self.assertEqual(ARP_DEVICES, scanner._get_arp())
        mocked_ssh.run_command.return_value = ''
        self.assertEqual({}, scanner._get_arp())

    @mock.patch(
        'homeassistant.components.device_tracker.asuswrt.SshConnection')
    def test_get_neigh(self, mocked_ssh):
        """Testing neigh."""
        mocked_ssh.run_command.return_value = NEIGH_DATA

        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.connection = mocked_ssh
        self.assertEqual(NEIGH_DEVICES, scanner._get_neigh(ARP_DEVICES.copy()))
        self.assertEqual(NEIGH_DEVICES, scanner._get_neigh({
            'UN:KN:WN:DE:VI:CE': Device('UN:KN:WN:DE:VI:CE', None, None),
        }))
        mocked_ssh.run_command.return_value = ''
        self.assertEqual({}, scanner._get_neigh(ARP_DEVICES.copy()))

    @mock.patch(
        'homeassistant.components.device_tracker.asuswrt.SshConnection')
    def test_get_leases(self, mocked_ssh):
        """Testing leases."""
        mocked_ssh.run_command.return_value = LEASES_DATA

        scanner = get_scanner(self.hass, VALID_CONFIG_ROUTER_SSH)
        scanner.connection = mocked_ssh
        self.assertEqual(
            LEASES_DEVICES, scanner._get_leases(NEIGH_DEVICES.copy()))
        mocked_ssh.run_command.return_value = ''
        self.assertEqual({}, scanner._get_leases(NEIGH_DEVICES.copy()))


@pytest.mark.skip(
    reason="These tests are performing actual failing network calls. They "
    "need to be cleaned up before they are re-enabled. They're frequently "
    "failing in Travis.")
class TestSshConnection(unittest.TestCase):
    """Testing SshConnection."""

    def setUp(self):
        """Set up test env."""
        self.connection = SshConnection(
            'fake', 'fake', 'fake', 'fake', 'fake')
        self.connection._connected = True

    def test_run_command_exception_eof(self):
        """Testing exception in run_command."""
        from pexpect import exceptions
        self.connection._ssh = mock.Mock()
        self.connection._ssh.sendline = mock.Mock()
        self.connection._ssh.sendline.side_effect = exceptions.EOF('except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)
        self.assertIsNone(self.connection._ssh)

    def test_run_command_exception_pxssh(self):
        """Testing exception in run_command."""
        from pexpect import pxssh
        self.connection._ssh = mock.Mock()
        self.connection._ssh.sendline = mock.Mock()
        self.connection._ssh.sendline.side_effect = pxssh.ExceptionPxssh(
            'except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)
        self.assertIsNone(self.connection._ssh)

    def test_run_command_assertion_error(self):
        """Testing exception in run_command."""
        self.connection._ssh = mock.Mock()
        self.connection._ssh.sendline = mock.Mock()
        self.connection._ssh.sendline.side_effect = AssertionError('except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)
        self.assertIsNone(self.connection._ssh)


@pytest.mark.skip(
    reason="These tests are performing actual failing network calls. They "
    "need to be cleaned up before they are re-enabled. They're frequently "
    "failing in Travis.")
class TestTelnetConnection(unittest.TestCase):
    """Testing TelnetConnection."""

    def setUp(self):
        """Set up test env."""
        self.connection = TelnetConnection(
            'fake', 'fake', 'fake', 'fake')
        self.connection._connected = True

    def test_run_command_exception_eof(self):
        """Testing EOFException in run_command."""
        self.connection._telnet = mock.Mock()
        self.connection._telnet.write = mock.Mock()
        self.connection._telnet.write.side_effect = EOFError('except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)

    def test_run_command_exception_connection_refused(self):
        """Testing ConnectionRefusedError in run_command."""
        self.connection._telnet = mock.Mock()
        self.connection._telnet.write = mock.Mock()
        self.connection._telnet.write.side_effect = ConnectionRefusedError(
            'except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)

    def test_run_command_exception_gaierror(self):
        """Testing socket.gaierror in run_command."""
        self.connection._telnet = mock.Mock()
        self.connection._telnet.write = mock.Mock()
        self.connection._telnet.write.side_effect = socket.gaierror('except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)

    def test_run_command_exception_oserror(self):
        """Testing OSError in run_command."""
        self.connection._telnet = mock.Mock()
        self.connection._telnet.write = mock.Mock()
        self.connection._telnet.write.side_effect = OSError('except')
        self.connection.run_command('test')
        self.assertFalse(self.connection._connected)
