<commit_before># encoding: utf-8

'''
@author: <NAME>
'''

import itertools

import pytest
import thutils
import tcconfig


@pytest.fixture
def subproc_wrapper():
    return thutils.subprocwrapper.SubprocessWrapper()


DEVICE = "eth0"


class NormalTestValue:
    RATE_LIST = [
        "",
        "--rate 100K",
        "--rate 0.5M",
        "--rate 0.1G",
    ]
    DELAY_LIST = [
        "",
        "--delay 0",
        "--delay 10",
        "--delay 100",
    ]
    LOSS_LIST = [
        "",
        "--loss 0",
        "--loss 0.1",
        "--loss 99",
    ]
    NETWORK_LIST = [
        "",
        "--network 192.168.0.10",
        "--network 192.168.0.10/24",
    ]
    PORT_LIST = [
        "",
        "--port 80",
    ]
    OVERWRITE_LIST = [
        "",
        "--overwrite",
    ]


class Test_tcconfig:

    # inappropriate test for Travis CI
    #   ruun locally: python setup.py test --addopts --runxfail
    @pytest.mark.xfail
    @pytest.mark.parametrize(["rate", "delay", "loss", "overwrite"], [
        opt_list
        for opt_list in itertools.product(
            NormalTestValue.RATE_LIST,
            NormalTestValue.DELAY_LIST,
            NormalTestValue.LOSS_LIST,
            NormalTestValue.OVERWRITE_LIST)
    ])
    def test_smoke(self, subproc_wrapper, rate, delay, loss, overwrite):
        option_list = [rate, delay, loss, overwrite]
        command = "tcset --device %s " % (DEVICE) + " ".join(option_list)
        assert subproc_wrapper.run(command) == 0

        command = "tcdel --device %s" % (DEVICE)
        assert subproc_wrapper.run(command) == 0


@pytest.fixture
def tc_obj():
    subproc_wrapper = thutils.subprocwrapper.SubprocessWrapper()
    return tcconfig.TrafficControl(subproc_wrapper, "eth0")


@pytest.mark.parametrize(["value"], [
    ["".join(opt_list)]
    for opt_list in itertools.product(
        ["0.1", "1", "2147483647"],
        [
            "k", " k", "K", " K",
            "m", " m", "M", " M",
            "g", " g", "G", " G",
        ]
    )
])
def test_TrafficControl_validate_rate_normal(tc_obj, value):
    tc_obj.rate = value
    tc_obj._TrafficControl__validate_rate()


@pytest.mark.parametrize(["value", "expected"], [
    ["".join(opt_list), ValueError]
    for opt_list in itertools.product(
        ["0.1", "1", "2147483647"],
        [
            "kb", "kbps", "KB",
            "mb", "mbps", "MB",
            "gb", "gbps", "GB"
        ]
    )
])
def test_TrafficControl_validate_rate_exception_1(tc_obj, value, expected):
    tc_obj.rate = value
    with pytest.raises(expected):
        tc_obj._TrafficControl__validate_rate()


@pytest.mark.parametrize(["value", "expected"], [
    ["".join(opt_list), ValueError]
    for opt_list in itertools.product(
        ["-1", "0", "0.0"],
        ["k", "K", "m", "M", "g", "G"]
    )
])
def test_TrafficControl_validate_rate_exception_2(tc_obj, value, expected):
    tc_obj.rate = value
    with pytest.raises(expected):
        tc_obj._TrafficControl__validate_rate()


class Test_TrafficControl_validate:

    @pytest.mark.parametrize(["rate", "delay", "loss", "network", "port"], [
        opt_list
        for opt_list in itertools.product(
            [None, "", "100K", "0.5M", "0.1G"],
            [None, 0, 10000],
            [None, 0, 99],
            [
                None,
                "",
                "192.168.0.1", "192.168.0.254",
                "192.168.0.1/32", "192.168.0.0/24"
            ],
            [None, 0, 65535],
        )
    ])
    def test_normal(self, tc_obj, rate, delay, loss, network, port):
        tc_obj.rate = rate
        tc_obj.delay_ms = delay
        tc_obj.loss_percent = loss
        tc_obj.network = network
        tc_obj.port = port

        tc_obj.validate()

    @pytest.mark.parametrize(["value", "expected"], [
        [{"delay_ms": -1}, ValueError],
        [{"delay_ms": 10001}, ValueError],

        [{"loss_percent": -0.1}, ValueError],
        [{"loss_percent": 99.1}, ValueError],

        [{"network": "192.168.0."}, ValueError],
        [{"network": "192.168.0.256"}, ValueError],
        [{"network": "192.168.0.0/0"}, ValueError],
        [{"network": "192.168.0.0/33"}, ValueError],
        [{"network": "192.168.0.2/24"}, ValueError],
        [{"network": "192.168.0.0000/24"}, ValueError],

        [{"port": -1}, ValueError],
        [{"port": 65536}, ValueError],
    ])
    def test_exception(self, tc_obj, value, expected):
        tc_obj.rate = value.get("rate")
        tc_obj.delay_ms = value.get("delay_ms")
        tc_obj.loss_percent = value.get("loss_percent")
        tc_obj.network = value.get("network")
        tc_obj.port = value.get("port")

        with pytest.raises(expected):
            tc_obj.validate()<commit_msg>Add support for incoming packet traffic control<commit_after># encoding: utf-8

'''
@author: <NAME>
'''

import itertools

import dataproperty
import pingparsing
import pytest
import thutils
import tcconfig


DEVICE = "eth0"
WAIT_TIME = 5  # [sec]


@pytest.fixture
def dst_host_option(request):
    return request.config.getoption("--dst-host")


@pytest.fixture
def dst_host_ex_option(request):
    return request.config.getoption("--dst-host-ex")


@pytest.fixture
def subproc_wrapper():
    return thutils.subprocwrapper.SubprocessWrapper()


@pytest.fixture
def pingparser():
    return pingparsing.PingParsing()


@pytest.fixture
def transmitter():
    transmitter = pingparsing.PingTransmitter()
    transmitter.ping_option = "-f -q"
    transmitter.waittime = WAIT_TIME

    return transmitter


class NormalTestValue:
    RATE_LIST = [
        "",
        "--rate 100K",
        "--rate 0.5M",
    ]
    DELAY_LIST = [
        "",
        "--delay 1",
        "--delay 100",
    ]
    LOSS_LIST = [
        "",
        "--loss 0.1",
        "--loss 99",
    ]
    DIRECTION_LIST = [
        "",
        "--direction outgoing",
        "--direction incoming",
    ]
    NETWORK_LIST = [
        "",
        "--network 192.168.0.10",
        "--network 192.168.0.0/24",
    ]
    PORT_LIST = [
        "",
        "--port 80",
    ]
    OVERWRITE_LIST = [
        "",
        "--overwrite",
    ]


class Test_tcconfig:
    """
    Inappropriate tests to Travis CI.
    Run locally with following command:
      python setup.py test --addopts --runxfail

    These tests expected to execute on Linux(debian) w/ iputils-ping package.
    """

    @pytest.mark.xfail
    @pytest.mark.parametrize(
        ["rate", "delay", "loss", "direction", "network", "port", "overwrite"],
        [
            opt_list
            for opt_list in itertools.product(
                NormalTestValue.RATE_LIST,
                NormalTestValue.DELAY_LIST,
                NormalTestValue.LOSS_LIST,
                NormalTestValue.DIRECTION_LIST,
                NormalTestValue.NETWORK_LIST,
                NormalTestValue.PORT_LIST,
                NormalTestValue.OVERWRITE_LIST)
        ])
    def test_smoke(
            self, subproc_wrapper, rate, delay, loss,
            direction, network, port, overwrite):
        command = " ".join([
            "tcset",
            "--device " + DEVICE,
            rate, delay, loss, direction, network, port, overwrite,
        ])
        assert subproc_wrapper.run(command) == 0

        assert subproc_wrapper.run("tcdel --device " + DEVICE) == 0


class Test_tcset_one_network:
    """
    Inappropriate tests to Travis CI.
    Run locally with following command:
      python setup.py test --addopts "--dst-host=<hostname or IP address>"

    These tests expected to execute on Linux(debian) w/ iputils-ping package.
    """

    @pytest.mark.parametrize(["delay"], [
        [100],
    ])
    def test_const_latency(
            self, dst_host_option, subproc_wrapper, transmitter, pingparser, delay):
        if dataproperty.is_empty_string(dst_host_option):
            # alternative to pytest.mark.skipif
            return

        subproc_wrapper.run("tcdel --device " + DEVICE)
        transmitter.destination_host = dst_host_option

        # w/o latency tc ---
        result = transmitter.ping()
        pingparser.parse(result)
        without_tc_rtt_avg = pingparser.rtt_avg

        # w/ latency tc ---
        command_list = [
            "tcset",
            "--device " + DEVICE,
            "--delay %d" % (delay),
        ]
        assert subproc_wrapper.run(" ".join(command_list)) == 0

        result = transmitter.ping()
        pingparser.parse(result)
        with_tc_rtt_avg = pingparser.rtt_avg

        # assertion ---
        rtt_diff = with_tc_rtt_avg - without_tc_rtt_avg
        assert rtt_diff > (delay / 2.0)

        # finalize ---
        subproc_wrapper.run("tcdel --device " + DEVICE)

    @pytest.mark.parametrize(["packet_loss"], [
        [10],
    ])
    def test_const_packet_loss(
            self, dst_host_option, subproc_wrapper,
            transmitter, pingparser, packet_loss):
        if dataproperty.is_empty_string(dst_host_option):
            # alternative to pytest.mark.skipif
            return

        subproc_wrapper.run("tcdel --device " + DEVICE)
        transmitter.destination_host = dst_host_option

        # w/o packet loss tc ---
        result = transmitter.ping()
        pingparser.parse(result)
        without_tc_loss = (
            pingparser.packet_receive / float(pingparser.packet_transmit)) * 100.0

        # w/ packet loss tc ---
        command_list = [
            "tcset",
            "--device " + DEVICE,
            "--loss %f" % (packet_loss),
        ]
        assert subproc_wrapper.run(" ".join(command_list)) == 0

        result = transmitter.ping()
        pingparser.parse(result)
        with_tc_loss = (
            pingparser.packet_receive / float(pingparser.packet_transmit)) * 100.0

        # assertion ---
        loss_diff = without_tc_loss - with_tc_loss
        assert loss_diff > (packet_loss / 2.0)

        # finalize ---
        subproc_wrapper.run("tcdel --device " + DEVICE)


class Test_tcset_two_network:
    """
    Inappropriate tests to Travis CI.
    Run locally with following command:
      python setup.py test --addopts \
        "--dst-host=<hostname or IP address> --dst-host-ex=<hostname or IP address>"

    These tests expected to execute on Linux(debian) w/ iputils-ping package.
    """

    def test_network(
            self, dst_host_option, dst_host_ex_option, subproc_wrapper,
            transmitter, pingparser):
        if any([
            dataproperty.is_empty_string(dst_host_option),
            dataproperty.is_empty_string(dst_host_ex_option),
        ]):
            # alternative to pytest.mark.skipif
            return

        subproc_wrapper.run("tcdel --device " + DEVICE)
        delay = 100

        # tc to specific network ---
        command_list = [
            "tcset",
            "--device " + DEVICE,
            "--delay %d" % (delay),
            "--network " + dst_host_ex_option,
        ]
        assert subproc_wrapper.run(" ".join(command_list)) == 0

        # w/o tc network ---
        transmitter.destination_host = dst_host_option
        result = transmitter.ping()
        pingparser.parse(result)
        without_tc_rtt_avg = pingparser.rtt_avg

        # w/ tc network ---
        transmitter.destination_host = dst_host_ex_option
        result = transmitter.ping()
        pingparser.parse(result)
        with_tc_rtt_avg = pingparser.rtt_avg

        # assertion ---
        rtt_diff = with_tc_rtt_avg - without_tc_rtt_avg
        assert rtt_diff > (delay / 2.0)

        # finalize ---
        subproc_wrapper.run("tcdel --device " + DEVICE)
{'fraction_non_alphanumeric': 0.13490013774104684, 'fraction_numerical': 0.02384641873278237, 'mean_word_length': 2.0403035854488354, 'pattern_counts': {'":': 12, '<': 8, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 13, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '316167', 'n_tokens_mistral': 3942, 'n_tokens_neox': 3724, 'n_words': 735}