cloudflared/component-tests/test_edge_discovery.py at master · SmartCodeSmith/cloudflared · GitHub

History

165 lines (145 loc) · 6.91 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

import ipaddress

import socket

import pytest

from constants import protocols

from cli import CloudflaredCli

from util import get_tunnel_connector_id, LOGGER, wait_tunnel_ready, write_config

class TestEdgeDiscovery:

def _extra_config(self, protocol, edge_ip_version):

config = {

"protocol": protocol,

}

if edge_ip_version:

config["edge-ip-version"] = edge_ip_version

return config

@pytest.mark.parametrize("protocol", protocols())

def test_default_only(self, tmp_path, component_tests_config, protocol):

"""

This test runs a tunnel to connect via IPv4-only edge addresses (default is unset "--edge-ip-version 4")

"""

if self.has_ipv6_only():

pytest.skip("Host has IPv6 only support and current default is IPv4 only")

self.expect_address_connections(

tmp_path, component_tests_config, protocol, None, self.expect_ipv4_address)

@pytest.mark.parametrize("protocol", protocols())

def test_ipv4_only(self, tmp_path, component_tests_config, protocol):

"""

This test runs a tunnel to connect via IPv4-only edge addresses

"""

if self.has_ipv6_only():

pytest.skip("Host has IPv6 only support")

self.expect_address_connections(

tmp_path, component_tests_config, protocol, "4", self.expect_ipv4_address)

@pytest.mark.parametrize("protocol", protocols())

def test_ipv6_only(self, tmp_path, component_tests_config, protocol):

"""

This test runs a tunnel to connect via IPv6-only edge addresses

"""

if self.has_ipv4_only():

pytest.skip("Host has IPv4 only support")

self.expect_address_connections(

tmp_path, component_tests_config, protocol, "6", self.expect_ipv6_address)

@pytest.mark.parametrize("protocol", protocols())

def test_auto_ip64(self, tmp_path, component_tests_config, protocol):

"""

This test runs a tunnel to connect via auto with a preference of IPv6 then IPv4 addresses for a dual stack host

This test also assumes that the host has IPv6 preference.

"""

if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET6):

pytest.skip("Host does not support dual stack with IPv6 preference")

self.expect_address_connections(

tmp_path, component_tests_config, protocol, "auto", self.expect_ipv6_address)

@pytest.mark.parametrize("protocol", protocols())

def test_auto_ip46(self, tmp_path, component_tests_config, protocol):

"""

This test runs a tunnel to connect via auto with a preference of IPv4 then IPv6 addresses for a dual stack host

This test also assumes that the host has IPv4 preference.

"""

if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET):

pytest.skip("Host does not support dual stack with IPv4 preference")

self.expect_address_connections(

tmp_path, component_tests_config, protocol, "auto", self.expect_ipv4_address)

def expect_address_connections(self, tmp_path, component_tests_config, protocol, edge_ip_version, assert_address_type):

config = component_tests_config(

self._extra_config(protocol, edge_ip_version))

config_path = write_config(tmp_path, config.full_config)

LOGGER.debug(config)

with CloudflaredCli(config, config_path, LOGGER):

wait_tunnel_ready(tunnel_url=config.get_url(),

require_min_connections=4)

cfd_cli = CloudflaredCli(config, config_path, LOGGER)

tunnel_id = config.get_tunnel_id()

info = cfd_cli.get_tunnel_info(tunnel_id)

connector_id = get_tunnel_connector_id()

connector = next(

(c for c in info["conns"] if c["id"] == connector_id), None)

assert connector, f"Expected connection info from get tunnel info for the connected instance: {info}"

conns = connector["conns"]

assert conns == None or len(

conns) == 4, f"There should be 4 connections registered: {conns}"

for conn in conns:

origin_ip = conn["origin_ip"]

assert origin_ip, f"No available origin_ip for this connection: {conn}"

assert_address_type(origin_ip)

def expect_ipv4_address(self, address):

assert type(ipaddress.ip_address(

address)) is ipaddress.IPv4Address, f"Expected connection from origin to be a valid IPv4 address: {address}"

def expect_ipv6_address(self, address):

assert type(ipaddress.ip_address(

address)) is ipaddress.IPv6Address, f"Expected connection from origin to be a valid IPv6 address: {address}"

def get_addresses(self):

"""

Returns a list of addresses for the host.

"""

host_addresses = socket.getaddrinfo(

"region1.v2.argotunnel.com", 7844, socket.AF_UNSPEC, socket.SOCK_STREAM)

assert len(

host_addresses) > 0, "No addresses returned from getaddrinfo"

return host_addresses

def has_dual_stack(self, address_family_preference=None):

"""

Returns true if the host has dual stack support and can optionally check

the provided IP family preference.

"""

dual_stack = not self.has_ipv6_only() and not self.has_ipv4_only()

if address_family_preference:

address = self.get_addresses()[0]

return dual_stack and address[0] == address_family_preference

return dual_stack

def has_ipv6_only(self):

"""

Returns True if the host has only IPv6 address support.

"""

return self.attempt_connection(socket.AddressFamily.AF_INET6) and not self.attempt_connection(socket.AddressFamily.AF_INET)

def has_ipv4_only(self):

"""

Returns True if the host has only IPv4 address support.

"""

return self.attempt_connection(socket.AddressFamily.AF_INET) and not self.attempt_connection(socket.AddressFamily.AF_INET6)

def attempt_connection(self, address_family):

"""

Returns True if a successful socket connection can be made to the

remote host with the provided address family to validate host support

for the provided address family.

"""

address = None

for a in self.get_addresses():

if a[0] == address_family:

address = a

break

if address is None:

# Couldn't even lookup the address family so we can't connect

return False

af, socktype, proto, canonname, sockaddr = address

s = None

try:

s = socket.socket(af, socktype, proto)

except OSError:

return False

try:

s.connect(sockaddr)

except OSError:

s.close()

return False

s.close()

return True

cloudflared/component-tests/test_edge_discovery.py at master · SmartCodeSmith/cloudflared · GitHub

Expand file tree

Breadcrumbs

test_edge_discovery.py

Latest commit

History

Breadcrumbs

test_edge_discovery.py

File metadata and controls