Merge branch 'WEEE-Open:master' into master

Author: Al3cr1s

.github/workflows/lint.yml

@@ -17,7 +17,7 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v1
         with:
-          python-version: 3.8
+          python-version: 3.12
 
       - name: Install Python dependencies
         run: pip install black

basilico.py

@@ -3,6 +3,7 @@
 import subprocess
 import stat
 import os
+import sys
 import time
 from collections import deque
 from typing import Optional, Callable, Dict, Set, List
@@ -96,7 +97,7 @@ def dequeue(self, cmd_runner):
                     next_in_line.start()
             else:
                 if cmd_runner.get_cmd() != "queued_sleep":
-                    cmd_runner.call_hdparm_for_sleep(self._path)
+                    cmd_runner._call_hdparm_for_sleep(self._path)
 
     def get_path(self):
         return self._path
@@ -320,6 +321,7 @@ def dispatch_command(self, cmd: str, args: str) -> (Optional[Callable[[str, str]
             "queued_badblocks": self.badblocks,
             "queued_cannolo": self.cannolo,
             "queued_sleep": self.sleep,
+            "queued_umount": self.umount,
             "upload_to_tarallo": self.upload_to_tarallo,
             "queued_upload_to_tarallo": self.queued_upload_to_tarallo,
             "get_disks": self.get_disks,
@@ -665,6 +667,47 @@ def cannolo(self, _cmd: str, dev_and_iso: str):
         else:
             self._queued_command.notify_finish()
 
+    def umount(self, _cmd: str, dev: str):
+        self._queued_command.notify_start("Calling umount")
+        success = self._umount_internal(dev)
+        self._queued_command.disk.update_mountpoints()
+        if success:
+            self._queued_command.notify_finish(f"Disk {dev} umounted")
+        else:
+            self._queued_command.notify_finish_with_error(f"umount failed")
+
+    def _umount_internal(self, dev):
+        try:
+            result = subprocess.run(["lsblk", "-J", dev], capture_output=True, text=True)
+
+            if result.returncode != 0:
+                return False
+
+            lsblk_output = json.loads(result.stdout)
+
+            partitions_to_unmount = []
+            blockdevices = lsblk_output.get("blockdevices", [])
+            for device in blockdevices:
+                if "children" in device:
+                    for partition in device["children"]:
+                        if "mountpoints" in partition and len(partition["mountpoints"]) > 0:
+                            partitions_to_unmount.append(f"/dev/{partition['name']}")
+                    break
+
+            if not partitions_to_unmount:
+                return True
+
+            for partition in partitions_to_unmount:
+                rc = self._call_shell_command(("sudo", "umount", partition))
+
+                if rc != 0:
+                    return False
+
+        except Exception as _:
+            return False
+
+        return True
+
     def _unswap(self) -> bool:
         if TEST_MODE:
             return True
@@ -694,26 +737,14 @@ def _unswap(self) -> bool:
 
     def sleep(self, _cmd: str, dev: str):
         self._queued_command.notify_start("Calling hdparm")
-        exitcode = self.call_hdparm_for_sleep(dev)
+        exitcode = self._call_hdparm_for_sleep(dev)
         if exitcode == 0:
             self._queued_command.notify_finish("Good night!")
         else:
             self._queued_command.notify_finish_with_error(f"hdparm exited with status {str(exitcode)}")
 
-    def call_hdparm_for_sleep(self, dev):
-        if TEST_MODE:
-            logging.debug(f"Fake putting {dev} to sleep")
-            return 0
-        logging.debug(f"[{self._the_id}] Putting {dev} to sleep")
-        res = subprocess.Popen(
-            ("sudo", "hdparm", "-Y", dev),
-            stderr=subprocess.DEVNULL,
-            stdout=subprocess.DEVNULL,
-        )
-        exitcode = res.wait()
-        if exitcode != 0:
-            logging.warning(f"[{self._the_id}] hdparm for {dev} returned {str(exitcode)}")
-        return exitcode
+    def _call_hdparm_for_sleep(self, dev: str):
+        return self._call_shell_command(("sudo", "hdparm", "-Y", dev))
 
     def get_smartctl(self, cmd: str, args: str):
         params = self._get_smartctl(args, False)
@@ -965,6 +996,26 @@ def stop_process(self, cmd: str, args: str):
         thread = find_thread_from_pid(args)
         thread.stop_asap()
 
+    def _call_shell_command(self, command: tuple):
+        if TEST_MODE:
+            logging.debug(f"Simulating command: {' '.join(command)}")
+            return 0
+        logging.debug(f"[{self._the_id}] Running command {' '.join(command)}")
+
+        try:
+            res = subprocess.Popen(
+                command,
+                stderr=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL,
+            )
+        except FileNotFoundError as _:
+            return sys.maxsize
+
+        exitcode = res.wait()
+        if exitcode != 0:
+            logging.warning(f"[{self._the_id}] Command '{' '.join(command)}' returned {str(exitcode)}")
+        return exitcode
+
 
 class QueuedCommand:
     def __init__(self, disk: Disk, command_runner: CommandRunner):
@@ -1274,7 +1325,7 @@ def main():
     ip = os.getenv("IP")
     port = os.getenv("PORT")
     global TEST_MODE
-    TEST_MODE = bool(os.getenv("TEST_MODE", False))
+    TEST_MODE = bool(int(os.getenv("TEST_MODE", False)))
 
     if TEST_MODE:
         logging.warning("Test mode is enabled, no destructive actions will be performed")

pinolo.py

@@ -65,7 +65,7 @@ def __init__(self, app: QApplication) -> None:
         self.settings = QSettings("WEEE-Open", "PESTO")
         self.client = ReactorThread(self.host, self.port, self.remoteMode)
 
-        " Utilities Widgets "
+        """ Utilities Widgets """
         self.latest_conf()
         self.diff_widgets = {}
         self.smart_widgets = {}
@@ -97,6 +97,7 @@ def __init__(self, app: QApplication) -> None:
         self.sleep_action = QAction("Sleep", self)
         self.uploadToTarallo_action = QAction("Upload to TARALLO", self)
         self.showSmartData_Action = QAction("Show SMART data", self)
+        self.umount_action = QAction("Umount disk", self)
         self.stop_action = QAction("Stop", self)
         self.remove_action = QAction("Remove", self)
         self.remove_all_action = QAction("Remove All", self)
@@ -112,19 +113,15 @@ def __init__(self, app: QApplication) -> None:
         self.setup()
 
     def setup(self):
-        """This method must be called in the __init__ function of the Ui class
-        to initialize the pinolo session"""
+        """This method must be called in __init__ function of Ui class
+        to initialize pinolo session"""
 
         # Check if the host and port field are set
         if self.host is None and self.port is None:
             message = "The host and port combination is not set.\nPlease visit the settings section."
             warning_dialog(message, dialog_type="ok")
 
-        """
-        The client try to connect to the BASILICO. If it can't and the client is in remote mode, then 
-        a critical error is shown and the client goes in idle. If the client is in local mode and it cannot reach a
-        BASILICO server, a new BASILICO process is instantiated.
-        """
+        # Start client thread
         self.client = ReactorThread(self.host, self.port, self.remoteMode)
         self.client.updateEvent.connect(self.gui_update)
         self.client.start()
@@ -197,6 +194,8 @@ def set_items_functions(self):
         self.showSmartData_Action.triggered.connect(self.show_smart_data)
         self.diskTable.addAction(self.showSmartData_Action)
         self.uploadToTarallo_action.setEnabled(False)
+        self.diskTable.addAction(self.umount_action)
+        self.umount_action.triggered.connect(self.umount_disk)
 
         self.diskTable.selectionModel().selectionChanged.connect(self.on_table_select)
 
@@ -381,18 +380,23 @@ def queue_info(self):
         info_dialog(message)
         self.deselect()
 
+    def umount_disk(self):
+        dialog = warning_dialog(
+            "Are you really sure you want to umount this disk?\nThis is generally not a good idea, proceed only if you are really sure of what are you doing.",
+            "yes_no",
+        )
+
+        if dialog == QMessageBox.Yes:
+            drives = self.get_multiple_drive_selection()
+            for drive in drives:
+                self.client.send("queued_umount " + drive)
+
     def get_multiple_drive_selection(self):
         """This method returns a list with the names of the selected drives on disk_table"""
         drives = []
-        selected_items = self.diskTable.selectedItems()
-        columns = self.diskTable.columnCount()
-        for idx in range(len(selected_items) // columns):
-            drives.append(
-                [
-                    selected_items[idx * columns].text(),
-                    selected_items[(idx * columns) + 1].text(),
-                ]
-            )
+        selected_rows = self.diskTable.selectionModel().selectedRows()
+        for row in selected_rows:
+            drives.append(row.data())
 
         return drives
 
@@ -518,9 +522,7 @@ def show_smart_data(self):
             else:
                 self.selected_drive = self.selected_drive.text()
             if self.selected_drive in self.smart_results:
-                self.smart_widgets[self.selected_drive] = SmartWidget(self.selected_drive,
-                                                                      self.smart_results[self.selected_drive]
-                                                                      )
+                self.smart_widgets[self.selected_drive] = SmartWidget(self.selected_drive, self.smart_results[self.selected_drive])
                 self.smart_widgets[self.selected_drive].close_signal.connect(self.remove_smart_widget)
 
         except BaseException as exc:
@@ -565,10 +567,10 @@ def cannolo(self, std=False, drives=None):
 
     def upload_to_tarallo_selection(self, std: bool = False):
         # TODO: check if it's really working
-        #for row in self.get_selected_drive_rows():
-            #if row[1] == "":
-                # self.upload_to_tarallo(row[0])
-        #self.selected_drive = self.selected_drive.text();
+        # for row in self.get_selected_drive_rows():
+        # if row[1] == "":
+        # self.upload_to_tarallo(row[0])
+        # self.selected_drive = self.selected_drive.text();
         self.selected_drive = self.diskTable.item(self.diskTable.currentRow(), 0)
 
         if not std:
@@ -581,7 +583,7 @@ def upload_to_tarallo_selection(self, std: bool = False):
                 return
         elif self.diskTable.item(self.diskTable.currentRow(), 1).text() != "":
             return
-        loc, ok = input_dialog("Location");
+        loc, ok = input_dialog("Location")
 
         # If no location is provided or cancel is selected,
         # cancel the operation
@@ -834,7 +836,7 @@ def gui_update(self, cmd: str, params: str):
             print(f"GUI: Ignored exception while parsing {cmd}, expected JSON but this isn't: {params}")
 
         match cmd:
-            case 'queue_status' | 'get_queue':
+            case "queue_status" | "get_queue":
                 if cmd == "queue_status":
                     params = [params]
                 for param in params:
@@ -891,7 +893,7 @@ def gui_update(self, cmd: str, params: str):
                     if "text" in param:
                         status_cell.setToolTip(param["text"])
 
-            case 'get_disks':
+            case "get_disks":
                 drives = params
                 if len(drives) <= 0:
                     self.diskTable.setRowCount(0)
@@ -903,10 +905,10 @@ def gui_update(self, cmd: str, params: str):
                 self.diskTable.resizeColumnToContents(0)
                 self.diskTable.resizeColumnToContents(1)
 
-            case 'smartctl' | 'queued_smartctl':
+            case "smartctl" | "queued_smartctl":
                 self.smart_results[params["disk"]] = {"output": params["output"], "status": params["status"]}
 
-            case ' connection_failed':
+            case " connection_failed":
                 message = params["reason"]
                 if not self.remoteMode:
                     print("GUI: Connection Failed: Local server not running.")
@@ -917,29 +919,29 @@ def gui_update(self, cmd: str, params: str):
                     message = "Cannot find BASILICO server.\nCheck if it's running in the " "targeted machine."
                 warning_dialog(message, dialog_type="ok")
 
-            case 'connection_lost':
+            case "connection_lost":
                 self.statusBar().showMessage(f"⚠ Connection lost. Press the reload button to reconnect.")
                 self.queueTable.setRowCount(0)
                 self.diskTable.setRowCount(0)
 
-            case 'connection_made':
+            case "connection_made":
                 self.statusBar().showMessage(f"Connected to {params['host']}:{params['port']}")
 
-            case 'list_iso':
+            case "list_iso":
                 self.dialog = CannoloDialog(self.settingsDialog, PATH, params)
                 if self.manual_cannolo:
                     self.dialog.update.connect(self.use_cannolo_img)
                     self.manual_cannolo = False
                 else:
                     self.dialog.update.connect(self.settingsDialog.set_default_cannolo)
 
-            case 'error':
+            case "error":
                 message = f"{params['message']}"
                 if "command" in params:
                     message += f":\n{params['command']}"
                 critical_dialog(message, dialog_type="ok")
 
-            case 'error_that_can_be_manually_fixed':
+            case "error_that_can_be_manually_fixed":
                 message = params["message"]
                 warning_dialog(message, dialog_type="ok")
 

utilities.py

@@ -1,4 +1,3 @@
-import re
 import subprocess
 import os
 import datetime