From fada718edf5080fbebe15de0ed7a7b942a9bead3 Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Sun, 23 Aug 2020 17:38:11 -1000
Subject: [PATCH 1/7] Unittests and example for portforwarding ability added in
 python-base.

---
 examples/pod_portforward.py        | 123 +++++++++++++++++++++++++++++
 kubernetes/e2e_test/test_client.py | 120 +++++++++++++++++++++++++++-
 2 files changed, 242 insertions(+), 1 deletion(-)
 create mode 100644 examples/pod_portforward.py

diff --git a/examples/pod_portforward.py b/examples/pod_portforward.py
new file mode 100644
index 0000000000..438bf57fad
--- /dev/null
+++ b/examples/pod_portforward.py
@@ -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()
diff --git a/kubernetes/e2e_test/test_client.py b/kubernetes/e2e_test/test_client.py
index 5fd1b5e64b..480e6928a5 100644
--- a/kubernetes/e2e_test/test_client.py
+++ b/kubernetes/e2e_test/test_client.py
@@ -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 @@ def test_pod_apis(self):
 
         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 @@ def test_exit_code(self):
         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)

From 49f3b6e0b51f3aaa9b6ac0a487eff6fb0bde73f7 Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Sun, 6 Sep 2020 09:28:31 -1000
Subject: [PATCH 2/7] Rework port forwarding unittest and example.

---
 examples/pod_portforward.py        | 133 ++++++++++++++++++++++-------
 kubernetes/e2e_test/test_client.py |  80 +++++++++++------
 2 files changed, 155 insertions(+), 58 deletions(-)

diff --git a/examples/pod_portforward.py b/examples/pod_portforward.py
index 438bf57fad..8b8e299f80 100644
--- a/examples/pod_portforward.py
+++ b/examples/pod_portforward.py
@@ -16,6 +16,7 @@
 Shows the functionality of portforward streaming using an nginx container.
 """
 
+import select
 import socket
 import time
 import urllib.request
@@ -26,6 +27,35 @@
 from kubernetes.client.rest import ApiException
 from kubernetes.stream import portforward
 
+##############################################################################
+# Kubernetes pod port forwarding works by directly providing a socket which
+# the python application uses to send and receive data on. This is in contrast
+# to the go client, which opens a local port that the go application then has
+# to open to get a socket to transmit data.
+#
+# This simplifies the python application, there is not local port to worry
+# about if that port number is available. Nor does the python application have
+# to then deal with opening this local port. The socket used to transmit data
+# is immediately provided to the python application.
+#
+# Below also is an example of monkey patching the socket.create_connection
+# function so that DNS names of the following formats will access kubernetes
+# ports:
+#
+#    <pod-name>.<namespace>.kubernetes
+#    <pod-name>.pod.<namespace>.kubernetes
+#    <service-name>.svc.<namespace>.kubernetes
+#    <service-name>.service.<namespace>.kubernetes
+#
+# These DNS name can be used to interact with pod ports using python libraries,
+# such as urllib.request and http.client. For example:
+#
+# response = urllib.request.urlopen(
+#     'https://metrics-server.service.kube-system.kubernetes/'
+# )
+#
+##############################################################################
+
 
 def portforward_commands(api_instance):
     name = 'portforward-example'
@@ -53,8 +83,8 @@ def portforward_commands(api_instance):
                 }]
             }
         }
-        resp = api_instance.create_namespaced_pod(body=pod_manifest,
-                                                  namespace='default')
+        api_instance.create_namespaced_pod(body=pod_manifest,
+                                           namespace='default')
         while True:
             resp = api_instance.read_namespaced_pod(name=name,
                                                     namespace='default')
@@ -63,46 +93,87 @@ def portforward_commands(api_instance):
             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()
+    pf = portforward(
+        api_instance.connect_get_namespaced_pod_portforward,
+        name, 'default',
+        ports='80',
+    )
+    http = pf.socket(80)
+    http.setblocking(True)
+    http.sendall(b'GET / HTTP/1.1\r\n')
+    http.sendall(b'Host: 127.0.0.1\r\n')
+    http.sendall(b'Connection: close\r\n')
+    http.sendall(b'Accept: */*\r\n')
+    http.sendall(b'\r\n')
+    response = b''
+    while True:
+        select.select([http], [], [])
+        data = http.recv(1024)
+        if not data:
+            break
+        response += data
+    http.close()
+    print(response.decode('utf-8'))
+    error = pf.error(80)
+    if error is None:
+        print("No port forward errors on port 80.")
+    else:
+        print("Port 80 has the following error: %s" % error)
 
     # 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.
+    socket_create_connection = socket.create_connection
     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":
+        if dns_name[-1] != 'kubernetes':
             return socket_create_connection(address, *args, **kwargs)
+        if len(dns_name) not in (3, 4):
+            raise RuntimeError("Unexpected kubernetes DNS name.")
+        namespace = dns_name[-2]
+        name = dns_name[0]
+        port = address[1]
+        if len(dns_name) == 4:
+            if dns_name[1] in ('svc', 'service'):
+                service = api_instance.read_namespaced_service(name, namespace)
+                for service_port in service.spec.ports:
+                    if service_port.port == port:
+                        port = service_port.target_port
+                        break
+                else:
+                    raise RuntimeError("Unable to find service port: %s" % port)
+                label_selector = []
+                for key, value in service.spec.selector.items():
+                    label_selector.append("%s=%s" % (key, value))
+                pods = api_instance.list_namespaced_pod(
+                    namespace, label_selector=",".join(label_selector)
+                )
+                if not pods.items:
+                    raise RuntimeError("Unable to find service pods.")
+                name = pods.items[0].metadata.name
+                if isinstance(port, str):
+                    for container in pods.items[0].spec.containers:
+                        for container_port in container.ports:
+                            if container_port.name == port:
+                                port = container_port.container_port
+                                break
+                        else:
+                            continue
+                        break
+                    else:
+                        raise RuntimeError("Unable to find service port name: %s" % port)
+            elif dns_name[1] != 'pod':
+                raise RuntimeError("Unsupported resource type: %s" % dns_name[1])
         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
+                         name, namespace, ports=str(port))
+        return pf.socket(port)
     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)
+    # Access the nginx http server using the "<pod-name>.pod.<namespace>.kubernetes" dns name.
+    response = urllib.request.urlopen('http://%s.pod.default.kubernetes' % name)
     html = response.read().decode('utf-8')
     response.close()
     print('Status:', response.status)
@@ -111,9 +182,9 @@ def kubernetes_create_connection(address, *args, **kwargs):
 
 def main():
     config.load_kube_config()
-    c = Configuration()
+    c = Configuration.get_default_copy()
     c.assert_hostname = False
-    #Configuration.set_default(c)
+    Configuration.set_default(c)
     core_v1 = core_v1_api.CoreV1Api()
 
     portforward_commands(core_v1)
diff --git a/kubernetes/e2e_test/test_client.py b/kubernetes/e2e_test/test_client.py
index 480e6928a5..cb405531b0 100644
--- a/kubernetes/e2e_test/test_client.py
+++ b/kubernetes/e2e_test/test_client.py
@@ -13,6 +13,7 @@
 # under the License.
 
 import json
+import select
 import socket
 import time
 import unittest
@@ -167,7 +168,10 @@ def test_portforward_raw(self):
         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")
+        pod_manifest = manifest_with_command(
+            name,
+            'for port in 1234 1235;do ((while true;do nc -l -p $port -e /bin/cat; done)&);done;sleep 60',
+        )
         resp = api.create_namespaced_pod(body=pod_manifest,
                                          namespace='default')
         self.assertEqual(name, resp.metadata.name)
@@ -182,39 +186,61 @@ def test_portforward_raw(self):
                 break
             time.sleep(1)
 
-        pf1234 = portforward(api.connect_get_namespaced_pod_portforward,
+        pf = portforward(api.connect_get_namespaced_pod_portforward,
                          name, 'default',
-                         ports='1234')
-        sock1234 = pf1234.socket(1234)
-        sock1234.settimeout(1)
+                         ports='1234,1235')
+        sock1234 = pf.socket(1234)
+        sock1235 = pf.socket(1235)
+        sock1234.setblocking(True)
+        sock1235.setblocking(True)
         sent1234 = b'Test port 1234 forwarding...'
+        sent1235 = b'Test port 1235 forwarding...'
         sock1234.sendall(sent1234)
+        sock1235.sendall(sent1235)
         reply1234 = b''
+        reply1235 = b''
         while True:
-            try:
-                reply1234 += sock1234.recv(1024)
-            except socket.timeout:
+            rlist = []
+            if sock1234.fileno() != -1:
+                rlist.append(sock1234)
+            if sock1235.fileno() != -1:
+                rlist.append(sock1235)
+            if not rlist:
                 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:
+            r, _w, _x = select.select(rlist, [], [], 1)
+            if not r:
                 break
-        self.assertEqual(reply9999, sent9999)
-        sock9999.close()
-        self.assertIsNone(pf9999.error(9999))
+            if sock1234 in r:
+                data = sock1234.recv(1024)
+                if data:
+                    reply1234 += data
+                else:
+                    assert False, 'Unexpected sock1234 close'
+            if sock1235 in r:
+                data = sock1235.recv(1024)
+                if data:
+                    reply1235 += data
+                else:
+                    assert False, 'Unexpected sock1235 close'
+        self.assertEqual(reply1234, sent1234)
+        self.assertEqual(reply1235, sent1235)
+        for sock in (sock1234, sock1235):
+            sent = b'Another test using fileno %s' % str(sock.fileno()).encode()
+            sock.sendall(sent)
+            reply = b''
+            while True:
+                r, _w, _x = select.select([sock], [], [], 1)
+                if not r:
+                    break
+                data = sock.recv(1024)
+                if data:
+                    reply += data
+                else:
+                    assert False, 'Unexpected sock close'
+            self.assertEqual(reply, sent)
+            sock.close()
+        self.assertIsNone(pf.error(1234))
+        self.assertIsNone(pf.error(1235))
 
         resp = api.delete_namespaced_pod(name=name, body={},
                                          namespace='default')

From 8afcebdf8d8ba42bb9a1e9f7c84731603a3cfbb4 Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Mon, 7 Sep 2020 13:10:25 -1000
Subject: [PATCH 3/7] Add test that checks for portforward port error return
 value.

---
 examples/pod_portforward.py        |  2 +-
 kubernetes/e2e_test/test_client.py | 36 ++++++++++++++++++------------
 2 files changed, 23 insertions(+), 15 deletions(-)

diff --git a/examples/pod_portforward.py b/examples/pod_portforward.py
index 8b8e299f80..a6e20217cc 100644
--- a/examples/pod_portforward.py
+++ b/examples/pod_portforward.py
@@ -33,7 +33,7 @@
 # to the go client, which opens a local port that the go application then has
 # to open to get a socket to transmit data.
 #
-# This simplifies the python application, there is not local port to worry
+# This simplifies the python application, there is not a local port to worry
 # about if that port number is available. Nor does the python application have
 # to then deal with opening this local port. The socket used to transmit data
 # is immediately provided to the python application.
diff --git a/kubernetes/e2e_test/test_client.py b/kubernetes/e2e_test/test_client.py
index cb405531b0..95977b21b6 100644
--- a/kubernetes/e2e_test/test_client.py
+++ b/kubernetes/e2e_test/test_client.py
@@ -170,7 +170,11 @@ def test_portforward_raw(self):
         name = 'portforward-raw-' + short_uuid()
         pod_manifest = manifest_with_command(
             name,
-            'for port in 1234 1235;do ((while true;do nc -l -p $port -e /bin/cat; done)&);done;sleep 60',
+            ' '.join((
+                '((while true;do nc -l -p 1234 -e /bin/cat; done)&);',
+                '((while true;do nc -l -p 1235 -e /bin/cat; done)&);',
+                'sleep 60',
+            ))
         )
         resp = api.create_namespaced_pod(body=pod_manifest,
                                          namespace='default')
@@ -188,7 +192,8 @@ def test_portforward_raw(self):
 
         pf = portforward(api.connect_get_namespaced_pod_portforward,
                          name, 'default',
-                         ports='1234,1235')
+                         ports='1234,1235,1236')
+        self.assertTrue(pf.connected)
         sock1234 = pf.socket(1234)
         sock1235 = pf.socket(1235)
         sock1234.setblocking(True)
@@ -212,19 +217,23 @@ def test_portforward_raw(self):
                 break
             if sock1234 in r:
                 data = sock1234.recv(1024)
-                if data:
-                    reply1234 += data
-                else:
-                    assert False, 'Unexpected sock1234 close'
+                self.assertNotEqual(data, b'', "Unexpected socket close")
+                reply1234 += data
             if sock1235 in r:
                 data = sock1235.recv(1024)
-                if data:
-                    reply1235 += data
-                else:
-                    assert False, 'Unexpected sock1235 close'
+                self.assertNotEqual(data, b'', "Unexpected socket close")
+                reply1235 += data
         self.assertEqual(reply1234, sent1234)
         self.assertEqual(reply1235, sent1235)
+        self.assertTrue(pf.connected)
+
+        sock = pf.socket(1236)
+        self.assertRaises(BrokenPipeError, sock.sendall, b'This should fail...')
+        self.assertIsNotNone(pf.error(1236))
+        sock.close()
+
         for sock in (sock1234, sock1235):
+            self.assertTrue(pf.connected)
             sent = b'Another test using fileno %s' % str(sock.fileno()).encode()
             sock.sendall(sent)
             reply = b''
@@ -233,12 +242,11 @@ def test_portforward_raw(self):
                 if not r:
                     break
                 data = sock.recv(1024)
-                if data:
-                    reply += data
-                else:
-                    assert False, 'Unexpected sock close'
+                self.assertNotEqual(data, b'', "Unexpected socket close")
+                reply += data
             self.assertEqual(reply, sent)
             sock.close()
+        self.assertFalse(pf.connected)
         self.assertIsNone(pf.error(1234))
         self.assertIsNone(pf.error(1235))
 

From d9ec734a6c7618496d02d5591257da39bceec0b1 Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Mon, 7 Sep 2020 18:22:10 -1000
Subject: [PATCH 4/7] Fix tox errors.

---
 examples/pod_portforward.py | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git a/examples/pod_portforward.py b/examples/pod_portforward.py
index a6e20217cc..aa95ada3e7 100644
--- a/examples/pod_portforward.py
+++ b/examples/pod_portforward.py
@@ -124,6 +124,7 @@ def portforward_commands(api_instance):
     # urllib.request. The same can be done with urllib3.util.connection.create_connection
     # if the "requests" package is used.
     socket_create_connection = socket.create_connection
+
     def kubernetes_create_connection(address, *args, **kwargs):
         dns_name = address[0]
         if isinstance(dns_name, bytes):
@@ -144,7 +145,8 @@ def kubernetes_create_connection(address, *args, **kwargs):
                         port = service_port.target_port
                         break
                 else:
-                    raise RuntimeError("Unable to find service port: %s" % port)
+                    raise RuntimeError(
+                        "Unable to find service port: %s" % port)
                 label_selector = []
                 for key, value in service.spec.selector.items():
                     label_selector.append("%s=%s" % (key, value))
@@ -164,16 +166,21 @@ def kubernetes_create_connection(address, *args, **kwargs):
                             continue
                         break
                     else:
-                        raise RuntimeError("Unable to find service port name: %s" % port)
+                        raise RuntimeError(
+                            "Unable to find service port name: %s" % port)
             elif dns_name[1] != 'pod':
-                raise RuntimeError("Unsupported resource type: %s" % dns_name[1])
+                raise RuntimeError(
+                    "Unsupported resource type: %s" %
+                    dns_name[1])
         pf = portforward(api_instance.connect_get_namespaced_pod_portforward,
                          name, namespace, ports=str(port))
         return pf.socket(port)
     socket.create_connection = kubernetes_create_connection
 
-    # Access the nginx http server using the "<pod-name>.pod.<namespace>.kubernetes" dns name.
-    response = urllib.request.urlopen('http://%s.pod.default.kubernetes' % name)
+    # Access the nginx http server using the
+    # "<pod-name>.pod.<namespace>.kubernetes" dns name.
+    response = urllib.request.urlopen(
+        'http://%s.pod.default.kubernetes' % name)
     html = response.read().decode('utf-8')
     response.close()
     print('Status:', response.status)

From c1249c9cf230af2b36a1883d5f919dae848c1a53 Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Mon, 7 Sep 2020 19:43:09 -1000
Subject: [PATCH 5/7] Support both python 2.7 and 3.x.

---
 examples/pod_portforward.py        | 7 ++++---
 kubernetes/e2e_test/test_client.py | 9 +++++----
 2 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/examples/pod_portforward.py b/examples/pod_portforward.py
index aa95ada3e7..9793fd3117 100644
--- a/examples/pod_portforward.py
+++ b/examples/pod_portforward.py
@@ -19,7 +19,8 @@
 import select
 import socket
 import time
-import urllib.request
+
+import six.moves.urllib.request as urllib_request
 
 from kubernetes import config
 from kubernetes.client import Configuration
@@ -179,11 +180,11 @@ def kubernetes_create_connection(address, *args, **kwargs):
 
     # Access the nginx http server using the
     # "<pod-name>.pod.<namespace>.kubernetes" dns name.
-    response = urllib.request.urlopen(
+    response = urllib_request.urlopen(
         'http://%s.pod.default.kubernetes' % name)
     html = response.read().decode('utf-8')
     response.close()
-    print('Status:', response.status)
+    print('Status Code: %s' % response.code)
     print(html)
 
 
diff --git a/kubernetes/e2e_test/test_client.py b/kubernetes/e2e_test/test_client.py
index 95977b21b6..00e46a3776 100644
--- a/kubernetes/e2e_test/test_client.py
+++ b/kubernetes/e2e_test/test_client.py
@@ -17,7 +17,6 @@
 import socket
 import time
 import unittest
-import urllib.request
 import uuid
 
 from kubernetes.client import api_client
@@ -26,6 +25,7 @@
 from kubernetes.stream import stream, portforward
 from kubernetes.stream.ws_client import ERROR_CHANNEL
 
+import six.moves.urllib.request as urllib_request
 
 def short_uuid():
     id = str(uuid.uuid4())
@@ -228,7 +228,7 @@ def test_portforward_raw(self):
         self.assertTrue(pf.connected)
 
         sock = pf.socket(1236)
-        self.assertRaises(BrokenPipeError, sock.sendall, b'This should fail...')
+        self.assertRaises(socket.error, sock.sendall, b'This should fail...')
         self.assertIsNotNone(pf.error(1236))
         sock.close()
 
@@ -246,6 +246,7 @@ def test_portforward_raw(self):
                 reply += data
             self.assertEqual(reply, sent)
             sock.close()
+        time.sleep(1)
         self.assertFalse(pf.connected)
         self.assertIsNone(pf.error(1234))
         self.assertIsNone(pf.error(1235))
@@ -300,12 +301,12 @@ def kubernetes_create_connection(address, *args, **kwargs):
         socket_create_connection = socket.create_connection
         try:
             socket.create_connection = kubernetes_create_connection
-            response = urllib.request.urlopen('http://%s.default.kubernetes/' % name)
+            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.assertEqual(response.code, 200)
         self.assertTrue('<h1>Welcome to nginx!</h1>' in html)
 
         resp = api.delete_namespaced_pod(name=name, body={},

From 3da49a207c4f5b0ccae2770cbbb74a40511d55ff Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Tue, 8 Sep 2020 16:11:25 -1000
Subject: [PATCH 6/7] Update kubernetes/base to latest master with
 portforwarding implementation.

---
 kubernetes/base | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/kubernetes/base b/kubernetes/base
index 54d188f89e..3dc7fe0b92 160000
--- a/kubernetes/base
+++ b/kubernetes/base
@@ -1 +1 @@
-Subproject commit 54d188f89e462e4c829aa9acb44d5c09cc5030ae
+Subproject commit 3dc7fe0b92cdb1fb06b565dc6a58d29b202701e1

From b1dd9c4f780bb86d0687c90132adf99e414458ab Mon Sep 17 00:00:00 2001
From: "Patrick J. McNerthney" <pat@mcnerthney.com>
Date: Tue, 8 Sep 2020 17:16:17 -1000
Subject: [PATCH 7/7] Install socat on Kubernetes node for use by port
 forwarding.

---
 scripts/kube-init.sh | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/scripts/kube-init.sh b/scripts/kube-init.sh
index ec2ecd2b27..f00fcbeea5 100755
--- a/scripts/kube-init.sh
+++ b/scripts/kube-init.sh
@@ -35,9 +35,10 @@ setenforce 0
 HOME=/home/travis
 sudo mount --make-rshared /
 
-# Install conntrack (required by minikube/K8s 1.18+)
+# Install conntrack (required by minikube/K8s 1.18+),
+# and socat, which is required for port forwarding.
 sudo apt-get update
-sudo apt-get install -y conntrack
+sudo apt-get install -y conntrack socat
 
 # Install docker if needed
 path_to_executable=$(which docker)