Unittests and example for portforwarding ability added in python-base.
This commit is contained in:
parent
ae3bce77fd
commit
d12a9a5717
123
examples/pod_portforward.py
Normal file
123
examples/pod_portforward.py
Normal file
@ -0,0 +1,123 @@
|
||||
# Copyright 2020 The Kubernetes Authors.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""
|
||||
Shows the functionality of portforward streaming using an nginx container.
|
||||
"""
|
||||
|
||||
import socket
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
from kubernetes import config
|
||||
from kubernetes.client import Configuration
|
||||
from kubernetes.client.api import core_v1_api
|
||||
from kubernetes.client.rest import ApiException
|
||||
from kubernetes.stream import portforward
|
||||
|
||||
|
||||
def portforward_commands(api_instance):
|
||||
name = 'portforward-example'
|
||||
resp = None
|
||||
try:
|
||||
resp = api_instance.read_namespaced_pod(name=name,
|
||||
namespace='default')
|
||||
except ApiException as e:
|
||||
if e.status != 404:
|
||||
print("Unknown error: %s" % e)
|
||||
exit(1)
|
||||
|
||||
if not resp:
|
||||
print("Pod %s does not exist. Creating it..." % name)
|
||||
pod_manifest = {
|
||||
'apiVersion': 'v1',
|
||||
'kind': 'Pod',
|
||||
'metadata': {
|
||||
'name': name
|
||||
},
|
||||
'spec': {
|
||||
'containers': [{
|
||||
'image': 'nginx',
|
||||
'name': 'nginx',
|
||||
}]
|
||||
}
|
||||
}
|
||||
resp = api_instance.create_namespaced_pod(body=pod_manifest,
|
||||
namespace='default')
|
||||
while True:
|
||||
resp = api_instance.read_namespaced_pod(name=name,
|
||||
namespace='default')
|
||||
if resp.status.phase != 'Pending':
|
||||
break
|
||||
time.sleep(1)
|
||||
print("Done.")
|
||||
|
||||
pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
|
||||
name, 'default',
|
||||
ports='80,8080:80')
|
||||
for port in (80, 8080):
|
||||
http = pf.socket(port)
|
||||
http.settimeout(1)
|
||||
http.sendall(b'GET / HTTP/1.1\r\n')
|
||||
http.sendall(b'Host: 127.0.0.1\r\n')
|
||||
http.sendall(b'Accept: */*\r\n')
|
||||
http.sendall(b'\r\n')
|
||||
response = b''
|
||||
while True:
|
||||
try:
|
||||
response += http.recv(1024)
|
||||
except socket.timeout:
|
||||
break
|
||||
print(response.decode('utf-8'))
|
||||
http.close()
|
||||
|
||||
# Monkey patch socket.create_connection which is used by http.client and
|
||||
# urllib.request. The same can be done with urllib3.util.connection.create_connection
|
||||
# if the "requests" package is used.
|
||||
def kubernetes_create_connection(address, *args, **kwargs):
|
||||
dns_name = address[0]
|
||||
if isinstance(dns_name, bytes):
|
||||
dns_name = dns_name.decode()
|
||||
# Look for "<pod-name>.<namspace>.kubernetes" dns names and if found
|
||||
# provide a socket that is port forwarded to the kuberntest pod.
|
||||
dns_name = dns_name.split(".")
|
||||
if len(dns_name) != 3 or dns_name[2] != "kubernetes":
|
||||
return socket_create_connection(address, *args, **kwargs)
|
||||
pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
|
||||
dns_name[0], dns_name[1], ports=str(address[1]))
|
||||
return pf.socket(address[1])
|
||||
|
||||
socket_create_connection = socket.create_connection
|
||||
socket.create_connection = kubernetes_create_connection
|
||||
|
||||
# Access the nginx http server using the "<pod-name>.<namespace>.kubernetes" dns name.
|
||||
response = urllib.request.urlopen('http://%s.default.kubernetes' % name)
|
||||
html = response.read().decode('utf-8')
|
||||
response.close()
|
||||
print('Status:', response.status)
|
||||
print(html)
|
||||
|
||||
|
||||
def main():
|
||||
config.load_kube_config()
|
||||
c = Configuration()
|
||||
c.assert_hostname = False
|
||||
#Configuration.set_default(c)
|
||||
core_v1 = core_v1_api.CoreV1Api()
|
||||
|
||||
portforward_commands(core_v1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@ -13,14 +13,16 @@
|
||||
# under the License.
|
||||
|
||||
import json
|
||||
import socket
|
||||
import time
|
||||
import unittest
|
||||
import urllib.request
|
||||
import uuid
|
||||
|
||||
from kubernetes.client import api_client
|
||||
from kubernetes.client.api import core_v1_api
|
||||
from kubernetes.e2e_test import base
|
||||
from kubernetes.stream import stream
|
||||
from kubernetes.stream import stream, portforward
|
||||
from kubernetes.stream.ws_client import ERROR_CHANNEL
|
||||
|
||||
|
||||
@ -119,6 +121,7 @@ class TestClient(unittest.TestCase):
|
||||
|
||||
resp = api.delete_namespaced_pod(name=name, body={},
|
||||
namespace='default')
|
||||
|
||||
def test_exit_code(self):
|
||||
client = api_client.ApiClient(configuration=self.config)
|
||||
api = core_v1_api.CoreV1Api(client)
|
||||
@ -159,6 +162,121 @@ class TestClient(unittest.TestCase):
|
||||
resp = api.delete_namespaced_pod(name=name, body={},
|
||||
namespace='default')
|
||||
|
||||
def test_portforward_raw(self):
|
||||
client = api_client.ApiClient(configuration=self.config)
|
||||
api = core_v1_api.CoreV1Api(client)
|
||||
|
||||
name = 'portforward-raw-' + short_uuid()
|
||||
pod_manifest = manifest_with_command(name, "while true;do nc -l -p 1234 -e /bin/cat; done")
|
||||
resp = api.create_namespaced_pod(body=pod_manifest,
|
||||
namespace='default')
|
||||
self.assertEqual(name, resp.metadata.name)
|
||||
self.assertTrue(resp.status.phase)
|
||||
|
||||
while True:
|
||||
resp = api.read_namespaced_pod(name=name,
|
||||
namespace='default')
|
||||
self.assertEqual(name, resp.metadata.name)
|
||||
self.assertTrue(resp.status.phase)
|
||||
if resp.status.phase != 'Pending':
|
||||
break
|
||||
time.sleep(1)
|
||||
|
||||
pf1234 = portforward(api.connect_get_namespaced_pod_portforward,
|
||||
name, 'default',
|
||||
ports='1234')
|
||||
sock1234 = pf1234.socket(1234)
|
||||
sock1234.settimeout(1)
|
||||
sent1234 = b'Test port 1234 forwarding...'
|
||||
sock1234.sendall(sent1234)
|
||||
reply1234 = b''
|
||||
while True:
|
||||
try:
|
||||
reply1234 += sock1234.recv(1024)
|
||||
except socket.timeout:
|
||||
break
|
||||
sock1234.close()
|
||||
self.assertEqual(reply1234, sent1234)
|
||||
self.assertIsNone(pf1234.error(1234))
|
||||
|
||||
pf9999 = portforward(api.connect_get_namespaced_pod_portforward,
|
||||
name, 'default',
|
||||
ports='9999:1234')
|
||||
sock9999 = pf9999.socket(9999)
|
||||
sock9999.settimeout(1)
|
||||
sent9999 = b'Test port 9999 forwarding...'
|
||||
sock9999.sendall(sent9999)
|
||||
reply9999 = b''
|
||||
while True:
|
||||
try:
|
||||
reply9999 += sock9999.recv(1024)
|
||||
except socket.timeout:
|
||||
break
|
||||
self.assertEqual(reply9999, sent9999)
|
||||
sock9999.close()
|
||||
self.assertIsNone(pf9999.error(9999))
|
||||
|
||||
resp = api.delete_namespaced_pod(name=name, body={},
|
||||
namespace='default')
|
||||
|
||||
def test_portforward_http(self):
|
||||
client = api_client.ApiClient(configuration=self.config)
|
||||
api = core_v1_api.CoreV1Api(client)
|
||||
|
||||
name = 'portforward-http-' + short_uuid()
|
||||
pod_manifest = {
|
||||
'apiVersion': 'v1',
|
||||
'kind': 'Pod',
|
||||
'metadata': {
|
||||
'name': name
|
||||
},
|
||||
'spec': {
|
||||
'containers': [{
|
||||
'name': 'nginx',
|
||||
'image': 'nginx',
|
||||
}]
|
||||
}
|
||||
}
|
||||
|
||||
resp = api.create_namespaced_pod(body=pod_manifest,
|
||||
namespace='default')
|
||||
self.assertEqual(name, resp.metadata.name)
|
||||
self.assertTrue(resp.status.phase)
|
||||
|
||||
while True:
|
||||
resp = api.read_namespaced_pod(name=name,
|
||||
namespace='default')
|
||||
self.assertEqual(name, resp.metadata.name)
|
||||
self.assertTrue(resp.status.phase)
|
||||
if resp.status.phase != 'Pending':
|
||||
break
|
||||
time.sleep(1)
|
||||
|
||||
def kubernetes_create_connection(address, *args, **kwargs):
|
||||
dns_name = address[0]
|
||||
if isinstance(dns_name, bytes):
|
||||
dns_name = dns_name.decode()
|
||||
dns_name = dns_name.split(".")
|
||||
if len(dns_name) != 3 or dns_name[2] != "kubernetes":
|
||||
return socket_create_connection(address, *args, **kwargs)
|
||||
pf = portforward(api.connect_get_namespaced_pod_portforward,
|
||||
dns_name[0], dns_name[1], ports=str(address[1]))
|
||||
return pf.socket(address[1])
|
||||
|
||||
socket_create_connection = socket.create_connection
|
||||
try:
|
||||
socket.create_connection = kubernetes_create_connection
|
||||
response = urllib.request.urlopen('http://%s.default.kubernetes/' % name)
|
||||
html = response.read().decode('utf-8')
|
||||
finally:
|
||||
socket.create_connection = socket_create_connection
|
||||
|
||||
self.assertEqual(response.status, 200)
|
||||
self.assertTrue('<h1>Welcome to nginx!</h1>' in html)
|
||||
|
||||
resp = api.delete_namespaced_pod(name=name, body={},
|
||||
namespace='default')
|
||||
|
||||
def test_service_apis(self):
|
||||
client = api_client.ApiClient(configuration=self.config)
|
||||
api = core_v1_api.CoreV1Api(client)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user