diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml new file mode 100644 index 0000000..6f9e1e6 --- /dev/null +++ b/.github/FUNDING.yml @@ -0,0 +1,3 @@ +# These are supported funding model platforms + +github: JeanExtreme002 diff --git a/PyMemoryEditor/app/cheat_poll_worker.py b/PyMemoryEditor/app/cheat_poll_worker.py index cf4b84b..0abcda1 100644 --- a/PyMemoryEditor/app/cheat_poll_worker.py +++ b/PyMemoryEditor/app/cheat_poll_worker.py @@ -62,12 +62,16 @@ class _CheatPollWorker(QThread): values_ready = Signal(object) # list[tuple[int, type, int, Any]] freeze_failed = Signal(object) # dict[_EntryKey, str] — current failing freezes + write_failed = Signal(object) # tuple[int, type, int, str] — a manual write that failed def __init__(self, process: AbstractProcess, parent=None): super().__init__(parent) self._process = process self._mutex = QMutex() self._snapshot: List[Tuple[int, type, int, Any, bool]] = [] + # One-shot manual writes queued from the UI (inline value edits). Drained + # at the top of each tick so the syscall runs here, not on the UI thread. + self._pending_writes: List[Tuple[int, type, int, Any]] = [] self._stop = False # Entries whose freeze write is currently failing → last error string. # Touched only by the worker thread (run() / _poll_once), so no lock. @@ -87,10 +91,52 @@ def update_snapshot( with QMutexLocker(self._mutex): self._snapshot = list(snapshot) + def request_write( + self, address: int, pytype: type, length: int, value: Any + ) -> None: + """Queue a one-shot manual write to be performed on the worker thread. + + The cheat table's inline value edits used to call + ``write_process_memory`` directly on the UI thread, freezing the UI when + the target was slow or the page faulted. Routing them here keeps the + syscall off the UI thread; a write that fails comes back via the + ``write_failed`` signal. Latency is at most one tick (~100 ms), which is + imperceptible for a manual edit. + """ + with QMutexLocker(self._mutex): + self._pending_writes.append((address, pytype, length, value)) + def stop(self) -> None: with QMutexLocker(self._mutex): self._stop = True + def _drain_pending_writes(self) -> List[Tuple[int, type, int, str]]: + """Perform and clear every queued manual write. + + Returns one ``(address, pytype, length, message)`` tuple per write that + failed, so ``run`` can surface it via ``write_failed``. Reads and clears + the queue under the lock, then runs the syscalls outside it so a slow + write never blocks a UI thread enqueuing the next edit. + """ + with QMutexLocker(self._mutex): + if not self._pending_writes: + return [] + pending = self._pending_writes + self._pending_writes = [] + + failures: List[Tuple[int, type, int, str]] = [] + for address, pytype, length, value in pending: + try: + self._process.write_process_memory(address, pytype, length, value) + except Exception as exc: # noqa: BLE001 + message = "%s: %s" % (type(exc).__name__, exc) + failures.append((address, pytype, length, message)) + _LOG.warning( + "Cheat-table write failed at 0x%X (%s, %dB): %s", + address, pytype.__name__, length, message, + ) + return failures + def run(self) -> None: while True: with QMutexLocker(self._mutex): @@ -98,18 +144,30 @@ def run(self) -> None: return snapshot = list(self._snapshot) - if snapshot: - results = self._poll_once(snapshot) - if results: - self.values_ready.emit(results) - - # Emit the freeze-failure state only when it changed since the last - # tick — a frozen page that keeps failing shouldn't fire the signal - # 10×/second, but the UI must learn the moment one starts or stops - # failing (including recovering back to an empty map). - if self._freeze_failures != self._last_emitted_failures: - self._last_emitted_failures = dict(self._freeze_failures) - self.freeze_failed.emit(dict(self._freeze_failures)) + # Never let an exception escape the loop body: a QThread whose + # run() raises dies silently, and the table would then stop + # auto-updating forever while the rest of the app (separate + # workers) keeps working. Catch, log the cause, and keep ticking. + try: + # Apply queued manual writes before reading, so a just-typed + # value lands and is read back as the new current value this tick. + for failure in self._drain_pending_writes(): + self.write_failed.emit(failure) + + if snapshot: + results = self._poll_once(snapshot) + if results: + self.values_ready.emit(results) + + # Emit the freeze-failure state only when it changed since the + # last tick — a frozen page that keeps failing shouldn't fire the + # signal 10×/second, but the UI must learn the moment one starts + # or stops failing (including recovering back to an empty map). + if self._freeze_failures != self._last_emitted_failures: + self._last_emitted_failures = dict(self._freeze_failures) + self.freeze_failed.emit(dict(self._freeze_failures)) + except Exception: # noqa: BLE001 + _LOG.exception("Cheat poll tick failed; continuing") QThread.msleep(TICK_INTERVAL_MS) diff --git a/PyMemoryEditor/app/cheat_table.py b/PyMemoryEditor/app/cheat_table.py index 580dedb..88878ef 100644 --- a/PyMemoryEditor/app/cheat_table.py +++ b/PyMemoryEditor/app/cheat_table.py @@ -17,7 +17,6 @@ """ import copy import json -import logging from typing import Dict, List, Optional, Tuple from PySide6.QtCore import Qt, QTimer, Signal @@ -56,9 +55,6 @@ # poll-interval constant from this module before the split. _TICK_INTERVAL_MS = TICK_INTERVAL_MS -# Child of "PyMemoryEditor" — the Log Console captures these via propagation. -_LOG = logging.getLogger(__name__) - class CheatTable(QWidget): """Bottom pane: saved addresses, freezing, manual edits.""" @@ -85,6 +81,7 @@ def __init__(self, process: AbstractProcess, parent=None): self._poller = _CheatPollWorker(process, self) self._poller.values_ready.connect(self._on_values_ready) self._poller.freeze_failed.connect(self._on_freeze_failed) + self._poller.write_failed.connect(self._on_write_failed) self._poller.start() # A short cadence to push fresh entry snapshots into the worker. This @@ -314,23 +311,15 @@ def _on_cell_changed(self, row: int, column: int) -> None: self._suspend_signals = False return - try: - self._process.write_process_memory( - entry.address, entry.spec.pytype, entry.length, value - ) - except Exception as exc: # noqa: BLE001 - QMessageBox.critical( - self, "Write Failed", f"{type(exc).__name__}: {exc}" - ) - _LOG.warning( - "Cheat-table write failed at 0x%X (%s, %dB): %s: %s", - entry.address, - entry.spec.pytype.__name__, - entry.length, - type(exc).__name__, - exc, - ) - return + # Route the write through the poll worker so the syscall never runs + # on the UI thread — a slow target or a page fault would otherwise + # freeze the interface. The local update below is optimistic: the + # next poll tick reads the value back and corrects the cell if the + # write didn't take, and an outright failure returns via + # write_failed → _on_write_failed. + self._poller.request_write( + entry.address, entry.spec.pytype, entry.length, value + ) entry.last_value = value if entry.frozen: @@ -424,6 +413,17 @@ def _on_freeze_failed(self, failures: dict) -> None: finally: self._suspend_signals = False + def _on_write_failed(self, failure) -> None: + """Surface a manual value write that failed on the worker thread. + + The write is now async (queued onto the poll worker), so its failure + arrives here via the ``write_failed`` signal rather than inline. The + optimistic cell update made when the edit was queued is corrected by the + next poll tick, which reads back the unchanged value. + """ + _address, _pytype, _length, message = failure + QMessageBox.critical(self, "Write Failed", message) + def _editing_row(self) -> int: """Return the row currently being edited, or -1 if none.""" if self._table.state() != QAbstractItemView.EditingState: diff --git a/PyMemoryEditor/app/main_window.py b/PyMemoryEditor/app/main_window.py index 6af63a7..bed4abe 100644 --- a/PyMemoryEditor/app/main_window.py +++ b/PyMemoryEditor/app/main_window.py @@ -344,21 +344,16 @@ def _on_first_scan(self, request: ScanRequest) -> None: if self._worker is not None: return - # Build a cached region snapshot the first time the user asks for one. - if self._scanner.use_snapshot_cache() and self._region_snapshot is None: - try: - self._region_snapshot = self._process.snapshot_memory_regions() - except Exception as exc: # noqa: BLE001 - QMessageBox.warning( - self, - "Memory regions", - f"Could not cache memory regions ({exc}). Continuing without cache.", - ) - self._region_snapshot = None - - request.memory_regions = ( - self._region_snapshot if self._scanner.use_snapshot_cache() else None - ) + # Decide how the worker sources its region list. When the user wants the + # snapshot cache and we don't have one yet, let the worker build it off + # the UI thread (region enumeration can be slow on a large target) and + # hand it back via snapshot_ready for reuse by the refine scans. + if self._scanner.use_snapshot_cache(): + request.memory_regions = self._region_snapshot + request.build_snapshot = self._region_snapshot is None + else: + request.memory_regions = None + request.build_snapshot = False self._results_model.clear() self._results_model.set_value_spec(request.spec) @@ -377,6 +372,9 @@ def _on_first_scan(self, request: ScanRequest) -> None: worker.status.connect(self._status.showMessage) worker.error.connect(self._on_worker_error) worker.finished_ok.connect(self._on_first_scan_done) + # The worker builds the region snapshot off the UI thread; cache it here + # for the refine/update scans that reuse it. + worker.snapshot_ready.connect(self._on_snapshot_ready) # Connection order matters: _cleanup_worker must clear self._worker # before _fill_initial_values runs, otherwise the busy guard in # _on_update_values rejects the auto-refresh. @@ -695,6 +693,10 @@ def _open_hex_viewer_with_size(self, address: int, size: int) -> None: self._hex_viewers.append(viewer) viewer.show() + def _on_snapshot_ready(self, snapshot) -> None: + """Cache the region snapshot the FirstScanWorker built off-thread.""" + self._region_snapshot = snapshot + def _refresh_region_snapshot(self) -> None: try: self._region_snapshot = self._process.snapshot_memory_regions() diff --git a/PyMemoryEditor/app/scan_worker.py b/PyMemoryEditor/app/scan_worker.py index 9906246..2d0b303 100644 --- a/PyMemoryEditor/app/scan_worker.py +++ b/PyMemoryEditor/app/scan_worker.py @@ -71,6 +71,11 @@ class ScanRequest: # Optional cached snapshot of memory regions, reused across scans to skip # the region enumeration step. Pass None to let the backend enumerate. memory_regions: Optional[Sequence[MemoryRegion]] = None + # When True and no snapshot is supplied, FirstScanWorker builds one itself + # (off the UI thread) and emits it via ``snapshot_ready`` so the caller can + # cache it for later scans. Building it here keeps the (potentially slow) + # region enumeration off the UI thread. + build_snapshot: bool = False def build_scan_request( @@ -172,6 +177,10 @@ def cancel(self) -> None: class FirstScanWorker(_BaseWorker): """Performs the very first scan, finding every address that matches.""" + # Emitted (off the UI thread) with the freshly built MemoryRegionSnapshot + # when ``request.build_snapshot`` was set, so the owner can cache it. + snapshot_ready = Signal(object) + def __init__(self, process: AbstractProcess, request: ScanRequest, parent=None): super().__init__(process, parent) self._request = request @@ -179,6 +188,23 @@ def __init__(self, process: AbstractProcess, request: ScanRequest, parent=None): def run(self) -> None: req = self._request try: + # Build the region snapshot here rather than on the UI thread: the + # enumeration can be slow on a large target and would otherwise + # stall the scan dialog. Emit it so the owner can cache it for the + # refine/update scans that follow. + if req.memory_regions is None and req.build_snapshot: + try: + req.memory_regions = self._process.snapshot_memory_regions() + self.snapshot_ready.emit(req.memory_regions) + except Exception as exc: # noqa: BLE001 + # Snapshot is an optimization; fall back to per-scan + # enumeration rather than failing the whole scan. + _LOG.warning( + "Region snapshot failed; scanning without cache: %s: %s", + type(exc).__name__, + exc, + ) + req.memory_regions = None # Pattern path: req.value is the IDA-style string or a bytes regex, # routed through search_by_pattern. req.length carries byte_length — # ignored for IDA strings (inferred from the token count), required diff --git a/PyMemoryEditor/macos/functions.py b/PyMemoryEditor/macos/functions.py index e64893b..5d5931a 100644 --- a/PyMemoryEditor/macos/functions.py +++ b/PyMemoryEditor/macos/functions.py @@ -9,7 +9,8 @@ import logging import os import warnings -from typing import Generator, List, Optional, Sequence, Tuple, Type, TypeVar, Union +from contextlib import contextmanager +from typing import Generator, Iterator, List, Optional, Sequence, Tuple, Type, TypeVar, Union from ..enums import ScanTypesEnum from ..process.module_info import ModuleInfo @@ -148,7 +149,17 @@ def get_task_for_pid(pid: int) -> int: def release_task(task: int) -> None: """Release a task port. No-op for mach_task_self_.""" if task and task != mach_task_self_.value: - libsystem.mach_port_deallocate(mach_task_self_.value, task) + kr = libsystem.mach_port_deallocate(mach_task_self_.value, task) + if kr != KERN_SUCCESS: + # A failed deallocate leaks the send right for the lifetime of the + # process. It shouldn't happen for a port we own, but swallowing it + # silently hid real leaks — log it so they're at least visible. + _logger.warning( + "mach_port_deallocate failed for task port %d: %s (kr=%d)", + task, + mach_error_message(kr), + kr, + ) def _region_user_tag(task: int, address: int) -> int: @@ -326,40 +337,33 @@ def _mach_read(task: int, address: int, local_buffer_address: int, size: int) -> return out_size.value -def _mach_write(task: int, address: int, local_buffer_address: int, size: int) -> None: +@contextmanager +def _elevated_write_protection( + task: int, address: int, size: int, *, original_kr: int +) -> Iterator[None]: """ - Write `size` bytes from `local_buffer_address` to `address`. - - On read-only pages, mach_vm_write returns KERN_PROTECTION_FAILURE. This - helper transparently elevates the page protection to RW (using VM_PROT_COPY - so the change is private to the target task), performs the write, and - restores the original protection. This mirrors the practical behavior of - WriteProcessMemory on Windows. - - ``mach_vm_write``'s ``data_count`` parameter is a 32-bit - ``mach_msg_type_number_t`` per the Mach interface; reject ``size`` values - that would silently truncate at the kernel boundary instead of letting - them slip through. + Temporarily make the page span covering ``(address, size)`` read+write, + restoring its original protection on exit — the protect-flip that lets a + write land on a read-only page (mirroring WriteProcessMemory on Windows). + + The elevate uses ``VM_PROT_COPY`` so the change is private to the target + task. ``original_kr`` is the kern_return_t of the write that triggered this + retry; it's woven into the error if the address turns out to be genuinely + invalid, so the caller still sees the *original* failure rather than a + misleading "protect failed". + + :raises OSError: if the region can't be queried (invalid address) or the + protection can't be elevated. A failure to *restore* on exit is logged + and warned (the write already happened) but never raised. """ - if size > _MACH_WRITE_MAX_SIZE: - raise OverflowError( - "mach_vm_write size %d exceeds UINT32_MAX — the Mach interface " - "would silently truncate. Split the write into smaller chunks." - % size - ) - - kr = libsystem.mach_vm_write(task, address, local_buffer_address, size) - if kr == KERN_SUCCESS: - return - - if kr not in _WRITE_RETRY_CODES: - raise OSError("mach_vm_write failed: %s (kr=%d)" % (mach_error_message(kr), kr)) - - # Try to discover the page's original protection so we can restore it. + # Discover the page's original protection so we can restore it afterwards. region = _query_region(task, address) if region is None: - # The address really is invalid — surface the original error. - raise OSError("mach_vm_write failed: %s (kr=%d)" % (mach_error_message(kr), kr)) + # The address really is invalid — surface the original write error. + raise OSError( + "mach_vm_write failed: %s (kr=%d)" + % (mach_error_message(original_kr), original_kr) + ) original_protection = region.struct.Protection @@ -376,16 +380,11 @@ def _mach_write(task: int, address: int, local_buffer_address: int, size: int) - raise OSError( "mach_vm_write failed (kr=%d) and mach_vm_protect could not elevate " "the protection (kr=%d, %s)." - % (kr, protect_kr, mach_error_message(protect_kr)) + % (original_kr, protect_kr, mach_error_message(protect_kr)) ) try: - kr = libsystem.mach_vm_write(task, address, local_buffer_address, size) - if kr != KERN_SUCCESS: - raise OSError( - "mach_vm_write failed after protect: %s (kr=%d)" - % (mach_error_message(kr), kr) - ) + yield finally: # Best-effort restore. The write itself already succeeded, so raising # here would discard the user's intended outcome; but a silent failure @@ -410,6 +409,44 @@ def _mach_write(task: int, address: int, local_buffer_address: int, size: int) - warnings.warn(message, ResourceWarning, stacklevel=2) +def _mach_write(task: int, address: int, local_buffer_address: int, size: int) -> None: + """ + Write `size` bytes from `local_buffer_address` to `address`. + + On read-only pages, mach_vm_write returns KERN_PROTECTION_FAILURE. This + helper retries under :func:`_elevated_write_protection`, which transparently + flips the page to RW, lets the write land, and restores the original + protection. This mirrors the practical behavior of WriteProcessMemory on + Windows. + + ``mach_vm_write``'s ``data_count`` parameter is a 32-bit + ``mach_msg_type_number_t`` per the Mach interface; reject ``size`` values + that would silently truncate at the kernel boundary instead of letting + them slip through. + """ + if size > _MACH_WRITE_MAX_SIZE: + raise OverflowError( + "mach_vm_write size %d exceeds UINT32_MAX — the Mach interface " + "would silently truncate. Split the write into smaller chunks." + % size + ) + + kr = libsystem.mach_vm_write(task, address, local_buffer_address, size) + if kr == KERN_SUCCESS: + return + + if kr not in _WRITE_RETRY_CODES: + raise OSError("mach_vm_write failed: %s (kr=%d)" % (mach_error_message(kr), kr)) + + with _elevated_write_protection(task, address, size, original_kr=kr): + retry_kr = libsystem.mach_vm_write(task, address, local_buffer_address, size) + if retry_kr != KERN_SUCCESS: + raise OSError( + "mach_vm_write failed after protect: %s (kr=%d)" + % (mach_error_message(retry_kr), retry_kr) + ) + + def _make_read_chunk(task: int): """ Build a ``read_chunk(address, size)`` closure bound to ``task``. @@ -578,8 +615,9 @@ def search_addresses_by_value( memory_regions: Optional[Sequence[MemoryRegion]] = None, ) -> Generator[Union[int, Tuple[int, dict]], None, None]: """ - Walk every readable region of the task and yield addresses whose value - matches the scan criteria. + Walk every readable, non-shared region of the task and yield addresses + whose value matches the scan criteria. Shared / file-backed mappings + (including the dyld shared cache) are skipped via ``default_scan_filter``. Passing a `memory_regions` snapshot skips region enumeration. """ @@ -947,7 +985,7 @@ def search_addresses_by_pattern( memory_regions: Optional[Sequence[MemoryRegion]] = None, ) -> Generator[Union[int, Tuple[int, dict]], None, None]: """ - AOB scan against every readable region of the target task. See + AOB scan against every readable, non-shared region of the target task. See :meth:`AbstractProcess.search_by_pattern`. """ compiled, length = compile_pattern(pattern, byte_length=byte_length) diff --git a/PyMemoryEditor/process/abstract.py b/PyMemoryEditor/process/abstract.py index 7343ca4..c3f6a5c 100644 --- a/PyMemoryEditor/process/abstract.py +++ b/PyMemoryEditor/process/abstract.py @@ -442,6 +442,10 @@ def read_process_memory( raising ``UnicodeDecodeError``. This matches ``search_by_addresses`` and ``convert_from_byte_array``. Callers that need the original bytes verbatim (no decoding) should pass ``pytype=bytes``. + + :raises ValueError: if ``pytype`` is unsupported. + :raises OSError: if the read fails, or returns fewer bytes than + requested (e.g. the range crosses an unreadable/freed page). """ raise NotImplementedError() @@ -479,6 +483,11 @@ def write_process_memory( Positional calls keep working unchanged (``write_process_memory(address, int, 4, 99)``). + :raises ValueError: if ``pytype`` is unsupported, or if an ``int`` + ``value`` does not fit in ``bufflength`` bytes (rejected up front + rather than silently truncated). + :raises OSError: if the write fails, or fewer bytes than requested are + written (e.g. the range crosses an unwritable/freed page). :return: the original ``value`` passed in. """ raise NotImplementedError() diff --git a/PyMemoryEditor/win32/functions.py b/PyMemoryEditor/win32/functions.py index 381ae62..af6a17f 100644 --- a/PyMemoryEditor/win32/functions.py +++ b/PyMemoryEditor/win32/functions.py @@ -337,13 +337,20 @@ def GetMemoryRegions(process_handle: int) -> Generator[MemoryRegion, None, None] walk keeps making progress. The walk bounds come from ``GetNativeSystemInfo`` rather than - ``GetSystemInfo``: from a 32-bit (WOW64) Python attached to a 64-bit target, - ``GetSystemInfo`` reports the *caller's* 2/4 GB ceiling, which would stop the - walk early and silently drop every region above it in the target. The native - info reports the true address-space ceiling. For a 32-bit target the extra - range is empty — ``VirtualQueryEx`` returns ERROR_INVALID_PARAMETER past the - 32-bit boundary and the loop terminates there — so using the native ceiling - never over-enumerates. (From a 64-bit Python the two infos are identical.) + ``GetSystemInfo`` so that, from a 64-bit Python attached to a 32-bit target, + the ceiling reflects the host's full address space. For a 32-bit target the + extra range is empty — ``VirtualQueryEx`` returns ERROR_INVALID_PARAMETER + past the 32-bit boundary and the loop terminates there — so using the native + ceiling never over-enumerates. (From a 64-bit Python the two infos are + identical for the address-range fields anyway.) + + Note this does *not* extend the walk for a 32-bit (WOW64) Python attached to + a 64-bit target. ``lpMaximumApplicationAddress`` is ``c_void_p`` (4 bytes in + a 32-bit interpreter), so it cannot hold a >4 GB ceiling regardless of which + *SystemInfo call fills it; and ``VirtualQueryEx`` takes a pointer-width + ``lpAddress``, so a WOW64 process cannot query the target's high memory at + all. Enumerating a 64-bit target's address space above 4 GB therefore + requires a 64-bit Python — an API limitation, not something this call fixes. """ mbi_class = mbi_class_for_handle(process_handle) mem_region_begin = _native_system_information.lpMinimumApplicationAddress @@ -535,8 +542,8 @@ def SearchAddressesByPattern( memory_regions: Optional[Sequence[MemoryRegion]] = None, ) -> Generator[Union[int, Tuple[int, dict]], None, None]: """ - AOB scan against every scannable region of the target process. See - :meth:`AbstractProcess.search_by_pattern`. + AOB scan against every readable, non-shared region of the target process. + See :meth:`AbstractProcess.search_by_pattern`. """ compiled, length = compile_pattern(pattern, byte_length=byte_length) diff --git a/pyproject.toml b/pyproject.toml index 329cdf8..f71d997 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,6 +5,9 @@ description = "Read, write and scan process memory in a few lines of Python — authors = [ { name = "Jean Loui Bernard Silva de Jesus", email = "contact@jeanloui.dev" }, ] +maintainers = [ + { name = "Jean Loui Bernard Silva de Jesus", email = "contact@jeanloui.dev" }, +] license = "MIT" license-files = ["LICENSE*"] readme = "README.md" @@ -99,6 +102,7 @@ Documentation = "https://pymemoryeditor.readthedocs.io" Repository = "https://github.com/JeanExtreme002/PyMemoryEditor" Issues = "https://github.com/JeanExtreme002/PyMemoryEditor/issues" Changelog = "https://github.com/JeanExtreme002/PyMemoryEditor/releases" +Funding = "https://github.com/sponsors/JeanExtreme002" [tool.mypy] # The Qt app uses dynamic types and depends on the optional PySide6 GUI @@ -142,8 +146,11 @@ exclude = [ "/scripts", ] +# hatchling>=1.27 is required for PEP 639 metadata: the SPDX ``license = "MIT"`` +# expression and ``license-files`` glob above are only understood from that +# version on. Older hatchling fails the build with a confusing error. [build-system] -requires = ["hatchling"] +requires = ["hatchling>=1.27"] build-backend = "hatchling.build" # Coverage scope: the Qt app is excluded — it's exercised manually, not by diff --git a/tests/test_cheat_poll_worker.py b/tests/test_cheat_poll_worker.py index 7dcff58..2028b8e 100644 --- a/tests/test_cheat_poll_worker.py +++ b/tests/test_cheat_poll_worker.py @@ -266,3 +266,40 @@ def test_empty_snapshot_yields_nothing(qapp): assert process.read_calls == [] assert process.batch_calls == [] assert process.write_calls == [] + + +def test_request_write_is_queued_and_drained(qapp): + """A manual write is queued and performed by the worker, not inline. + + Routing the cheat-table's inline value edit through the worker keeps the + write_process_memory syscall off the UI thread. + """ + process = _FakeProcess() + worker = _make_worker(process) + + # Queuing alone performs no syscall. + worker.request_write(0x1000, int, 4, 77) + assert process.write_calls == [] + + failures = worker._drain_pending_writes() + + assert failures == [] + assert process.write_calls == [(0x1000, int, 4, 77)] + # The queue is cleared after draining. + assert worker._drain_pending_writes() == [] + assert process.write_calls == [(0x1000, int, 4, 77)] + + +def test_request_write_failure_is_reported(qapp): + """A manual write that fails comes back as a (key, message) failure tuple + so the UI can surface it via the write_failed signal.""" + process = _FakeProcess(raise_on_write=True) + worker = _make_worker(process) + + worker.request_write(0x2000, int, 4, 5) + failures = worker._drain_pending_writes() + + assert len(failures) == 1 + address, pytype, length, message = failures[0] + assert (address, pytype, length) == (0x2000, int, 4) + assert "OSError" in message