#! /usr/bin/python3

# SPDX-License-Identifier: GPL-2.0-or-later
#
# tasst - Test A Simple Socket Transport
#         library of test helpers for passt & pasta
#
# tasst/ndp.py - Helpers for testing NDP
#
# Copyright Red Hat
# Author: David Gibson <david@gibson.dropbear.id.au>

import contextlib
import ipaddress
import os

from tasst import Tasst, TasstSubData
from tasst.address import IpiAllocator, TEST_NET6_TASST_A
from tasst.nstool import UnshareSite
from tasst.site import Site
from tasst.typing import typecheck


class NdpTasstInfo:
    def __init__(self, site, ifname, net, gw):
        self.site = typecheck(site, Site)
        self.ifname = typecheck(ifname, str)
        self.net = typecheck(net, ipaddress.IPv6Network)
        self.gw = typecheck(gw, ipaddress.IPv6Address)

        site.require_cmds('ip')


class BaseNdpTasst(Tasst):
    """
    Test NDP behaviour.
    
    :avocado: disable
    """

    def setup_ndp(self):
        raise NotImplementedError("{} must implement setup_ndp() method".format(type(self).__name__))

    @contextlib.contextmanager
    def check_setup_ndp(self):
        with self.setup_ndp() as ndp:
            if not isinstance(ndp, NdpTasstInfo):
                raise TypeError("{}.setup_ndp() must yield a NdpTasstInfo instance".format(type(self).__name__))
            yield ndp
    
    def test_addr(self):
        with self.check_setup_ndp() as ndp:
            # Wait for NDP to do its thing
            (addr,) = ndp.site.addr_wait(ndp.ifname, family='inet6', scope='global')

            # The SLAAC address is derived from the guest ns MAC, so
            # probably won't exactly match the host address (we need
            # DHCPv6 for that).  It should be in the right network though.
            self.assertEquals(addr.network, ndp.net)

    def test_route(self):
        with self.check_setup_ndp() as ndp:
            defroutes = ndp.site.routes6(dst='default')
            while not defroutes:
                defroutes = ndp.site.routes6(dst='default')

            self.assertEquals(len(defroutes), 1)
            gateway = ipaddress.ip_address(defroutes[0]['gateway'])
            self.assertEquals(gateway, ndp.gw)


class MetaNdpTasst(BaseNdpTasst):
    """Ugly workaround for
    https://github.com/avocado-framework/avocado/issues/5680.
    Explicitly apply the "meta" tag to inherited tests

    :avocado: disable
    :avocado: tags=meta

    """

    def test_addr(self):
        super().test_addr()

    def test_route(self):
        super().test_route()


class RadvdNdpTasst(MetaNdpTasst):
    timeout = 15.0
    
    @contextlib.contextmanager
    def setup_ndp(self):
        ifname = 'clientif'
        router_ifname = 'routerif'
        prefix = TEST_NET6_TASST_A

        with UnshareSite(type(self).__name__ + '.client', '-Un') as client, \
             UnshareSite(type(self).__name__ + '.router', '-n', parent=client, sudo=True) as router:
            router.require_cmds('radvd')

            client.veth(ifname, router_ifname, router)

            # Configure the simulated router
            ipa = IpiAllocator(prefix)
            (router_ip6,) = ipa.next_ipis()

            confpath = os.path.join(self.workdir, 'radvd.conf')
            pidpath = os.path.join(self.workdir, 'radvd.pid')
            open(confpath, 'w').write(
            '''
            interface {} {{
                AdvSendAdvert on;
                prefix {} {{
                }};
            }};
            '''.format(router_ifname, prefix))

            router.ifup('lo')
            router.ifup('routerif', router_ip6)

            # Configure the client
            client.ifup('lo')
            client.ifup(ifname)

            # Get the router's link-local-address
            (router_ll,) = router.addr_wait(router_ifname, family='inet6', scope='link')

            # Run radvd
            router.fg('radvd -c -C {}'.format(confpath))
            with router.bg('radvd -C {} -p {} -n -d 5'.format(confpath, pidpath), sudo=True) as radvd:
                yield NdpTasstInfo(client, ifname, prefix, router_ll.ip)

                pid = int(open(pidpath).read())
                router.fg('kill {}'.format(pid))
                status = radvd.wait()
                self.assertEquals(status, 0)