test_network_isolation.py 文件源码

def test_SNAT_rules(cluster):
    container_ids, container_ips, pods, specs = setup_pods(cluster)
    # --------- Test that SNAT rules are applied correctly --------
    jenkins_ip = _get_jenkins_ip(cluster)

    LOG.debug('{}Test that SNAT rules work properly{}'.format(
        Fore.CYAN, Style.RESET_ALL))
    LOG_MSG = "Check SNAT rules for pod '{}' public IP: '{}' host node: '{}'"

    BIND_IP = '0.0.0.0'

    POD_TCP_CMD = 'nc -z -v {} {}'.format(jenkins_ip, JENKINS_TCP_SERVER_PORT)
    POD_UDP_CMD = 'nc -u -z -v {} {}'.format(jenkins_ip,
                                             JENKINS_UDP_SERVER_PORT)

    for name, pod in pods.items():
        msg = LOG_MSG.format(name, pod.public_ip, specs[name]['host'])

        # Check if pod can ping jenkins
        ping(pod, container_ids[name], jenkins_ip)

        LOG.debug('{}TCP check {}{}'.format(Style.DIM, msg, Style.RESET_ALL))
        # Check if SNAT rules work properly for TCP connections
        with jenkins_accept_connections(
                SocketServer.TCPServer, MyRequestHandler, BIND_IP,
                JENKINS_TCP_SERVER_PORT) as connection_list:
            pod.docker_exec(container_ids[name], POD_TCP_CMD)
            _check_visible_ip(pod, specs, connection_list)

        LOG.debug('{}UDP check {}{}'.format(Style.DIM, msg, Style.RESET_ALL))
        # Check if SNAT rules work properly for UDP connections
        with jenkins_accept_connections(
                SocketServer.UDPServer, MyRequestHandler, BIND_IP,
                JENKINS_UDP_SERVER_PORT) as connection_list:
            pod.docker_exec(container_ids[name], POD_UDP_CMD)
            _check_visible_ip(pod, specs, connection_list)