Implement back end raw port test container using python.

kubernetes/e2e_test/port_server.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+
+import select
+import socketserver
+import sys
+import threading
+import time
+
+
+class PortServer:
+    def __init__(self, port):
+        self.port = port
+        self.server = socketserver.ThreadingTCPServer(('0.0.0.0', port), self.handler)
+        self.server.daemon_threads = True
+        self.thread = threading.Thread(target=self.server.serve_forever,
+                                       name='Port %s Server' % port)
+        self.thread.daemon = True
+        self.thread.start()
+
+
+    def handler(self, request, address, server):
+        threading.current_thread().name = 'Port %s Handler' % self.port
+        rlist = [request]
+        echo = b''
+        while True:
+            r, w, _x = select.select(rlist, [request] if echo else [], [])
+            if r:
+                data = request.recv(1024)
+                if not data:
+                    break
+                echo += data
+            if w:
+                echo = echo[request.send(echo):]
+
+
+if __name__ == '__main__':
+    ports = []
+    for port in sys.argv[1:]:
+        ports.append(PortServer(int(port)))
+    time.sleep(10 * 60)
+

kubernetes/e2e_test/test_client.py

@@ -13,6 +13,7 @@
 # under the License.
 
 import json
+import os
 import select
 import socket
 import time
@@ -167,17 +168,64 @@ def test_portforward_raw(self):
         client = api_client.ApiClient(configuration=self.config)
         api = core_v1_api.CoreV1Api(client)
 
+        with open(os.path.join(os.path.dirname(__file__), 'port_server.py')) as fh:
+            port_server_py = fh.read()
         name = 'portforward-raw-' + short_uuid()
-        pod_manifest = manifest_with_command(
-            name,
-            ' '.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_config_map(
+            body={
+                'apiVersion': 'v1',
+                'kind': 'ConfigMap',
+                'metadata': {
+                    'name': name,
+                },
+                'data': {
+                    'port-server.py': port_server_py,
+                }
+            },
+            namespace='default',
+        )
+        resp = api.create_namespaced_pod(
+            body={
+                'apiVersion': 'v1',
+                'kind': 'Pod',
+                'metadata': {
+                    'name': name
+                },
+                'spec': {
+                    'containers': [
+                        {
+                            'name': 'port-server',
+                            'image': 'python',
+                            'command': [
+                                '/opt/port-server.py', '1234', '1235',
+                            ],
+                            'volumeMounts': [
+                                {
+                                    'name': 'port-server',
+                                    'mountPath': '/opt',
+                                    'readOnly': True,
+                                },
+                            ],
+                            'startupProbe': {
+                                'tcpSocket': {
+                                    'port': 1234,
+                                },
+                            },
+                        },
+                    ],
+                    'volumes': [
+                        {
+                            'name': 'port-server',
+                            'configMap': {
+                                'name': name,
+                                'defaultMode': 0o777,
+                            },
+                        },
+                    ],
+                },
+            },
+            namespace='default',
         )
-        resp = api.create_namespaced_pod(body=pod_manifest,
-                                         namespace='default')
         self.assertEqual(name, resp.metadata.name)
         self.assertTrue(resp.status.phase)
 
@@ -189,6 +237,7 @@ def test_portforward_raw(self):
             if resp.status.phase != 'Pending':
                 break
             time.sleep(1)
+        self.assertEqual(resp.status.phase, 'Running')
 
         pf = portforward(api.connect_get_namespaced_pod_portforward,
                          name, 'default',
@@ -251,8 +300,8 @@ def test_portforward_raw(self):
         self.assertIsNone(pf.error(1234))
         self.assertIsNone(pf.error(1235))
 
-        resp = api.delete_namespaced_pod(name=name, body={},
-                                         namespace='default')
+        resp = api.delete_namespaced_pod(name=name, namespace='default')
+        resp = api.delete_namespaced_config_map(name=name, namespace='default')
 
     def test_portforward_http(self):
         client = api_client.ApiClient(configuration=self.config)