Merge pull request #7612 from kozlovsky/fix/unable_to_open_processes_…

…database

Add the reason for the OperationalError 'unable to open database file' when opening processes.sqlite

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

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import logging
+import os
 import sqlite3
 import sys
+import time
 from contextlib import contextmanager
 from pathlib import Path
 from threading import Lock
@@ -72,12 +74,37 @@ def connect(self) -> ContextManager[sqlite3.Connection]:
 
         except Exception as e:
             logger.exception(f'{e.__class__.__name__}: {e}')
+
             if connection:
                 connection.close()
+
             if isinstance(e, sqlite3.DatabaseError):
                 self.db_filepath.unlink(missing_ok=True)
+
+            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
 
+    def _unable_to_open_db_file_get_reason(self):
+        dir_path = self.db_filepath.parent
+        if not dir_path.exists():
+            return f'parent directory `{dir_path}` does not exist'
+
+        if not os.access(dir_path, os.W_OK):
+            return f'the process does not have write permissions to the directory `{dir_path}`'
+
+        try:
+            tmp_filename = dir_path / f'tmp_{int(time.time())}.txt'
+            with tmp_filename.open('w') as f:
+                f.write('test')
+            tmp_filename.unlink()
+        except Exception as e2:  # pylint: disable=broad-except
+            return f'{e2.__class__.__name__}: {e2}'
+
+        return 'unknown reason'
+
     def primary_process_rowid(self, kind: ProcessKind) -> Optional[int]:
         """
         A helper method to load the current primary process of the specified kind from the database.

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

@@ -1,10 +1,16 @@
+import sqlite3
 import time
 from unittest.mock import Mock, patch
 
-from tribler.core.utilities.process_manager.manager import ProcessManager, logger
+import pytest
+
+from tribler.core.utilities.process_manager.manager import DB_FILENAME, ProcessManager, logger
 from tribler.core.utilities.process_manager.process import ProcessKind, TriblerProcess
 
 
+# pylint: disable=protected-access
+
+
 def test_become_primary(process_manager: ProcessManager):
     # Initially process manager fixture creates a primary current process that is a single process in DB
     p1 = process_manager.current_process
@@ -137,3 +143,39 @@ def test_delete_old_records_2(process_manager):
     with process_manager.connect() as connection:
         # Only the current primary process and the last 100 processes should remain
         assert connection.execute("select count(*) from processes").fetchone()[0] == 101
+
+
+def test_unable_to_open_db_file_get_reason_unknown_reason(process_manager):
+    reason = process_manager._unable_to_open_db_file_get_reason()
+    assert reason == 'unknown reason'
+
+
+def test_unable_to_open_db_file_get_reason_unable_to_write(process_manager):
+    class TestException(Exception):
+        pass
+
+    with patch('pathlib.Path.open', side_effect=TestException('exception text')):
+        reason = process_manager._unable_to_open_db_file_get_reason()
+        assert reason == 'TestException: exception text'
+
+
+def test_unable_to_open_db_file_get_reason_no_write_permissions(process_manager):
+    with patch('os.access', return_value=False):
+        reason = process_manager._unable_to_open_db_file_get_reason()
+        assert reason.startswith('the process does not have write permissions to the directory')
+
+
+def test_unable_to_open_db_file_get_reason_directory_does_not_exist(process_manager):
+    process_manager.root_dir /= 'non_existent_subdir'
+    process_manager.db_filepath = process_manager.root_dir / DB_FILENAME
+    reason = process_manager._unable_to_open_db_file_get_reason()
+    assert reason.startswith('parent directory') and reason.endswith('does not exist')
+
+
+def test_unable_to_open_db_file_reason_added(process_manager):
+    process_manager.root_dir /= 'non_existent_subdir'
+    process_manager.db_filepath = process_manager.root_dir / DB_FILENAME
+    with pytest.raises(sqlite3.OperationalError,
+                       match=r'^unable to open database file: parent directory.*does not exist$'):
+        with process_manager.connect():
+            pass