diff options
author | Jonathan Weth <git@jonathanweth.de> | 2020-07-22 15:43:26 +0200 |
---|---|---|
committer | Jonathan Weth <git@jonathanweth.de> | 2020-07-22 15:43:26 +0200 |
commit | b3d8e355abd81eee87b28b1dfd96f556ce6e0216 (patch) | |
tree | c4928a6ee0247c728d0ea2dcf74b58faaefd8e9d | |
parent | 773c749f582ff8fd9c1f106d5de8af34588cd3f8 (diff) | |
download | RWA.Support.SessionService-b3d8e355abd81eee87b28b1dfd96f556ce6e0216.tar.gz RWA.Support.SessionService-b3d8e355abd81eee87b28b1dfd96f556ce6e0216.tar.bz2 RWA.Support.SessionService-b3d8e355abd81eee87b28b1dfd96f556ce6e0216.zip |
Reformat
-rwxr-xr-x | service.py | 38 | ||||
-rw-r--r-- | session.py | 42 |
2 files changed, 50 insertions, 30 deletions
@@ -1,16 +1,15 @@ #!/usr/bin/env python3 import argparse - import json import time from threading import Thread +from typing import Union import dbus import dbus.service from session import Session -from typing import Union class RWAService(dbus.service.Object): def __init__(self, mockup_mode: bool): @@ -90,33 +89,40 @@ class RWAService(dbus.service.Object): self.update_service_running = False # TODO Probably kill daemon here (quit main loop) + def str2bool(v: Union[str, bool, int]) -> bool: """Return true or false if the given string can be interpreted as a boolean otherwise raise an exception.""" if isinstance(v, bool): - return v - if v.lower() in ('yes', 'true', 't', 'y', '1', 1): + return v + if v.lower() in ("yes", "true", "t", "y", "1", 1): return True - elif v.lower() in ('no', 'false', 'f', 'n', '0', 0): + elif v.lower() in ("no", "false", "f", "n", "0", 0): return False else: - raise argparse.ArgumentTypeError('Boolean value expected.') + raise argparse.ArgumentTypeError("Boolean value expected.") + if __name__ == "__main__": - parser = argparse.ArgumentParser(description='DBus session service for ' + - 'ArcticaProject\'s ' + - 'Remote Web App') - parser.add_argument("-m", "--mockup-mode", type=str2bool, nargs='?', - const=True, default=False, - help="Activate mockup mode. Act like the session " + - "service but don\'t do changes or call other " + - "parts of RWA.") + parser = argparse.ArgumentParser( + description="DBus session service for " + "ArcticaProject's " + "Remote Web App" + ) + parser.add_argument( + "-m", + "--mockup-mode", + type=str2bool, + nargs="?", + const=True, + default=False, + help="Activate mockup mode. Act like the session " + + "service but don't do changes or call other " + + "parts of RWA.", + ) args = parser.parse_args() mockup_mode = args.mockup_mode if mockup_mode: - print("All API responses are faked and should NOT BE USED IN " + - "PRODUCTION!") + print("All API responses are faked and should NOT BE USED IN " + "PRODUCTION!") import dbus.mainloop.glib from gi.repository import GLib @@ -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" |