Minor search module refactoring

Author: PabloLec

recoverpy/lib/search/grep_consumer.py

@@ -8,12 +8,12 @@
 from recoverpy.models.search_params import SearchParams
 
 
-class ResultProcessor:
+class ResultFilter:
     def __init__(self, search_params: SearchParams):
         self.search_params = search_params
 
-    def get_new_results(self, queue_object: Queue[bytes]) -> List[str]:
-        """Consume grep output queue and format results."""
+    def filter_results(self, queue_object: Queue[bytes]) -> List[str]:
+        """Consume raw grep results, filter out false positives and return valid results."""
 
         queue_list: List[bytes] = list(queue_object.queue)
         queue_size = len(queue_list)
@@ -27,12 +27,13 @@ def get_new_results(self, queue_object: Queue[bytes]) -> List[str]:
             return decoded_results
 
         final_results = [
-            result for result in decoded_results if self.is_result_format_valid(result)
+            result for result in decoded_results if self._is_result_valid(result)
         ]
         log.debug(f"result_processor - Found {len(final_results)} new results")
         return final_results
 
-    def is_result_format_valid(self, result: str) -> bool:
+    def _is_result_valid(self, result: str) -> bool:
+        """Check if result contains all searched lines."""
         inode = int(get_inode(result))
         block_factor = self.search_params.block_size * 8
 
@@ -42,6 +43,7 @@ def is_result_format_valid(self, result: str) -> bool:
         )
 
     def _get_combined_block_output(self, inode: int, block_factor: int) -> str:
+        """Get combined output of current and next block."""
         block_index = inode // block_factor
         block_output = get_dd_output(
             self.search_params.partition, block_factor, block_index
@@ -51,22 +53,3 @@ def _get_combined_block_output(self, inode: int, block_factor: int) -> str:
         )
 
         return decode_result(block_output) + decode_result(next_block_output)
-
-    def fix_line_start(self, line: str) -> str:
-        result_index: int = line.find(self.search_params.searched_lines[0])
-        return line[min(result_index, 15) :]
-
-    def fix_inode(self, inode: int) -> int:
-        inode //= self.search_params.block_size
-
-        for _ in range(10):
-            dd_output = self._get_dd_output(inode)
-            if self.search_params.searched_lines[0] in decode_result(dd_output):
-                return inode
-            inode += 1
-        return inode
-
-    def _get_dd_output(self, inode: int) -> bytes:
-        return get_dd_output(
-            self.search_params.partition, self.search_params.block_size, inode
-        )

recoverpy/lib/search/result_processor.py recoverpy/lib/search/result_filter.pyrecoverpy/lib/search/result_processor.py renamed to recoverpy/lib/search/result_filter.py

@@ -3,24 +3,33 @@
 from asyncio import AbstractEventLoop
 from asyncio import Queue as AsyncQueue
 from asyncio import new_event_loop
+from io import BufferedReader
 from queue import Queue
 from subprocess import Popen
 from time import sleep
 from typing import List
 
-from recoverpy.lib.search.result_processor import ResultProcessor
+from recoverpy.lib.helper import get_dd_output, decode_result
+from recoverpy.lib.search.result_filter import ResultFilter
 from recoverpy.lib.search.thread_factory import (
     start_grep_process,
     start_progress_monitoring_thread,
-    start_result_dequeue_thread,
-    start_result_enqueue_thread,
+    start_result_formatter_thread,
+    start_grep_stdout_consumer_thread,
 )
 from recoverpy.log.logger import log
 from recoverpy.models.grep_result import GrepResult
 from recoverpy.models.search_params import SearchParams
 from recoverpy.models.search_progress import SearchProgress
 
 
+def _consume_grep_stdout(out: BufferedReader, queue: Queue[bytes]) -> None:
+    log.debug("grep_consumer - Grep output enqueue thread started")
+    for line in iter(out.readline, b""):
+        queue.put(line)
+    out.close()
+
+
 class SearchEngine:
     _grep_process: Popen[bytes]
     _seen_inodes: set[int] = set()
@@ -35,35 +44,37 @@ def _initialize_search_components(
     ) -> None:
         self.search_params = SearchParams(partition, searched_string)
         self.search_progress = SearchProgress()
-        self.result_processor = ResultProcessor(self.search_params)
+        self.result_processor = ResultFilter(self.search_params)
 
     async def start_search(self) -> None:
-        self._initialize_grep_process()
+        self._start_grep_process()
         self._start_auxiliary_threads()
 
     def stop_search(self) -> None:
         self._grep_process.kill()
 
-    def _initialize_grep_process(self) -> None:
+    def _start_grep_process(self) -> None:
         self._grep_process = start_grep_process(
             searched_string=self.search_params.searched_lines[0],
             partition=self.search_params.partition,
         )
 
     def _start_auxiliary_threads(self) -> None:
-        start_result_enqueue_thread(self._grep_process, self.results_queue)
-        start_result_dequeue_thread(self._dequeue_results)
+        start_grep_stdout_consumer_thread(
+            _consume_grep_stdout, self._grep_process, self.results_queue
+        )
+        start_result_formatter_thread(self._format_results)
         start_progress_monitoring_thread(self._grep_process, self.search_progress)
 
-    def _dequeue_results(self) -> None:
+    def _format_results(self) -> None:
         loop = new_event_loop()
         while True:
-            results = self.result_processor.get_new_results(self.results_queue)
-            self._add_new_results(results, loop)
+            results = self.result_processor.filter_results(self.results_queue)
+            self._process_new_results(results, loop)
             log.debug(f"search_engine - Dequeued {len(results)} results")
             sleep(0.1)
 
-    def _add_new_results(self, results: List[str], loop: AbstractEventLoop) -> None:
+    def _process_new_results(self, results: List[str], loop: AbstractEventLoop) -> None:
         for result in results:
             grep_result = self._create_grep_result(
                 result, self.search_progress.result_count
@@ -83,8 +94,24 @@ def _create_grep_result(self, result: str, result_index: int) -> GrepResult:
     def _configure_grep_result(
         self, grep_result: GrepResult, result_index: int
     ) -> None:
-        grep_result.inode = self.result_processor.fix_inode(grep_result.inode)
-        grep_result.line = self.result_processor.fix_line_start(grep_result.line)
+        grep_result.inode = self._fix_inode(grep_result.inode)
+        grep_result.line = self._fix_line_start(grep_result.line)
         grep_result.css_class = (
             "grep-result-odd" if result_index % 2 == 0 else "grep-result-even"
         )
+
+    def _fix_line_start(self, line: str) -> str:
+        result_index: int = line.find(self.search_params.searched_lines[0])
+        return line[min(result_index, 15) :]
+
+    def _fix_inode(self, inode: int) -> int:
+        inode //= self.search_params.block_size
+
+        for _ in range(10):
+            dd_output = get_dd_output(
+                self.search_params.partition, self.search_params.block_size, inode
+            )
+            if self.search_params.searched_lines[0] in decode_result(dd_output):
+                return inode
+            inode += 1
+        return inode

recoverpy/lib/search/search_engine.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
+from io import BufferedReader
 from queue import Queue
 from subprocess import PIPE, Popen
 from threading import Thread
 from typing import Callable
 
 from recoverpy.lib.helper import is_dependency_installed
-from recoverpy.lib.search.grep_consumer import enqueue_grep_output
 from recoverpy.lib.search.progress_monitoring import monitor_search_progress
 from recoverpy.log.logger import log
 from recoverpy.models.search_progress import SearchProgress
@@ -22,22 +22,24 @@ def start_grep_process(searched_string: str, partition: str) -> Popen[bytes]:
     )
 
 
-def start_result_enqueue_thread(
-    grep_process: Popen[bytes], queue: Queue[bytes]
+def start_grep_stdout_consumer_thread(
+    consume_function: Callable[[BufferedReader, Queue[bytes]], None],
+    grep_process: Popen[bytes],
+    queue: Queue[bytes],
 ) -> None:
-    log.debug("thread_factory - Starting result enqueue thread")
+    log.debug("thread_factory - Starting grep stdout consumer thread")
     Thread(
-        target=enqueue_grep_output,
+        target=consume_function,
         args=(grep_process.stdout, queue),
         daemon=True,
         name="enqueue-grep-output-thread",
     ).start()
 
 
-def start_result_dequeue_thread(dequeue_results: Callable[[], None]) -> None:
-    log.debug("thread_factory - Starting result dequeue thread")
+def start_result_formatter_thread(format_function: Callable[[], None]) -> None:
+    log.debug("thread_factory - Starting result formatter thread")
     Thread(
-        target=dequeue_results,
+        target=format_function,
         daemon=True,
         name="dequeue-results-thread",
     ).start()

recoverpy/lib/search/thread_factory.py

@@ -64,4 +64,6 @@ def on_resize(self) -> None:
     def _resize_item(self, index: int) -> None:
         max_item_width = self.size.width - self.size.width // 20
         grep_result_line = self.grep_results[index].line
-        cast(Label, self.grep_results[index].label).update(grep_result_line[:max_item_width])
+        cast(Label, self.grep_results[index].label).update(
+            grep_result_line[:max_item_width]
+        )

recoverpy/ui/widgets/grep_result_list.py

@@ -3,7 +3,6 @@
 import pytest
 
 from recoverpy.lib.search.search_engine import SearchEngine
-
 from .fixtures import (
     mock_dd_output,
     mock_grep_process,
@@ -21,7 +20,11 @@ def system_calls_mock(session_mocker):
         new=mock_grep_process.mock_start_grep_process,
     )
     session_mocker.patch(
-        "recoverpy.lib.search.result_processor.get_dd_output",
+        "recoverpy.lib.search.result_filter.get_dd_output",
+        side_effect=mock_dd_output.mock_dd_string_output,
+    )
+    session_mocker.patch(
+        "recoverpy.lib.search.search_engine.get_dd_output",
         side_effect=mock_dd_output.mock_dd_string_output,
     )
     session_mocker.patch(