<commit_before># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for :py:mod:`flocker.route._iptables`.
"""

from __future__ import print_function

from os import getuid
from socket import socket

from netifaces import AF_INET, interfaces, ifaddresses
from ipaddr import IPAddress

from twisted.trial.unittest import SkipTest, TestCase

from .. import create


def connect_nonblocking(ip, port):
    """
    Attempt a TCP connection to the given address without blocking.
    """
    client = socket()
    client.setblocking(False)
    client.connect_ex((ip.exploded, port))
    return client



def is_environment_configured():
    """
    Determine whether it is possible to exercise the proxy setup functionality
    in the current execution environment.

    :return: :obj:`True` if the proxy setup functionality could work given the
        underlying system and the privileges of this process, :obj:`False`
        otherwise.
    """
    # TODO: A nicer approach would be to create a new network namespace,
    # configure a couple interfaces with a couple addresses in it, and run the
    # test in the context of that network namespace.
    #
    # Something like:
    #
    #    ip netns create flocker-testing
    #    ip link add veth0 type veth peer name veth1
    #    ip link set veth1 netns flocker-testing
    #    ip netns exec flocker-testing ip link set dev veth1 up
    #    ip netns exec flocker-testing ip address add 10.0.0.1/24 dev veth1
    #    ip netns exec flocker-testing ip link set dev lo up
    #
    # Or, require such to be configured already.  That setup requires
    # privileged capabilities (probably CAP_SYS_ADMIN?) though.
    #
    # The functionality under test probably also requires privileged
    # capabilities (at least CAP_NET_ADMIN I think?) though.
    #
    # So for now just require root and crap on the host system. :/
    #
    # You might want to run these tests in a container! ;)
    #
    # -exarkun
    return getuid() == 0


class CreateTests(TestCase):
    """
    Tests for the creation of new external routing rules.
    """
    def setUp(self):
        """
        Select some addresses between which to proxy and set up a server to act
        as the target of the proxying.
        """
        if not is_environment_configured():
            raise SkipTest(
                "Cannot test port forwarding without suitable test environment.")

        self.addresses = [
            IPAddress(address['addr'])
            for name in interfaces()
            for address in ifaddresses(name).get(AF_INET, [])
        ]
        if len(self.addresses) < 2:
            raise SkipTest(
                "Cannot test proxying without at least two addresses.")

        self.serverAddress = self.addresses[0]
        self.proxyAddress = self.addresses[1]


        # This is the target of the proxy which will be created.
        self.server = socket()
        self.server.bind((self.serverAddress.exploded, 0))
        self.server.listen(1)<commit_msg>Use Python2.7 unittest's new skipping APIs.<commit_after># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

"""
Tests for :py:mod:`flocker.route._iptables`.
"""

from __future__ import print_function

from os import getuid
from socket import socket
from unittest import skipUnless

from netifaces import AF_INET, interfaces, ifaddresses
from ipaddr import IPAddress

from twisted.trial.unittest import TestCase

from .. import create


def connect_nonblocking(ip, port):
    """
    Attempt a TCP connection to the given address without blocking.
    """
    client = socket()
    client.setblocking(False)
    client.connect_ex((ip.exploded, port))
    return client



def is_environment_configured():
    """
    Determine whether it is possible to exercise the proxy setup functionality
    in the current execution environment.

    :return: :obj:`True` if the proxy setup functionality could work given the
        underlying system and the privileges of this process, :obj:`False`
        otherwise.
    """
    # TODO: A nicer approach would be to create a new network namespace,
    # configure a couple interfaces with a couple addresses in it, and run the
    # test in the context of that network namespace.
    #
    # Something like:
    #
    #    ip netns create flocker-testing
    #    ip link add veth0 type veth peer name veth1
    #    ip link set veth1 netns flocker-testing
    #    ip netns exec flocker-testing ip link set dev veth1 up
    #    ip netns exec flocker-testing ip address add 10.0.0.1/24 dev veth1
    #    ip netns exec flocker-testing ip link set dev lo up
    #
    # Or, require such to be configured already.  That setup requires
    # privileged capabilities (probably CAP_SYS_ADMIN?) though.
    #
    # The functionality under test probably also requires privileged
    # capabilities (at least CAP_NET_ADMIN I think?) though.
    #
    # So for now just require root and crap on the host system. :/
    #
    # You might want to run these tests in a container! ;)
    #
    # -exarkun
    return getuid() == 0


ADDRESSES = [
    IPAddress(address['addr'])
    for name in interfaces()
    for address in ifaddresses(name).get(AF_INET, [])
]


class CreateTests(TestCase):
    """
    Tests for the creation of new external routing rules.
    """
    @skipUnless(
        is_environment_configured(),
        "Cannot test port forwarding without suitable test environment.")
    @skipUnless(
        len(ADDRESSES) >= 2,
        "Cannot test proxying without at least two addresses.")
    def setUp(self):
        """
        Select some addresses between which to proxy and set up a server to act
        as the target of the proxying.
        """
        self.serverAddress = ADDRESSES[0]
        self.proxyAddress = ADDRESSES[1]


        # This is the target of the proxy which will be created.
        self.server = socket()
        self.server.bind((self.serverAddress.exploded, 0))
        self.server.listen(1)
{'fraction_non_alphanumeric': 0.07305708104509902, 'fraction_numerical': 0.006323847561990348, 'mean_word_length': 2.687116564417178, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 2, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5318163', 'n_tokens_mistral': 1686, 'n_tokens_neox': 1601, 'n_words': 668}