#! /usr/bin/env avocado-runner-avocado-classless

# SPDX-License-Identifier: GPL-2.0-or-later
#
# Copyright Red Hat
# Author: David Gibson <david@gibson.dropbear.id.au>

"""
avocado/pasta.py - Basic tests for pasta mode
"""

import contextlib
import ipaddress
import os
import tempfile

from avocado_classless.test import assert_eq, assert_eq_unordered, test

from tasst.address import LOOPBACK4, LOOPBACK6
from tasst.dhcp import test_dhcp
from tasst.dhcpv6 import test_dhcpv6
from tasst.ndp import test_ndp
from tasst.nstool import unshare_site
from tasst.pasta import Pasta
from tasst.scenario.simple import (
    simple_net, IFNAME, HOST_NET6, IP4, IP6, GW_IP4, REMOTE_IP4, REMOTE_IP6
)
from tasst.transfer import test_transfers


IN_FWD_PORT = 10002
SPLICE_FWD_PORT = 10003
FWD_OPTS = f'-t {IN_FWD_PORT} -u {IN_FWD_PORT} \
             -T {SPLICE_FWD_PORT} -U {SPLICE_FWD_PORT}'


@contextlib.contextmanager
def pasta_unconfigured(opts=FWD_OPTS):
    with simple_net() as simnet:
        with unshare_site('pastans', '-Ucnpf --mount-proc',
                          parent=simnet.simhost, sudo=True) as guestns:
            with tempfile.TemporaryDirectory() as tmpdir:
                pidfile = os.path.join(tmpdir, 'pasta.pid')

                with Pasta(simnet.simhost, pidfile, opts, ns=guestns) as pasta:
                    yield simnet, pasta.ns


@test
def test_ifname():
    with pasta_unconfigured() as (_, ns):
        assert_eq_unordered(ns.ifs(), ('lo', IFNAME))


@test_ndp(IFNAME, HOST_NET6)
@contextlib.contextmanager
def pasta_ndp():
    with pasta_unconfigured() as (simnet, guestns):
        guestns.ifup(IFNAME)
        yield guestns, simnet.gw_ip6_ll.ip


@test_dhcp(IFNAME, IP4.ip, GW_IP4.ip, 65520)
@test_dhcpv6(IFNAME, IP6.ip)
@contextlib.contextmanager
def pasta_dhcp():
    with pasta_unconfigured() as (_, guestns):
        yield guestns


@contextlib.contextmanager
def pasta_configured():
    with pasta_unconfigured(FWD_OPTS + ' --config-net') as (simnet, ns):
        # Wait for DAD to complete on the --config-net address
        ns.addr_wait(IFNAME, family='inet6', scope='global')
        yield simnet, ns


@test
def test_config_net_addr():
    with pasta_configured() as (_, ns):
        addrs = ns.addrs(IFNAME, scope='global')
        assert_eq_unordered(addrs, [IP4, IP6])


@test
def test_config_net_route4():
    with pasta_configured() as (_, ns):
        (defroute,) = ns.routes4(dst='default')
        gateway = ipaddress.ip_address(defroute['gateway'])
        assert_eq(gateway, GW_IP4.ip)


@test
def test_config_net_route6():
    with pasta_configured() as (simnet, ns):
        (defroute,) = ns.routes6(dst='default')
        gateway = ipaddress.ip_address(defroute['gateway'])
        assert_eq(gateway, simnet.gw_ip6_ll.ip)


@test
def test_config_net_mtu():
    with pasta_configured() as (_, ns):
        mtu = ns.mtu(IFNAME)
        assert_eq(mtu, 65520)


@test_transfers(ip4=REMOTE_IP4.ip, ip6=REMOTE_IP6.ip, port=10000)
@contextlib.contextmanager
def outward_transfer():
    with pasta_configured() as (simnet, ns):
        yield ns, simnet.gw


@test_transfers(ip4=IP4.ip, ip6=IP6.ip, port=IN_FWD_PORT,
                fromip4=REMOTE_IP4.ip, fromip6=REMOTE_IP6.ip)
@contextlib.contextmanager
def inward_transfer():
    with pasta_configured() as (simnet, ns):
        yield simnet.gw, ns


@test_transfers(ip4=LOOPBACK4, ip6=LOOPBACK6, port=SPLICE_FWD_PORT,
                listenip4=LOOPBACK4, listenip6=LOOPBACK6)
@contextlib.contextmanager
def spliced_transfer():
    with pasta_configured() as (simnet, ns):
        yield ns, simnet.simhost