From ebfddda782ca7c20bb125c01b4d74fbeb2e2f76a Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 12:13:57 -0800 Subject: [PATCH 1/7] Lint/Reg Fixes --- tool/scripts/setup_dev.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tool/scripts/setup_dev.py b/tool/scripts/setup_dev.py index a47423c..1abd639 100644 --- a/tool/scripts/setup_dev.py +++ b/tool/scripts/setup_dev.py @@ -27,7 +27,7 @@ def main() -> None: logging.basicConfig(level=logging.WARNING) if not _EXAMPLES_DIR.is_dir(): - console.print(f' [red]✗[/red] examples/ directory not found at {_EXAMPLES_DIR}') + console.print(f' [red]x[/red] examples/ directory not found at {_EXAMPLES_DIR}') sys.exit(1) local_config = LocalConfiguration() @@ -43,7 +43,8 @@ def main() -> None: if _EXAMPLES_DIR.resolve() in resolved.parents and resolved not in example_dirs: try: porringer.cache.remove_directory(entry.path) - console.print(f' [yellow]−[/yellow] [bold]{entry.name or resolved.name}[/bold] [dim](removed)[/dim]') + name = entry.name or resolved.name + console.print(f' [yellow]-[/yellow] [bold]{name}[/bold] [dim](removed)[/dim]') pruned += 1 except ValueError: pass @@ -59,7 +60,7 @@ def main() -> None: try: porringer.cache.add_directory(child, name=child.name) - console.print(f' [green]✓[/green] [bold]{child.name}[/bold] [dim]({child})[/dim]') + console.print(f' [green]+[/green] [bold]{child.name}[/bold] [dim]({child})[/dim]') added += 1 except ValueError: skipped += 1 From cab40f23bfd4558f0cbbcb80ed805dd0586ec280 Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 12:32:00 -0800 Subject: [PATCH 2/7] File Reorg --- synodic_client/application/qt.py | 23 +- synodic_client/application/uri.py | 24 ++ tests/unit/qt/__init__.py | 1 + tests/unit/qt/conftest.py | 10 + tests/unit/qt/test_install_preview.py | 483 ++++++++++++++++++++++++++ tests/unit/{ => qt}/test_log_panel.py | 0 tests/unit/{ => qt}/test_logging.py | 0 tests/unit/test_examples.py | 2 +- tests/unit/test_install_preview.py | 2 +- tests/unit/test_protocol.py | 10 - tests/unit/test_uri.py | 2 +- tests/unit/windows/__init__.py | 1 + tests/unit/windows/conftest.py | 9 + tests/unit/windows/test_protocol.py | 193 ++++++++++ 14 files changed, 725 insertions(+), 35 deletions(-) create mode 100644 synodic_client/application/uri.py create mode 100644 tests/unit/qt/__init__.py create mode 100644 tests/unit/qt/conftest.py create mode 100644 tests/unit/qt/test_install_preview.py rename tests/unit/{ => qt}/test_log_panel.py (100%) rename tests/unit/{ => qt}/test_logging.py (100%) create mode 100644 tests/unit/windows/__init__.py create mode 100644 tests/unit/windows/conftest.py create mode 100644 tests/unit/windows/test_protocol.py diff --git a/synodic_client/application/qt.py b/synodic_client/application/qt.py index a608e54..e50332d 100644 --- a/synodic_client/application/qt.py +++ b/synodic_client/application/qt.py @@ -7,7 +7,6 @@ import sys import types from collections.abc import Callable -from urllib.parse import parse_qs, urlparse from porringer.api import API from porringer.schema import LocalConfiguration @@ -19,6 +18,7 @@ from synodic_client.application.screen.install import InstallPreviewWindow from synodic_client.application.screen.screen import Screen from synodic_client.application.screen.tray import TrayScreen +from synodic_client.application.uri import parse_uri from synodic_client.client import Client from synodic_client.config import GlobalConfiguration, set_dev_mode from synodic_client.logging import configure_logging @@ -27,27 +27,6 @@ from synodic_client.updater import initialize_velopack -def parse_uri(uri: str) -> dict[str, str | list[str]]: - """Parse a ``synodic://`` URI into its components. - - Example: - ``synodic://install?manifest=https://example.com/foo.toml`` - returns ``{'action': 'install', 'manifest': ['https://example.com/foo.toml']}``. - - Args: - uri: A ``synodic://`` URI string. - - Returns: - A dict with ``'action'`` (the host/path) and any query parameters. - """ - parsed = urlparse(uri) - result: dict[str, str | list[str]] = { - 'action': parsed.netloc or parsed.path.strip('/'), - } - result.update(parse_qs(parsed.query)) - return result - - def _init_services(logger: logging.Logger) -> tuple[Client, API, GlobalConfiguration]: """Create and configure core services. diff --git a/synodic_client/application/uri.py b/synodic_client/application/uri.py new file mode 100644 index 0000000..a2c2867 --- /dev/null +++ b/synodic_client/application/uri.py @@ -0,0 +1,24 @@ +"""URI parsing utilities for the ``synodic://`` scheme.""" + +from urllib.parse import parse_qs, urlparse + + +def parse_uri(uri: str) -> dict[str, str | list[str]]: + """Parse a ``synodic://`` URI into its components. + + Example: + ``synodic://install?manifest=https://example.com/foo.toml`` + returns ``{'action': 'install', 'manifest': ['https://example.com/foo.toml']}``. + + Args: + uri: A ``synodic://`` URI string. + + Returns: + A dict with ``'action'`` (the host/path) and any query parameters. + """ + parsed = urlparse(uri) + result: dict[str, str | list[str]] = { + 'action': parsed.netloc or parsed.path.strip('/'), + } + result.update(parse_qs(parsed.query)) + return result diff --git a/tests/unit/qt/__init__.py b/tests/unit/qt/__init__.py new file mode 100644 index 0000000..5217c96 --- /dev/null +++ b/tests/unit/qt/__init__.py @@ -0,0 +1 @@ +"""Qt-dependent tests.""" diff --git a/tests/unit/qt/conftest.py b/tests/unit/qt/conftest.py new file mode 100644 index 0000000..b342242 --- /dev/null +++ b/tests/unit/qt/conftest.py @@ -0,0 +1,10 @@ +"""Configuration for Qt-dependent tests. + +Tests in this directory require PySide6. When the Qt runtime libraries +are not available (e.g. on headless Linux CI), the entire directory is +skipped automatically. +""" + +import pytest + +pytest.importorskip('PySide6', reason='PySide6 requires system Qt libraries') diff --git a/tests/unit/qt/test_install_preview.py b/tests/unit/qt/test_install_preview.py new file mode 100644 index 0000000..8310154 --- /dev/null +++ b/tests/unit/qt/test_install_preview.py @@ -0,0 +1,483 @@ +"""Tests for the install preview window and URI-based install flow.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +from porringer.schema import ( + CancellationToken, + DownloadResult, + PluginKind, + ProgressEvent, + ProgressEventKind, + SetupActionResult, + SetupResults, + SkipReason, +) + +from synodic_client.application.screen import ( + ACTION_KIND_LABELS, + SKIP_REASON_LABELS, + skip_reason_label, +) +from synodic_client.application.screen.install import ( + InstallWorker, + PreviewWorker, + format_cli_command, + resolve_local_path, +) +from synodic_client.application.uri import parse_uri + + +class TestParseUriInstall: + """Tests for parsing install URIs.""" + + @staticmethod + def test_install_action_parsed() -> None: + """Verify the action is 'install' for an install URI.""" + result = parse_uri('synodic://install?manifest=https://example.com/porringer.json') + assert result['action'] == 'install' + + @staticmethod + def test_manifest_key_present() -> None: + """Verify the manifest query parameter is extracted.""" + result = parse_uri('synodic://install?manifest=https://example.com/porringer.json') + assert 'manifest' in result + assert isinstance(result['manifest'], list) + assert result['manifest'][0] == 'https://example.com/porringer.json' + + @staticmethod + def test_multiple_manifests() -> None: + """Verify multiple manifest values are captured.""" + result = parse_uri('synodic://install?manifest=https://a.com/a.json&manifest=https://b.com/b.json') + manifests = result['manifest'] + assert isinstance(manifests, list) + assert len(manifests) == 2 # noqa: PLR2004 + + @staticmethod + def test_unknown_action() -> None: + """Verify unknown actions are still parsed without error.""" + result = parse_uri('synodic://unknown?foo=bar') + assert result['action'] == 'unknown' + + +class TestInstallPreviewWindow: + """Tests for InstallPreviewWindow table population logic.""" + + @staticmethod + def _make_action( + kind: str = 'PACKAGE', + description: str = 'Install test', + installer: str = 'pip', + package: str = 'requests', + ) -> MagicMock: + """Create a mock SetupAction.""" + action = MagicMock() + action.kind = getattr(PluginKind, kind) + action.description = description + action.installer = installer + action.package = package + action.command = None + action.cli_command = None + return action + + @staticmethod + def test_action_kind_labels() -> None: + """Verify action kind label mapping covers all kinds plus None.""" + for plugin_kind in PluginKind: + assert plugin_kind in ACTION_KIND_LABELS + assert None in ACTION_KIND_LABELS + + @staticmethod + def test_skip_reason_labels() -> None: + """Verify skip reason label mapping covers all reasons.""" + for reason in SkipReason: + assert reason in SKIP_REASON_LABELS + + @staticmethod + def test_skip_reason_label_human_readable() -> None: + """Verify skip reason labels are human-readable, not raw enum names.""" + assert skip_reason_label(SkipReason.ALREADY_INSTALLED) == 'Already installed' + assert skip_reason_label(None) == 'Skipped' + + +class TestFormatCliCommand: + """Tests for format_cli_command helper.""" + + @staticmethod + def _make_action(**overrides: Any) -> MagicMock: + """Create a mock SetupAction with optional attribute overrides.""" + defaults: dict[str, Any] = { + 'kind': 'PACKAGE', + 'description': 'Install test', + 'installer': 'pip', + 'package': 'requests', + 'cli_command': None, + 'command': None, + } + defaults.update(overrides) + action = MagicMock() + kind = defaults['kind'] + action.kind = getattr(PluginKind, kind) if isinstance(kind, str) else kind + action.description = defaults['description'] + action.installer = defaults['installer'] + action.package = defaults['package'] + action.command = defaults['command'] + action.cli_command = defaults['cli_command'] + return action + + def test_prefers_cli_command(self) -> None: + """Verify cli_command takes precedence over command and fallback.""" + action = self._make_action(cli_command=['uv', 'pip', 'install', 'requests']) + assert format_cli_command(action) == 'uv pip install requests' + + def test_falls_back_to_command(self) -> None: + """Verify command is used when cli_command is absent.""" + action = self._make_action( + kind='TOOL', + command=['echo', 'hello'], + ) + assert format_cli_command(action) == 'echo hello' + + def test_synthesises_package_command(self) -> None: + """Verify package actions synthesise installer + package.""" + action = self._make_action(installer='pip', package='ruff') + assert format_cli_command(action) == 'pip install ruff' + + def test_synthesises_default_installer(self) -> None: + """Verify pip is used as default installer for package actions.""" + action = self._make_action(installer=None, package='ruff') + assert format_cli_command(action) == 'pip install ruff' + + def test_description_fallback(self) -> None: + """Verify description is returned when nothing else is available.""" + action = self._make_action( + kind='TOOL', + description='Custom step', + package=None, + ) + assert format_cli_command(action) == 'Custom step' + + +class TestInstallWorker: + """Tests for InstallWorker signal emission.""" + + @staticmethod + def test_worker_emits_finished_on_success() -> None: + """Verify worker emits finished signal with results.""" + porringer = MagicMock() + manifest_path = Path('/tmp/test/porringer.json') + + action = MagicMock() + manifest = SetupResults(actions=[action]) + manifest_event = ProgressEvent(kind=ProgressEventKind.MANIFEST_LOADED, manifest=manifest) + + result = MagicMock(spec=SetupActionResult) + completed_event = ProgressEvent( + kind=ProgressEventKind.ACTION_COMPLETED, + action=action, + result=result, + ) + + async def mock_stream(*args, **kwargs): # noqa: ANN002, ANN003 + yield manifest_event + yield completed_event + + porringer.sync.execute_stream = mock_stream + + token = CancellationToken() + worker = InstallWorker(porringer, manifest_path, token) + + received: list[SetupResults] = [] + worker.finished.connect(received.append) + worker.run() + + assert len(received) == 1 + assert received[0].actions == manifest.actions + + @staticmethod + def test_worker_emits_error_on_failure() -> None: + """Verify worker emits error signal on exception.""" + porringer = MagicMock() + manifest_path = Path('/tmp/test/porringer.json') + + async def mock_stream(*args, **kwargs): # noqa: ANN002, ANN003 + if False: + yield # pragma: no cover — establishes async generator protocol + msg = 'boom' + raise RuntimeError(msg) + + porringer.sync.execute_stream = mock_stream + + token = CancellationToken() + worker = InstallWorker(porringer, manifest_path, token) + + errors: list[str] = [] + worker.error.connect(errors.append) + worker.run() + + assert len(errors) == 1 + assert 'boom' in errors[0] + + +class TestResolveLocalPath: + """Tests for _resolve_local_path helper.""" + + @staticmethod + def test_http_url_returns_none() -> None: + """HTTP URLs should not resolve to a local path.""" + assert resolve_local_path('https://example.com/porringer.json') is None + + @staticmethod + def test_absolute_path_returns_path() -> None: + """Absolute OS paths should resolve.""" + result = resolve_local_path('C:\\Users\\test\\porringer.json') + assert result is not None + assert result == Path('C:\\Users\\test\\porringer.json') + + @staticmethod + def test_file_uri_returns_path() -> None: + """file:// URIs should resolve to a local path.""" + result = resolve_local_path('file:///C:/Users/test/porringer.json') + assert result is not None + assert 'porringer.json' in str(result) + + @staticmethod + def test_existing_relative_path(tmp_path: Path) -> None: + """Relative paths that exist on disk should resolve.""" + manifest = tmp_path / 'porringer.json' + manifest.write_text('{}') + result = resolve_local_path(str(manifest)) + assert result is not None + + +class TestPreviewWorkerLocal: + """Tests for PreviewWorker with local manifest files.""" + + @staticmethod + def test_local_manifest_skips_download(tmp_path: Path) -> None: + """Verify PreviewWorker skips download for local files.""" + manifest = tmp_path / 'porringer.json' + manifest.write_text('{}') + + porringer = MagicMock() + expected = SetupResults(actions=[]) + manifest_event = ProgressEvent(kind=ProgressEventKind.MANIFEST_LOADED, manifest=expected) + + async def mock_stream(*args: Any, **kwargs: Any) -> Any: + yield manifest_event + + porringer.sync.execute_stream = mock_stream + + worker = PreviewWorker(porringer, str(manifest)) + + results: list[tuple[object, str, str]] = [] + worker.preview_ready.connect(lambda r, p, t: results.append((r, p, t))) + worker.run() + + assert len(results) == 1 + assert results[0][0] is expected + # download should NOT have been called + porringer.sync.download.assert_not_called() + + @staticmethod + def test_local_manifest_not_found() -> None: + """Verify PreviewWorker emits error for missing local file.""" + porringer = MagicMock() + worker = PreviewWorker(porringer, 'C:\\nonexistent\\porringer.json') + + errors: list[str] = [] + worker.error.connect(errors.append) + worker.run() + + assert len(errors) == 1 + assert 'not found' in errors[0].lower() + + +class TestPreviewWorker: + """Tests for PreviewWorker download and preview flow.""" + + @staticmethod + def test_emits_error_on_download_failure() -> None: + """Verify PreviewWorker emits error when download fails.""" + porringer = MagicMock() + porringer.sync.download.return_value = DownloadResult( + success=False, + path=None, + verified=False, + size=0, + message='Network error', + ) + + worker = PreviewWorker(porringer, 'https://example.com/bad.json') + + errors: list[str] = [] + worker.error.connect(errors.append) + worker.run() + + assert len(errors) == 1 + assert 'Network error' in errors[0] + + @staticmethod + def test_emits_preview_ready_on_success() -> None: + """Verify PreviewWorker emits preview_ready with SetupResults.""" + porringer = MagicMock() + porringer.sync.download.return_value = DownloadResult( + success=True, + path=Path('/tmp/test/porringer.json'), + verified=True, + size=100, + message='OK', + ) + expected = SetupResults(actions=[]) + manifest_event = ProgressEvent(kind=ProgressEventKind.MANIFEST_LOADED, manifest=expected) + + async def mock_stream(*args: Any, **kwargs: Any) -> Any: + yield manifest_event + + porringer.sync.execute_stream = mock_stream + + worker = PreviewWorker(porringer, 'https://example.com/good.json') + + results: list[tuple[object, str, str]] = [] + worker.preview_ready.connect(lambda r, p, t: results.append((r, p, t))) + worker.run() + + assert len(results) == 1 + assert results[0][0] is expected + + +class TestPreviewWorkerSignals: + """Tests for PreviewWorker signal emission and dry-run status check.""" + + @staticmethod + def test_emits_preview_ready_and_finished(tmp_path: Path) -> None: + """Verify worker emits preview_ready then finished for a local manifest.""" + manifest = tmp_path / 'porringer.json' + manifest.write_text('{}') + + porringer = MagicMock() + action = MagicMock() + action.kind = PluginKind.PACKAGE + preview = SetupResults(actions=[action]) + + # Dry-run stream yields manifest loaded then one completed event + manifest_event = ProgressEvent(kind=ProgressEventKind.MANIFEST_LOADED, manifest=preview) + result = SetupActionResult(action=action, success=True, skipped=False, skip_reason=None) + completed_event = ProgressEvent(kind=ProgressEventKind.ACTION_COMPLETED, action=action, result=result) + + async def mock_stream(*args: Any, **kwargs: Any) -> Any: + yield manifest_event + yield completed_event + + porringer.sync.execute_stream = mock_stream + + worker = PreviewWorker(porringer, str(manifest)) + + ready_calls: list[tuple[object, str, str]] = [] + checked: list[tuple[int, SetupActionResult]] = [] + finished_count: list[int] = [] + worker.preview_ready.connect(lambda p, m, t: ready_calls.append((p, m, t))) + worker.action_checked.connect(lambda row, r: checked.append((row, r))) + worker.finished.connect(lambda: finished_count.append(1)) + worker.run() + + assert len(ready_calls) == 1 + assert ready_calls[0][0] is preview + assert len(checked) == 1 + assert checked[0] == (0, result) + assert len(finished_count) == 1 + + @staticmethod + def test_emits_finished_for_empty_actions(tmp_path: Path) -> None: + """Verify worker emits finished signal even with no actions.""" + manifest = tmp_path / 'porringer.json' + manifest.write_text('{}') + + porringer = MagicMock() + preview = SetupResults(actions=[]) + manifest_event = ProgressEvent(kind=ProgressEventKind.MANIFEST_LOADED, manifest=preview) + + async def mock_stream(*args: Any, **kwargs: Any) -> Any: + yield manifest_event + + porringer.sync.execute_stream = mock_stream + + worker = PreviewWorker(porringer, str(manifest)) + + finished_count: list[int] = [] + worker.finished.connect(lambda: finished_count.append(1)) + worker.run() + + assert len(finished_count) == 1 + + @staticmethod + def test_action_checked_maps_correct_rows(tmp_path: Path) -> None: + """Verify action_checked emits correct row indices via identity matching.""" + manifest = tmp_path / 'porringer.json' + manifest.write_text('{}') + + porringer = MagicMock() + action_a = MagicMock() + action_a.kind = PluginKind.RUNTIME + action_b = MagicMock() + action_b.kind = PluginKind.PACKAGE + preview = SetupResults(actions=[action_a, action_b]) + + manifest_event = ProgressEvent(kind=ProgressEventKind.MANIFEST_LOADED, manifest=preview) + result_b = SetupActionResult( + action=action_b, success=True, skipped=True, skip_reason=SkipReason.ALREADY_INSTALLED + ) + result_a = SetupActionResult(action=action_a, success=True, skipped=False, skip_reason=None) + + # Stream returns in execution order (b before a), not preview order + event_b = ProgressEvent(kind=ProgressEventKind.ACTION_COMPLETED, action=action_b, result=result_b) + event_a = ProgressEvent(kind=ProgressEventKind.ACTION_COMPLETED, action=action_a, result=result_a) + + async def mock_stream(*args: Any, **kwargs: Any) -> Any: + yield manifest_event + yield event_b + yield event_a + + porringer.sync.execute_stream = mock_stream + + worker = PreviewWorker(porringer, str(manifest)) + + checked: list[tuple[int, SetupActionResult]] = [] + worker.action_checked.connect(lambda row, r: checked.append((row, r))) + worker.run() + + assert len(checked) == 2 # noqa: PLR2004 + # action_b is at index 1 in preview, action_a at index 0 + assert checked[0] == (1, result_b) + assert checked[1] == (0, result_a) + + @staticmethod + def test_emits_error_when_dry_run_fails(tmp_path: Path) -> None: + """Verify worker emits error when dry-run raises.""" + manifest = tmp_path / 'porringer.json' + manifest.write_text('{}') + + porringer = MagicMock() + + async def mock_stream(*args: Any, **kwargs: Any) -> Any: + if False: + yield # pragma: no cover — establishes async generator protocol + msg = 'dry-run boom' + raise RuntimeError(msg) + + porringer.sync.execute_stream = mock_stream + + worker = PreviewWorker(porringer, str(manifest)) + + errors: list[str] = [] + finished_count: list[int] = [] + worker.error.connect(errors.append) + worker.finished.connect(lambda: finished_count.append(1)) + worker.run() + + assert len(errors) == 1 + assert 'dry-run boom' in errors[0] + assert len(finished_count) == 0 diff --git a/tests/unit/test_log_panel.py b/tests/unit/qt/test_log_panel.py similarity index 100% rename from tests/unit/test_log_panel.py rename to tests/unit/qt/test_log_panel.py diff --git a/tests/unit/test_logging.py b/tests/unit/qt/test_logging.py similarity index 100% rename from tests/unit/test_logging.py rename to tests/unit/qt/test_logging.py diff --git a/tests/unit/test_examples.py b/tests/unit/test_examples.py index f743378..ae17120 100644 --- a/tests/unit/test_examples.py +++ b/tests/unit/test_examples.py @@ -6,7 +6,7 @@ import re from pathlib import Path -from synodic_client.application.qt import parse_uri +from synodic_client.application.uri import parse_uri _URI_PATTERN = re.compile(r'synodic://\S+') diff --git a/tests/unit/test_install_preview.py b/tests/unit/test_install_preview.py index 2178a59..8310154 100644 --- a/tests/unit/test_install_preview.py +++ b/tests/unit/test_install_preview.py @@ -17,7 +17,6 @@ SkipReason, ) -from synodic_client.application.qt import parse_uri from synodic_client.application.screen import ( ACTION_KIND_LABELS, SKIP_REASON_LABELS, @@ -29,6 +28,7 @@ format_cli_command, resolve_local_path, ) +from synodic_client.application.uri import parse_uri class TestParseUriInstall: diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py index 6349185..a145ed4 100644 --- a/tests/unit/test_protocol.py +++ b/tests/unit/test_protocol.py @@ -1,6 +1,5 @@ """Tests for URI protocol handler registration.""" -import sys import winreg from pathlib import Path from unittest.mock import MagicMock, patch @@ -19,7 +18,6 @@ class TestRegisterProtocol: """Tests for register_protocol.""" @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_writes_registry_keys() -> None: """Verify correct registry keys are written on Windows.""" mock_key = MagicMock() @@ -44,7 +42,6 @@ def test_writes_registry_keys() -> None: assert second_call_args[1].endswith('shell\\open\\command') @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_sets_url_protocol_value() -> None: """Verify the 'URL Protocol' value is set.""" mock_key = MagicMock() @@ -71,7 +68,6 @@ class TestRemoveProtocol: """Tests for remove_protocol.""" @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_deletes_registry_key() -> None: """Verify the protocol key is deleted.""" with patch('synodic_client.protocol._delete_key_recursive') as mock_delete: @@ -83,7 +79,6 @@ def test_deletes_registry_key() -> None: ) @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_handles_missing_key_gracefully() -> None: """Verify no error when protocol key doesn't exist.""" with patch( @@ -105,7 +100,6 @@ class TestProtocolIntegration: """Integration tests that read/write real registry keys under a test protocol name.""" @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_register_creates_valid_registry_entries() -> None: """Register under a test key, verify values, then clean up.""" test_exe = r'C:\test\synodic_test.exe' @@ -137,7 +131,6 @@ def test_register_creates_valid_registry_entries() -> None: remove_protocol() @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_remove_deletes_registry_entries() -> None: """Register then remove under a test key, verify the key is gone.""" key_path = f'Software\\Classes\\{_TEST_PROTOCOL}' @@ -150,7 +143,6 @@ def test_remove_deletes_registry_entries() -> None: winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_register_is_idempotent() -> None: """Calling register twice with a different exe updates the command.""" key_path = f'Software\\Classes\\{_TEST_PROTOCOL}\\shell\\open\\command' @@ -176,7 +168,6 @@ class TestProtocolLive: """Verify the live protocol registration on this machine.""" @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_protocol_is_registered() -> None: """Verify that the synodic:// protocol handler is currently registered.""" key_path = f'Software\\Classes\\{PROTOCOL_NAME}' @@ -188,7 +179,6 @@ def test_protocol_is_registered() -> None: pytest.fail(f'Protocol handler not registered. Run the application once to register HKCU\\{key_path}') @staticmethod - @pytest.mark.skipif(sys.platform != 'win32', reason='Windows only') def test_command_points_to_existing_exe() -> None: """Verify the registered command points to an exe path (may not exist in CI).""" key_path = f'Software\\Classes\\{PROTOCOL_NAME}\\shell\\open\\command' diff --git a/tests/unit/test_uri.py b/tests/unit/test_uri.py index 3270cf8..6d66f63 100644 --- a/tests/unit/test_uri.py +++ b/tests/unit/test_uri.py @@ -1,6 +1,6 @@ """Tests for URI parsing in the qt application module.""" -from synodic_client.application.qt import parse_uri +from synodic_client.application.uri import parse_uri class TestParseUri: diff --git a/tests/unit/windows/__init__.py b/tests/unit/windows/__init__.py new file mode 100644 index 0000000..7dedca1 --- /dev/null +++ b/tests/unit/windows/__init__.py @@ -0,0 +1 @@ +"""Windows-only tests.""" diff --git a/tests/unit/windows/conftest.py b/tests/unit/windows/conftest.py new file mode 100644 index 0000000..199e0ba --- /dev/null +++ b/tests/unit/windows/conftest.py @@ -0,0 +1,9 @@ +"""Configuration for Windows-only tests. + +Tests in this directory require the ``winreg`` stdlib module and are +skipped automatically on non-Windows platforms. +""" + +import pytest + +pytest.importorskip('winreg', reason='winreg is only available on Windows') diff --git a/tests/unit/windows/test_protocol.py b/tests/unit/windows/test_protocol.py new file mode 100644 index 0000000..a145ed4 --- /dev/null +++ b/tests/unit/windows/test_protocol.py @@ -0,0 +1,193 @@ +"""Tests for URI protocol handler registration.""" + +import winreg +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from synodic_client.protocol import PROTOCOL_NAME, register_protocol, remove_protocol + +_EXPECTED_REGISTRY_KEY_COUNT = 2 + +_TEST_PROTOCOL = f'{PROTOCOL_NAME}_test' +"""Temporary protocol name used by integration tests to avoid clobbering the real registration.""" + + +class TestRegisterProtocol: + """Tests for register_protocol.""" + + @staticmethod + def test_writes_registry_keys() -> None: + """Verify correct registry keys are written on Windows.""" + mock_key = MagicMock() + mock_key.__enter__ = MagicMock(return_value=mock_key) + mock_key.__exit__ = MagicMock(return_value=False) + + with ( + patch.object(winreg, 'CreateKey', return_value=mock_key) as mock_create, + patch.object(winreg, 'SetValueEx'), + ): + register_protocol(r'C:\Program Files\Synodic\synodic.exe') + + # Should create the protocol key and command key + assert mock_create.call_count == _EXPECTED_REGISTRY_KEY_COUNT + + # Verify protocol key path + first_call_args = mock_create.call_args_list[0][0] + assert first_call_args[1] == f'Software\\Classes\\{PROTOCOL_NAME}' + + # Verify command key path + second_call_args = mock_create.call_args_list[1][0] + assert second_call_args[1].endswith('shell\\open\\command') + + @staticmethod + def test_sets_url_protocol_value() -> None: + """Verify the 'URL Protocol' value is set.""" + mock_key = MagicMock() + mock_key.__enter__ = MagicMock(return_value=mock_key) + mock_key.__exit__ = MagicMock(return_value=False) + + with patch.object(winreg, 'CreateKey', return_value=mock_key), patch.object(winreg, 'SetValueEx') as mock_set: + register_protocol(r'C:\Program Files\Synodic\synodic.exe') + + # Check that SetValueEx was called with 'URL Protocol' + url_protocol_calls = [c for c in mock_set.call_args_list if c[0][1] == 'URL Protocol'] + assert len(url_protocol_calls) == 1 + + @staticmethod + def test_noop_on_non_windows() -> None: + """Verify register_protocol is a no-op on non-Windows platforms.""" + with patch('synodic_client.protocol.sys') as mock_sys: + mock_sys.platform = 'linux' + # Should not raise and should not attempt winreg import + register_protocol('/usr/bin/synodic') + + +class TestRemoveProtocol: + """Tests for remove_protocol.""" + + @staticmethod + def test_deletes_registry_key() -> None: + """Verify the protocol key is deleted.""" + with patch('synodic_client.protocol._delete_key_recursive') as mock_delete: + remove_protocol() + + mock_delete.assert_called_once_with( + winreg.HKEY_CURRENT_USER, + f'Software\\Classes\\{PROTOCOL_NAME}', + ) + + @staticmethod + def test_handles_missing_key_gracefully() -> None: + """Verify no error when protocol key doesn't exist.""" + with patch( + 'synodic_client.protocol._delete_key_recursive', + side_effect=FileNotFoundError, + ): + # Should not raise + remove_protocol() + + @staticmethod + def test_noop_on_non_windows() -> None: + """Verify remove_protocol is a no-op on non-Windows platforms.""" + with patch('synodic_client.protocol.sys') as mock_sys: + mock_sys.platform = 'linux' + remove_protocol() + + +class TestProtocolIntegration: + """Integration tests that read/write real registry keys under a test protocol name.""" + + @staticmethod + def test_register_creates_valid_registry_entries() -> None: + """Register under a test key, verify values, then clean up.""" + test_exe = r'C:\test\synodic_test.exe' + key_path = f'Software\\Classes\\{_TEST_PROTOCOL}' + + try: + # Register using the test protocol name + with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): + register_protocol(test_exe) + + # Verify the protocol key + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: + description, _ = winreg.QueryValueEx(key, '') + assert description == 'Synodic Client Protocol' + + url_protocol, _ = winreg.QueryValueEx(key, 'URL Protocol') + assert not url_protocol + + # Verify the command key + command_path = f'{key_path}\\shell\\open\\command' + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, command_path) as key: + command, _ = winreg.QueryValueEx(key, '') + assert test_exe in command + assert '"%1"' in command + + finally: + # Clean up the test key + with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): + remove_protocol() + + @staticmethod + def test_remove_deletes_registry_entries() -> None: + """Register then remove under a test key, verify the key is gone.""" + key_path = f'Software\\Classes\\{_TEST_PROTOCOL}' + + with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): + register_protocol(r'C:\test\synodic_test.exe') + remove_protocol() + + with pytest.raises(FileNotFoundError): + winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) + + @staticmethod + def test_register_is_idempotent() -> None: + """Calling register twice with a different exe updates the command.""" + key_path = f'Software\\Classes\\{_TEST_PROTOCOL}\\shell\\open\\command' + exe_v1 = r'C:\test\v1\synodic.exe' + exe_v2 = r'C:\test\v2\synodic.exe' + + try: + with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): + register_protocol(exe_v1) + register_protocol(exe_v2) + + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: + command, _ = winreg.QueryValueEx(key, '') + assert exe_v2 in command + assert exe_v1 not in command + + finally: + with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): + remove_protocol() + + +class TestProtocolLive: + """Verify the live protocol registration on this machine.""" + + @staticmethod + def test_protocol_is_registered() -> None: + """Verify that the synodic:// protocol handler is currently registered.""" + key_path = f'Software\\Classes\\{PROTOCOL_NAME}' + try: + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: + _, reg_type = winreg.QueryValueEx(key, 'URL Protocol') + assert reg_type == winreg.REG_SZ + except FileNotFoundError: + pytest.fail(f'Protocol handler not registered. Run the application once to register HKCU\\{key_path}') + + @staticmethod + def test_command_points_to_existing_exe() -> None: + """Verify the registered command points to an exe path (may not exist in CI).""" + key_path = f'Software\\Classes\\{PROTOCOL_NAME}\\shell\\open\\command' + try: + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: + command, _ = winreg.QueryValueEx(key, '') + # Command format: "C:\...\synodic.exe" "%1" + exe_path = command.split('"')[1] + assert exe_path.endswith('.exe'), f'Expected .exe path, got: {exe_path}' + assert Path(exe_path).name in {'synodic.exe', 'python.exe'}, f'Unexpected exe: {exe_path}' + except FileNotFoundError: + pytest.skip('Protocol handler not registered on this machine') From 43fbd5b835162d4600d3cd04a57ca50ed6d058bf Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 14:02:44 -0800 Subject: [PATCH 3/7] ye --- pyproject.toml | 2 +- synodic_client/application/qt.py | 13 +- synodic_client/application/screen/install.py | 15 ++- synodic_client/application/screen/screen.py | 10 +- synodic_client/application/screen/tray.py | 21 ++-- synodic_client/application/threading.py | 82 ------------- synodic_client/protocol.py | 123 ++++++++++--------- synodic_client/updater.py | 10 +- tests/unit/test_protocol.py | 2 +- tests/unit/test_updater.py | 6 +- tests/unit/windows/test_protocol.py | 2 +- 11 files changed, 104 insertions(+), 182 deletions(-) delete mode 100644 synodic_client/application/threading.py diff --git a/pyproject.toml b/pyproject.toml index e5328cf..703022e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,7 +68,7 @@ quote-style = "single" skip_empty = true [tool.pyrefly] -search_path = ["synodic_client/..."] +replace_imports_with_any = ["velopack"] [tool.pdm.version] source = "scm" diff --git a/synodic_client/application/qt.py b/synodic_client/application/qt.py index e50332d..6188c7c 100644 --- a/synodic_client/application/qt.py +++ b/synodic_client/application/qt.py @@ -7,6 +7,7 @@ import sys import types from collections.abc import Callable +from typing import Any from porringer.api import API from porringer.schema import LocalConfiguration @@ -78,12 +79,12 @@ def _suppress_subprocess_consoles() -> None: _original_init = subprocess.Popen.__init__ - def _patched_init(self: subprocess.Popen, *args: object, **kwargs: object) -> None: # type: ignore[override] + def _patched_init(self: subprocess.Popen, *args: Any, **kwargs: Any) -> None: if 'creationflags' not in kwargs: - kwargs['creationflags'] = subprocess.CREATE_NO_WINDOW - _original_init(self, *args, **kwargs) # type: ignore[arg-type] + kwargs['creationflags'] = getattr(subprocess, 'CREATE_NO_WINDOW', 0) + _original_init(self, *args, **kwargs) - subprocess.Popen.__init__ = _patched_init # type: ignore[assignment] + subprocess.Popen.__init__ = _patched_init def _install_exception_hook(logger: logging.Logger) -> None: @@ -110,7 +111,9 @@ def _init_app() -> QApplication: # Set the App User Model ID so Windows uses our icon on the taskbar # instead of the generic python.exe icon. if sys.platform == 'win32': - ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID('synodic.client') # type: ignore[union-attr] + windll = getattr(ctypes, 'windll', None) + if windll is not None: + windll.shell32.SetCurrentProcessExplicitAppUserModelID('synodic.client') app = QApplication([]) app.setQuitOnLastWindowClosed(False) diff --git a/synodic_client/application/screen/install.py b/synodic_client/application/screen/install.py index d86229b..217c606 100644 --- a/synodic_client/application/screen/install.py +++ b/synodic_client/application/screen/install.py @@ -31,7 +31,7 @@ SetupResults, SubActionProgress, ) -from PySide6.QtCore import QObject, Qt, QTimer, Signal +from PySide6.QtCore import Qt, QThread, QTimer, Signal from PySide6.QtGui import QFont, QKeySequence, QShortcut from PySide6.QtWidgets import ( QApplication, @@ -69,7 +69,6 @@ MUTED_STYLE, NO_MARGINS, ) -from synodic_client.application.threading import ThreadRunner logger = logging.getLogger(__name__) @@ -83,7 +82,7 @@ def format_cli_command(action: SetupAction) -> str: return action.description -class InstallWorker(QObject): +class InstallWorker(QThread): """Background worker that executes setup actions via porringer. Uses the ``execute_stream`` async generator to consume progress events @@ -275,7 +274,7 @@ def __init__(self, porringer: API, parent: QWidget | None = None, *, show_close: self._preview: SetupResults | None = None self._manifest_path: Path | None = None self._project_directory: Path | None = None - self._runner: ThreadRunner | None = None + self._runner: QThread | None = None self._cancellation_token: CancellationToken | None = None self._completed_count = 0 self._action_statuses: list[str] = [] @@ -586,7 +585,7 @@ def _on_install(self) -> None: worker.finished.connect(self._on_install_finished) worker.error.connect(self._on_install_error) - self._runner = ThreadRunner(worker) + self._runner = worker self._runner.start() def _on_action_started(self, action: SetupAction) -> None: @@ -686,7 +685,7 @@ def __init__(self, porringer: API, manifest_url: str, parent: QWidget | None = N self._porringer = porringer self._manifest_url = manifest_url self._temp_dir_path: str | None = None - self._runner: ThreadRunner | None = None + self._runner: QThread | None = None # Default project directory to the current working directory self._project_directory: Path = Path.cwd() @@ -788,7 +787,7 @@ def start(self) -> None: preview_worker.finished.connect(self._preview_widget.on_preview_finished) preview_worker.error.connect(self._preview_widget.on_preview_error) - self._runner = ThreadRunner(preview_worker) + self._runner = preview_worker self._runner.start() # --- Preview callback (intercepts to capture temp dir) --- @@ -804,7 +803,7 @@ def _on_preview_ready(self, preview: SetupResults, manifest_path: str, temp_dir_ self._preview_widget.on_preview_ready(preview, manifest_path, temp_dir_path) -class PreviewWorker(QObject): +class PreviewWorker(QThread): """Background worker that downloads a manifest and performs a dry-run. Combines two stages into a single background pipeline: diff --git a/synodic_client/application/screen/screen.py b/synodic_client/application/screen/screen.py index 41498ed..7bb2054 100644 --- a/synodic_client/application/screen/screen.py +++ b/synodic_client/application/screen/screen.py @@ -10,7 +10,7 @@ from porringer.api import API from porringer.schema import PluginInfo, PluginKind -from PySide6.QtCore import Qt, Signal +from PySide6.QtCore import Qt, QThread, Signal from PySide6.QtGui import QStandardItem from PySide6.QtWidgets import ( QComboBox, @@ -44,7 +44,6 @@ PLUGIN_TOGGLE_STYLE, PLUGIN_UPDATE_STYLE, ) -from synodic_client.application.threading import ThreadRunner from synodic_client.config import GlobalConfiguration, save_config if TYPE_CHECKING: @@ -490,7 +489,7 @@ def __init__(self, porringer: API, parent: QWidget | None = None) -> None: """ super().__init__(parent) self._porringer = porringer - self._runner: ThreadRunner | None = None + self._runner: QThread | None = None self._init_ui() def _init_ui(self) -> None: @@ -548,7 +547,8 @@ def refresh(self) -> None: if not exists: # Grey out entries whose directory no longer exists on disk - item = self._combo.model().item(idx) # type: ignore[union-attr] + model = self._combo.model() + item = model.item(idx) if hasattr(model, 'item') else None if isinstance(item, QStandardItem): item.setForeground(self.palette().placeholderText()) item.setToolTip(f'{tooltip} \u2014 directory not found' if tooltip else 'Directory not found') @@ -645,7 +645,7 @@ def _load_preview(self) -> None: preview_worker.finished.connect(self._preview.on_preview_finished) preview_worker.error.connect(self._on_preview_error) - self._runner = ThreadRunner(preview_worker) + self._runner = preview_worker self._runner.start() def _on_preview_error(self, message: str) -> None: diff --git a/synodic_client/application/screen/tray.py b/synodic_client/application/screen/tray.py index 86c56af..291baaa 100644 --- a/synodic_client/application/screen/tray.py +++ b/synodic_client/application/screen/tray.py @@ -6,7 +6,7 @@ from porringer.api import API from porringer.schema import SetupParameters, SyncStrategy -from PySide6.QtCore import QObject, QTimer, Signal +from PySide6.QtCore import QThread, QTimer, Signal from PySide6.QtGui import QAction from PySide6.QtWidgets import ( QApplication, @@ -27,7 +27,6 @@ from synodic_client.application.icon import app_icon from synodic_client.application.screen.screen import MainWindow from synodic_client.application.theme import UPDATE_SOURCE_DIALOG_MIN_WIDTH -from synodic_client.application.threading import ThreadRunner from synodic_client.client import Client from synodic_client.config import GlobalConfiguration from synodic_client.logging import open_log @@ -37,7 +36,7 @@ logger = logging.getLogger(__name__) -class UpdateCheckWorker(QObject): +class UpdateCheckWorker(QThread): """Worker for checking updates in a background thread.""" finished = Signal(object) # UpdateInfo @@ -58,7 +57,7 @@ def run(self) -> None: self.error.emit(str(e)) -class UpdateDownloadWorker(QObject): +class UpdateDownloadWorker(QThread): """Worker for downloading updates in a background thread.""" finished = Signal(bool) # success status @@ -84,7 +83,7 @@ def progress_callback(percentage: int) -> None: self.error.emit(str(e)) -class ToolUpdateWorker(QObject): +class ToolUpdateWorker(QThread): """Worker for re-syncing manifest-declared tools in a background thread.""" finished = Signal(int) # number of manifests processed @@ -209,8 +208,8 @@ def __init__( self._client = client self._window = window self._config = config - self._runner: ThreadRunner | None = None - self._tool_runner: ThreadRunner | None = None + self._runner: QThread | None = None + self._tool_runner: QThread | None = None self._progress_dialog: QProgressDialog | None = None self._pending_update_info: UpdateInfo | None = None self._download_cancelled = False @@ -429,7 +428,7 @@ def _do_check_updates(self, *, silent: bool) -> None: worker.finished.connect(lambda result: self._on_update_check_finished(result, silent=silent)) worker.error.connect(lambda error: self._on_update_check_error(error, silent=silent)) - self._runner = ThreadRunner(worker) + self._runner = worker self._runner.start() def _on_update_check_finished(self, result: UpdateInfo | None, *, silent: bool = False) -> None: @@ -509,7 +508,7 @@ def _on_tool_update(self) -> None: worker.finished.connect(self._on_tool_update_finished) worker.error.connect(self._on_tool_update_error) - self._tool_runner = ThreadRunner(worker) + self._tool_runner = worker self._tool_runner.start() def _on_single_plugin_update(self, plugin_name: str) -> None: @@ -525,7 +524,7 @@ def _on_single_plugin_update(self, plugin_name: str) -> None: worker.finished.connect(self._on_tool_update_finished) worker.error.connect(self._on_tool_update_error) - self._tool_runner = ThreadRunner(worker) + self._tool_runner = worker self._tool_runner.start() def _on_tool_update_finished(self, count: int) -> None: @@ -570,7 +569,7 @@ def _start_download(self) -> None: worker.progress.connect(self._on_download_progress) worker.error.connect(self._on_download_error) - self._runner = ThreadRunner(worker) + self._runner = worker self._runner.start() def _on_download_cancelled(self) -> None: diff --git a/synodic_client/application/threading.py b/synodic_client/application/threading.py deleted file mode 100644 index 474c9da..0000000 --- a/synodic_client/application/threading.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Reusable QThread worker runner. - -Encapsulates the boilerplate for moving a ``QObject`` worker onto a -``QThread``, wiring lifecycle signals, and starting execution. - -.. note:: - - The caller **must** store the returned ``ThreadRunner`` as an instance - attribute to prevent premature garbage collection. -""" - -from __future__ import annotations - -from PySide6.QtCore import QObject, QThread, Signal - - -class ThreadRunner(QObject): - """Manages a worker ``QObject`` on a dedicated ``QThread``. - - Usage:: - - runner = ThreadRunner(my_worker) - runner.start() - - The runner connects ``thread.started`` → ``worker.run()`` and ensures - both the thread and worker are cleaned up via ``deleteLater`` when the - thread finishes. - - Callers should connect domain signals (``finished``, ``error``, etc.) - on the *worker* before calling :meth:`start`. - """ - - #: Emitted when the managed thread finishes (for external cleanup). - thread_finished = Signal() - - def __init__(self, worker: QObject, parent: QObject | None = None) -> None: - """Initialise the runner. - - Args: - worker: A ``QObject`` with a ``run()`` slot. Must **not** - already be parented — ``moveToThread`` requires this. - parent: Optional parent for preventing GC. - """ - super().__init__(parent) - self._thread = QThread() - self._worker = worker - worker.moveToThread(self._thread) - - # Start the worker when the thread begins - self._thread.started.connect(worker.run) # type: ignore[attr-defined] - - # Quit-and-cleanup wiring: connect any ``finished`` / ``error`` - # signals on the worker so the thread stops automatically. - for signal_name in ('finished', 'error'): - signal = getattr(worker, signal_name, None) - if signal is not None: - signal.connect(self._thread.quit) - - self._thread.finished.connect(self._thread.deleteLater) - self._thread.finished.connect(worker.deleteLater) - self._thread.finished.connect(self.thread_finished) - - # -- public helpers -- - - def start(self) -> None: - """Start the background thread.""" - self._thread.start() - - def quit_and_wait(self) -> None: - """Ask the thread to quit and block until it finishes.""" - self._thread.quit() - self._thread.wait() - - @property - def managed_thread(self) -> QThread: - """Return the underlying ``QThread``.""" - return self._thread - - @property - def worker(self) -> QObject: - """Return the managed worker.""" - return self._worker diff --git a/synodic_client/protocol.py b/synodic_client/protocol.py index 8bfb60a..99c95e6 100644 --- a/synodic_client/protocol.py +++ b/synodic_client/protocol.py @@ -10,74 +10,75 @@ import logging import sys -if sys.platform == 'win32': - import winreg - logger = logging.getLogger(__name__) PROTOCOL_NAME = 'synodic' _PROTOCOL_DESCRIPTION = 'Synodic Client Protocol' -def register_protocol(exe_path: str) -> None: - """Register the ``synodic://`` URI protocol handler. +if sys.platform == 'win32': + import winreg - Args: - exe_path: Absolute path to the application executable. - """ - if sys.platform != 'win32': + def register_protocol(exe_path: str) -> None: + """Register the ``synodic://`` URI protocol handler. + + Args: + exe_path: Absolute path to the application executable. + """ + key_path = f'Software\\Classes\\{PROTOCOL_NAME}' + + try: + with winreg.CreateKey(winreg.HKEY_CURRENT_USER, key_path) as key: + winreg.SetValueEx(key, '', 0, winreg.REG_SZ, _PROTOCOL_DESCRIPTION) + winreg.SetValueEx(key, 'URL Protocol', 0, winreg.REG_SZ, '') + + command_path = f'{key_path}\\shell\\open\\command' + with winreg.CreateKey(winreg.HKEY_CURRENT_USER, command_path) as key: + winreg.SetValueEx(key, '', 0, winreg.REG_SZ, f'"{exe_path}" "%1"') + + logger.info('Registered synodic:// protocol handler -> %s', exe_path) + except OSError: + logger.exception('Failed to register synodic:// protocol handler') + + def remove_protocol() -> None: + """Remove the ``synodic://`` URI protocol handler registration.""" + key_path = f'Software\\Classes\\{PROTOCOL_NAME}' + + try: + _delete_key_recursive(winreg.HKEY_CURRENT_USER, key_path) + logger.info('Removed synodic:// protocol handler registration') + except FileNotFoundError: + logger.debug('Protocol handler registration not found, nothing to remove') + except OSError: + logger.exception('Failed to remove synodic:// protocol handler') + + def _delete_key_recursive(root: int, key_path: str) -> None: + """Recursively delete a registry key and all its subkeys. + + Args: + root: Registry root (e.g. ``winreg.HKEY_CURRENT_USER``). + key_path: Path to the key to delete. + """ + with winreg.OpenKey(root, key_path, 0, winreg.KEY_ALL_ACCESS) as key: + while True: + try: + subkey_name = winreg.EnumKey(key, 0) + _delete_key_recursive(root, f'{key_path}\\{subkey_name}') + except OSError: + break + + winreg.DeleteKey(root, key_path) + +else: + + def register_protocol(exe_path: str) -> None: + """Register the ``synodic://`` URI protocol handler (no-op on non-Windows). + + Args: + exe_path: Absolute path to the application executable. + """ logger.warning('Protocol registration is only supported on Windows (current: %s)', sys.platform) - return - - key_path = f'Software\\Classes\\{PROTOCOL_NAME}' - - try: - # Create the protocol key - with winreg.CreateKey(winreg.HKEY_CURRENT_USER, key_path) as key: - winreg.SetValueEx(key, '', 0, winreg.REG_SZ, _PROTOCOL_DESCRIPTION) - winreg.SetValueEx(key, 'URL Protocol', 0, winreg.REG_SZ, '') - - # Create the shell\open\command key with the exe path - command_path = f'{key_path}\\shell\\open\\command' - with winreg.CreateKey(winreg.HKEY_CURRENT_USER, command_path) as key: - winreg.SetValueEx(key, '', 0, winreg.REG_SZ, f'"{exe_path}" "%1"') - - logger.info('Registered synodic:// protocol handler -> %s', exe_path) - except OSError: - logger.exception('Failed to register synodic:// protocol handler') - -def remove_protocol() -> None: - """Remove the ``synodic://`` URI protocol handler registration.""" - if sys.platform != 'win32': + def remove_protocol() -> None: + """Remove the ``synodic://`` URI protocol handler registration (no-op on non-Windows).""" logger.warning('Protocol removal is only supported on Windows (current: %s)', sys.platform) - return - - key_path = f'Software\\Classes\\{PROTOCOL_NAME}' - - try: - _delete_key_recursive(winreg.HKEY_CURRENT_USER, key_path) - logger.info('Removed synodic:// protocol handler registration') - except FileNotFoundError: - logger.debug('Protocol handler registration not found, nothing to remove') - except OSError: - logger.exception('Failed to remove synodic:// protocol handler') - - -def _delete_key_recursive(root: int, key_path: str) -> None: - """Recursively delete a registry key and all its subkeys. - - Args: - root: Registry root (e.g. ``winreg.HKEY_CURRENT_USER``). - key_path: Path to the key to delete. - """ - with winreg.OpenKey(root, key_path, 0, winreg.KEY_ALL_ACCESS) as key: - # Enumerate and delete all subkeys first - while True: - try: - subkey_name = winreg.EnumKey(key, 0) - _delete_key_recursive(root, f'{key_path}\\{subkey_name}') - except OSError: - break - - winreg.DeleteKey(root, key_path) diff --git a/synodic_client/updater.py b/synodic_client/updater.py index 791bb35..3bddf69 100644 --- a/synodic_client/updater.py +++ b/synodic_client/updater.py @@ -14,8 +14,10 @@ from enum import Enum, StrEnum, auto from typing import Any -import velopack from packaging.version import Version +from velopack import App as _VelopackApp +from velopack import UpdateManager as _UpdateManager +from velopack import UpdateOptions as _UpdateOptions from synodic_client.protocol import register_protocol, remove_protocol @@ -294,11 +296,11 @@ def _get_velopack_manager(self) -> Any: return self._velopack_manager try: - options = velopack.UpdateOptions() # type: ignore[attr-defined] + options = _UpdateOptions() options.allow_version_downgrade = False options.explicit_channel = self._config.channel_name - self._velopack_manager = velopack.UpdateManager( # type: ignore[attr-defined] + self._velopack_manager = _UpdateManager( self._config.repo_url, options, ) @@ -340,7 +342,7 @@ def initialize_velopack() -> None: On Windows, install/uninstall hooks register the ``synodic://`` URI protocol. """ try: - app = velopack.App() # type: ignore[attr-defined] + app = _VelopackApp() app.on_after_install_fast_callback(_on_after_install) app.on_before_uninstall_fast_callback(_on_before_uninstall) app.run() diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py index a145ed4..9c06e9c 100644 --- a/tests/unit/test_protocol.py +++ b/tests/unit/test_protocol.py @@ -176,7 +176,7 @@ def test_protocol_is_registered() -> None: _, reg_type = winreg.QueryValueEx(key, 'URL Protocol') assert reg_type == winreg.REG_SZ except FileNotFoundError: - pytest.fail(f'Protocol handler not registered. Run the application once to register HKCU\\{key_path}') + pytest.skip('Protocol handler not registered on this machine') @staticmethod def test_command_points_to_existing_exe() -> None: diff --git a/tests/unit/test_updater.py b/tests/unit/test_updater.py index cd2c1dc..99b661f 100644 --- a/tests/unit/test_updater.py +++ b/tests/unit/test_updater.py @@ -439,9 +439,9 @@ class TestInitializeVelopack: @staticmethod def test_initialize_success() -> None: - """Verify initialize_velopack calls velopack.App().run().""" + """Verify initialize_velopack calls _VelopackApp().run().""" mock_app = MagicMock() - with patch('synodic_client.updater.velopack.App', return_value=mock_app) as mock_app_class: + with patch('synodic_client.updater._VelopackApp', return_value=mock_app) as mock_app_class: initialize_velopack() mock_app_class.assert_called_once() mock_app.run.assert_called_once() @@ -451,6 +451,6 @@ def test_initialize_handles_exception() -> None: """Verify initialize_velopack handles exceptions gracefully.""" mock_app = MagicMock() mock_app.run.side_effect = Exception('Test') - with patch('synodic_client.updater.velopack.App', return_value=mock_app): + with patch('synodic_client.updater._VelopackApp', return_value=mock_app): # Should not raise initialize_velopack() diff --git a/tests/unit/windows/test_protocol.py b/tests/unit/windows/test_protocol.py index a145ed4..9c06e9c 100644 --- a/tests/unit/windows/test_protocol.py +++ b/tests/unit/windows/test_protocol.py @@ -176,7 +176,7 @@ def test_protocol_is_registered() -> None: _, reg_type = winreg.QueryValueEx(key, 'URL Protocol') assert reg_type == winreg.REG_SZ except FileNotFoundError: - pytest.fail(f'Protocol handler not registered. Run the application once to register HKCU\\{key_path}') + pytest.skip('Protocol handler not registered on this machine') @staticmethod def test_command_points_to_existing_exe() -> None: From 0bcd2c98f2c0c730ade0b1f22cb12a3cdb32b983 Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 16:01:19 -0800 Subject: [PATCH 4/7] Windows Registration Cleanup --- pyproject.toml | 1 - synodic_client/application/qt.py | 32 +++--------------------- synodic_client/protocol.py | 35 ++++++++++----------------- synodic_client/updater.py | 10 +++----- tests/unit/test_protocol.py | 7 ++---- tests/unit/test_updater.py | 6 ++--- tests/unit/windows/test_protocol.py | 7 ++---- tool/pyinstaller/rthook_no_console.py | 24 ++++++++++++++++++ tool/pyinstaller/synodic.spec | 2 +- 9 files changed, 52 insertions(+), 72 deletions(-) create mode 100644 tool/pyinstaller/rthook_no_console.py diff --git a/pyproject.toml b/pyproject.toml index 703022e..2c13973 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,7 +68,6 @@ quote-style = "single" skip_empty = true [tool.pyrefly] -replace_imports_with_any = ["velopack"] [tool.pdm.version] source = "scm" diff --git a/synodic_client/application/qt.py b/synodic_client/application/qt.py index 6188c7c..79b25ea 100644 --- a/synodic_client/application/qt.py +++ b/synodic_client/application/qt.py @@ -3,11 +3,9 @@ import ctypes import logging import signal -import subprocess import sys import types from collections.abc import Callable -from typing import Any from porringer.api import API from porringer.schema import LocalConfiguration @@ -65,28 +63,6 @@ def _process_uri(uri: str, handler: Callable[[str], None]) -> None: handler(manifests[0]) -def _suppress_subprocess_consoles() -> None: - """Monkey-patch ``subprocess.Popen`` to hide console windows on Windows. - - When the application is built as a windowed executable (``console=False`` - in PyInstaller), every ``subprocess.Popen`` call that launches a console - program (pip, pipx, uv, winget, etc.) would briefly flash a visible - console window. This patch adds ``CREATE_NO_WINDOW`` to *creationflags* - for all calls that don't already set it, suppressing those flashes. - """ - if sys.platform != 'win32': - return - - _original_init = subprocess.Popen.__init__ - - def _patched_init(self: subprocess.Popen, *args: Any, **kwargs: Any) -> None: - if 'creationflags' not in kwargs: - kwargs['creationflags'] = getattr(subprocess, 'CREATE_NO_WINDOW', 0) - _original_init(self, *args, **kwargs) - - subprocess.Popen.__init__ = _patched_init - - def _install_exception_hook(logger: logging.Logger) -> None: """Redirect unhandled exceptions to the log file. @@ -146,12 +122,10 @@ def application(*, uri: str | None = None, dev_mode: bool = False) -> None: # Activate dev-mode namespacing before anything reads config paths. set_dev_mode(dev_mode) - # Suppress console window flashes from subprocess calls (e.g. porringer - # running pip, pipx, uv) before any subprocesses are spawned. Skipped - # in dev mode because the source-run process already has a console. if not dev_mode: - _suppress_subprocess_consoles() - # Initialize Velopack early, before any UI + # Initialize Velopack early, before any UI. + # Console window suppression for subprocesses is handled by the + # PyInstaller runtime hook (rthook_no_console.py). initialize_velopack() register_protocol(sys.executable) diff --git a/synodic_client/protocol.py b/synodic_client/protocol.py index 99c95e6..4bc4d20 100644 --- a/synodic_client/protocol.py +++ b/synodic_client/protocol.py @@ -17,8 +17,16 @@ if sys.platform == 'win32': + import ctypes import winreg + # Bind RegDeleteTreeW for recursive registry key deletion in a single call. + _reg_delete_tree = ctypes.windll.advapi32.RegDeleteTreeW + _reg_delete_tree.argtypes = [ctypes.c_void_p, ctypes.c_wchar_p] + _reg_delete_tree.restype = ctypes.c_long + + _ERROR_FILE_NOT_FOUND = 2 + def register_protocol(exe_path: str) -> None: """Register the ``synodic://`` URI protocol handler. @@ -44,30 +52,13 @@ def remove_protocol() -> None: """Remove the ``synodic://`` URI protocol handler registration.""" key_path = f'Software\\Classes\\{PROTOCOL_NAME}' - try: - _delete_key_recursive(winreg.HKEY_CURRENT_USER, key_path) + result = _reg_delete_tree(winreg.HKEY_CURRENT_USER, key_path) + if result == 0: logger.info('Removed synodic:// protocol handler registration') - except FileNotFoundError: + elif result == _ERROR_FILE_NOT_FOUND: logger.debug('Protocol handler registration not found, nothing to remove') - except OSError: - logger.exception('Failed to remove synodic:// protocol handler') - - def _delete_key_recursive(root: int, key_path: str) -> None: - """Recursively delete a registry key and all its subkeys. - - Args: - root: Registry root (e.g. ``winreg.HKEY_CURRENT_USER``). - key_path: Path to the key to delete. - """ - with winreg.OpenKey(root, key_path, 0, winreg.KEY_ALL_ACCESS) as key: - while True: - try: - subkey_name = winreg.EnumKey(key, 0) - _delete_key_recursive(root, f'{key_path}\\{subkey_name}') - except OSError: - break - - winreg.DeleteKey(root, key_path) + else: + logger.error('Failed to remove synodic:// protocol handler (error code %d)', result) else: diff --git a/synodic_client/updater.py b/synodic_client/updater.py index 3bddf69..ad8adbb 100644 --- a/synodic_client/updater.py +++ b/synodic_client/updater.py @@ -14,10 +14,8 @@ from enum import Enum, StrEnum, auto from typing import Any +import velopack # type: ignore[import] from packaging.version import Version -from velopack import App as _VelopackApp -from velopack import UpdateManager as _UpdateManager -from velopack import UpdateOptions as _UpdateOptions from synodic_client.protocol import register_protocol, remove_protocol @@ -296,11 +294,11 @@ def _get_velopack_manager(self) -> Any: return self._velopack_manager try: - options = _UpdateOptions() + options = velopack.UpdateOptions() options.allow_version_downgrade = False options.explicit_channel = self._config.channel_name - self._velopack_manager = _UpdateManager( + self._velopack_manager = velopack.UpdateManager( self._config.repo_url, options, ) @@ -342,7 +340,7 @@ def initialize_velopack() -> None: On Windows, install/uninstall hooks register the ``synodic://`` URI protocol. """ try: - app = _VelopackApp() + app = velopack.App() app.on_after_install_fast_callback(_on_after_install) app.on_before_uninstall_fast_callback(_on_before_uninstall) app.run() diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py index 9c06e9c..91a230d 100644 --- a/tests/unit/test_protocol.py +++ b/tests/unit/test_protocol.py @@ -70,7 +70,7 @@ class TestRemoveProtocol: @staticmethod def test_deletes_registry_key() -> None: """Verify the protocol key is deleted.""" - with patch('synodic_client.protocol._delete_key_recursive') as mock_delete: + with patch('synodic_client.protocol._reg_delete_tree', return_value=0) as mock_delete: remove_protocol() mock_delete.assert_called_once_with( @@ -81,10 +81,7 @@ def test_deletes_registry_key() -> None: @staticmethod def test_handles_missing_key_gracefully() -> None: """Verify no error when protocol key doesn't exist.""" - with patch( - 'synodic_client.protocol._delete_key_recursive', - side_effect=FileNotFoundError, - ): + with patch('synodic_client.protocol._reg_delete_tree', return_value=2): # ERROR_FILE_NOT_FOUND # Should not raise remove_protocol() diff --git a/tests/unit/test_updater.py b/tests/unit/test_updater.py index 99b661f..76df689 100644 --- a/tests/unit/test_updater.py +++ b/tests/unit/test_updater.py @@ -439,9 +439,9 @@ class TestInitializeVelopack: @staticmethod def test_initialize_success() -> None: - """Verify initialize_velopack calls _VelopackApp().run().""" + """Verify initialize_velopack calls App().run().""" mock_app = MagicMock() - with patch('synodic_client.updater._VelopackApp', return_value=mock_app) as mock_app_class: + with patch('synodic_client.updater.velopack.App', return_value=mock_app) as mock_app_class: initialize_velopack() mock_app_class.assert_called_once() mock_app.run.assert_called_once() @@ -451,6 +451,6 @@ def test_initialize_handles_exception() -> None: """Verify initialize_velopack handles exceptions gracefully.""" mock_app = MagicMock() mock_app.run.side_effect = Exception('Test') - with patch('synodic_client.updater._VelopackApp', return_value=mock_app): + with patch('synodic_client.updater.velopack.App', return_value=mock_app): # Should not raise initialize_velopack() diff --git a/tests/unit/windows/test_protocol.py b/tests/unit/windows/test_protocol.py index 9c06e9c..91a230d 100644 --- a/tests/unit/windows/test_protocol.py +++ b/tests/unit/windows/test_protocol.py @@ -70,7 +70,7 @@ class TestRemoveProtocol: @staticmethod def test_deletes_registry_key() -> None: """Verify the protocol key is deleted.""" - with patch('synodic_client.protocol._delete_key_recursive') as mock_delete: + with patch('synodic_client.protocol._reg_delete_tree', return_value=0) as mock_delete: remove_protocol() mock_delete.assert_called_once_with( @@ -81,10 +81,7 @@ def test_deletes_registry_key() -> None: @staticmethod def test_handles_missing_key_gracefully() -> None: """Verify no error when protocol key doesn't exist.""" - with patch( - 'synodic_client.protocol._delete_key_recursive', - side_effect=FileNotFoundError, - ): + with patch('synodic_client.protocol._reg_delete_tree', return_value=2): # ERROR_FILE_NOT_FOUND # Should not raise remove_protocol() diff --git a/tool/pyinstaller/rthook_no_console.py b/tool/pyinstaller/rthook_no_console.py new file mode 100644 index 0000000..3f08fa1 --- /dev/null +++ b/tool/pyinstaller/rthook_no_console.py @@ -0,0 +1,24 @@ +"""PyInstaller runtime hook: suppress console windows for child processes. + +When the application is built as a windowed executable (``console=False``), +every ``subprocess.Popen`` call that launches a console program (pip, pipx, +uv, winget, etc.) would briefly flash a visible console window. This hook +merges ``CREATE_NO_WINDOW`` into *creationflags* for every call, suppressing +those flashes while preserving any flags the caller already set. + +Placed as a runtime hook so the patch is active before any application or +library code spawns subprocesses. +""" + +import subprocess +import sys +from typing import Any + +if sys.platform == 'win32': + _original_init = subprocess.Popen.__init__ + + def _patched_init(self: subprocess.Popen, *args: Any, **kwargs: Any) -> None: + kwargs['creationflags'] = kwargs.get('creationflags', 0) | subprocess.CREATE_NO_WINDOW + _original_init(self, *args, **kwargs) + + subprocess.Popen.__init__ = _patched_init diff --git a/tool/pyinstaller/synodic.spec b/tool/pyinstaller/synodic.spec index 39fe5ba..3426e20 100644 --- a/tool/pyinstaller/synodic.spec +++ b/tool/pyinstaller/synodic.spec @@ -40,7 +40,7 @@ a = Analysis( hiddenimports=hiddenimports, hookspath=[], hooksconfig={}, - runtime_hooks=[], + runtime_hooks=['rthook_no_console.py'], excludes=[], noarchive=False, ) From 15a1c2f46767041a55c29a3a6f29aac37ae4ecf3 Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 16:12:03 -0800 Subject: [PATCH 5/7] Resolve Linux Lint --- pyproject.toml | 1 + synodic_client/updater.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 2c13973..c3ffba9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,6 +68,7 @@ quote-style = "single" skip_empty = true [tool.pyrefly] +replace-imports-with-any = ["velopack", "winreg"] [tool.pdm.version] source = "scm" diff --git a/synodic_client/updater.py b/synodic_client/updater.py index ad8adbb..a24d6ea 100644 --- a/synodic_client/updater.py +++ b/synodic_client/updater.py @@ -14,7 +14,7 @@ from enum import Enum, StrEnum, auto from typing import Any -import velopack # type: ignore[import] +import velopack from packaging.version import Version from synodic_client.protocol import register_protocol, remove_protocol From 63e6d7d2fec1a5d77cdadea13a538b344cfffa70 Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 16:21:39 -0800 Subject: [PATCH 6/7] Fix Per-Platform Tests --- tests/unit/qt/conftest.py | 2 +- tests/unit/test_cli.py | 4 + tests/unit/test_install_preview.py | 4 + tests/unit/test_protocol.py | 190 ----------------------------- 4 files changed, 9 insertions(+), 191 deletions(-) delete mode 100644 tests/unit/test_protocol.py diff --git a/tests/unit/qt/conftest.py b/tests/unit/qt/conftest.py index b342242..6163ef3 100644 --- a/tests/unit/qt/conftest.py +++ b/tests/unit/qt/conftest.py @@ -7,4 +7,4 @@ import pytest -pytest.importorskip('PySide6', reason='PySide6 requires system Qt libraries') +pytest.importorskip("PySide6.QtWidgets", reason="PySide6 requires system Qt libraries") diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 5d6dd01..cf89a91 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -2,6 +2,10 @@ from unittest.mock import patch +import pytest + +pytest.importorskip('PySide6.QtWidgets', reason='PySide6 requires system Qt libraries') + from typer.testing import CliRunner from synodic_client.cli import app diff --git a/tests/unit/test_install_preview.py b/tests/unit/test_install_preview.py index 8310154..e71cac4 100644 --- a/tests/unit/test_install_preview.py +++ b/tests/unit/test_install_preview.py @@ -2,6 +2,10 @@ from __future__ import annotations +import pytest + +pytest.importorskip('PySide6.QtWidgets', reason='PySide6 requires system Qt libraries') + from pathlib import Path from typing import Any from unittest.mock import MagicMock diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py deleted file mode 100644 index 91a230d..0000000 --- a/tests/unit/test_protocol.py +++ /dev/null @@ -1,190 +0,0 @@ -"""Tests for URI protocol handler registration.""" - -import winreg -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest - -from synodic_client.protocol import PROTOCOL_NAME, register_protocol, remove_protocol - -_EXPECTED_REGISTRY_KEY_COUNT = 2 - -_TEST_PROTOCOL = f'{PROTOCOL_NAME}_test' -"""Temporary protocol name used by integration tests to avoid clobbering the real registration.""" - - -class TestRegisterProtocol: - """Tests for register_protocol.""" - - @staticmethod - def test_writes_registry_keys() -> None: - """Verify correct registry keys are written on Windows.""" - mock_key = MagicMock() - mock_key.__enter__ = MagicMock(return_value=mock_key) - mock_key.__exit__ = MagicMock(return_value=False) - - with ( - patch.object(winreg, 'CreateKey', return_value=mock_key) as mock_create, - patch.object(winreg, 'SetValueEx'), - ): - register_protocol(r'C:\Program Files\Synodic\synodic.exe') - - # Should create the protocol key and command key - assert mock_create.call_count == _EXPECTED_REGISTRY_KEY_COUNT - - # Verify protocol key path - first_call_args = mock_create.call_args_list[0][0] - assert first_call_args[1] == f'Software\\Classes\\{PROTOCOL_NAME}' - - # Verify command key path - second_call_args = mock_create.call_args_list[1][0] - assert second_call_args[1].endswith('shell\\open\\command') - - @staticmethod - def test_sets_url_protocol_value() -> None: - """Verify the 'URL Protocol' value is set.""" - mock_key = MagicMock() - mock_key.__enter__ = MagicMock(return_value=mock_key) - mock_key.__exit__ = MagicMock(return_value=False) - - with patch.object(winreg, 'CreateKey', return_value=mock_key), patch.object(winreg, 'SetValueEx') as mock_set: - register_protocol(r'C:\Program Files\Synodic\synodic.exe') - - # Check that SetValueEx was called with 'URL Protocol' - url_protocol_calls = [c for c in mock_set.call_args_list if c[0][1] == 'URL Protocol'] - assert len(url_protocol_calls) == 1 - - @staticmethod - def test_noop_on_non_windows() -> None: - """Verify register_protocol is a no-op on non-Windows platforms.""" - with patch('synodic_client.protocol.sys') as mock_sys: - mock_sys.platform = 'linux' - # Should not raise and should not attempt winreg import - register_protocol('/usr/bin/synodic') - - -class TestRemoveProtocol: - """Tests for remove_protocol.""" - - @staticmethod - def test_deletes_registry_key() -> None: - """Verify the protocol key is deleted.""" - with patch('synodic_client.protocol._reg_delete_tree', return_value=0) as mock_delete: - remove_protocol() - - mock_delete.assert_called_once_with( - winreg.HKEY_CURRENT_USER, - f'Software\\Classes\\{PROTOCOL_NAME}', - ) - - @staticmethod - def test_handles_missing_key_gracefully() -> None: - """Verify no error when protocol key doesn't exist.""" - with patch('synodic_client.protocol._reg_delete_tree', return_value=2): # ERROR_FILE_NOT_FOUND - # Should not raise - remove_protocol() - - @staticmethod - def test_noop_on_non_windows() -> None: - """Verify remove_protocol is a no-op on non-Windows platforms.""" - with patch('synodic_client.protocol.sys') as mock_sys: - mock_sys.platform = 'linux' - remove_protocol() - - -class TestProtocolIntegration: - """Integration tests that read/write real registry keys under a test protocol name.""" - - @staticmethod - def test_register_creates_valid_registry_entries() -> None: - """Register under a test key, verify values, then clean up.""" - test_exe = r'C:\test\synodic_test.exe' - key_path = f'Software\\Classes\\{_TEST_PROTOCOL}' - - try: - # Register using the test protocol name - with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): - register_protocol(test_exe) - - # Verify the protocol key - with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: - description, _ = winreg.QueryValueEx(key, '') - assert description == 'Synodic Client Protocol' - - url_protocol, _ = winreg.QueryValueEx(key, 'URL Protocol') - assert not url_protocol - - # Verify the command key - command_path = f'{key_path}\\shell\\open\\command' - with winreg.OpenKey(winreg.HKEY_CURRENT_USER, command_path) as key: - command, _ = winreg.QueryValueEx(key, '') - assert test_exe in command - assert '"%1"' in command - - finally: - # Clean up the test key - with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): - remove_protocol() - - @staticmethod - def test_remove_deletes_registry_entries() -> None: - """Register then remove under a test key, verify the key is gone.""" - key_path = f'Software\\Classes\\{_TEST_PROTOCOL}' - - with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): - register_protocol(r'C:\test\synodic_test.exe') - remove_protocol() - - with pytest.raises(FileNotFoundError): - winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) - - @staticmethod - def test_register_is_idempotent() -> None: - """Calling register twice with a different exe updates the command.""" - key_path = f'Software\\Classes\\{_TEST_PROTOCOL}\\shell\\open\\command' - exe_v1 = r'C:\test\v1\synodic.exe' - exe_v2 = r'C:\test\v2\synodic.exe' - - try: - with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): - register_protocol(exe_v1) - register_protocol(exe_v2) - - with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: - command, _ = winreg.QueryValueEx(key, '') - assert exe_v2 in command - assert exe_v1 not in command - - finally: - with patch('synodic_client.protocol.PROTOCOL_NAME', _TEST_PROTOCOL): - remove_protocol() - - -class TestProtocolLive: - """Verify the live protocol registration on this machine.""" - - @staticmethod - def test_protocol_is_registered() -> None: - """Verify that the synodic:// protocol handler is currently registered.""" - key_path = f'Software\\Classes\\{PROTOCOL_NAME}' - try: - with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: - _, reg_type = winreg.QueryValueEx(key, 'URL Protocol') - assert reg_type == winreg.REG_SZ - except FileNotFoundError: - pytest.skip('Protocol handler not registered on this machine') - - @staticmethod - def test_command_points_to_existing_exe() -> None: - """Verify the registered command points to an exe path (may not exist in CI).""" - key_path = f'Software\\Classes\\{PROTOCOL_NAME}\\shell\\open\\command' - try: - with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path) as key: - command, _ = winreg.QueryValueEx(key, '') - # Command format: "C:\...\synodic.exe" "%1" - exe_path = command.split('"')[1] - assert exe_path.endswith('.exe'), f'Expected .exe path, got: {exe_path}' - assert Path(exe_path).name in {'synodic.exe', 'python.exe'}, f'Unexpected exe: {exe_path}' - except FileNotFoundError: - pytest.skip('Protocol handler not registered on this machine') From 0c40a2dba86db62eae75a0cacdd469ad7c73df04 Mon Sep 17 00:00:00 2001 From: Asher Norland Date: Sun, 15 Feb 2026 16:35:16 -0800 Subject: [PATCH 7/7] Fix Paths --- tests/unit/qt/test_install_preview.py | 12 +++++++----- tests/unit/test_install_preview.py | 10 +++++++--- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/tests/unit/qt/test_install_preview.py b/tests/unit/qt/test_install_preview.py index 8310154..7973ac4 100644 --- a/tests/unit/qt/test_install_preview.py +++ b/tests/unit/qt/test_install_preview.py @@ -231,11 +231,12 @@ def test_http_url_returns_none() -> None: assert resolve_local_path('https://example.com/porringer.json') is None @staticmethod - def test_absolute_path_returns_path() -> None: + def test_absolute_path_returns_path(tmp_path: Path) -> None: """Absolute OS paths should resolve.""" - result = resolve_local_path('C:\\Users\\test\\porringer.json') + path = str(tmp_path / 'porringer.json') + result = resolve_local_path(path) assert result is not None - assert result == Path('C:\\Users\\test\\porringer.json') + assert result == Path(path) @staticmethod def test_file_uri_returns_path() -> None: @@ -283,10 +284,11 @@ async def mock_stream(*args: Any, **kwargs: Any) -> Any: porringer.sync.download.assert_not_called() @staticmethod - def test_local_manifest_not_found() -> None: + def test_local_manifest_not_found(tmp_path: Path) -> None: """Verify PreviewWorker emits error for missing local file.""" + path = str(tmp_path / 'nonexistent' / 'porringer.json') porringer = MagicMock() - worker = PreviewWorker(porringer, 'C:\\nonexistent\\porringer.json') + worker = PreviewWorker(porringer, path) errors: list[str] = [] worker.error.connect(errors.append) diff --git a/tests/unit/test_install_preview.py b/tests/unit/test_install_preview.py index e71cac4..0b0b95d 100644 --- a/tests/unit/test_install_preview.py +++ b/tests/unit/test_install_preview.py @@ -2,6 +2,8 @@ from __future__ import annotations +import sys + import pytest pytest.importorskip('PySide6.QtWidgets', reason='PySide6 requires system Qt libraries') @@ -237,9 +239,10 @@ def test_http_url_returns_none() -> None: @staticmethod def test_absolute_path_returns_path() -> None: """Absolute OS paths should resolve.""" - result = resolve_local_path('C:\\Users\\test\\porringer.json') + path = 'C:\\Users\\test\\porringer.json' if sys.platform == 'win32' else '/Users/test/porringer.json' + result = resolve_local_path(path) assert result is not None - assert result == Path('C:\\Users\\test\\porringer.json') + assert result == Path(path) @staticmethod def test_file_uri_returns_path() -> None: @@ -289,8 +292,9 @@ async def mock_stream(*args: Any, **kwargs: Any) -> Any: @staticmethod def test_local_manifest_not_found() -> None: """Verify PreviewWorker emits error for missing local file.""" + path = 'C:\\nonexistent\\porringer.json' if sys.platform == 'win32' else '/nonexistent/porringer.json' porringer = MagicMock() - worker = PreviewWorker(porringer, 'C:\\nonexistent\\porringer.json') + worker = PreviewWorker(porringer, path) errors: list[str] = [] worker.error.connect(errors.append)