Merge pull request #7716 from drew2a/fix/7712

Refactor disk usage tracking and handling

src/tribler/core/components/resource_monitor/implementation/core.py

@@ -1,5 +1,6 @@
 import time
 from collections import deque
+from typing import NamedTuple, Optional
 
 import psutil
 from ipv8.taskmanager import TaskManager
@@ -64,24 +65,25 @@ def record_disk_usage(self, recorded_at=None):
         recorded_at = recorded_at or time.time()
 
         # Check for available disk space
-        disk_usage = self.get_free_disk_space()
-        self.disk_usage_data.append({"time": recorded_at,
-                                     "total": disk_usage.total,
-                                     "used": disk_usage.used,
-                                     "free": disk_usage.free,
-                                     "percent": disk_usage.percent})
+        if disk_usage := self.get_disk_usage():
+            self.disk_usage_data.append(
+                {
+                    "time": recorded_at,
+                    "total": disk_usage.total,
+                    "used": disk_usage.used,
+                    "free": disk_usage.free,
+                    "percent": disk_usage.percent
+                }
+            )
 
         # Notify session if less than 100MB of disk space is available
         if disk_usage.free < FREE_DISK_THRESHOLD:
             self._logger.warning("Warning! Less than 100MB of disk space available")
             if self.notifier:
                 self.notifier[notifications.low_space](self.disk_usage_data[-1])
 
-    def get_free_disk_space(self):
-        return psutil.disk_usage(str(self.state_dir))
-
-    def get_disk_usage(self):
-        """
-        Return a list containing the history of free disk space
-        """
-        return self.disk_usage_data
+    def get_disk_usage(self) -> Optional[NamedTuple]:
+        try:
+            return psutil.disk_usage(str(self.state_dir))
+        except OSError:
+            return None

src/tribler/core/components/resource_monitor/implementation/tests/test_resource_monitor.py

@@ -1,6 +1,6 @@
 import os
 from collections import deque, namedtuple
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 
@@ -52,7 +52,7 @@ def test_get_history_dicts(resource_monitor):
     memory_dict = resource_monitor.get_memory_history_dict()
     assert isinstance(memory_dict, list)
 
-    disk_usage_history = resource_monitor.get_disk_usage()
+    disk_usage_history = resource_monitor.disk_usage_data
     assert isinstance(disk_usage_history, deque)
 
 
@@ -83,12 +83,19 @@ def fake_get_free_disk_space():
 
     resource_monitor.notifier = MagicMock()
 
-    resource_monitor.get_free_disk_space = fake_get_free_disk_space
+    resource_monitor.get_disk_usage = fake_get_free_disk_space
     resource_monitor.check_resources()
     resource_monitor.notifier[notifications.low_space].assert_called()
     resource_monitor.notifier[notifications.tribler_shutdown_state].assert_called()
 
 
+@patch('tribler.core.components.resource_monitor.implementation.core.psutil.disk_usage',
+       Mock(side_effect=FileNotFoundError))
+def test_get_disk_usage_exception(resource_monitor: CoreResourceMonitor):
+    # Test if disk usage is None when psutil.disk_usage raises an exception
+    assert not resource_monitor.get_disk_usage()
+
+
 def test_enable_resource_log(resource_monitor):
     resource_monitor.set_resource_log_enabled(True)
     assert resource_monitor.is_resource_log_enabled()