framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase Class Reference

Inheritance diagram for framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase:

Classes
class	SecurityMode

Public Member Functions
def	assertClientCannotReachServer (self, XdsTestClient test_client)
def	assertClientCannotReachServerRepeatedly (self, XdsTestClient test_client, *Optional[int] times=None, Optional[_timedelta] delay=None)
def	assertClientChannelFailed (self, XdsTestClient test_client)
def	assertSecurityMtls (self, grpc_channelz.Security client_security, grpc_channelz.Security server_security)
def	assertSecurityPlaintext (self, client_security, server_security)
def	assertSecurityTls (self, grpc_channelz.Security client_security, grpc_channelz.Security server_security)
def	assertTestAppSecurity (self, SecurityMode mode, XdsTestClient test_client, XdsTestServer test_server)
def	debug_sock_certs (cls, grpc_channelz.Security security)
KubernetesClientRunner	initKubernetesClientRunner (self)
KubernetesServerRunner	initKubernetesServerRunner (self)
TrafficDirectorSecureManager	initTrafficDirectorManager (self)
def	setUpClass (cls)
def	setupSecurityPolicies (self, *server_tls, server_mtls, client_tls, client_mtls)
XdsTestClient	startSecureTestClient (self, XdsTestServer test_server, *wait_for_active_server_channel=True, **kwargs)
XdsTestServer	startSecureTestServer (self, replica_count=1, **kwargs)
Public Member Functions inherited from framework.xds_k8s_testcase.IsolatedXdsKubernetesTestCase
def	setUp (self)
def	tearDown (self)
Public Member Functions inherited from framework.xds_k8s_testcase.XdsKubernetesBaseTestCase
def	assertAllBackendsReceivedRpcs (self, lb_stats)
def	assertFailedRpcs (self, XdsTestClient test_client, Optional[int] num_rpcs=100)
def	assertRouteConfigUpdateTrafficHandoff (self, XdsTestClient test_client, str previous_route_config_version, int retry_wait_second, int timeout_second)
def	assertRpcsEventuallyGoToGivenServers (self, XdsTestClient test_client, List[XdsTestServer] servers, int num_rpcs=100)
None	assertRpcStatusCodes (self, XdsTestClient test_client, *grpc.StatusCode status_code, _timedelta duration, str method)
def	assertSuccessfulRpcs (self, XdsTestClient test_client, int num_rpcs=100)
def	assertXdsConfigExists (self, XdsTestClient test_client)
def	removeServerBackends (self, *server_runner=None)
def	setupServerBackends (self, *wait_for_healthy_status=True, server_runner=None, Optional[int] max_rate_per_endpoint=None)
def	setupTrafficDirectorGrpc (self)
def	tearDownClass (cls)

Static Public Member Functions
def	debug_cert (cert)
Tuple[grpc_channelz.Socket, grpc_channelz.Socket]	getConnectedSockets (XdsTestClient test_client, XdsTestServer test_server)
Static Public Member Functions inherited from framework.xds_k8s_testcase.XdsKubernetesBaseTestCase
def	diffAccumulatedStatsPerMethod (grpc_testing.LoadBalancerAccumulatedStatsResponse before, grpc_testing.LoadBalancerAccumulatedStatsResponse after)
LoadBalancerStatsResponse	getClientRpcStats (XdsTestClient test_client, int num_rpcs)
bool	is_supported (skips.TestConfig config)

Public Attributes
	server_maintenance_port
	server_namespace
Public Attributes inherited from framework.xds_k8s_testcase.IsolatedXdsKubernetesTestCase
	client_namespace
	client_runner
	resource_suffix
	server_namespace
	server_runner
	server_xds_port
	td
Public Attributes inherited from framework.xds_k8s_testcase.XdsKubernetesBaseTestCase
	check_local_certs
	client_image
	client_name
	client_port
	compute_api_version
	debug_use_port_forwarding
	enable_workload_identity
	ensure_firewall
	firewall_allowed_ports
	force_cleanup
	gcp_api_manager
	gcp_service_account
	k8s_api_manager
	network
	project
	resource_prefix
	resource_suffix
	resource_suffix_randomize
	secondary_k8s_api_manager
	server_image
	server_maintenance_port
	server_name
	server_port
	server_xds_host
	server_xds_port
	td_bootstrap_image
	xds_server_uri

Additional Inherited Members
Static Public Attributes inherited from framework.xds_k8s_testcase.XdsKubernetesBaseTestCase
	bool
	str

Detailed Description

Test case base class for testing PSM security features in isolation.

Definition at line 543 of file xds_k8s_testcase.py.

Member Function Documentation

assertClientCannotReachServer()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertClientCannotReachServer	(		self,
		XdsTestClient	test_client
	)

Definition at line 782 of file xds_k8s_testcase.py.

assertClientCannotReachServerRepeatedly()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertClientCannotReachServerRepeatedly	(		self,
		XdsTestClient	test_client,
		*Optional[int]	times = None,
		Optional[_timedelta]	delay = None
	)

Asserts that the client repeatedly cannot reach the server.

With negative tests we can't be absolutely certain expected failure
state is not caused by something else.
To mitigate for this, we repeat the checks several times, and expect
all of them to succeed.

This is useful in case the channel eventually stabilizes, and RPCs pass.

Args:
    test_client: An instance of XdsTestClient
    times: Optional; A positive number of times to confirm that
the server is unreachable. Defaults to `3` attempts.
    delay: Optional; Specifies how long to wait before the next check.
Defaults to `10` seconds.

Definition at line 747 of file xds_k8s_testcase.py.

assertClientChannelFailed()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertClientChannelFailed	(		self,
		XdsTestClient	test_client
	)

Definition at line 786 of file xds_k8s_testcase.py.

assertSecurityMtls()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertSecurityMtls	(		self,
		grpc_channelz.Security	client_security,
		grpc_channelz.Security	server_security
	)

Definition at line 661 of file xds_k8s_testcase.py.

assertSecurityPlaintext()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertSecurityPlaintext	(		self,
			client_security,
			server_security
	)

Definition at line 729 of file xds_k8s_testcase.py.

assertSecurityTls()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertSecurityTls	(		self,
		grpc_channelz.Security	client_security,
		grpc_channelz.Security	server_security
	)

Definition at line 697 of file xds_k8s_testcase.py.

assertTestAppSecurity()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.assertTestAppSecurity	(		self,
		SecurityMode	mode,
		XdsTestClient	test_client,
		XdsTestServer	test_server
	)

Definition at line 642 of file xds_k8s_testcase.py.

debug_cert()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.debug_cert (   cert )

static

Definition at line 813 of file xds_k8s_testcase.py.

debug_sock_certs()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.debug_sock_certs	(		cls,
		grpc_channelz.Security	security
	)

Definition at line 805 of file xds_k8s_testcase.py.

getConnectedSockets()

Tuple[grpc_channelz.Socket, grpc_channelz.Socket] framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.getConnectedSockets ( XdsTestClient  test_client, XdsTestServer   test_server  )

static

Definition at line 797 of file xds_k8s_testcase.py.

initKubernetesClientRunner()

KubernetesClientRunner framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.initKubernetesClientRunner

(

self

)

Reimplemented from framework.xds_k8s_testcase.IsolatedXdsKubernetesTestCase.

Definition at line 591 of file xds_k8s_testcase.py.

initKubernetesServerRunner()

KubernetesServerRunner framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.initKubernetesServerRunner

(

self

)

Reimplemented from framework.xds_k8s_testcase.IsolatedXdsKubernetesTestCase.

Definition at line 576 of file xds_k8s_testcase.py.

initTrafficDirectorManager()

TrafficDirectorSecureManager framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.initTrafficDirectorManager

(

self

)

Reimplemented from framework.xds_k8s_testcase.IsolatedXdsKubernetesTestCase.

Definition at line 567 of file xds_k8s_testcase.py.

setUpClass()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.setUpClass

(

cls

)

Hook method for setting up class fixture before running tests in
the class.

Reimplemented from framework.xds_k8s_testcase.XdsKubernetesBaseTestCase.

Definition at line 553 of file xds_k8s_testcase.py.

setupSecurityPolicies()

def framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.setupSecurityPolicies	(		self,
		*	server_tls,
			server_mtls,
			client_tls,
			client_mtls
	)

Definition at line 618 of file xds_k8s_testcase.py.

startSecureTestClient()

XdsTestClient framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.startSecureTestClient	(		self,
		XdsTestServer	test_server,
		*	wait_for_active_server_channel = True,
		**	kwargs
	)

Definition at line 630 of file xds_k8s_testcase.py.

startSecureTestServer()

XdsTestServer framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.startSecureTestServer	(		self,
			replica_count = 1,
		**	kwargs
	)

Definition at line 608 of file xds_k8s_testcase.py.

Member Data Documentation

server_maintenance_port

framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.server_maintenance_port

Definition at line 564 of file xds_k8s_testcase.py.

server_namespace

framework.xds_k8s_testcase.SecurityXdsKubernetesTestCase.server_namespace

Definition at line 605 of file xds_k8s_testcase.py.

---

The documentation for this class was generated from the following file:

• xds_k8s_testcase.py