Introduce and apply reformat/lint

5 files changed, 77 insertions, 71 deletions

diff --git a/rwa/support/sessionservice/lock.py b/rwa/support/sessionservice/lock.py
index b4e384c..49be2f5 100644
--- a/rwa/support/sessionservice/lock.py
+++ b/rwa/support/sessionservice/lock.py
@@ -26,13 +26,13 @@
 import os
 import tempfile
 
-lock_dir_path = os.path.join(tempfile.gettempdir(), "rwa.support.sessionservice")
-lock_file_path = os.path.join(lock_dir_path, "rwa.support.sessionservice.lock")
+TEMP_DIR_PATH = os.path.join(tempfile.gettempdir(), "rwa.support.sessionservice")
+lock_file_path = os.path.join(TEMP_DIR_PATH, "rwa.support.sessionservice.lock")
 
 
 def lock():
     """Create lock file."""
-    os.makedirs(lock_dir_path, exist_ok=True)
+    os.makedirs(TEMP_DIR_PATH, exist_ok=True)
     with open(lock_file_path, "w") as f:
         f.write("lock")
 
@@ -48,3 +48,7 @@ def unlock():
         os.remove(lock_file_path)
     except FileNotFoundError:
         pass
+
+
+# Always ensure temp dir
+os.makedirs(TEMP_DIR_PATH, exist_ok=True)
diff --git a/rwa/support/sessionservice/service.py b/rwa/support/sessionservice/service.py
index ae634c8..c50da26 100755
--- a/rwa/support/sessionservice/service.py
+++ b/rwa/support/sessionservice/service.py
@@ -28,12 +28,10 @@
 import argparse
 import json
 import logging
-import os
 import signal
-import tempfile
 import time
 from threading import Thread
-from typing import Union, Any
+from typing import Union
 
 import dbus
 import dbus.mainloop.glib
@@ -58,9 +56,7 @@ class RWASupportSessionService(dbus.service.Object):
     :param mockup_mode: Starts the service in mock up mode
     """
 
-    def __init__(
-        self, loop: GLib.MainLoop, mockup_mode: bool = False, one_time: bool = False
-    ):
+    def __init__(self, loop: GLib.MainLoop, mockup_mode: bool = False, one_time: bool = False):
         self.loop = loop
         self.mockup_mode = mockup_mode
         self.one_time = one_time
@@ -102,14 +98,15 @@ class RWASupportSessionService(dbus.service.Object):
         """
         if ALLOW_ONLY_ONE_SESSION and len(self.sessions.values()) > 0:
             logging.warning(
-                "There is already one session running and the service is configured to allow only one "
+                "There is already one session running and the service "
+                "is configured to allow only one "
                 "session, so this session won't be started."
             )
             return json.dumps({"status": "error", "type": "multiple"})
 
         # Start session
         try:
-            session = Session(self.trigger_service.port, mockup_mode)
+            session = Session(self.trigger_service.port, self.mockup_mode)
 
             # Add session to sessions list
             self.sessions[session.pid] = session
@@ -120,9 +117,7 @@ class RWASupportSessionService(dbus.service.Object):
             return_json = session.client_meta
             return_json["status"] = "success"
 
-            logging.info(
-                f"New session #{session.pid} was started with meta {return_json}."
-            )
+            logging.info(f"New session #{session.pid} was started with meta {return_json}.")
 
             return json.dumps(return_json)
         except ConnectionError:
@@ -130,7 +125,9 @@ class RWASupportSessionService(dbus.service.Object):
 
         return json.dumps({"status": "error", "type": "connection"})
 
-    @dbus.service.method("org.ArcticaProject.RWASupportSessionService", in_signature="i", out_signature="s")
+    @dbus.service.method(
+        "org.ArcticaProject.RWASupportSessionService", in_signature="i", out_signature="s"
+    )
     def status(self, pid: int) -> str:
         """Return the status of a session.
 
@@ -158,14 +155,17 @@ class RWASupportSessionService(dbus.service.Object):
         """
         return self._get_status(pid)
 
-    @dbus.service.method("org.ArcticaProject.RWASupportSessionService", in_signature="i", out_signature="s")
+    @dbus.service.method(
+        "org.ArcticaProject.RWASupportSessionService", in_signature="i", out_signature="s"
+    )
     def refresh_status(self, pid: int) -> str:
-        """Same as :meth:`status`, but updates status from RWA.WebApp before returning it here.
-        """
+        """Update status from WebApp before returning it here like :meth:`status`."""
         self._update_session(pid)
         return self._get_status(pid)
 
-    @dbus.service.method("org.ArcticaProject.RWASupportSessionService", in_signature="i", out_signature="s")
+    @dbus.service.method(
+        "org.ArcticaProject.RWASupportSessionService", in_signature="i", out_signature="s"
+    )
     def stop(self, pid: int) -> str:
         """Stop a remote session.
 
@@ -200,7 +200,6 @@ class RWASupportSessionService(dbus.service.Object):
 
     def _update_session(self, pid: int):
         """Update the status of a session."""
-
         try:
             session = self.sessions[pid]
         except KeyError:
@@ -238,7 +237,7 @@ class RWASupportSessionService(dbus.service.Object):
         if self.one_time:
             self._stop_all()
 
-    def _trigger(self, session_id: int,  data: dict, method: str = "trigger") -> Union[dict, bool]:
+    def _trigger(self, session_id: int, data: dict, method: str = "trigger") -> Union[dict, bool]:
         """Trigger a specific session via trigger token."""
         logging.info(f"Triggered with session ID {session_id} and {data}")
 
@@ -271,7 +270,10 @@ class RWASupportSessionService(dbus.service.Object):
 
 
 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."""
+    """Return true or false if the given string can be interpreted as a boolean.
+
+    If it fails, raise an exception.
+    """
     if isinstance(v, bool):
         return v
     if v.lower() in ("yes", "true", "t", "y", "1", 1):
@@ -281,6 +283,7 @@ def str2bool(v: Union[str, bool, int]) -> bool:
     else:
         raise argparse.ArgumentTypeError("Boolean value expected.")
 
+
 def main():
     # Check for lock file
     if is_locked():
@@ -298,7 +301,8 @@ def main():
         nargs="?",
         const=True,
         default=False,
-        help="Activates mock up mode. Acts like the real session service but don't do changes or call RWA.",
+        help="Activates mock up mode. Acts like the real session service "
+        "but don't do changes or call RWA.",
     )
     parser.add_argument(
         "-o",
@@ -315,18 +319,16 @@ def main():
     one_time = args.one_time
 
     if mockup_mode:
-        logging.warning(
-            "All API responses are faked and should NOT BE USED IN PRODUCTION!"
-        )
+        logging.warning("All API responses are faked and should NOT BE USED IN PRODUCTION!")
 
     dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 
     loop = GLib.MainLoop()
-    object = RWASupportSessionService(loop, mockup_mode, one_time)
+    service_object = RWASupportSessionService(loop, mockup_mode, one_time)
 
     def signal_handler(sig, frame):
         logging.info("Service was terminated.")
-        object._stop_all()
+        service_object._stop_all()
 
     signal.signal(signal.SIGINT, signal_handler)
 
@@ -337,4 +339,4 @@ def main():
 
 
 if __name__ == "__main__":
-   main()
\ No newline at end of file
+    main()
diff --git a/rwa/support/sessionservice/session.py b/rwa/support/sessionservice/session.py
index 1dfb6bb..d6dbc82 100644
--- a/rwa/support/sessionservice/session.py
+++ b/rwa/support/sessionservice/session.py
@@ -24,18 +24,18 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import os
-import random
 import secrets
 import signal
 import string
-import subprocess
-from typing import Dict, Union, Any
+import subprocess  # noqa
+from typing import Dict, Union
 
 import port_for
 import psutil
 import requests
 
 from .config import settings
+from .lock import TEMP_DIR_PATH
 from .log import logging
 from .vnc import run_vnc, save_password
 
@@ -50,17 +50,13 @@ logging.info(f"Load API config: {API_SERVER}")
 
 
 def random_digits(length: int):
-    return "".join(random.choice(string.digits) for _ in range(length))
+    return "".join(secrets.choice(string.digits) for _ in range(length))
 
 
 def get_desktop_dir():
     """Get desktop directory from xdg vars."""
-    return (
-        subprocess.check_output(["xdg-user-dir", "DESKTOP"])
-        .decode()
-        .strip()
-        .replace("\n", "")
-    )
+    output = subprocess.check_output(["xdg-user-dir", "DESKTOP"]).decode()  # noqa
+    return output.strip().replace("\n", "")
 
 
 class Session:
@@ -94,6 +90,12 @@ class Session:
     def _api_headers(self) -> Dict[str, str]:
         return {"Authorization": f"Token {self.api_token}"}
 
+    @property
+    def _mock_lock_file_path(self) -> str:
+        return os.path.join(
+            TEMP_DIR_PATH, f"{self.ws_port}-{self.vnc_port}-{self.ws_pid}-{self.vnc_pid}.lock",
+        )
+
     def _generate_password(self):
         """Generate password for x11vnc and save it."""
         self.password = secrets.token_urlsafe(20)
@@ -125,8 +127,7 @@ class Session:
             logging.info("The lock file for mocking a VNC server has been created.")
 
             # 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 = open(self._mock_lock_file_path, "w")
             new_file.write("this session is running")
 
     def _register_session(self):
@@ -145,7 +146,8 @@ class Session:
                 raise ConnectionError()
 
             logging.info(
-                f"The session has been registered in RWA.Support.WebApp with status code {r.status_code} and response {r.content.decode()}."
+                "The session has been registered in RWA.Support.WebApp "
+                f"with status code {r.status_code} and response {r.content.decode()}."
             )
 
             if r.status_code != 200:
@@ -156,10 +158,10 @@ class Session:
             self.web_url = self.meta["url"]
             self.api_token = self.meta["token"]
         else:
-            logging.info(f"The session has pretended that he had created a session.")
+            logging.info("The session has pretended that he had created a session.")
             self.meta = {}
             self.session_id = int(random_digits(10))
-            self.web_url = "http://example.com:" + random_digits(5) + "/app/rwasupport/test/"
+            self.web_url = f"http://example.com:{random_digits(5)}/app/rwasupport/test/"
             self.api_token = secrets.token_urlsafe(10)
         self.pin = int(random_digits(4))
 
@@ -174,8 +176,8 @@ class Session:
                 "trigger_token": self.trigger_token,
             }
 
-
         return False
+
     def pull(self):
         """Update status: Get status from Django."""
         if not self.mockup_session:
@@ -185,7 +187,8 @@ class Session:
                 )
 
                 logging.info(
-                    f"The session has received its status from RWA.Support.WebApp with status code {r.status_code} and response {r.content.decode()}."
+                    "The session has received its status from RWA.Support.WebApp "
+                    f"with status code {r.status_code} and response {r.content.decode()}."
                 )
             except requests.ConnectionError:
                 pass
@@ -208,10 +211,8 @@ class Session:
 
     def _do_file_job(self, job):
         """Download a file from server to the user's desktop."""
-        logging.info(
-            f"The session has received a file job and is downloading it now ({job}):"
-        )
-        subprocess.Popen(["wget", job["file"], "-P", self.desktop_dir])
+        logging.info(f"The session has received a file job and is downloading it now ({job}):")
+        subprocess.Popen(["wget", job["file"], "-P", self.desktop_dir])  # noqa
         self._mark_job_as_done(job)
 
     def _mark_job_as_done(self, job):
@@ -219,12 +220,11 @@ class Session:
         self.done_jobs.append(job["job_id"])
         try:
             r = requests.post(
-                MARK_JOB_AS_DONE_URL,
-                params={"id": job["job_id"]},
-                headers=self._api_headers,
+                MARK_JOB_AS_DONE_URL, params={"id": job["job_id"]}, headers=self._api_headers,
             )
             logging.info(
-                f"The session has marked the job {job} as done in RWA.Support.WebApp with status code {r.status_code} and response {r.content.decode()}."
+                f"The session has marked the job {job} as done in RWA.Support.WebApp "
+                f"with status code {r.status_code} and response {r.content.decode()}."
             )
         except requests.ConnectionError:
             pass
@@ -237,9 +237,8 @@ class Session:
         """Stop session and clean up."""
         if self.mockup_session:
             logging.info("Mock session has been stopped by deleting its lock file.")
-            filename = f"/tmp/rwa/{str(self.ws_port) + str(self.vnc_port) + str(self.ws_pid) + str(self.vnc_pid)}.lock"
-            if os.path.isfile(filename):
-                os.remove(filename)
+            if os.path.isfile(self._mock_lock_file_path):
+                os.remove(self._mock_lock_file_path)
 
             # Delete self
             del self
@@ -268,7 +267,8 @@ class Session:
                     STOP_URL, params={"id": self.session_id}, headers=self._api_headers
                 )
                 logging.info(
-                    f"The stop action has been registered in RWA.Support.WebApp with status code {r.status_code} and response {r.content.decode()}."
+                    "The stop action has been registered in RWA.Support.WebApp "
+                    f"with status code {r.status_code} and response {r.content.decode()}."
                 )
             except requests.ConnectionError:
                 pass
@@ -282,8 +282,7 @@ class Session:
     def vnc_process_running(self) -> bool:
         """Check if the VNC process is still running."""
         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"
-            return os.path.isfile(filename)
+            return os.path.isfile(self._mock_lock_file_path)
 
         if self.vnc_pid in psutil.pids():
             p = psutil.Process(self.vnc_pid)
diff --git a/rwa/support/sessionservice/trigger.py b/rwa/support/sessionservice/trigger.py
index fa9734a..a9643f7 100644
--- a/rwa/support/sessionservice/trigger.py
+++ b/rwa/support/sessionservice/trigger.py
@@ -24,11 +24,11 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import threading
-from typing import Any, Callable, Optional, Union
+from typing import Callable, Optional, Union
 from wsgiref.simple_server import make_server
 
 import port_for
-from flask import Flask, abort, request, jsonify
+from flask import Flask, abort, jsonify, request
 
 
 class TriggerServerThread(threading.Thread):
@@ -68,8 +68,7 @@ class TriggerServerThread(threading.Thread):
             except (ValueError, TypeError):
                 return abort(404)
 
-
-        self.srv = make_server("0.0.0.0", self.port, app)
+        self.srv = make_server("0.0.0.0", self.port, app)  # noqa
         self.ctx = app.app_context()
         self.ctx.push()
 
diff --git a/rwa/support/sessionservice/vnc.py b/rwa/support/sessionservice/vnc.py
index 1725532..f41efce 100644
--- a/rwa/support/sessionservice/vnc.py
+++ b/rwa/support/sessionservice/vnc.py
@@ -24,18 +24,20 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import os
-import subprocess
+import secrets
+import subprocess  # noqa
 from typing import Dict
-from uuid import uuid4
 
 import port_for
 
+from .lock import TEMP_DIR_PATH
+
 
 def save_password(pw: str) -> str:
     """Save password in x11vnc format in temporary directory."""
-    filename = f"/tmp/rwa/{uuid4()}.pw"
-    os.makedirs("/tmp/rwa/", exist_ok=True)
-    p = subprocess.Popen(["x11vnc", "-storepasswd", f"{pw}", filename])
+    filename = os.path.join(TEMP_DIR_PATH, f"{secrets.token_urlsafe(20)}.pw")
+    os.makedirs(TEMP_DIR_PATH, exist_ok=True)
+    p = subprocess.Popen(["x11vnc", "-storepasswd", f"{pw}", filename])  # noqa
     p.communicate()
     return filename
 
@@ -46,10 +48,10 @@ def run_vnc(pw_filename: str) -> Dict[str, Dict[str, int]]:
     port_vnc = port_for.select_random()
 
     # Start VNC process
-    p = subprocess.Popen(["x11vnc", "-rfbauth", pw_filename, "-rfbport", f"{port_vnc}"])
+    p = subprocess.Popen(["x11vnc", "-rfbauth", pw_filename, "-rfbport", f"{port_vnc}"])  # noqa
 
     # Start websockify
-    p2 = subprocess.Popen(f"websockify {port} 127.0.0.1:{port_vnc}", shell=True,)
+    p2 = subprocess.Popen(["websockify", str(port), f"127.0.0.1:{port_vnc}"])  # noqa
 
     return {
         "ws": {"pid": p2.pid, "port": port},