diff --git a/.forgejo/workflows/ci.yml b/.forgejo/workflows/ci.yml index c599614..ef0b2ff 100644 --- a/.forgejo/workflows/ci.yml +++ b/.forgejo/workflows/ci.yml @@ -20,16 +20,25 @@ jobs: run: | set -eux sudo apt-get update - # libegl1 / libxkbcommon0 / libdbus-1-3 cover the shared libs PyQt6 - # wheels dlopen at import time even under QT_QPA_PLATFORM=offscreen. + # libgl1 / libegl1 / libxkbcommon0 / libdbus-1-3 cover the shared libs + # PyQt6 wheels dlopen at import time, even under QT_QPA_PLATFORM=offscreen. + # libgl1 is needed because PyQt6.QtGui transitively links libGL.so.1 — + # discovered when CI first ran on a Forgejo runner (prior queue-never- + # executed runs masked this). sudo apt-get install -y --no-install-recommends \ - libegl1 libxkbcommon0 libdbus-1-3 libfontconfig1 libxcb-cursor0 + libgl1 libegl1 libxkbcommon0 libdbus-1-3 libfontconfig1 libxcb-cursor0 python -m pip install --upgrade pip pip install -e ".[dev]" - name: ruff run: ruff check . + - name: mypy + # Strict on safety-critical modules (flash_ops, udev_check, + # app_logging, app_settings); signature-typing on main + + # internet_panel. See pyproject [tool.mypy] for rationale. + run: mypy + - name: gitleaks # Scan full history on push to main; staged diff is already covered # by the pre-commit hook. Use --no-git for forks/runners that don't @@ -42,7 +51,7 @@ jobs: /tmp/gitleaks detect --source . --redact --verbose - name: py_compile - run: python -m py_compile flash_ops.py main.py internet_panel.py udev_check.py app_logging.py + run: python -m py_compile flash_ops.py main.py internet_panel.py udev_check.py app_logging.py app_settings.py - name: pytest # --cov measures coverage; no --cov-fail-under yet — first establish @@ -75,7 +84,7 @@ jobs: PY - name: Upload coverage report - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v3 with: name: coverage-xml path: coverage.xml @@ -98,13 +107,18 @@ jobs: sudo apt-get update # appimagetool needs FUSE at runtime; CI containers usually lack # /dev/fuse, so we set APPIMAGE_EXTRACT_AND_RUN=1 below to bypass. + # libgl1/libegl1/libxkbcommon0/libdbus-1-3/libfontconfig1/libxcb-cursor0 + # are needed at AppImage runtime by PyQt6 (the `--version` smoke + # step launches the AppImage, which imports PyQt6.QtGui before + # argparse runs). Same set as the smoke job. sudo apt-get install -y --no-install-recommends \ - file desktop-file-utils zsync wget + file desktop-file-utils zsync wget squashfs-tools \ + libgl1 libegl1 libxkbcommon0 libdbus-1-3 libfontconfig1 libxcb-cursor0 python -m pip install --upgrade pip pip install pipx # Pin python-appimage so a future upstream change can't silently # break the build. Bump deliberately when validated. - pipx install 'python-appimage==0.34' + pipx install 'python-appimage>=1.4,<2' - name: Build AppImage env: @@ -141,7 +155,7 @@ jobs: ./dist/HDZeroProgrammer-x86_64.AppImage --check-rule - name: Upload AppImage - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v3 with: name: HDZeroProgrammer-x86_64.AppImage path: dist/HDZeroProgrammer-x86_64.AppImage diff --git a/.forgejo/workflows/hw-test.yml b/.forgejo/workflows/hw-test.yml new file mode 100644 index 0000000..1d0b921 --- /dev/null +++ b/.forgejo/workflows/hw-test.yml @@ -0,0 +1,160 @@ +# Real-hardware flash regression test. +# +# Runs only on push to main (NOT on PRs from contributors — too slow, +# too risky for branches that haven't been reviewed) AND only on a +# runner labeled `hdzero-hw`. Without such a runner registered the +# job queues forever and is skipped on Forgejo housekeeping; the +# default `lab01-hdzero` runner advertises no `hdzero-hw` label so +# this never blocks day-to-day CI. +# +# Setup expectations for an `hdzero-hw` runner (see docs/HARDWARE-CI.md +# for the long form): +# - CH341A USB SPI programmer wired to an HDZero VTX via the SOIC +# clip on the W25Q80 chip. +# - `flashrom` >= 1.4 installed on the runner host. +# - Production firmware blob committed to the runner host at +# /var/lib/hdzero-hw/known-good-firmware.bin (NOT to this repo — +# keeps binary blobs out of git). +# - Runner registered with labels: `ubuntu-22.04`, `hdzero-hw`. +# - The CH341A udev rule installed (see packaging/99-ch341a.rules) +# so the runner user can talk to the device without sudo. +# +# Why the round-trip matters: every other test in this repo mocks +# flashrom via tests/fixtures/fake_flashrom.sh. A flashrom upgrade +# that breaks the argv contract or the stdout phase markers +# FlashWorker greps for would ship a broken AppImage with green CI. +# This job is the only thing that catches those classes of regression. + +name: hw-test + +on: + push: + branches: [main] + # Allow on-demand kicks from the Forgejo Actions UI when debugging + # without forcing a code change. + workflow_dispatch: + +# IMPORTANT: do NOT add `pull_request:` here. PR-from-fork workflows +# can ship arbitrary build-script changes; running them on real +# hardware on every PR is a footgun. PR validation stays in ci.yml. + +jobs: + flash-roundtrip: + # The label gate. A runner without `hdzero-hw` won't claim the + # job. The fallback `lab01-hdzero` runner only advertises + # `ubuntu-22.04` so this matrix is naturally inert in clean + # development setups. + runs-on: [ubuntu-22.04, hdzero-hw] + timeout-minutes: 15 + env: + # Production firmware committed on the runner host, NOT in the + # repo. This is the image we leave the device in after the test + # (idempotent teardown — see "restore" step). + KNOWN_GOOD_FIRMWARE: /var/lib/hdzero-hw/known-good-firmware.bin + # Where flash transcripts land for the artifact upload. + HW_TRANSCRIPT_DIR: ${{ github.workspace }}/dist/hw-transcripts + steps: + - uses: actions/checkout@v4 + + - name: Pre-flight — confirm hardware is connected + run: | + set -eux + # Vendor:product matches udev_check.CH341A_VENDOR/PRODUCT. + lsusb | grep -E "1a86:5512" || { + echo "::error::CH341A not enumerated. Re-seat USB or check SOIC clip." >&2 + exit 1 + } + test -r "$KNOWN_GOOD_FIRMWARE" || { + echo "::error::Known-good firmware missing at $KNOWN_GOOD_FIRMWARE." >&2 + exit 1 + } + mkdir -p "$HW_TRANSCRIPT_DIR" + + - name: Read current chip — backup before mutating + id: backup + run: | + set -eux + BACKUP="$HW_TRANSCRIPT_DIR/pre-test-backup-$(date +%s).bin" + flashrom -p ch341a_spi -r "$BACKUP" \ + 2>&1 | tee "$HW_TRANSCRIPT_DIR/01-read.log" + echo "backup=$BACKUP" >> "$GITHUB_OUTPUT" + # 1 MiB exact, per FLASH_SIZE_BYTES in flash_ops.py. + actual=$(stat -c%s "$BACKUP") + [ "$actual" = "1048576" ] || { + echo "::error::Backup is $actual bytes, expected 1048576 (1 MiB)." >&2 + exit 1 + } + + - name: Set up Python + install package + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install runtime deps + run: | + set -eux + sudo apt-get update + sudo apt-get install -y --no-install-recommends \ + libgl1 libegl1 libxkbcommon0 libdbus-1-3 libfontconfig1 libxcb-cursor0 + python -m pip install --upgrade pip + pip install -e ".[dev]" + + - name: Pad the known-good firmware to 1 MiB + # Reuses the same make_padded_image_1mib path the GUI takes — + # this validates that our padding logic is what flashrom + # actually accepts on real silicon, not just bytes that + # happen to be the right size. + run: | + set -eux + python - <&1 | tee "$HW_TRANSCRIPT_DIR/02-write.log" + + - name: Verify chip matches what we wrote + env: + HDZERO_NO_ESCALATE: "1" + run: | + set -eux + PADDED="$(ls /tmp/hdzero_pad_*.bin | head -1)" + flashrom -p ch341a_spi -v "$PADDED" \ + 2>&1 | tee "$HW_TRANSCRIPT_DIR/03-verify.log" + + - name: Restore — leave the VTX in known-good state + # Always runs (success OR failure) so an aborted job still + # leaves the VTX bootable. Acceptance: idempotent teardown. + if: always() + env: + HDZERO_NO_ESCALATE: "1" + run: | + set -eux + # Even on failure, the production blob is what we want + # the chip holding when the human comes to debug. Re-flash + # from the known-good source, not from the (possibly + # corrupt) backup we took at the start. + PADDED="$(ls /tmp/hdzero_pad_*.bin | head -1 || true)" + if [ -n "$PADDED" ] && [ -r "$PADDED" ]; then + flashrom -p ch341a_spi -w "$PADDED" \ + 2>&1 | tee "$HW_TRANSCRIPT_DIR/04-restore.log" || true + fi + + - name: Upload flashrom transcripts + if: always() + uses: actions/upload-artifact@v3 + with: + name: hw-test-transcripts + path: dist/hw-transcripts/ + if-no-files-found: warn diff --git a/.forgejo/workflows/release.yml b/.forgejo/workflows/release.yml index 18b7ca4..d8a36bd 100644 --- a/.forgejo/workflows/release.yml +++ b/.forgejo/workflows/release.yml @@ -20,12 +20,14 @@ jobs: run: | set -eux sudo apt-get update + # Full Qt runtime set needed at AppImage launch — see ci.yml. sudo apt-get install -y --no-install-recommends \ - file desktop-file-utils zsync wget jq + file desktop-file-utils zsync wget jq squashfs-tools \ + libgl1 libegl1 libxkbcommon0 libdbus-1-3 libfontconfig1 libxcb-cursor0 python -m pip install --upgrade pip pip install pipx # Match the pin used in ci.yml so release builds are reproducible. - pipx install 'python-appimage==0.34' + pipx install 'python-appimage>=1.4,<2' - name: Build AppImage env: @@ -98,7 +100,7 @@ jobs: done - name: Upload AppImage as workflow artifact too - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v3 with: name: HDZeroProgrammer-x86_64.AppImage path: dist/HDZeroProgrammer-x86_64.AppImage diff --git a/CHANGELOG.md b/CHANGELOG.md index 590b641..596aee6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -Post-`v0.2.0` work on `main` not yet tagged. Bumps the next release; until -then, `__version__` and `pyproject` still read `0.2.0`. +No changes on `main` since `v0.3.0` yet. + +## [0.3.0] - 2026-04-30 + +Post-acquisition-audit batch. Closes the audit's documentation and +operational blockers (`#22` HW-CI gate, `#23` SECURITY.md), lands the +ADR corpus, the mypy gate, the legacy-settings migration, and the +HTTP-worker refactor. Firmware-trust trio (`#19`/`#20`/`#21`) remains +open and is the gating set for `v0.4.0`. ### Added - ADR scaffolding under `docs/adr/` with template + index; ADR-0001 @@ -26,6 +33,22 @@ then, `__version__` and `pyproject` still read `0.2.0`. - pytest-cov coverage measurement in CI (no fail-under threshold yet). - Contributing guide, code of conduct, issue + PR templates. - 2026-04-29 acquisition audit report archived under `docs/audits/`. +- `app_settings.py`: `QSettings` reverse-DNS scope + (`lab.squatch / hdzero-programmer-linux`) plus a one-shot migration + from the legacy `HDZero/Programmer` scope, sentinel-guarded so a + downgrade still finds its data. (#33) +- mypy gate in CI: strict on safety-critical modules (`flash_ops`, + `udev_check`, `app_logging`, `app_settings`); `disallow_untyped_defs` + + `check_untyped_defs` on UI plumbing (`main`, `internet_panel`). + PyQt6 namespace ignored for missing imports — no first-party stubs. + (#26, #68) +- Real-hardware regression workflow `.forgejo/workflows/hw-test.yml` + gated on the `hdzero-hw` runner label, plus `docs/HARDWARE-CI.md` + documenting host-side CH341A pass-through, runner registration, and + the idempotent restore step. (#22) +- `docs/legal/trademark.md` capturing HDZero brand exposure and a + `vtx-programmer-linux` rename contingency if the policy turns + restrictive. (#43) ### Changed - Chip backups (manual + pre-flash auto) now land in @@ -36,11 +59,28 @@ then, `__version__` and `pyproject` still read `0.2.0`. - Narrowed broad `except Exception` handlers across flash/UI paths. - LICENSE: SPDX header, dual fork copyright line, PyQt6 GPL-binary position documented in `LICENSE-NOTES.md`. -- `actions/upload-artifact` bumped to v4. +- Collapsed four near-duplicate `InternetPanel` HTTP workers into a + single `HttpWorker` with retry + exponential backoff (`requests`-based, + bounded by `timeout` + `retries` kwargs). (#25, #37) +- python-appimage pinned to `>=1.4,<2` — 0.34 imports `distutils` which + was removed in Python 3.12. +- AppImage + smoke jobs now apt-install the full PyQt6 runtime lib set + (`libgl1`, `libegl1`, `libxkbcommon0`, `libdbus-1-3`, `libfontconfig1`, + `libxcb-cursor0`) plus `squashfs-tools` for `appimagetool`. ### Fixed - Temp firmware files (downloaded `.bin` and 1 MiB padded image) now unlinked after flash success and failure. +- `run_admin_streaming` now polls stdout via `select.select()` instead + of blocking on `for line in proc.stdout:`. The wall-clock deadline + fires even when a wedged `flashrom` produces no output, so a hung + CH341A no longer burns the full timeout budget waiting on readline. + +### Reverted +- `actions/upload-artifact` rolled back from v4 to v3 — Forgejo's + Actions implementation does not yet support the v4 upload protocol + (returns `GHESNotSupportedError`). Bump again once Forgejo lands v4 + compatibility. ## [0.2.0] - 2026-04-29 diff --git a/CLAUDE.md b/CLAUDE.md index ab10702..36cce8d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,11 +12,14 @@ under MIT. ## Architecture -Five top-level Python modules (all listed in `[tool.setuptools].py-modules`; +Six top-level Python modules (all listed in `[tool.setuptools].py-modules`; the AppImage build copies them into `_hdzero_app/`), one Qt event loop, worker threads for blocking I/O. ADRs for non-trivial decisions live in [`docs/adr/`](docs/adr/); the 2026-04-29 acquisition audit and its -open-blocker backlog (`#19`–`#23`) are under [`docs/audits/`](docs/audits/). +remaining firmware-trust blockers (`#19`–`#21`) are under +[`docs/audits/`](docs/audits/). `#22` (HW-CI gate) and `#23` (SECURITY.md) +were closed in the post-audit batch — see `docs/HARDWARE-CI.md` and +`SECURITY.md`. - `main.py` — `MainWindow` owns the three tabs and routes flash signals. Holds `FlashWorker` / `BackupWorker` references on `self.worker` / @@ -37,6 +40,9 @@ open-blocker backlog (`#19`–`#23`) are under [`docs/audits/`](docs/audits/). - `udev_check.py` — `99-ch341a.rules` install state, CH341A vendor/product walk of `/sys/bus/usb/devices`, reads `HDZERO_NO_ESCALATE`. Drives the install banner and the `--check-rule` CLI flag. +- `app_settings.py` — `settings()` (reverse-DNS QSettings scope + `lab.squatch / hdzero-programmer-linux`) and `migrate_settings_once()` + (one-shot copy of the legacy `HDZero/Programmer` scope into the new one). - `app_logging.py` — `state_dir()` (`HDZERO_STATE_DIR` > `$XDG_STATE_HOME/hdzero-programmer` > `~/.local/state/hdzero-programmer`) and `backup_dir()` (`HDZERO_BACKUP_DIR` > `state_dir()/backups/`). @@ -108,10 +114,15 @@ in `MainWindow`. Do not call `flash_ops` from panels directly. before constructing `QApplication`; `parse_known_args` lets Qt's platform flags pass through. CI `appimage` job execs both flags as a launch smoke test. -- **Autobackup persistence.** `QSettings("HDZero", "Programmer")` key - `autobackup`, shared between Local and Internet panels — last toggle - on either becomes the next-launch default. File: - `~/.config/HDZero/Programmer.conf`. +- **Autobackup persistence.** `app_settings.settings()` returns a + `QSettings` handle in the reverse-DNS scope `lab.squatch / + hdzero-programmer-linux` (file `~/.config/lab.squatch/hdzero-programmer-linux.conf`). + Key `autobackup` is shared between Local and Internet panels — last + toggle on either becomes the next-launch default. + `app_settings.migrate_settings_once()` runs in `main()` after + `_install_excepthook()` and copies the legacy `HDZero/Programmer` keys + forward exactly once (sentinel `_migrated_from_legacy_org`); the legacy + file is left intact so a downgrade still finds its data. - **Test harness.** `tests/fixtures/fake_flashrom.sh` is a drop-in stand-in: emits the same stdout markers `FlashWorker` greps and supports `FAKE_FLASHROM_FAIL=read|write|verify`. Integration tests call @@ -127,8 +138,11 @@ in `MainWindow`. Do not call `flash_ops` from panels directly. `.forgejo/workflows/ci.yml` (push to `main` + PRs): -- **smoke** — `pip install -e ".[dev]"`, `ruff check`, `gitleaks detect`, - `py_compile` the five modules, `pytest --cov` (no fail-under yet), +- **smoke** — `pip install -e ".[dev]"`, `ruff check`, `mypy` (strict on + `flash_ops` / `udev_check` / `app_logging` / `app_settings`; + `disallow_untyped_defs` on `main` / `internet_panel` — see + `[tool.mypy]` overrides in `pyproject.toml`), `gitleaks detect`, + `py_compile` the six modules, `pytest --cov` (no fail-under yet), headless `MainWindow` boot under `QT_QPA_PLATFORM=offscreen` against an unroutable `HDZERO_API_BASE`. - **appimage** — `python-appimage` (pinned), runs @@ -136,6 +150,12 @@ in `MainWindow`. Do not call `flash_ops` from panels directly. lacks `/dev/fuse`), uploads the artifact, smoke-execs `--version` and `--check-rule` (with `HDZERO_NO_ESCALATE=1`). +`.forgejo/workflows/hw-test.yml` (manual / `workflow_dispatch`): real-HW +flash regression gated on `[ubuntu-22.04, hdzero-hw]`. Runner setup +is host-specific opt-in work; see `docs/HARDWARE-CI.md`. Without an +`hdzero-hw`-labeled runner the job sits unclaimed forever, which is the +intended default. + `.forgejo/workflows/release.yml` triggers on `v*` tags: builds the AppImage, generates `SHA256SUMS`, creates/updates a Forgejo release with both files attached. Cut a release: `git tag -a vX.Y.Z -m "vX.Y.Z" && diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fb18b9a..87fa659 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -107,4 +107,8 @@ the content is a normal architecture/onboarding doc — read it on day one. By contributing, you agree your work is licensed under the project's [MIT License](LICENSE), with the binary distribution implications -documented in [`LICENSE-NOTES.md`](LICENSE-NOTES.md). +documented in [`LICENSE-NOTES.md`](LICENSE-NOTES.md). The HDZero brand +is owned by Divimath Inc.; this project is unaffiliated, and the +ongoing trademark exposure is tracked in +[`docs/legal/trademark.md`](docs/legal/trademark.md) — including a +contingency rename plan if the policy ever turns restrictive. diff --git a/README.md b/README.md index 62ebc81..fd85865 100644 --- a/README.md +++ b/README.md @@ -39,21 +39,37 @@ Gunther. This fork only adapts it for Linux. ## Run -Install via pip (recommended): +### Option A — AppImage (recommended) + +Download the latest `HDZeroProgrammer-x86_64.AppImage` from +[Releases](https://forgejo.squatch.lab/bmags/hdzero-programmer-linux/releases), +verify, and run: + +```bash +sha256sum -c SHA256SUMS # both files in the same directory +chmod +x HDZeroProgrammer-x86_64.AppImage +./HDZeroProgrammer-x86_64.AppImage +``` + +The AppImage bundles Python 3.12, PyQt6, and `requests`. It still +requires `flashrom` on the host (the app shells out to it) and the +CH341A access setup below (udev rule or polkit agent). + +### Option B — From source ```bash pip install --user . hdzero-programmer ``` -Or run directly out of a checkout: +Or run directly out of a checkout without installing: ```bash pip install --user PyQt6 requests python3 main.py ``` -Override the firmware index API base if needed: +### Override the firmware index API base ```bash HDZERO_API_BASE=https://your-mirror.example hdzero-programmer @@ -102,11 +118,10 @@ script installed by `pip install`. If you run from a checkout without installing, edit the installed `.desktop` file's `Exec=` line to `python3 /absolute/path/to/main.py`. -## Build a self-contained AppImage (optional) +## Build the AppImage locally (maintainers) -For distribution to users without a Python toolchain, build a single-file -AppImage that bundles a relocatable Python 3.12, PyQt6, `requests`, and -the application source: +CI builds the AppImage on every push and on every `v*` tag (see +[Releases](#releases)). Local rebuild for testing packaging changes: ```bash pipx install python-appimage @@ -114,10 +129,6 @@ pipx install python-appimage # → dist/HDZeroProgrammer-x86_64.AppImage ``` -The resulting AppImage still requires `flashrom` on the host (the AppImage -shells out to it) and the same CH341A access setup described above (udev -rule or polkit agent). - ## Usage 1. **Internet tab** — pick a device from the dropdown, pick a firmware @@ -169,7 +180,7 @@ choices. [`CLAUDE.md`](CLAUDE.md) is the day-one onboarding doc. ## How it works -Five Python modules, one Qt event loop, blocking I/O isolated to QThread +Six Python modules, one Qt event loop, blocking I/O isolated to QThread workers: - `main.py` — `MainWindow` owns the three tabs and routes signals. @@ -182,6 +193,9 @@ workers: whether the CH341A is currently attached; drives the first-run banner. - `app_logging.py` — resolves the per-flash transcript dir under `$XDG_STATE_HOME/hdzero-programmer/` and opens line-buffered log files. +- `app_settings.py` — `QSettings` reverse-DNS scope + (`lab.squatch / hdzero-programmer-linux`) plus a one-shot migration + from the legacy `HDZero/Programmer` scope. See `CLAUDE.md` for deeper architecture notes. @@ -189,10 +203,13 @@ See `CLAUDE.md` for deeper architecture notes. `.forgejo/workflows/ci.yml` runs on every push and pull request: -- **smoke** — `py_compile`, `pytest` against `tests/`, and a headless - `MainWindow` boot under `QT_QPA_PLATFORM=offscreen`. +- **smoke** — `ruff check`, `mypy` (strict on the safety-critical + modules; typed signatures on UI plumbing), `gitleaks`, `py_compile`, + `pytest --cov` against `tests/`, and a headless `MainWindow` boot + under `QT_QPA_PLATFORM=offscreen`. - **appimage** — builds the AppImage and uploads it as a workflow - artifact (downloadable from the run page). + artifact (downloadable from the run page). Smoke-execs `--version` + and `--check-rule` against the built image. Requires a registered Forgejo Actions runner labeled `ubuntu-22.04`. @@ -201,7 +218,7 @@ Requires a registered Forgejo Actions runner labeled `ubuntu-22.04`. ```bash pip install --user -e ".[dev]" # pytest + ruff pip install --user pre-commit && pre-commit install -pytest # 43 tests, mostly hardware-mock +pytest # 64 tests, mostly hardware-mock ruff check . # lint ``` diff --git a/SECURITY.md b/SECURITY.md index 2331a0e..b94860c 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -45,8 +45,8 @@ with `SHA256SUMS` attached; signed-release support is tracked in `#21`. | Version | Supported | | ------- | ------------------ | -| `0.2.x` | :white_check_mark: | -| `< 0.2` | :x: | +| `0.3.x` | :white_check_mark: | +| `< 0.3` | :x: | ## Scope @@ -55,7 +55,7 @@ In scope: - The packaged AppImage and its build pipeline (`packaging/`, `.forgejo/workflows/`). - The Python source modules (`main.py`, `internet_panel.py`, - `flash_ops.py`, `udev_check.py`, `app_logging.py`). + `flash_ops.py`, `udev_check.py`, `app_logging.py`, `app_settings.py`). - The bundled udev rule (`packaging/99-ch341a.rules`) and install helpers (`packaging/install-udev.sh`). - Default trust assumptions about the firmware index API base diff --git a/app_settings.py b/app_settings.py new file mode 100644 index 0000000..eac15ea --- /dev/null +++ b/app_settings.py @@ -0,0 +1,64 @@ +# app_settings.py +"""Reverse-DNS QSettings scope + one-shot legacy migration. + +The macOS upstream and pre-v0.3 Linux fork both wrote QSettings under +the bare brand `HDZero/Programmer`. On a host that runs both, one tool +overwrites the other's `autobackup` toggle. New scope is reverse-DNS +under the maintainer's domain so the two stores never collide. +`migrate_settings_once()` copies the legacy keys forward exactly once; +the legacy file is left intact so a downgrade still finds its data. +""" +from typing import Optional + +from PyQt6.QtCore import QSettings + +LEGACY_SETTINGS_ORG = "HDZero" +LEGACY_SETTINGS_APP = "Programmer" + +SETTINGS_ORG = "lab.squatch" +SETTINGS_APP = "hdzero-programmer-linux" + +SETTINGS_KEY_AUTOBACKUP = "autobackup" + +# Sentinel written into the new scope after the first successful migration. +# Idempotent — a second call sees the sentinel and short-circuits. +_KEY_MIGRATED = "_migrated_from_legacy_org" + + +def settings() -> QSettings: + """Construct a QSettings handle in the canonical reverse-DNS scope.""" + return QSettings(SETTINGS_ORG, SETTINGS_APP) + + +def migrate_settings_once( + new: Optional[QSettings] = None, + old: Optional[QSettings] = None, +) -> None: + """Copy legacy `HDZero/Programmer` keys into the new scope on first run. + + Skipped if the sentinel `_migrated_from_legacy_org` is already set in + the new scope, so reruns are free. Existing keys in the new scope are + not overwritten — if a user sets a value under the new scope before + migration runs (unlikely but possible across versions), their choice + wins. + + The optional `new` and `old` arguments exist so tests can inject + QSettings instances pinned to explicit on-disk paths, sidestepping + PyQt6's QStandardPaths cache that otherwise makes XDG-based isolation + flaky in a same-process pytest run. + """ + if new is None: + new = settings() + if old is None: + old = QSettings(LEGACY_SETTINGS_ORG, LEGACY_SETTINGS_APP) + if new.value(_KEY_MIGRATED, False, type=bool): + return + for key in old.allKeys(): + if not new.contains(key): + new.setValue(key, old.value(key)) + new.setValue(_KEY_MIGRATED, True) + # Force flush. QSettings auto-syncs on a timer / on destruction, but a + # crash in the same launch — or another QSettings instance reading the + # new scope before the timer fires — would miss the migrated keys + # otherwise. Explicit sync also makes the test suite hermetic. + new.sync() diff --git a/docs/HARDWARE-CI.md b/docs/HARDWARE-CI.md new file mode 100644 index 0000000..a68698f --- /dev/null +++ b/docs/HARDWARE-CI.md @@ -0,0 +1,149 @@ +# Hardware CI runner setup + +The default Forgejo Actions runner (`lab01-hdzero`) advertises only the +`ubuntu-22.04` label. The `hw-test` workflow +([`.forgejo/workflows/hw-test.yml`](../.forgejo/workflows/hw-test.yml)) +gates on `[ubuntu-22.04, hdzero-hw]`, so without a runner that +advertises `hdzero-hw` the job sits unclaimed forever — which is the +right default. Adding a hardware runner is opt-in, host-specific work. + +This doc captures the steps for the maintainer (or anyone replicating +the lab setup) to bring an `hdzero-hw` runner online. + +## Hardware + +- **CH341A USB SPI programmer** wired to the W25Q80 chip on an HDZero + VTX via a SOIC-8 clip. The udev rule at + [`packaging/99-ch341a.rules`](../packaging/99-ch341a.rules) grants + `uaccess` + `plugdev` group access so the runner can talk to the + device without `sudo`. +- **HDZero VTX** wired permanently to the programmer. The workflow + re-flashes the production firmware on every run (idempotent + teardown), so the VTX stays bootable across runs. +- **Host** running Linux with USB pass-through (or a bare-metal lab + box). Tested on Fedora 43 + Ubuntu 22.04. The runner image itself + is `catthehacker/ubuntu:act-22.04` so the kernel underneath only + needs to expose `/dev/bus/usb/...` to the container. + +## Software + +```bash +# Install flashrom on the host (the act_runner container shares +# /dev/bus/usb but the binary lives in the runner image — see below). +sudo dnf install flashrom # Fedora +sudo apt install flashrom # Debian/Ubuntu + +# Install the udev rule + reload so the runner user can access the +# CH341A without root. +sudo install -m 0644 packaging/99-ch341a.rules \ + /etc/udev/rules.d/99-ch341a.rules +sudo udevadm control --reload-rules +sudo udevadm trigger +# unplug + replug the CH341A + +# Verify the rule took effect — uaccess group should be present. +ls -l /dev/bus/usb/*/$(lsusb | awk '/1a86:5512/ {print $4}' | tr -d :) +``` + +## Stage the known-good firmware + +The hw-test workflow re-flashes the production firmware on every run +so the VTX is left in a known state. Stage the blob *on the runner +host* — NOT in the repo (binary blobs don't belong in git): + +```bash +sudo mkdir -p /var/lib/hdzero-hw +sudo cp /path/to/hdzero-production-firmware.bin \ + /var/lib/hdzero-hw/known-good-firmware.bin +sudo chmod 0644 /var/lib/hdzero-hw/known-good-firmware.bin +``` + +The workflow padding step (`make_padded_image_1mib`) handles the +1 MiB target, so the blob can be the raw firmware binary — anything +≤ 64 KiB will work. + +## Register the runner + +The default Quadlet at +`~/.config/containers/systemd/act_runner.container` (see +[`docs/HARDWARE-CI-RUNNER-QUADLET.md`](HARDWARE-CI-RUNNER-QUADLET.md) +when written) advertises one label. Either add a second runner that +advertises `hdzero-hw`, OR replace the existing runner's labels. + +For a dedicated hardware runner, register a fresh one: + +```bash +TOKEN=$(cat ~/.forgejo-token) +REG_TOKEN=$(curl -s -H "Authorization: token $TOKEN" \ + "https://forgejo.squatch.lab/api/v1/repos/bmags/hdzero-programmer-linux/actions/runners/registration-token" \ + | python3 -c "import json,sys; print(json.load(sys.stdin)['token'])") + +# In a separate data dir so the hw runner can be torn down +# independently of the everyday runner. +mkdir -p ~/.local/share/act_runner-hw + +# Auto-register via env vars (same env-var auto-flow as the default +# runner). Note the labels: matches the workflow's `runs-on:` matrix. +cat > ~/.local/share/act_runner-hw/registration.env < Optional[List[str]]: return None -def run_admin(cmd: str) -> subprocess.CompletedProcess: +def run_admin(cmd: str) -> "subprocess.CompletedProcess[str]": """Buffered elevated execution. Returns a CompletedProcess so callers see a uniform contract regardless of which escalation path was taken. """ @@ -102,34 +103,52 @@ def run_admin_streaming( ) assert proc.stdout is not None deadline = time.monotonic() + timeout + + def _kill_after_timeout(reason: str) -> int: + line_cb(f"(timeout: {reason}; terminating flashrom)\n") + proc.terminate() + try: + return proc.wait(timeout=5) + except subprocess.TimeoutExpired: + line_cb("(timeout: SIGTERM ignored, sending SIGKILL)\n") + proc.kill() + return proc.wait() + try: - for line in proc.stdout: - line_cb(line) - if time.monotonic() >= deadline: - # Streaming is past the cap. Break out, drop into the - # kill path below. Don't read further lines — the process - # may be producing them at a slow drip and we want out now. - line_cb(f"(timeout: exceeded {timeout:.0f}s; terminating flashrom)\n") - proc.terminate() - try: - return proc.wait(timeout=5) - except subprocess.TimeoutExpired: - line_cb("(timeout: SIGTERM ignored, sending SIGKILL)\n") - proc.kill() - return proc.wait() - # stdout closed normally; let proc.wait() pick up the exit code, - # still bounded by the remaining wall-clock budget. + # Poll the stdout fd with `select` so the deadline check fires + # even when the child is wedged with no output (e.g. a hung + # flashrom mid-erase). Plain `for line in proc.stdout:` blocks + # on readline until EOF, which would let a wedged process burn + # through the entire wall-clock budget before we noticed. + buf = "" + fd = proc.stdout.fileno() + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return _kill_after_timeout(f"exceeded {timeout:.0f}s") + # 1s poll keeps the deadline check responsive without + # busy-looping. select returns ready fds OR timeout-empty. + ready, _, _ = select.select([fd], [], [], min(1.0, remaining)) + if not ready: + continue + chunk = os.read(fd, 4096) + if not chunk: # EOF — process closed stdout + if buf: + line_cb(buf) + break + buf += chunk.decode(errors="replace") + while "\n" in buf: + line, buf = buf.split("\n", 1) + line_cb(line + "\n") + if time.monotonic() >= deadline: + return _kill_after_timeout(f"exceeded {timeout:.0f}s") + + # stdout closed; bound proc.wait() by what's left of the budget. remaining = max(0.0, deadline - time.monotonic()) try: return proc.wait(timeout=remaining) except subprocess.TimeoutExpired: - line_cb(f"(timeout: exceeded {timeout:.0f}s after stdout close; terminating)\n") - proc.terminate() - try: - return proc.wait(timeout=5) - except subprocess.TimeoutExpired: - proc.kill() - return proc.wait() + return _kill_after_timeout(f"exceeded {timeout:.0f}s after stdout close") finally: # If we returned via kill, proc.stdout might still be open. Close # it so we don't leak the FD. @@ -173,7 +192,7 @@ def __init__(self, flashrom_path: str, fw_path: str, backup_path: Optional[str] # firmware blob); LocalPanel's user-selected .bin must stay False. self.cleanup_fw = cleanup_fw - def run(self): + def run(self) -> None: padded: Optional[str] = None try: self.status.emit("Wait - Prepare firmware") @@ -214,7 +233,7 @@ def run(self): # per phase is sufficient. seen = {"backup": False, "write": False, "verify": False} - def on_line(line: str): + def on_line(line: str) -> None: self.log.emit(line) low = line.lower() if not seen["backup"] and self.backup_path and "reading flash" in low: @@ -273,7 +292,7 @@ def __init__(self, flashrom_path: str, out_path: str): self.flashrom = flashrom_path self.out = out_path - def run(self): + def run(self) -> None: try: cmd = ( f"{shlex.quote(self.flashrom)} -p ch341a_spi -r " diff --git a/internet_panel.py b/internet_panel.py index ce4cdf5..29206a3 100644 --- a/internet_panel.py +++ b/internet_panel.py @@ -3,12 +3,13 @@ import os import sys import tempfile +import time from pathlib import Path -from typing import Callable, List, Optional, Tuple +from typing import Any, Callable, List, Optional, Tuple import requests from PyQt6 import QtCore -from PyQt6.QtCore import QSettings, Qt, QThread, pyqtSignal +from PyQt6.QtCore import Qt, QThread, pyqtSignal from PyQt6.QtGui import QIcon, QPixmap, QTextCursor from PyQt6.QtWidgets import ( QCheckBox, @@ -23,18 +24,15 @@ ) from requests.exceptions import RequestException +from app_settings import SETTINGS_KEY_AUTOBACKUP +from app_settings import settings as _qsettings + # Narrow exception class for HTTP-worker run(). Catches network/HTTP # failures + malformed JSON. Anything else (AttributeError, KeyError on a # changed schema, programmer error) propagates so it surfaces as a real # bug rather than a misleading "connectivity error" string. _HTTP_FAILURES = (RequestException, json.JSONDecodeError, OSError) -# Keep these in sync with the constants in main.py — single shared key so -# toggling on either tab persists app-wide. -_SETTINGS_ORG = "HDZero" -_SETTINGS_APP = "Programmer" -_SETTINGS_KEY_AUTOBACKUP = "autobackup" - API_BASE = os.environ.get("HDZERO_API_BASE", "https://hdzero.go-next.co").rstrip("/") def resource_path(relpath: str) -> str: @@ -48,81 +46,99 @@ def resource_path(relpath: str) -> str: base = getattr(sys.modules["__main__"], "_MEIPASS", Path(__file__).parent) return str(Path(base) / relpath) -# HTTP workers local to the Internet panel -class LoadDevicesWorker(QThread): - ok = pyqtSignal(list); fail = pyqtSignal(str) - def run(self): - try: - url = f"{API_BASE}/api/devices" - r = requests.get(url, timeout=15) - r.raise_for_status() - self.ok.emit(r.json().get("devices", [])) - except _HTTP_FAILURES as e: - self.fail.emit(str(e)) - -class LoadFirmwaresWorker(QThread): - ok = pyqtSignal(list); fail = pyqtSignal(str) - def __init__(self, device_id: int): - super().__init__() - self.device_id = device_id - def run(self): - try: - url = f"{API_BASE}/api/firmwares/{self.device_id}" - r = requests.get(url, timeout=15) - r.raise_for_status() - self.ok.emit(r.json().get("firmwares", [])) - except _HTTP_FAILURES as e: - self.fail.emit(str(e)) - -class LoadImageWorker(QThread): - ok = pyqtSignal(bytes); fail = pyqtSignal(str) - def __init__(self, url: str): - super().__init__() - self.url = url - def run(self): - try: - r = requests.get(self.url, timeout=10) - r.raise_for_status() - self.ok.emit(r.content) - except _HTTP_FAILURES as e: - self.fail.emit(str(e)) - -class DownloadFirmwareWorker(QThread): - progress = pyqtSignal(int); ok = pyqtSignal(str); fail = pyqtSignal(str) - def __init__(self, url: str): +# Generic HTTP worker for InternetPanel — single QThread shape with hooks +# for response shape (raw bytes, JSON parser, streamed-to-tempfile). All +# four legacy workers (LoadDevices/LoadFirmwares/LoadImage/DownloadFirmware) +# collapse onto this; uniform retry/backoff/timeout policy lives in one +# place. See ticket #25 + #37. +class HttpWorker(QThread): + """One QThread for every flavour of HTTP fetch the panel needs. + + - JSON-parsed response: pass `parser=lambda r: r.json().get("foo", [])`. + Emitted via `ok` as whatever the parser returns. + - Raw bytes: leave `parser` and `stream_to_temp_bin` unset; `ok` + emits the response body bytes. + - Streamed download to a temp .bin: set `stream_to_temp_bin=True`. + `progress` fires with percent (0..100); `ok` emits the temp path. + + Retries on `_HTTP_FAILURES` with exponential backoff + (`backoff * 2^attempt`), default 2 retries → 1s, 2s. Each retry is + surfaced via `log` so a user investigating a flake sees the + attempts. `time.sleep` runs on this worker thread, not the GUI. + """ + ok = pyqtSignal(object) + fail = pyqtSignal(str) + progress = pyqtSignal(int) + log = pyqtSignal(str) + + def __init__( + self, + url: str, + *, + parser: Optional[Callable[[requests.Response], Any]] = None, + stream_to_temp_bin: bool = False, + timeout: float = 15.0, + retries: int = 2, + backoff: float = 1.0, + ): super().__init__() self.url = url - def run(self): - try: - r = requests.get(self.url, stream=True, timeout=30) - r.raise_for_status() - total = int(r.headers.get("Content-Length") or 0) - tmp = tempfile.NamedTemporaryFile(prefix="hdzero_dl_", suffix=".bin", delete=False) - tmp_path = tmp.name - tmp.close() - read = 0 - with open(tmp_path, "wb") as f: - for chunk in r.iter_content(chunk_size=8192): - if not chunk: - continue - f.write(chunk) - read += len(chunk) - if total > 0: - self.progress.emit(int(read * 100 / total)) - self.progress.emit(100) - self.ok.emit(tmp_path) - except _HTTP_FAILURES as e: - self.fail.emit(str(e)) + self.parser = parser + self.stream_to_temp_bin = stream_to_temp_bin + self.timeout = timeout + self.retries = retries + self.backoff = backoff + + def run(self) -> None: + last_err: Optional[str] = None + for attempt in range(1, self.retries + 2): + try: + if self.stream_to_temp_bin: + self._stream_download() + else: + r = requests.get(self.url, timeout=self.timeout) + r.raise_for_status() + self.ok.emit(self.parser(r) if self.parser else r.content) + return + except _HTTP_FAILURES as e: + last_err = str(e) + if attempt <= self.retries: + wait = self.backoff * (2 ** (attempt - 1)) + self.log.emit( + f"http attempt {attempt} failed: {last_err}; " + f"retrying in {wait:.1f}s" + ) + time.sleep(wait) + self.fail.emit(last_err or "unknown error") + + def _stream_download(self) -> None: + r = requests.get(self.url, stream=True, timeout=self.timeout) + r.raise_for_status() + total = int(r.headers.get("Content-Length") or 0) + tmp = tempfile.NamedTemporaryFile(prefix="hdzero_dl_", suffix=".bin", delete=False) + tmp_path = tmp.name + tmp.close() + read = 0 + with open(tmp_path, "wb") as f: + for chunk in r.iter_content(chunk_size=8192): + if not chunk: + continue + f.write(chunk) + read += len(chunk) + if total > 0: + self.progress.emit(int(read * 100 / total)) + self.progress.emit(100) + self.ok.emit(tmp_path) class InternetPanel(QWidget): firmwareSelected = pyqtSignal(str) # local downloaded path (consumed by Local panel) log = pyqtSignal(str) # log lines forwarded to Local panel flashRequested = pyqtSignal(str, bool) # (local path, autobackup) - def __init__(self): + def __init__(self) -> None: super().__init__() - self.devices: List[dict] = [] - self.firmwares: List[dict] = [] + self.devices: List[dict[str, Any]] = [] + self.firmwares: List[dict[str, Any]] = [] # (label shown on the button, callable that re-runs the failed op). # Set by every loader before kicking; cleared on success; consulted # by the Retry button which sits next to lbl_state. @@ -196,9 +212,8 @@ def __init__(self): self.btn_flash.clicked.connect(self.download_selected_fw) self.cb_autobackup = QCheckBox("Backup chip before flashing") - _s = QSettings(_SETTINGS_ORG, _SETTINGS_APP) self.cb_autobackup.setChecked( - _s.value(_SETTINGS_KEY_AUTOBACKUP, True, type=bool) + _qsettings().value(SETTINGS_KEY_AUTOBACKUP, True, type=bool) ) self.cb_autobackup.toggled.connect(self._persist_autobackup) @@ -235,24 +250,22 @@ def __init__(self): @staticmethod def _persist_autobackup(checked: bool) -> None: - QSettings(_SETTINGS_ORG, _SETTINGS_APP).setValue( - _SETTINGS_KEY_AUTOBACKUP, checked - ) + _qsettings().setValue(SETTINGS_KEY_AUTOBACKUP, checked) - def set_phase(self, text: str): + def set_phase(self, text: str) -> None: self.lbl_phase.setText(text) # ===== Status helpers ===== - def status_append(self, text: str): + def status_append(self, text: str) -> None: self.status_box.moveCursor(QTextCursor.MoveOperation.End) self.status_box.insertPlainText(text if text.endswith("\n") else text + "\n") self.status_box.moveCursor(QTextCursor.MoveOperation.End) - def status_set(self, text: str): + def status_set(self, text: str) -> None: self.status_box.setPlainText(text) self.status_box.moveCursor(QTextCursor.MoveOperation.End) - def set_loading(self, msg: str): + def set_loading(self, msg: str) -> None: self.lbl_state.setText(msg) def _arm_loader(self, label: str, fn: Callable[[], None]) -> None: @@ -273,7 +286,7 @@ def _retry_last(self) -> None: self.btn_retry.setVisible(False) fn() - def on_fail(self, msg: str): + def on_fail(self, msg: str) -> None: self.set_loading(f"Error: {msg}") self.status_append(f"ERROR: {msg}") self.log.emit(f"[Internet] ERROR: {msg}\n") @@ -283,17 +296,21 @@ def on_fail(self, msg: str): self.btn_retry.setVisible(True) # ===== HTTP logic ===== - def load_devices(self): + def load_devices(self) -> None: self._arm_loader("device list", self.load_devices) self.set_loading("Loading devices…") self.cb_devices.clear() - w = LoadDevicesWorker() + w = HttpWorker( + f"{API_BASE}/api/devices", + parser=lambda r: r.json().get("devices", []), + ) w.ok.connect(self.on_devices_ok) w.fail.connect(self.on_fail) + w.log.connect(self.log.emit) w.start() self._w_dev = w - def on_devices_ok(self, devices: list): + def on_devices_ok(self, devices: list[dict[str, Any]]) -> None: self._clear_loader() self.devices = devices or [] self.cb_devices.clear() @@ -305,20 +322,24 @@ def on_devices_ok(self, devices: list): self.cb_devices.setCurrentIndex(0) self.on_device_changed() - def _set_device_image(self, url: Optional[str]): + def _set_device_image(self, url: Optional[str]) -> None: if not url: self.device_img.setText("No image") self.device_img.setPixmap(QPixmap()) return self.device_img.setText("Loading image…") self.device_img.setPixmap(QPixmap()) - w = LoadImageWorker(url) + # Image worker keeps the historic 10s timeout (vs 15s default); a + # device thumbnail is small enough that a longer wait is just a UX + # delay. Retries default to 2. + w = HttpWorker(url, timeout=10.0) w.ok.connect(self._on_image_loaded) w.fail.connect(self._on_image_failed) + w.log.connect(self.log.emit) w.start() self._w_img = w - def _on_image_loaded(self, data: bytes): + def _on_image_loaded(self, data: bytes) -> None: pix = QPixmap() if not pix.loadFromData(data): self.device_img.setText("Image load error") @@ -328,11 +349,11 @@ def _on_image_loaded(self, data: bytes): self.device_img.setPixmap(scaled) self.device_img.setText("") - def _on_image_failed(self, _msg: str): + def _on_image_failed(self, _msg: str) -> None: self.device_img.setText("Image load error") self.device_img.setPixmap(QPixmap()) - def on_device_changed(self): + def on_device_changed(self) -> None: data = self.cb_devices.currentData() if not data: self.cb_fw.clear() @@ -344,13 +365,17 @@ def on_device_changed(self): self._arm_loader("firmware list", self.on_device_changed) self.set_loading("Loading firmwares…") self.cb_fw.clear() - w = LoadFirmwaresWorker(device_id) + w = HttpWorker( + f"{API_BASE}/api/firmwares/{device_id}", + parser=lambda r: r.json().get("firmwares", []), + ) w.ok.connect(self.on_fw_ok) w.fail.connect(self.on_fail) + w.log.connect(self.log.emit) w.start() self._w_fw = w - def on_fw_ok(self, firmwares: list): + def on_fw_ok(self, firmwares: list[dict[str, Any]]) -> None: self._clear_loader() self.firmwares = firmwares or [] self.cb_fw.clear() @@ -362,12 +387,12 @@ def on_fw_ok(self, firmwares: list): self.cb_fw.setCurrentIndex(0) self.on_fw_changed() - def on_fw_changed(self): + def on_fw_changed(self) -> None: fw = self.cb_fw.currentData() self.notes.setPlainText("" if not fw else (fw.get("notes") or "")) # ===== Download and request flash ===== - def download_selected_fw(self): + def download_selected_fw(self) -> None: fw = self.cb_fw.currentData() if not fw: self.on_fail("No firmware selected.") @@ -380,14 +405,17 @@ def download_selected_fw(self): self.set_phase("Wait - Downloading.") self.status_set(f"Downloading: {url}") self._arm_loader("download", self.download_selected_fw) - w = DownloadFirmwareWorker(url) + # Firmware download keeps the historic 30s timeout — payloads are + # under 64 KiB but the per-CDN handshake can drag. + w = HttpWorker(url, stream_to_temp_bin=True, timeout=30.0) w.progress.connect(lambda p: self.set_loading(f"Downloading… {p}%")) w.ok.connect(self.on_download_ok_then_flash) w.fail.connect(self.on_fail) + w.log.connect(self.log.emit) w.start() self._w_dl = w - def on_download_ok_then_flash(self, local_path: str): + def on_download_ok_then_flash(self, local_path: str) -> None: self._clear_loader() self.firmwareSelected.emit(local_path) self.status_append(f"Downloaded: {local_path}") diff --git a/main.py b/main.py index 5913182..d3c012e 100644 --- a/main.py +++ b/main.py @@ -4,10 +4,10 @@ import sys import time from pathlib import Path -from typing import Optional +from typing import IO, Callable, Optional from PyQt6 import QtCore -from PyQt6.QtCore import QSettings, Qt +from PyQt6.QtCore import Qt from PyQt6.QtGui import QIcon, QPixmap, QTextCursor from PyQt6.QtWidgets import ( QApplication, @@ -26,6 +26,8 @@ ) from app_logging import backup_dir, open_flash_log +from app_settings import SETTINGS_KEY_AUTOBACKUP, migrate_settings_once +from app_settings import settings as _settings from flash_ops import HDZERO_MAX, BackupWorker, FlashWorker, find_flashrom from internet_panel import InternetPanel, resource_path from udev_check import ( @@ -40,11 +42,7 @@ # Hard-coded so AppImage runs (which graft source into site-packages # without dist-info) can still report a version. Kept in sync with # pyproject.toml by tests/test_version_consistency.py. -__version__ = "0.2.0" - -SETTINGS_ORG = "HDZero" -SETTINGS_APP = "Programmer" -SETTINGS_KEY_AUTOBACKUP = "autobackup" +__version__ = "0.3.0" APP_TITLE = "HDZero Programmer Tool – by Gunther_FPV" APP_HEADER_TITLE = "HDZero Programmer (Linux)" @@ -57,7 +55,11 @@ ) class LocalPanel(QWidget): - def __init__(self, start_backup_cb, start_flash_cb): + def __init__( + self, + start_backup_cb: "Callable[[], None]", + start_flash_cb: "Callable[[str, bool], None]", + ) -> None: super().__init__() self.start_backup_cb = start_backup_cb self.start_flash_cb = start_flash_cb @@ -79,9 +81,8 @@ def __init__(self, start_backup_cb, start_flash_cb): self.log = QTextEdit(); self.log.setReadOnly(True); layout.addWidget(self.log, 1) self.cb_autobackup = QCheckBox("Backup chip before flashing (rollback image in ~)") - settings = QSettings(SETTINGS_ORG, SETTINGS_APP) self.cb_autobackup.setChecked( - settings.value(SETTINGS_KEY_AUTOBACKUP, True, type=bool) + _settings().value(SETTINGS_KEY_AUTOBACKUP, True, type=bool) ) self.cb_autobackup.toggled.connect(self._persist_autobackup) layout.addWidget(self.cb_autobackup) @@ -107,11 +108,11 @@ def __init__(self, start_backup_cb, start_flash_cb): if not self.flashrom: self.append_log(FLASHROM_INSTALL_HINT) - def set_fw_path(self, path: str): + def set_fw_path(self, path: str) -> None: self.fw_path = Path(path); self.path_edit.setText(path) self.status.setText("Ready to flash (from Internet).") - def append_log(self, text: str): + def append_log(self, text: str) -> None: cursor = self.log.textCursor() cursor.movePosition(QTextCursor.MoveOperation.End) self.log.setTextCursor(cursor) @@ -119,7 +120,7 @@ def append_log(self, text: str): self.log.setTextCursor(cursor) self.log.ensureCursorVisible() - def pick_bin(self): + def pick_bin(self) -> None: path, _ = QFileDialog.getOpenFileName(self, "Select firmware .bin", "", "BIN (*.bin)") if not path: return if not path.lower().endswith(".bin"): @@ -131,18 +132,16 @@ def pick_bin(self): @staticmethod def _persist_autobackup(checked: bool) -> None: - QSettings(SETTINGS_ORG, SETTINGS_APP).setValue( - SETTINGS_KEY_AUTOBACKUP, checked - ) + _settings().setValue(SETTINGS_KEY_AUTOBACKUP, checked) - def on_backup_pressed(self): self.start_backup_cb() - def on_flash_pressed(self): + def on_backup_pressed(self) -> None: self.start_backup_cb() + def on_flash_pressed(self) -> None: if not self.fw_path or not self.fw_path.exists(): QMessageBox.critical(self, "Error", "Select or download a .bin file first."); return self.start_flash_cb(str(self.fw_path), self.cb_autobackup.isChecked()) class HelpPanel(QWidget): - def __init__(self): + def __init__(self) -> None: super().__init__() layout = QVBoxLayout(self); layout.setContentsMargins(10,10,10,10); layout.setSpacing(10) @@ -170,7 +169,7 @@ def __init__(self): layout.addWidget(logo, 0, Qt.AlignmentFlag.AlignHCenter) class MainWindow(QWidget): - def __init__(self): + def __init__(self) -> None: super().__init__() self.setWindowTitle(APP_TITLE) self.resize(700, 620) @@ -178,6 +177,11 @@ def __init__(self): self.flashrom = find_flashrom() or "" self.fw_path: Optional[Path] = None + # Per-flash transcript handle. Set by _open_flash_log on flash + # start, cleared by _close_flash_log on completion. Optional + # because the open path degrades silently on OSError. + self._flash_log_path: Optional[Path] = None + self._flash_log_fh: Optional[IO[str]] = None # Dark style + gray tabs self.setStyleSheet(""" @@ -369,12 +373,12 @@ def _confirm_ch341a_present(self) -> bool: return reply == QMessageBox.StandardButton.Yes # ==== High-level handlers (reused by both tabs) ==== - def on_fw_downloaded_set_local(self, path: str): + def on_fw_downloaded_set_local(self, path: str) -> None: self.fw_path = Path(path) self.panel_local.set_fw_path(path) self.panel_local.append_log(f"Downloaded from Internet → {path}\n") - def start_backup(self): + def start_backup(self) -> None: if not self.flashrom or not os.path.exists(self.flashrom): QMessageBox.critical(self, "Error", FLASHROM_INSTALL_HINT) return @@ -401,7 +405,7 @@ def start_backup(self): self.bkw.fail.connect(self.on_backup_fail) self.bkw.start() - def on_backup_ok(self, out_path: str): + def on_backup_ok(self, out_path: str) -> None: self._close_flash_log(f"OK: backup saved {out_path}") self.panel_local.status.setText("✅ Backup done") self.panel_local.pb.setRange(0, 100) @@ -411,7 +415,7 @@ def on_backup_ok(self, out_path: str): self.panel_local.btn_backup.setEnabled(True) self.panel_local.flash_btn.setEnabled(True) - def on_backup_fail(self, msg: str): + def on_backup_fail(self, msg: str) -> None: self._close_flash_log(f"FAIL: {msg}") self.panel_local.status.setText("❌ Backup error") self.panel_local.pb.setRange(0, 100) @@ -421,7 +425,7 @@ def on_backup_fail(self, msg: str): self.panel_local.btn_backup.setEnabled(True) self.panel_local.flash_btn.setEnabled(True) - def start_flash(self, fw_path: str, autobackup: bool = True, *, cleanup_fw: bool = False): + def start_flash(self, fw_path: str, autobackup: bool = True, *, cleanup_fw: bool = False) -> None: if not fw_path or not Path(fw_path).exists(): QMessageBox.critical(self, "Error", "Select a .bin file.") return @@ -468,7 +472,7 @@ def start_flash(self, fw_path: str, autobackup: bool = True, *, cleanup_fw: bool self.worker.start() - def on_flash_ok(self): + def on_flash_ok(self) -> None: self._close_flash_log("OK: flash completed and verified") self.panel_local.status.setText("✅ Done") self.panel_local.pb.setValue(100) @@ -477,7 +481,7 @@ def on_flash_ok(self): self.panel_local.flash_btn.setEnabled(True) self.panel_local.btn_backup.setEnabled(True) - def on_flash_fail(self, msg: str): + def on_flash_fail(self, msg: str) -> None: self._close_flash_log(f"FAIL: {msg}") self.panel_local.status.setText("❌ Error") self.panel_local.pb.setValue(100) @@ -518,11 +522,21 @@ def _install_excepthook() -> None: re-raise via the original hook. """ import traceback as _tb + from types import TracebackType original = sys.excepthook in_hook = [False] - def hook(exc_type, exc, tb): + def hook( + exc_type: type[BaseException], + exc: BaseException, + tb: Optional[TracebackType], + ) -> None: + if issubclass(exc_type, KeyboardInterrupt): + # Deliberate quit (Ctrl-C). Not a crash — skip the dialog and + # let the event loop unwind cleanly. + QApplication.quit() + return if in_hook[0]: original(exc_type, exc, tb) return @@ -566,7 +580,7 @@ def _run_check_rule() -> int: return 0 if (installed or no_escalate) else 1 -def main(argv=None) -> int: +def main(argv: Optional[list[str]] = None) -> int: parser = _build_argparser() # argparse splits its own flags off; everything else is forwarded to # QApplication so platform plugin args (-style, -platform, …) still work. @@ -576,6 +590,7 @@ def main(argv=None) -> int: app = QApplication([sys.argv[0], *qt_argv]) _install_excepthook() + migrate_settings_once() app_icon_path = resource_path("icon256.png") if Path(app_icon_path).exists(): app.setWindowIcon(QIcon(app_icon_path)) diff --git a/packaging/build-appimage.sh b/packaging/build-appimage.sh index 948ea13..aeb155b 100755 --- a/packaging/build-appimage.sh +++ b/packaging/build-appimage.sh @@ -80,6 +80,7 @@ cp "$REPO_ROOT/internet_panel.py" "$INJECT_ABS/" cp "$REPO_ROOT/flash_ops.py" "$INJECT_ABS/" cp "$REPO_ROOT/udev_check.py" "$INJECT_ABS/" cp "$REPO_ROOT/app_logging.py" "$INJECT_ABS/" +cp "$REPO_ROOT/app_settings.py" "$INJECT_ABS/" # Image assets used by HelpPanel, LocalPanel, InternetPanel, MainWindow cp "$REPO_ROOT"/*.png "$INJECT_ABS/" diff --git a/pyproject.toml b/pyproject.toml index 28cf8f7..e2fdcf1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "hdzero-programmer" -version = "0.2.0" +version = "0.3.0" description = "Linux desktop GUI for flashing HDZero VTX firmware via a CH341A USB SPI programmer." readme = "README.md" license = { file = "LICENSE" } @@ -39,6 +39,8 @@ dev = [ "pytest>=8", "pytest-cov>=5", "ruff>=0.5", + "mypy>=1.10", + "types-requests", ] [project.urls] @@ -49,12 +51,57 @@ Upstream = "https://github.com/gvotteler/HDZero-Programmer-Tool-Mac" hdzero-programmer = "main:main" [tool.setuptools] -py-modules = ["main", "internet_panel", "flash_ops", "udev_check", "app_logging"] +py-modules = ["main", "internet_panel", "flash_ops", "udev_check", "app_logging", "app_settings"] [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] +[tool.mypy] +# Safety-critical modules carry strict typing. main + internet_panel sit +# on PyQt6 (no first-party stubs) and run too much UI glue to be +# `--strict` without an `Any` ocean. The first pass keeps them at +# import-and-explicit-signature checking only; tightening to +# `disallow_untyped_defs` is a deliberate follow-up. +python_version = "3.10" +files = [ + "flash_ops.py", + "udev_check.py", + "app_logging.py", + "app_settings.py", + "main.py", + "internet_panel.py", +] +strict = false +warn_unused_ignores = true + +# PyQt6 ships no stubs (community PyQt6-stubs lags the wheel cadence +# and breaks CI when out of date). Treating the namespace as untyped +# is a deliberate trade-off; safety-critical code calls QThread / +# pyqtSignal sparingly so the Any blast radius is small. +[[tool.mypy.overrides]] +module = ["PyQt6.*"] +ignore_missing_imports = true + +# Strict surface — flashrom argv contracts, privilege escalation, +# udev probe, transcript file path resolution, settings migration. A +# single None-where-Path or wrong-shaped tuple here ships as an +# AttributeError at flash time; mypy is cheap insurance. +[[tool.mypy.overrides]] +module = ["flash_ops", "udev_check", "app_logging", "app_settings"] +strict = true + +# main + internet_panel are UI plumbing on top of PyQt6 (which has no +# first-party stubs). disallow_untyped_defs keeps the bar at "every +# def has a typed signature" so cross-module signature drift surfaces. +# Full strict isn't worth the Any-cast cost on Qt boilerplate, but +# typed defs catch the bug class that matters: a slot connected with +# the wrong arg shape, or a helper that loses an argument silently. +[[tool.mypy.overrides]] +module = ["main", "internet_panel"] +disallow_untyped_defs = true +check_untyped_defs = true + [tool.ruff] target-version = "py310" line-length = 120 diff --git a/tests/test_app_settings.py b/tests/test_app_settings.py new file mode 100644 index 0000000..2db8591 --- /dev/null +++ b/tests/test_app_settings.py @@ -0,0 +1,128 @@ +# tests/test_app_settings.py +"""Reverse-DNS QSettings scope + one-shot legacy migration. + +QSettings + pytest is finicky: PyQt6's QStandardPaths caches the +config dir per-process, so an `XDG_CONFIG_HOME` set via +`monkeypatch.setenv` after the runtime cache warms up doesn't always +reach a QSettings instance constructed later. To stay hermetic we +construct QSettings with an *explicit file path* (the +`QSettings(filename, IniFormat)` overload) and inject those instances +into `migrate_settings_once()` via its optional kwargs. This bypasses +QStandardPaths entirely and is the same pattern the Qt docs recommend +for tests. +""" +import os + +import pytest + + +@pytest.fixture(autouse=True) +def _qapp(): + """QSettings on some bindings needs a QCoreApplication for org/app + fallbacks. Construct one for the duration of the test if absent.""" + from PyQt6.QtCore import QCoreApplication + + app = QCoreApplication.instance() + created = False + if app is None: + app = QCoreApplication([os.environ.get("PYTEST_CURRENT_TEST", "test")]) + created = True + yield + if created: + app.quit() + + +@pytest.fixture +def settings_pair(tmp_path): + """Return ``(new, old, new_path, old_path)`` — two empty QSettings + instances backed by explicit IniFormat files under tmp_path.""" + from PyQt6.QtCore import QSettings + + new_path = tmp_path / "new.conf" + old_path = tmp_path / "legacy.conf" + new = QSettings(str(new_path), QSettings.Format.IniFormat) + old = QSettings(str(old_path), QSettings.Format.IniFormat) + return new, old, new_path, old_path + + +def test_settings_uses_reverse_dns_scope(): + """Constants pin the reverse-DNS scope so a downgrade or a side-by- + side macOS install doesn't share QSettings storage.""" + import app_settings + + assert app_settings.SETTINGS_ORG == "lab.squatch" + assert app_settings.SETTINGS_APP == "hdzero-programmer-linux" + + +def test_migrate_copies_legacy_keys_once(settings_pair): + import app_settings + + new, old, new_path, old_path = settings_pair + old.setValue("autobackup", False) + old.setValue("some_other_key", "preserved") + old.sync() + + app_settings.migrate_settings_once(new=new, old=old) + + assert new.value("autobackup", True, type=bool) is False + assert new.value("some_other_key") == "preserved" + assert new.value("_migrated_from_legacy_org", False, type=bool) is True + assert old_path.exists(), "legacy file should be left intact" + + +def test_migrate_is_idempotent(settings_pair): + import app_settings + + new, old, _, _ = settings_pair + new.setValue("autobackup", True) + new.setValue("_migrated_from_legacy_org", True) + new.sync() + + # Plant a contradicting legacy value. A second migrate must NOT + # clobber the new scope's value. + old.setValue("autobackup", False) + old.sync() + + app_settings.migrate_settings_once(new=new, old=old) + + assert new.value("autobackup", False, type=bool) is True + + +def test_migrate_skips_keys_already_in_new_scope(settings_pair): + import app_settings + + new, old, _, _ = settings_pair + # User set the new scope (somehow — e.g. opened a build, toggled + # checkbox, then a later build added migration). Migration must not + # overwrite their explicit choice. + new.setValue("autobackup", True) + new.sync() + + old.setValue("autobackup", False) + old.sync() + + app_settings.migrate_settings_once(new=new, old=old) + + assert new.value("autobackup", False, type=bool) is True + + +def test_migrate_no_legacy_data_is_noop(settings_pair): + import app_settings + + new, old, _, _ = settings_pair + # No keys in `old` at all. Migration should run cleanly and write + # the sentinel so a future legacy file doesn't accidentally migrate + # later. + app_settings.migrate_settings_once(new=new, old=old) + + assert new.value("_migrated_from_legacy_org", False, type=bool) is True + + +def test_settings_helper_returns_correct_scope(): + """`settings()` returns a QSettings with the reverse-DNS org/app.""" + import app_settings + + s = app_settings.settings() + # organizationName + applicationName are the canonical readbacks. + assert s.organizationName() == app_settings.SETTINGS_ORG + assert s.applicationName() == app_settings.SETTINGS_APP diff --git a/tests/test_excepthook.py b/tests/test_excepthook.py index 49d5285..f6307f9 100644 --- a/tests/test_excepthook.py +++ b/tests/test_excepthook.py @@ -46,6 +46,36 @@ def fake_critical(parent, title, text): sys.excepthook = sys.__excepthook__ +def test_excepthook_keyboardinterrupt_quits_without_dialog(monkeypatch, capsys): + """Ctrl-C on the terminal must quit cleanly, not pop the crash dialog.""" + dialog_calls = [] + quit_calls = [] + + def fake_critical(parent, title, text): + dialog_calls.append((title, text)) + + def fake_quit(): + quit_calls.append(True) + + from PyQt6 import QtWidgets + monkeypatch.setattr(QtWidgets.QMessageBox, "critical", fake_critical) + monkeypatch.setattr(QtWidgets.QApplication, "quit", staticmethod(fake_quit)) + + main._install_excepthook() + try: + try: + raise KeyboardInterrupt() + except KeyboardInterrupt: + et, e, tb = sys.exc_info() + sys.excepthook(et, e, tb) + + assert dialog_calls == [], "crash dialog must not pop on Ctrl-C" + assert quit_calls == [True], "QApplication.quit() must be called" + assert capsys.readouterr().err == "" + finally: + sys.excepthook = sys.__excepthook__ + + def test_excepthook_reentrancy_falls_back(monkeypatch): """If the hook is invoked while already running (e.g. its own QMessageBox somehow re-enters), the second call must not loop.""" diff --git a/tests/test_http_worker.py b/tests/test_http_worker.py new file mode 100644 index 0000000..7cb6c33 --- /dev/null +++ b/tests/test_http_worker.py @@ -0,0 +1,196 @@ +# tests/test_http_worker.py +"""HttpWorker — generic HTTP fetch with retry/backoff for InternetPanel. + +Tests run synchronously via `worker.run()` (not `start()`) so signal +slots fire inline and assertions stay simple — same pattern the +flash-integration tests use. `requests.get` is monkeypatched per-test. +""" +import os + +import pytest + + +@pytest.fixture(autouse=True) +def _qapp(): + from PyQt6.QtCore import QCoreApplication + + app = QCoreApplication.instance() + created = False + if app is None: + app = QCoreApplication([os.environ.get("PYTEST_CURRENT_TEST", "test")]) + created = True + yield + if created: + app.quit() + + +class _FakeResponse: + def __init__(self, *, content=b"", json_data=None, headers=None, status=200): + self.content = content + self._json = json_data + self.headers = headers or {} + self.status_code = status + + def json(self): + return self._json + + def raise_for_status(self): + if self.status_code >= 400: + import requests + + raise requests.exceptions.HTTPError(f"{self.status_code}") + + def iter_content(self, chunk_size=8192): + yield self.content + + +def _captured(slot_holder, key): + """Connect-helper: stash the emitted value into slot_holder[key].""" + + def _slot(value): + slot_holder[key] = value + + return _slot + + +def test_ok_path_with_parser(monkeypatch): + from internet_panel import HttpWorker + + monkeypatch.setattr( + "requests.get", + lambda url, timeout=None: _FakeResponse(json_data={"devices": [{"id": 1}]}), + ) + captured = {} + w = HttpWorker( + "https://example/api/devices", + parser=lambda r: r.json().get("devices", []), + retries=0, + ) + w.ok.connect(_captured(captured, "ok")) + w.fail.connect(_captured(captured, "fail")) + w.run() + + assert captured.get("ok") == [{"id": 1}] + assert "fail" not in captured + + +def test_ok_path_raw_bytes(monkeypatch): + from internet_panel import HttpWorker + + monkeypatch.setattr( + "requests.get", + lambda url, timeout=None: _FakeResponse(content=b"\x89PNG..."), + ) + captured = {} + w = HttpWorker("https://example/img.png", retries=0) + w.ok.connect(_captured(captured, "ok")) + w.fail.connect(_captured(captured, "fail")) + w.run() + + assert captured.get("ok") == b"\x89PNG..." + + +def test_stream_to_temp_bin(monkeypatch, tmp_path): + from pathlib import Path + + from internet_panel import HttpWorker + + payload = b"firmware-bytes" * 100 + + def fake_get(url, timeout=None, stream=False): + # `stream=True` path must produce iter_content; both return paths + # share the FakeResponse class for simplicity. + return _FakeResponse( + content=payload, + headers={"Content-Length": str(len(payload))}, + ) + + monkeypatch.setattr("requests.get", fake_get) + captured = {} + progress_log = [] + w = HttpWorker( + "https://example/fw.bin", stream_to_temp_bin=True, retries=0 + ) + w.ok.connect(_captured(captured, "ok")) + w.fail.connect(_captured(captured, "fail")) + w.progress.connect(progress_log.append) + w.run() + + out_path = captured.get("ok") + assert out_path and Path(out_path).exists() + assert Path(out_path).read_bytes() == payload + assert 100 in progress_log + + +def test_retry_then_success(monkeypatch): + from internet_panel import HttpWorker + + calls = {"n": 0} + + def fake_get(url, timeout=None): + calls["n"] += 1 + if calls["n"] < 3: + import requests + + raise requests.exceptions.ConnectionError("boom") + return _FakeResponse(content=b"ok") + + monkeypatch.setattr("requests.get", fake_get) + monkeypatch.setattr("internet_panel.time.sleep", lambda _s: None) + + captured = {} + log_messages = [] + w = HttpWorker("https://example", retries=3, backoff=0.01) + w.ok.connect(_captured(captured, "ok")) + w.fail.connect(_captured(captured, "fail")) + w.log.connect(log_messages.append) + w.run() + + assert captured.get("ok") == b"ok" + assert "fail" not in captured + assert calls["n"] == 3 + # Two retry attempts logged before the third succeeded. + assert sum(1 for m in log_messages if "retrying" in m) == 2 + + +def test_retry_exhausted_then_fail(monkeypatch): + from internet_panel import HttpWorker + + def fake_get(url, timeout=None): + import requests + + raise requests.exceptions.ConnectionError("permanent failure") + + monkeypatch.setattr("requests.get", fake_get) + monkeypatch.setattr("internet_panel.time.sleep", lambda _s: None) + + captured = {} + log_messages = [] + w = HttpWorker("https://example", retries=2, backoff=0.01) + w.ok.connect(_captured(captured, "ok")) + w.fail.connect(_captured(captured, "fail")) + w.log.connect(log_messages.append) + w.run() + + assert "ok" not in captured + assert "permanent failure" in captured.get("fail", "") + # 1 initial attempt + 2 retries = 3 total; 2 retry log lines. + assert sum(1 for m in log_messages if "retrying" in m) == 2 + + +def test_default_retry_count_is_two(): + """ARC-1 + #37 acceptance: retries default to 2 (so 3 total attempts). + Pin in code so a future tweak is deliberate.""" + from internet_panel import HttpWorker + + w = HttpWorker("https://example") + assert w.retries == 2 + + +def test_default_timeout_is_15s(): + """Default timeout matches the legacy LoadDevicesWorker / LoadFirmwaresWorker + value — not an arbitrary number, those were the load-list endpoints.""" + from internet_panel import HttpWorker + + w = HttpWorker("https://example") + assert w.timeout == 15.0