Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 32 additions & 2 deletions .github/workflows/python-package.yml
Original file line number Diff line number Diff line change
Expand Up @@ -110,8 +110,29 @@ jobs:
- name: Test with pytest
env:
QT_QPA_PLATFORM: offscreen
# --cov-fail-under is the real, in-repo coverage gate (deterministic and
# independent of any external service). It's a conservative floor: the
# full suite measures ~87% on a given OS today (coverage.py only counts
# the host backend + shared code — the two foreign backends aren't
# imported, so they don't drag the number down per-platform). 65 leaves
# headroom for a less-covered backend / Python-version variance while
# still catching any large regression. Ratchet it up once the per-OS
# numbers settle on Codecov.
run: |
pytest tests -v -s -x --cov=PyMemoryEditor --cov-report=term
pytest tests -v -s -x --cov=PyMemoryEditor --cov-report=term --cov-report=xml --cov-fail-under=65
- name: Upload coverage to Codecov
# Upload from every matrix cell so Codecov merges each OS's backend
# coverage into one combined view (each platform exercises a different
# win32/linux/macos backend). Informational only — see codecov.yml — so a
# flaky upload never blocks the merge; the hard gate is --cov-fail-under
# above. Runs even when tests fail so partial coverage stays visible.
if: always()
uses: codecov/codecov-action@v5
with:
files: ./coverage.xml
flags: ${{ runner.os }}-py${{ matrix.python-version }}
token: ${{ secrets.CODECOV_TOKEN }}
fail_ci_if_error: false

# Isolated job for the optional NumPy-accelerated scan fast path (the
# `[speed]` extra). The matrix above deliberately runs WITHOUT NumPy so the
Expand Down Expand Up @@ -158,5 +179,14 @@ jobs:
- name: Test with pytest (NumPy fast path)
env:
QT_QPA_PLATFORM: offscreen
# Same conservative coverage gate as the main matrix (see that step's note).
run: |
pytest tests -v -s -x --cov=PyMemoryEditor --cov-report=term
pytest tests -v -s -x --cov=PyMemoryEditor --cov-report=term --cov-report=xml --cov-fail-under=65
- name: Upload coverage to Codecov
if: always()
uses: codecov/codecov-action@v5
with:
files: ./coverage.xml
flags: speed-Linux-py3.12
token: ${{ secrets.CODECOV_TOKEN }}
fail_ci_if_error: false
2 changes: 1 addition & 1 deletion CONTRIBUTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ source venv/bin/activate # On Windows: venv\Scripts\activate
make install-dev # pip install -e ".[dev]"
```

The `dev` extra includes `pytest`, `pytest-cov`, `flake8`, `mypy`, `build` and `twine`.
The `dev` extra installs the full toolchain and the GUI deps.

The `Makefile` is the single source of truth for the dev commands below — run
`make help` to see every target (docs, build, coverage, etc.). The raw command
Expand Down
2 changes: 2 additions & 0 deletions PyMemoryEditor/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from .process.abstract import AbstractProcess
from .process.errors import (
AmbiguousProcessNameError,
BitnessDetectionError,
ClosedProcess,
ProcessIDNotExistsError,
ProcessNotFoundError,
Expand Down Expand Up @@ -85,6 +86,7 @@
__all__ = (
"AbstractProcess",
"AmbiguousProcessNameError",
"BitnessDetectionError",
"ClosedProcess",
"MemoryRegion",
"MemoryRegionSnapshot",
Expand Down
131 changes: 131 additions & 0 deletions PyMemoryEditor/app/_auto_refresh_dialog.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
# -*- coding: utf-8 -*-
"""
Shared base for the read-only, auto-refreshing table dialogs (Memory Map,
Modules, Threads).

All three did the same dance — run a one-shot read off the UI thread, repopulate
a sortable table when it returns, poll on a timer, and tear the worker down
safely on close — each with its own near-identical ``_*Worker(QThread)`` and
lifecycle boilerplate. That triplicated the fiddly bits (the in-flight guard
that self-throttles the poll, the detach-on-shutdown safety) where a fix to one
copy wouldn't reach the others. This owns the lifecycle once; subclasses provide
only what differs: how to fetch the data, and how to render it.
"""
from typing import Callable, Optional

from PySide6.QtCore import QThread, QTimer, Signal
from PySide6.QtWidgets import QDialog

from ._widgets import shutdown_worker_thread


class _DataWorker(QThread):
"""Runs a no-arg ``fetch`` callable off the UI thread, once."""

ready = Signal(object)
failed = Signal(str)

def __init__(self, fetch: Callable[[], object], parent=None):
super().__init__(parent)
self._fetch = fetch

def run(self) -> None:
try:
data = self._fetch()
except Exception as exc: # noqa: BLE001 — surfaced to the UI as a string
self.failed.emit(str(exc))
return
self.ready.emit(data)


class AutoRefreshTableDialog(QDialog):
"""Owns the fetch-worker lifecycle + auto-refresh timer for a table dialog.

Subclasses must:

* call ``super().__init__(process, refresh_interval_ms=..., parent=...)``,
then build their UI, then ``refresh()``, then ``_start_auto_refresh()``;
* implement :meth:`_fetch_data` (runs in the worker thread) and
:meth:`_on_data_ready` (runs on the UI thread);
* implement :meth:`_on_data_failed` and (optionally) :meth:`_set_loading_hint`.
"""

def __init__(self, process, *, refresh_interval_ms: int, parent=None):
super().__init__(parent)
self._process = process
self._worker: Optional[_DataWorker] = None
self._has_data = False
self._refresh_interval_ms = refresh_interval_ms
self._auto_timer: Optional[QTimer] = None

def _start_auto_refresh(self) -> None:
"""Begin polling. Call once, after the first ``refresh()``.

The ``refresh()`` in-flight guard self-throttles the cadence to however
long a fetch actually takes, so the timer can't stack workers on a slow
target.
"""
self._auto_timer = QTimer(self)
self._auto_timer.setInterval(self._refresh_interval_ms)
self._auto_timer.timeout.connect(self.refresh)
self._auto_timer.start()

# ------------------------------------------------------------------ #
# Hooks for subclasses
# ------------------------------------------------------------------ #

def _fetch_data(self):
"""Read the data to display. Runs in the worker thread; may raise."""
raise NotImplementedError

def _on_data_ready(self, data) -> None:
"""Render ``data`` into the table. Runs on the UI thread."""
raise NotImplementedError

def _on_data_failed(self, message: str) -> None:
"""Report a failed fetch. Runs on the UI thread."""
raise NotImplementedError

def _set_loading_hint(self) -> None:
"""Optionally show a one-time loading message before the first fetch."""

# ------------------------------------------------------------------ #
# Lifecycle (shared)
# ------------------------------------------------------------------ #

def refresh(self) -> None:
# Skip if a fetch is already in flight — the timer would otherwise stack
# workers on a slow target. This self-throttles to the real fetch time.
if self._worker is not None and self._worker.isRunning():
return

# Loading hint only before the first successful fetch; the periodic
# refresh updates silently to avoid flicker.
if not self._has_data:
self._set_loading_hint()

worker = _DataWorker(self._fetch_data, self)
worker.ready.connect(self._handle_ready)
worker.failed.connect(self._on_data_failed)
worker.finished.connect(self._on_worker_finished)
self._worker = worker
worker.start()

def _handle_ready(self, data) -> None:
self._has_data = True
self._on_data_ready(data)

def _on_worker_finished(self) -> None:
worker = self._worker
self._worker = None
if worker is not None:
worker.deleteLater()

def closeEvent(self, event): # noqa: N802 — Qt naming
if self._auto_timer is not None:
self._auto_timer.stop()
# Unhook + join the worker; if it can't stop in time it's detached
# rather than destroyed under us.
shutdown_worker_thread(self._worker, wait_ms=1000)
self._worker = None
super().closeEvent(event)
72 changes: 71 additions & 1 deletion PyMemoryEditor/app/_widgets.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
previously appeared duplicated across several dialog modules.
"""

from typing import List, Optional
from typing import Callable, Iterable, List, Optional, Tuple

from PySide6.QtCore import Qt, QThread
from PySide6.QtGui import QStandardItem
Expand Down Expand Up @@ -90,3 +90,73 @@ def parse_hex_address(text: str) -> Optional[int]:
return int(cleaned, 16)
except (TypeError, ValueError):
return None


def parse_offsets(texts: Iterable[str]) -> Optional[List[int]]:
"""Parse pointer-chain offset tokens (in order) from raw field strings.

Empty tokens are skipped; every remaining token is read as hex (with or
without ``0x``). Returns ``None`` if any non-empty token can't be parsed —
the caller treats that as "invalid chain, do nothing". Pure (no Qt) so it
can be unit-tested directly.
"""
offsets: List[int] = []
for text in texts:
if not text:
continue
parsed = parse_hex_address(text)
if parsed is None:
# parse_hex_address handles the 0x form; fall back to a plain hex
# int for ambiguous tokens like "10" (the dialog treats offsets as
# hex throughout).
try:
parsed = int(text, 16)
except ValueError:
return None
offsets.append(parsed)
return offsets


def resolve_base_address(
text: str, module_lookup: Callable[[str], Optional[int]]
) -> Tuple[Optional[int], Optional[str]]:
"""Resolve a pointer-chain base field into an absolute address.

Accepts either a plain hex address (``0x14010F4F4``) or Cheat-Engine's
``"module"+0xoffset`` form (``"libpython3.12.dylib"+0x4ED3D0``); for the
latter the module's current load base is looked up via ``module_lookup``
(a ``name -> base | None`` callable) and the offset added, so a saved
pointer-scan path resolves correctly despite ASLR.

Returns ``(address, None)`` on success or ``(None, error_message)`` on
failure — the caller renders ``error_message`` in a dialog. Pure (no Qt).
"""
text = text.strip()

if "+" in text:
name_part, _, offset_part = text.partition("+")
module_name = name_part.strip().strip('"').strip("'").strip()
offset = parse_hex_address(offset_part)
if offset is None:
try:
offset = int(offset_part.strip(), 16)
except ValueError:
return None, (
"The offset after '+' must be hex "
'(e.g. "game.exe"+0x10F4F4).'
)
module_base = module_lookup(module_name)
if module_base is None:
return None, (
f"Module {module_name!r} is not loaded in this process.\n\n"
"Open Tools → Modules to see the exact names available."
)
return module_base + offset, None

base = parse_hex_address(text)
if base is None:
return None, (
'Base must be hex (0x14010F4F4) or "module"+0xoffset '
'(e.g. "game.exe"+0x10F4F4).'
)
return base, None
Loading
Loading