#!/usr/bin/env python3
# SPDX-License-Identifier: LGPL-2.1+
# ~~~
#   Description: Tests for Dynamic Host Configuration v6 Server (dhcp6s)
#
#   Author: Susant Sahani <susant@redhat.com>
#   Copyright (c) 2018 Red Hat, Inc.
# ~~~

import errno
import os
import sys
import time
import unittest
import subprocess
import signal
import shutil
import psutil
import socket
from pyroute2 import IPRoute

RADVD_TCP_DUMP_FILE='/tmp/radvd-tcp-dump-ipv6-ra.pcap'

RADVD_CONFIG_FILE ='/var/run/wide-dhcpv6-ci/radvd-ci.conf'
RADVD_PID_FILE='/var/run/wide-dhcpv6-ci/radvd.pid'
RADVD_LOG_FILE='/var/run/wide-dhcpv6-ci/radvd.log'

DHCP6S_CONFIG_FILE='/var/run/wide-dhcpv6-ci/dhcp6s-ci.conf'
DHCP6S_PID_FILE='/var/run/wide-dhcpv6-ci/dhcp6s.pid'

DHCP6C_CONFIG_FILE='/var/run/wide-dhcpv6-ci/dhcp6c.conf'
DHCP6C_PID_FILE='/var/run/wide-dhcpv6-ci/dhcp6c.pid'
DHCP6C_DUID_FILE='/var/db/dhcp6c_duid'

DHCLIENT_PID_FILE='/var/run/wide-dhcpv6-ci/test-dhclient.pid'

def setUpModule():
    """Initialize the environment, and perform sanity checks on it."""

    if shutil.which('radvd') is None:
        raise OSError(errno.ENOENT, 'radvd not found')

    if shutil.which('dhcp6s') is None:
        raise OSError(errno.ENOENT, 'dhcp6s not found')

    if shutil.which('dhclient') is None:
        raise OSError(errno.ENOENT, 'dhclient not found')

def tearDownModule():
    pass

class GenericUtilities():
    """Provide a set of utility functions start stop daemons. write config files etc """

    def StartRadvd(self):
        """Start radvd"""

        subprocess.check_output(['radvd', '-d5', '-C', RADVD_CONFIG_FILE, '-p', RADVD_PID_FILE, '-l', RADVD_LOG_FILE])

    def StartDhcp6s(self):
        """Start dhcp6s"""

        subprocess.check_output(['dhcp6s', '-c', DHCP6S_CONFIG_FILE, 'veth-peer', '-dD', '-P', DHCP6S_PID_FILE])

    def StartDhclient(self, conf_file):
        """Start dhclient"""

        path=os.path.join('/var/run/wide-dhcpv6-ci', conf_file)
        subprocess.check_output(['dhclient', '-i', 'veth-test', '-6','-pf', DHCLIENT_PID_FILE, 'veth-peer', '-cf', path])

    def StopDaemon(self, pid_file):

        with open(pid_file, 'r') as f:
                 pid = f.read().rstrip(' \t\r\n\0')
                 os.kill(int(pid), signal.SIGKILL)

        os.remove(pid_file)

    def SetupVethInterface(self):
        """Setup veth interface"""

        ip = IPRoute()

        ip.link('add', ifname='veth-test', peer='veth-peer', kind='veth')
        idx_veth_test= ip.link_lookup(ifname='veth-test')[0]
        idx_veth_peer = ip.link_lookup(ifname='veth-peer')[0]

        ip.link('set', index=idx_veth_test, address='02:01:02:03:04:08')
        ip.link('set', index=idx_veth_peer, address='02:01:02:03:04:09')
        ip.link('set', index=idx_veth_test, state='up')
        ip.link('set', index=idx_veth_peer, state='up')

        ip.close()
        time.sleep(4)

    def TearDownVethInterface(self):

        ip = IPRoute()
        ip.link('del', index=ip.link_lookup(ifname='veth-test')[0])
        ip.close()

class Dhcp6sTests(unittest.TestCase, GenericUtilities):

    def setUp(self):
        """ setup veth and write radvd and dhcpv6configs """
        self.SetupVethInterface()

    def tearDown(self):
        self.StopDaemon(DHCLIENT_PID_FILE)
        self.StopDaemon(DHCP6S_PID_FILE)
        self.StopDaemon(RADVD_PID_FILE)

        self.TearDownVethInterface()

    def test_dhcp6s_assigns_static_address_using_duid1(self):
        """ DHCP6S assigns same (static) addresses to same hosts using known DUID values"""

        self.StartRadvd()
        self.StartDhcp6s()

        self.StartDhclient('dhclient-ci-duid2.conf')

        time.sleep(10)
        output=subprocess.check_output(['ip','address', 'show', 'veth-test']).rstrip().decode('utf-8')

        # Address prefix
        self.assertRegex(output, "2001:888:db8:1::c")

    def test_dhcp6s_assigns_static_address_using_duid2(self):
        """ DHCP6S assigns same (static) addresses to same hosts using known DUID values"""

        self.StartRadvd()
        self.StartDhcp6s()

        self.StartDhclient('dhclient-ci-duid1.conf')

        time.sleep(15)
        output=subprocess.check_output(['ip','address', 'show', 'veth-test']).rstrip().decode('utf-8')

        # Address prefix
        self.assertRegex(output, "2001:888:db8:1::b")

if __name__ == '__main__':
    unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout,
                                                     verbosity=3))