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

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

from __future__ import print_function

from time import sleep
from errno import ECONNREFUSED
from os import getuid, getpid
from socket import error, socket
from unittest import skipUnless
from subprocess import check_call

from ipaddr import IPAddress, IPNetwork
from eliot.testing import LoggedAction, validateLogging, assertHasAction

from twisted.trial.unittest import TestCase

from .. import create_proxy_to, delete_proxy, enumerate_proxies
from .._logging import CREATE_PROXY_TO, DELETE_PROXY, IPTABLES
from .iptables import create_network_namespace, get_iptables_rules



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 create_user_rule():
    """
    Create an iptables rule which simulates an existing (or otherwise
    configured beyond flocker's control) rule on the system and needs to be
    ignored by :py:func:`enumerate_proxies`.
    """
    check_call([
            b"iptables",
            # Stick it in the PREROUTING chain based on our knowledge that the
            # implementation inspects this chain to enumerate proxies.
            b"--table", b"nat", b"--append", b"PREROUTING",

            b"--protocol", b"tcp", b"--dport", b"12345",
            b"--match", b"addrtype", b"--dst-type", b"LOCAL",

            b"--jump", b"DNAT", b"--to-destination", b"10.7.8.9",
            ])


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

<commit_msg>Get rid of redundant comment.<commit_after># Copyright Hybrid Logic Ltd.  See LICENSE file for details.

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

from __future__ import print_function

from time import sleep
from errno import ECONNREFUSED
from os import getuid, getpid
from socket import error, socket
from unittest import skipUnless
from subprocess import check_call

from ipaddr import IPAddress, IPNetwork
from eliot.testing import LoggedAction, validateLogging, assertHasAction

from twisted.trial.unittest import TestCase

from .. import create_proxy_to, delete_proxy, enumerate_proxies
from .._logging import CREATE_PROXY_TO, DELETE_PROXY, IPTABLES
from .iptables import create_network_namespace, get_iptables_rules



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 create_user_rule():
    """
    Create an iptables rule which simulates an existing (or otherwise
    configured beyond flocker's control) rule on the system and needs to be
    ignored by :py:func:`enumerate_proxies`.
    """
    check_call([
            b"iptables",
            # Stick it in the PREROUTING chain based on our knowledge that the
            # implementation inspects this chain to enumerate proxies.
            b"--table", b"nat", b"--append", b"PREROUTING",

            b"--protocol", b"tcp", b"--dport", b"12345",
            b"--match", b"addrtype", b"--dst-type", b"LOCAL",

            b"--jump", b"DNAT", b"--to-destination", b"10.7.8.9",
            ])


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.
    """
    return getuid() == 0


{'fraction_non_alphanumeric': 0.0841897233201581, 'fraction_numerical': 0.0067193675889328066, 'mean_word_length': 3.1348039215686274, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 1, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4256836', 'n_tokens_mistral': 1480, 'n_tokens_neox': 1401, 'n_words': 541}