diff options
Diffstat (limited to 'session.py')
| -rw-r--r-- | session.py | 42 |
1 files changed, 28 insertions, 14 deletions
@@ -1,10 +1,10 @@ import os +import random import secrets import signal +import string from multiprocessing import Process -import random, string - import psutil import requests @@ -62,19 +62,22 @@ class Session: self.ws_pid = process_info["ws"]["pid"] self.ws_port = process_info["ws"]["port"] else: - self.ws_port = port_for.select_random() + self.ws_port = port_for.select_random() self.vnc_port = port_for.select_random() # Use negative values to ensure we don't do something harmful # to random processes - self.ws_pid = int('-' + ''.join(random.choice(string.digits) for _ in range(5))) - self.vnc_pid = int('-' + ''.join(random.choice(string.digits) for _ in range(5))) + self.ws_pid = int( + "-" + "".join(random.choice(string.digits) for _ in range(5)) + ) + self.vnc_pid = int( + "-" + "".join(random.choice(string.digits) for _ in range(5)) + ) # Create a temporary file to indicate that this process is still 'Running' filename = f"/tmp/rwa/{str(self.ws_port) + str(self.vnc_port) + str(self.ws_pid) + str(self.vnc_pid)}.lock" - new_file = open(filename, "w"); - new_file.write('this session is running') - + new_file = open(filename, "w") + new_file.write("this session is running") def _register_session(self): """Register session in RWA if not in mockup_session mode.""" @@ -96,12 +99,23 @@ class Session: self.api_token = self.meta["token"] self.pin = self.meta["pin"] else: - print("\"Registered\" in RWA") + print('"Registered" in RWA') self.meta = {} - self.session_id = int(''.join(random.choice(string.digits) for _ in range(10))) - self.web_url = "testhostname:" + ''.join(random.choice(string.digits) for _ in range(5)) + "/RWA/test/" - self.api_token = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(10)) - self.pin = int(''.join(random.choice(string.digits) for _ in range(5))) + self.session_id = int( + "".join(random.choice(string.digits) for _ in range(10)) + ) + self.web_url = ( + "testhostname:" + + "".join(random.choice(string.digits) for _ in range(5)) + + "/RWA/test/" + ) + self.api_token = "".join( + random.choice( + string.ascii_uppercase + string.ascii_lowercase + string.digits + ) + for _ in range(10) + ) + self.pin = int("".join(random.choice(string.digits) for _ in range(5))) def _start_trigger_service(self): self.trigger_port = port_for.select_random() @@ -136,7 +150,7 @@ class Session: """Update status: Push status to Django.""" pass - def stop(self, triggered: bool =False): + def stop(self, triggered: bool = False): """Stop session and clean up.""" if self.mockup_session: filename = f"/tmp/rwa/{str(self.ws_port) + str(self.vnc_port) + str(self.ws_pid) + str(self.vnc_pid)}.lock" |