Add -m/--mockup-mode feature to service.py and session.py

3 files changed, 100 insertions, 30 deletions

diff --git a/service.py b/service.py
index 45cb17a..eb6d311 100755
--- a/service.py
+++ b/service.py
@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+import argparse
+
 import json
 import time
 from threading import Thread
@@ -8,9 +10,12 @@ import dbus.service
 
 from session import Session
 
+from typing import Union
 
 class RWAService(dbus.service.Object):
-    def __init__(self):
+    def __init__(self, mockup_mode: bool):
+        self.mockup_mode = mockup_mode
+
         self.bus = dbus.SessionBus()
         name = dbus.service.BusName("org.ArcticaProject.RWA", bus=self.bus)
 
@@ -22,7 +27,7 @@ class RWAService(dbus.service.Object):
     def start(self):
         """Start a new remote session."""
         # Start session
-        session = Session()
+        session = Session(mockup_mode)
 
         # Add session to sessions list
         self.sessions[session.pid] = session
@@ -85,13 +90,38 @@ 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 True
+    elif v.lower() in ('no', 'false', 'f', 'n', '0', 0):
+        return False
+    else:
+        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.")
+
+    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!")
     import dbus.mainloop.glib
     from gi.repository import GLib
 
     dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 
     loop = GLib.MainLoop()
-    object = RWAService()
+    object = RWAService(mockup_mode)
     loop.run()
diff --git a/session.py b/session.py
index f7ca9fe..f98c86f 100644
--- a/session.py
+++ b/session.py
@@ -3,6 +3,8 @@ import secrets
 import signal
 from multiprocessing import Process
 
+import random, string
+
 import psutil
 import requests
 
@@ -21,7 +23,8 @@ class Session:
     #: Remote has joined the session
     STATUS_JOINED = "joined"
 
-    def __init__(self):
+    def __init__(self, mockup_session: bool):
+        self.mockup_session = mockup_session
         self._generate_password()
         self._start_vnc()
         self._start_trigger_service()
@@ -38,35 +41,61 @@ class Session:
     def _generate_password(self):
         """Generate password for x11vnc and save it."""
         self.password = secrets.token_urlsafe(20)
-        self.pw_filename = save_password(self.password)
+
+        # Don't actually save a password if we just pretend to be a session.
+        if not self.mockup_session:
+            self.pw_filename = save_password(self.password)
 
     def _start_vnc(self):
-        """Start x11vnc server."""
-        process_info = run_vnc(self.pw_filename)
+        """Start x11vnc server if not in mockup_session mode."""
+        if not self.mockup_session:
+            process_info = run_vnc(self.pw_filename)
+
+            self.vnc_pid = process_info["vnc"]["pid"]
+            self.vnc_port = process_info["vnc"]["port"]
+            self.ws_pid = process_info["ws"]["pid"]
+            self.ws_port = process_info["ws"]["port"]
+        else:
+            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)))
+
+            # 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')
 
-        self.vnc_pid = process_info["vnc"]["pid"]
-        self.vnc_port = process_info["vnc"]["port"]
-        self.ws_pid = process_info["ws"]["pid"]
-        self.ws_port = process_info["ws"]["port"]
 
     def _register_session(self):
-        """Register session in RWA."""
-        r = requests.post(
-            REGISTER_URL,
-            json={
-                "port": self.ws_port,
-                "password": self.password,
-                "pid": self.vnc_pid,
-                "trigger_port": self.trigger_port,
-                "trigger_token": self.trigger_token,
-            },
-        )
-        print(r)
-        self.meta = r.json()
-        self.session_id = self.meta["session_id"]
-        self.web_url = self.meta["url"]
-        self.api_token = self.meta["token"]
-        self.pin = self.meta["pin"]
+        """Register session in RWA if not in mockup_session mode."""
+        if not self.mockup_session:
+            r = requests.post(
+                REGISTER_URL,
+                json={
+                    "port": self.ws_port,
+                    "password": self.password,
+                    "pid": self.vnc_pid,
+                    "trigger_port": self.trigger_port,
+                    "trigger_token": self.trigger_token,
+                },
+            )
+            print(r)
+            self.meta = r.json()
+            self.session_id = self.meta["session_id"]
+            self.web_url = self.meta["url"]
+            self.api_token = self.meta["token"]
+            self.pin = self.meta["pin"]
+        else:
+            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)))
 
     def _start_trigger_service(self):
         self.trigger_port = port_for.select_random()
@@ -102,6 +131,15 @@ class Session:
         pass
 
     def stop(self):
+        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"
+            if os.path.isfile(filename):
+                os.remove(filename);
+
+            # Delete self
+            del self
+            return
+
         """Stop session and clean up."""
         # Kill VNC
         if self.vnc_pid in psutil.pids():
@@ -129,6 +167,10 @@ class Session:
 
     @property
     def vnc_process_running(self):
+        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)
+
         """Check if the VNC process is still running."""
         if self.vnc_pid in psutil.pids():
             p = psutil.Process(self.vnc_pid)
diff --git a/test_client.py b/test_client.py
index 9cc6b68..4e60c29 100755
--- a/test_client.py
+++ b/test_client.py
@@ -3,9 +3,7 @@ import dbus
 
 bus = dbus.SessionBus()
 
-
 time = bus.get_object("org.ArcticaProject.RWA", "/RWA")
 
-
 curr = time.start()
 print("Your VNC session is", curr)