Using sockets instead of netcat

share/trick/pymods/trick/tests/civet_server/test_http.py

@@ -18,15 +18,18 @@ path.append("../..")
 from utils import is_web_server_started, params
 
 def open_connections(numConnections):
-    processes = []
+    sockets = []
     for _ in range(numConnections):
-        processes.append(subprocess.Popen(f"nc -s 127.0.0.1 localhost {params.get_var_server_port()}".split())) #Connect to webserver to establish connections.
+        sockets.append(socket.socket(socket.AF_INET, socket.SOCK_STREAM))
+        sockets[-1].connect(("127.0.0.1", params.get_var_server_port()))
     sleep(1) #Wait for the connection to persist.
-    return processes
+    return sockets
 
-def kill_processes(processes):
-    for p in processes:
-        p.kill()
+def close_sockets(sockets):
+    for s in sockets:
+        port = s.getsockname()[1]
+        s.close()
+        # is_web_server_started(port, "")
     sleep(1)
 
 def get_vs_open_connections():
@@ -65,15 +68,15 @@ class TestWebserverHttp:
         assert res.json()["alloc_total"] == 48, "Expecting 48 memory allocations."
 
     def test_vs_connections(self):
-        processes = open_connections(1)
+        sockets = open_connections(1)
         webResponse=get_vs_open_connections()
-        kill_processes(processes)
+        close_sockets(sockets)
         assert webResponse.json()["variable_server_connections"][0]["connection"]["client_IP_address"] == "127.0.0.1"
         assert len(webResponse.json()["variable_server_connections"]) == 1
 
-        processes = open_connections(50)
+        sockets = open_connections(50)
         webResponse=get_vs_open_connections()
-        kill_processes(processes)
+        close_sockets(sockets)
         assert webResponse.json()["variable_server_connections"][0]["connection"]["client_IP_address"] == "127.0.0.1"
         assert len(webResponse.json()["variable_server_connections"]) == 50, "Should be able to open more than 1 connection." #Todo: determine appropriate number of simultaneous connections to test
 

share/trick/pymods/trick/utils.py

@@ -85,12 +85,12 @@ class Params:
 
 params = Params()
 
-def is_web_server_started():
+def is_web_server_started(port=params.get_port(), status_method="LISTEN"):
 	isConnectionOpen = False
-	for _ in range(20): #Wait 2 seconds i.e 20 * .1 seconds, must wait for service to get to listening state.
+	for _ in range(20): #Wait up to 2 seconds i.e 20 * .1 seconds, must wait for service to get to listening state.
 		for connection in psutil.net_connections():
 			local_address = connection.laddr
-			if len(local_address) > 1 and local_address[1] == params.get_port() and connection.status == "LISTEN":
+			if len(local_address) > 1 and local_address[1] == port and connection.status == status_method:
 				isConnectionOpen = True
 				break
 		if isConnectionOpen: