Merge pull request #7932 from kozlovsky/fix/process_manager_connect_p…

…ermission_error

Process manager should not attempt to unlink the processes database file in case of OperationalError

src/tribler/core/utilities/process_manager/manager.py

@@ -104,14 +104,21 @@ def connect(self) -> ContextManager[sqlite3.Connection]:
             if connection:
                 connection.close()
 
-            if isinstance(e, sqlite3.DatabaseError):
-                self.db_filepath.unlink(missing_ok=True)
+            if isinstance(e, sqlite3.OperationalError):
+
+                if str(e) == 'unable to open database file':
+                    msg = f"{e}: {self._unable_to_open_db_file_get_reason()}"
+                    raise sqlite3.OperationalError(msg) from e
 
-            if isinstance(e, sqlite3.OperationalError) and str(e) == 'unable to open database file':
-                msg = f"{e}: {self._unable_to_open_db_file_get_reason()}"
-                raise sqlite3.OperationalError(msg) from e
+                raise e
+
+            if isinstance(e, sqlite3.DatabaseError):
+                try:
+                    self.db_filepath.unlink(missing_ok=True)
+                except PermissionError:
+                    logger.warning(f'Unable to delete the processes database file: {self.db_filepath}')
 
-            raise
+            raise e
 
     def _unable_to_open_db_file_get_reason(self):
         dir_path = self.db_filepath.parent

src/tribler/core/utilities/process_manager/tests/test_manager.py

@@ -282,3 +282,33 @@ def test_unable_to_open_db_file_reason_added(process_manager):
                        match=r'^unable to open database file: parent directory.*does not exist$'):
         with process_manager.connect():
             pass
+
+
+@patch('sqlite3.connect', MagicMock(side_effect=sqlite3.OperationalError('error text')))
+def test_connect_operational_error(process_manager):
+    # All OperationalError exceptions except "unable to open database file" should be re-raised
+    with pytest.raises(sqlite3.OperationalError, match=r'^error text$'):
+        with process_manager.connect():
+            pass  # pragma: no cover
+
+
+@patch('sqlite3.connect', MagicMock(side_effect=sqlite3.DatabaseError('error text')))
+@patch('pathlib.Path.unlink', MagicMock(side_effect=PermissionError))
+def test_connect_database_error_suppress_permission_error(process_manager):
+    # All OperationalError exceptions except "unable to open database file" should be re-raised
+    with pytest.raises(sqlite3.DatabaseError, match=r'^error text$'):
+        with process_manager.connect():
+            pass  # pragma: no cover
+
+
+@patch('sqlite3.connect', MagicMock(side_effect=sqlite3.DatabaseError('error text')))
+def test_connect_database_error_raise_non_permission_error(process_manager):
+
+    class TestError(Exception):
+        pass
+
+    with patch('pathlib.Path.unlink', MagicMock(side_effect=TestError)):
+        # All OperationalError exceptions except "unable to open database file" should be re-raised
+        with pytest.raises(TestError):
+            with process_manager.connect():
+                pass  # pragma: no cover