#!/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))