🚧 WIP

Ackee-Blockchain · Nov 28, 2024 · caa771c · caa771c
1 parent eafae47
commit caa771c
Show file tree

Hide file tree

Showing 2 changed files with 54 additions and 0 deletions.
diff --git a/wake/cli/__main__.py b/wake/cli/__main__.py
@@ -21,6 +21,7 @@
 from .run import run_run
 from .svm import run_svm
 from .test import run_test
+from .qr import run_qr
 
 if platform.system() != "Windows":
     try:
@@ -211,6 +212,7 @@ def exit():
 main.add_command(run_run)
 main.add_command(run_svm)
 main.add_command(run_test)
+main.add_command(run_qr)
 
 
 @main.command(name="config")

diff --git a/wake/testing/pytest_plugin_multiprocess_server.py b/wake/testing/pytest_plugin_multiprocess_server.py
@@ -24,6 +24,8 @@
 
 from .pytest_plugin_multiprocess import PytestWakePluginMultiprocess
 
+from firebase_admin import credentials, messaging, initialize_app, db
+
 
 class PytestWakePluginMultiprocessServer:
     _config: WakeConfig
@@ -41,6 +43,8 @@ class PytestWakePluginMultiprocessServer:
         int, Dict[Path, Dict[IdePosition, IdeFunctionCoverageRecord]]
     ]
 
+    _firebase_token: Optional[str]
+
     def __init__(
         self,
         config: WakeConfig,
@@ -63,6 +67,18 @@ def __init__(
         self._pytest_args = pytest_args
         self._exported_coverages = {i: {} for i in range(self._proc_count)}
 
+        if config.wake_remote.uuid is not None:
+            cred = credentials.Certificate(config.global_data_path / "fuzz-firebase-adminsdk.json")
+            initialize_app(
+                cred,
+                {
+                    "databaseURL": "https://wake-remote-default-rtdb.europe-west1.firebasedatabase.app/"
+                }
+            )
+
+            x = db.reference("/")
+            self._firebase_token = x.child("tokens").child(config.wake_remote.uuid).get()
+
     def pytest_sessionstart(self, session: pytest.Session):
         if self._coverage != 0:
             empty_coverage = CoverageHandler(self._config)
@@ -105,6 +121,16 @@ def pytest_sessionstart(self, session: pytest.Session):
             p.start()
         signal.signal(signal.SIGINT, signal.SIG_IGN)
 
+        if self._firebase_token is not None:
+            message = messaging.Message(
+                token=self._firebase_token,
+                notification=messaging.Notification(
+                    title="New fuzzing session started",
+                    body=f"{self._proc_count} tasks",
+                )
+            )
+            messaging.send(message)
+
     def pytest_sessionfinish(self, session: pytest.Session):
         self._queue.cancel_join_thread()
         for p, conn in self._processes.values():
@@ -220,6 +246,15 @@ def pytest_runtestloop(self, session: pytest.Session):
                             exception_info[2],
                         )
 
+                        message = messaging.Message(
+                            token=self._firebase_token,
+                            notification=messaging.Notification(
+                                title=f"Exception {}",
+                                body=f"Process #{index}",
+                            )
+                        )
+                        messaging.send(message)
+
                         if progress is not None:
                             progress.stop()
 
@@ -249,6 +284,14 @@ def pytest_runtestloop(self, session: pytest.Session):
                         if progress is not None:
                             progress.start()
                     elif msg[0] == "breakpoint":
+                        message = messaging.Message(
+                            token=self._firebase_token,
+                            notification=messaging.Notification(
+                                title="Breakpoint reached",
+                                body=f"Process #{index}",
+                            )
+                        )
+                        messaging.send(message)
 
                         (filename, lineno, function_name, syntax) = pickle.loads(msg[2])
                         if progress is not None:
@@ -314,6 +357,15 @@ def pytest_runtestloop(self, session: pytest.Session):
 
                         self._processes.pop(index)
                     elif msg[0] == "pytest_internalerror":
+                        message = messaging.Message(
+                            token=self._firebase_token,
+                            notification=messaging.Notification(
+                                title="Internal pytest error",
+                                body=f"Process #{index}",
+                            )
+                        )
+                        messaging.send(message)
+
                         exc_info = pytest.ExceptionInfo.from_exc_info(
                             pickle.loads(msg[2])
                         )