#! /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/site.py - Manage simulated network sites for testing
#
# Copyright Red Hat
# Author: David Gibson <david@gibson.dropbear.id.au>

import ipaddress
import json

import avocado
from avocado.utils.process import CmdError

from tasst import Tasst


class Site:
    """
    A (usually virtual or simulated) location where we can execute
    commands and configure networks.

    """

    def __init__(self, name):
        self.name = name # For debugging

    # Shut down the site and release any resources it's using.  We
    # can't use __del__() for this (RAII like), because Python doesn't
    # guarantee that will get called, and that's pretty easy to hit in
    # practice.  The modern Pythonic way of doing this is 'with' and
    # ContextManagers, but that doesn't work with Avocado's jUnit
    # derived format for setUp() and tearDown().  Oh well, do it
    # manually.
    def close(self):
        pass

    def output(self, cmd, **kwargs):
        raise NotImplementedError

    def fg(self, cmd, **kwargs):
        self.output(cmd, **kwargs)

    def bg(self, cmd, **kwargs):
        raise NotImplementedError

    def require_cmds(self, *cmds):
        missing = [c for c in cmds
                   if self.fg('type {}'.format(c), ignore_status=True) != 0]
        if missing:
            raise avocado.TestCancel("Missing commands {} on {}"
                                     .format(', '.join(missing), self.name))

    def ifs(self):
        self.require_cmds('ip')
        info = json.loads(self.output('ip -j link show'))
        return [i['ifname'] for i in info]

    def ifup(self, ifname):
        self.require_cmds('ip')
        self.fg('ip link set {} up'.format(ifname), sudo=True)

    def addrinfos(self, ifname, **filter):
        self.require_cmds('ip')
        info = json.loads(self.output('ip -j addr show {}'.format(ifname)))
        assert len(info) == 1 # We specified a specific interface

        ais = [ai for ai in info[0]['addr_info']]
        for key, value in filter.items():
            ais = [ai for ai in ais if key in ai and ai[key] == value]

        return ais

    def addrs(self, ifname, **filter):
        self.require_cmds('ip')
        # Return just the parsed, non-tentative addresses
        return [ipaddress.ip_interface('{}/{}'.format(ai['local'], ai['prefixlen']))
                for ai in self.addrinfos(ifname, **filter)
                if not 'tentative' in ai]


class BaseSiteTasst(Tasst):
    """
    Basic tests for executing commands on sites

    :avocado: disable
    :avocado: tags=meta
    """
    timeout = 1.0

    # Derived classes must call this from setUp()
    def subsetup(self, site):
        assert isinstance(site, Site)
        site.require_cmds('true', 'false', 'echo')
        Tasst.subsetup(self, BaseSiteTasst, site)

    def test_true(self):
        site = self.get_subsetup(BaseSiteTasst)
        site.fg('true')

    def test_false(self):
        site = self.get_subsetup(BaseSiteTasst)
        self.assertRaises(CmdError, site.fg, 'false')

    def test_echo(self):
        site = self.get_subsetup(BaseSiteTasst)
        s = 'Hello tasst'
        out = site.output('echo {}'.format(s))
        self.assertEquals(out, s.encode('utf-8'))

    def test_bg_true(self):
        site = self.get_subsetup(BaseSiteTasst)
        proc = site.bg('true')
        status = proc.wait()
        self.assertEquals(status, 0)

    def test_bg_false(self):
        site = self.get_subsetup(BaseSiteTasst)
        proc = site.bg('false')
        status = proc.wait()
        self.assertNotEquals(status, 0)

    def test_has_lo(self):
        site = self.get_subsetup(BaseSiteTasst)
        self.assertIn('lo', site.ifs())

    def test_lo_addrs(self):
        site = self.get_subsetup(BaseSiteTasst)
        expected = [ipaddress.ip_interface(a)
                    for a in ['127.0.0.1/8', '::1/128']]
        self.assertCountEqual(site.addrs('lo'), expected)


# Represents the host on which the tests are running, as opposed to
# some simulated host created by the tests
class RealHost(Site):
    def __init__(self):
        super().__init__('REAL_HOST')

    def output(self, cmd, sudo=False, **kwargs):
        assert not sudo, "BUG: Shouldn't run commands with privilege on host"
        return avocado.utils.process.system_output(cmd, **kwargs)

    def fg(self, cmd, sudo=False, **kwargs):
        assert not sudo, "BUG: Shouldn't run commands with privilege on host"
        return avocado.utils.process.system(cmd, **kwargs)

    def bg(self, cmd, sudo=False, **kwargs):
        assert not sudo, "BUG: Shouldn't run commands with privilege on host"
        proc = avocado.utils.process.SubProcess(cmd, **kwargs)
        proc.start()
        return proc


REAL_HOST = RealHost()


class RealHostTasst(BaseSiteTasst):
    def setUp(self):
        super().setUp()
        BaseSiteTasst.subsetup(self, REAL_HOST)