#! /usr/bin/python3

# SPDX-License-Identifier: GPL-2.0-or-later
#
# avocado/pasta.py - Basic tests for pasta mode
#
# Copyright Red Hat
# Author: David Gibson <david@gibson.dropbear.id.au>

import ipaddress
import os

from tasst.address import LOOPBACK4, LOOPBACK6
from tasst.scenario.simple import SimpleNetTasst
from tasst.dhcp import BaseDhcpTasst
from tasst.dhcpv6 import BaseDhcpv6Tasst
from tasst.ndp import BaseNdpTasst
from tasst.nstool import UnshareSite
from tasst.transfer import BaseTransferTasst, TcpUploadTasst, UdpTransferTasst


PASTA_BIN = './pasta'


class UnconfiguredPastaTasst(SimpleNetTasst):
    EXTRA_OPTS = ''
    INBOUND_PORT = 10002
    OUTBOUND_PORT = 10003

    def setUp(self):
        super().setUp()

        self.guestns = UnshareSite(self.__class__.__name__ + '.guestns',
                                   '-Ucnpf --mount-proc',
                                   parent=self.simhost, sudo=True)
        self.guestns_relpid = self.guestns.relative_pid(self.simhost)

        pidfile = os.path.join(self.workdir, 'pasta.pid')
        pastacmd = ('{} -f -t {} -u {} -T {} -U {} -P {} {} {}'
                    .format(PASTA_BIN, self.INBOUND_PORT, self.INBOUND_PORT,
                            self.OUTBOUND_PORT, self.OUTBOUND_PORT,
                            pidfile, self.EXTRA_OPTS, self.guestns_relpid))

        self.pasta = self.simhost.bg(pastacmd)

        # Wait for the pidfile to be written
        while not os.path.exists(pidfile) or not open(pidfile).read():
            pass
        # PID of pasta process in simhost namespace
        self.pastapid = int(open(pidfile).read())

    def tearDown(self):
        self.simhost.fg('kill -TERM {}'.format(self.pastapid))
        rc = self.pasta.wait(timeout=1.0)
        self.assertEqual(rc, 0)
        self.guestns.close()
        super().tearDown()


class PastaIfnameTasst(UnconfiguredPastaTasst):
    def test_ifname(self):
        self.assertCountEqual(self.guestns.ifs(), ('lo', self.ifname))


class PastaNdpTasst(UnconfiguredPastaTasst, BaseNdpTasst):
    def setUp(self):
        super().setUp()
        self.guestns.ifup(self.ifname)
        BaseNdpTasst.subsetup(self, self.guestns, self.ifname, self.ip6.network, self.gw_ip6_ll.ip)


class PastaDhcpTasst(UnconfiguredPastaTasst, BaseDhcpTasst):
    def setUp(self):
        super().setUp()
        BaseDhcpTasst.subsetup(self, self.guestns, self.ifname, self.ip4.ip, self.gw_ip4.ip, 65520)


class PastaDhcpv6Tasst(UnconfiguredPastaTasst, BaseDhcpv6Tasst):
    def setUp(self):
        super().setUp()
        BaseDhcpv6Tasst.subsetup(self, self.guestns, self.ifname, self.ip6.ip)


class ConfiguredPastaTasst(UnconfiguredPastaTasst):
    EXTRA_OPTS = '--config-net'


class ConfigNetTasst(ConfiguredPastaTasst):
    def test_addr(self):
        addrs = self.guestns.addrs(self.ifname, scope='global')
        self.assertCountEqual(addrs, [self.ip4, self.ip6])

    def test_route4(self):
        (defroute,) = self.guestns.routes4(dst='default')
        gateway = ipaddress.ip_address(defroute['gateway'])
        self.assertEquals(gateway, self.gw_ip4.ip)

    def test_route6(self):
        (defroute,) = self.guestns.routes6(dst='default')
        gateway = ipaddress.ip_address(defroute['gateway'])
        self.assertEquals(gateway, self.gw_ip6_ll.ip)

    def test_mtu(self):
        mtu = self.guestns.mtu(self.ifname)
        self.assertEquals(mtu, 65520)


class PastaOutwardTransferTasst(ConfiguredPastaTasst, BaseTransferTasst):
    """
    :avocado: disable
    """

    PORT = 10000
    def setUp(self):
        super().setUp()

        BaseTransferTasst.subsetup(self, self.DATAFILE, self.guestns, self.gw,
                                   self.remote_ip4.ip, self.remote_ip6.ip, self.PORT)


class PastaOutwardSmallTcpUploadTasst(PastaOutwardTransferTasst, TcpUploadTasst):
    DATAFILE = './test/small.bin'


class PastaOutwardBigTcpUploadTasst(PastaOutwardSmallTcpUploadTasst):
    DATAFILE = './test/big.bin'
    timeout = 30.0


class PastaOutwardUdpTransferTasst(PastaOutwardTransferTasst, UdpTransferTasst):
    DATAFILE = './test/medium.bin'


class PastaInwardTransferTasst(ConfiguredPastaTasst, BaseTransferTasst):
    """
    :avocado: disable
    """

    def setUp(self):
        super().setUp()

        BaseTransferTasst.subsetup(self, self.DATAFILE, self.gw, self.guestns,
                                   self.ip4.ip, self.ip6.ip, self.INBOUND_PORT,
                                   from_ip4=self.remote_ip4.ip, from_ip6=self.remote_ip6.ip)


class PastaInwardSmallTcpUploadTasst(PastaInwardTransferTasst, TcpUploadTasst):
    DATAFILE = './test/small.bin'


class PastaInwardBigTcpUploadTasst(PastaInwardSmallTcpUploadTasst):
    DATAFILE = './test/big.bin'
    timeout = 30.0


class PastaInwardUdpTransferTasst(PastaInwardTransferTasst, UdpTransferTasst):
    DATAFILE = './test/medium.bin'


class PastaSplicedTransferTasst(ConfiguredPastaTasst, BaseTransferTasst):
    """
    :avocado: disable
    """

    def setUp(self):
        super().setUp()

        BaseTransferTasst.subsetup(self, self.DATAFILE, self.guestns, self.simhost,
                                   LOOPBACK4, LOOPBACK6, self.OUTBOUND_PORT,
                                   listen_ip4=LOOPBACK4, listen_ip6=LOOPBACK6)


class PastaSplicedSmallTcpUploadTasst(PastaSplicedTransferTasst, TcpUploadTasst):
    DATAFILE = './test/small.bin'


class PastaSplicedBigTcpUploadTasst(PastaSplicedSmallTcpUploadTasst):
    DATAFILE = './test/big.bin'
    timeout = 30.0


class PastaSplicedUdpTransferTasst(PastaSplicedTransferTasst, UdpTransferTasst):
    DATAFILE = './test/medium.bin'