refactor: improve exception handling and type hints across modules

- Simplified exception handling in `bridge.py`, `connection.py`, and `suite_leasing.py`.
- Updated type hints in `suite_leasing.py` for better clarity.
- Refactored `find_revit_path` in `discovery.py` to use `Path` for file existence checks.
- Enhanced readability by cleaning up commented-out code and improving function definitions in `dialog_resolver.py` and `plugin.py`.

Author: trgiangv

src/revitdevtool_pytest/bridge.py

@@ -66,7 +66,7 @@ def connect(self) -> None:
                     self._handle, win32pipe.PIPE_READMODE_BYTE, None, None,
                 )
                 return
-            except Exception:
+            except Exception: # noqa
                 if time.monotonic() >= deadline:
                     raise ConnectionError(
                         f"Cannot connect to pipe '{self._pipe_name}' "
@@ -82,11 +82,11 @@ def disconnect(self) -> None:
 
         try:
             win32file.FlushFileBuffers(handle)
-        except Exception:  # noqa: BLE001
+        except Exception:  # noqa
             pass
         try:
             win32file.CloseHandle(handle)
-        except Exception:  # noqa: BLE001
+        except Exception:  # noqa
             pass
 
     @property
@@ -98,7 +98,7 @@ def connected(self) -> bool:
 
             win32pipe.PeekNamedPipe(self._handle, 0)
             return True
-        except Exception:  # noqa: BLE001
+        except Exception:  # noqa
             log.debug("Pipe health check failed, marking as disconnected")
             return False
 
@@ -120,7 +120,7 @@ def run_tests(
             nodeids=nodeids,
             pytest_args=pytest_args or [],
         )
-        response = self._request(
+        response: BridgeResponse = self._request(
             BridgeRequest(method=BRIDGE_METHOD_TESTS_RUN, params=request.to_params()),
             timeout_s,
             on_notification=on_notification,
@@ -138,20 +138,17 @@ def _request(
     ) -> BridgeResponse:
         self._write_frame(req.to_json_bytes())
         deadline = time.monotonic() + timeout_s
-        while True:
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
-                raise TimeoutError(f"Timed out waiting for response to {req.id}")
-            data = self._read_frame(remaining)
+        while time.monotonic() < deadline:
+            data = self._read_frame(deadline - time.monotonic())
             parsed = json.loads(data)
 
             if parsed.get("type") == BRIDGE_MSG_TYPE_NOTIFICATION:
                 _dispatch_notification(parsed, on_notification)
                 continue
 
-            resp = BridgeResponse.from_json(parsed)
-            if resp.id == req.id:
-                return resp
+            return BridgeResponse.from_json(parsed)
+
+        raise TimeoutError(f"Timed out waiting for response to {req.id}")
 
     def _write_frame(self, body: bytes) -> None:
         import win32file  # type: ignore[import-untyped]
@@ -211,5 +208,5 @@ def _dispatch_notification(
     params = parsed.get("params")
     try:
         callback(method, params)
-    except Exception:  # noqa: BLE001
+    except Exception:  # noqa
         log.debug("Notification callback error for method=%s", method, exc_info=True)

src/revitdevtool_pytest/connection.py

@@ -294,7 +294,7 @@ def _resolve_launch_version(
     if version is not None:
         return version
     if instances:
-        return max(instances, key=lambda i: i.version).version
+        return max(instances, key=lambda i: i.version).version  # noqa
     pytest.exit(
         f"{PLUGIN_NAME}: --revit-version is required when no existing instances are available.",
         returncode=EXIT_CODE_CONFIG_ERROR,

src/revitdevtool_pytest/dialog_resolver.py

@@ -112,8 +112,7 @@ def _is_whitelisted(self, title: str) -> bool:
     def _find_button(self, parent: int) -> int | None:
         result: list[tuple[int, int]] = []
 
-        @_EnumWindowsProc
-        def _cb(child: int, _: int) -> bool:
+        def _enum_cb(child: int, _: int) -> bool:
             score = _get_button_score(
                 child,
                 self._opts.preferred_button_keywords,
@@ -123,7 +122,7 @@ def _cb(child: int, _: int) -> bool:
                 result.append((score, child))
             return True
 
-        _user32.EnumChildWindows(parent, _cb, 0)
+        _user32.EnumChildWindows(parent, _EnumWindowsProc(_enum_cb), 0)
         if not result:
             return None
 
@@ -133,13 +132,12 @@ def _cb(child: int, _: int) -> bool:
     def _enum_dialog_windows(self) -> list[int]:
         windows: list[int] = []
 
-        @_EnumWindowsProc
-        def _cb(hwnd: int, _: int) -> bool:
+        def _enum_cb(hwnd: int, _: int) -> bool:
             if _is_target_dialog(hwnd, self._pid):
                 windows.append(hwnd)
             return True
 
-        _user32.EnumWindows(_cb, 0)
+        _user32.EnumWindows(_EnumWindowsProc(_enum_cb), 0)
         return windows
 
 

src/revitdevtool_pytest/discovery.py

@@ -8,6 +8,7 @@
 import time
 import winreg
 from dataclasses import dataclass
+from pathlib import Path
 
 from .constants import (
     DEFAULT_LAUNCH_TIMEOUT_S,
@@ -70,7 +71,7 @@ def find_revit_path(version: int) -> str | None:
     if path:
         return path
     default = f"C:\\Program Files\\Autodesk\\Revit {version}\\Revit.exe"
-    return default if os.path.isfile(default) else None
+    return default if Path(default).is_file() else None
 
 
 def start_revit(version: int) -> int:

src/revitdevtool_pytest/plugin.py

@@ -117,7 +117,7 @@ def pytest_runtestloop(session: pytest.Session) -> bool:
 
 
 @pytest.hookimpl(tryfirst=True)
-def pytest_runtest_protocol(item: pytest.Item, nextitem: pytest.Item | None) -> bool:  # noqa: ARG001
+def pytest_runtest_protocol(item: pytest.Item, nextitem: pytest.Item | None) -> bool:  # noqa
     results_by_nodeid = item.session.stash.get(_remote_results_key, None)
     if results_by_nodeid is None:
         return False
@@ -145,8 +145,8 @@ def _count_failures(item: pytest.Item, results: list[CaseResult]) -> None:
             item.session.testsfailed += 1
 
 
-def pytest_unconfigure(config: pytest.Config) -> None:  # noqa: ARG001
-    global _bridge, _dialog_resolver, _lease_store  # noqa: PLW0603
+def pytest_unconfigure(config: pytest.Config) -> None:  # noqa
+    global _bridge, _dialog_resolver, _lease_store
     _suite_mutex.release()
     if _dialog_resolver is not None:
         _dialog_resolver.stop()

src/revitdevtool_pytest/suite_leasing.py

@@ -8,6 +8,7 @@
 import time
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Any
 
 from .discovery import RevitInstance
 
@@ -28,7 +29,7 @@ class SuiteLease:
     assigned_at: float
     last_seen_at: float
 
-    def to_dict(self) -> dict[str, object]:
+    def to_dict(self) -> dict[str, Any]:
         return {
             "suite_key": self.suite_key,
             "suite_path": self.suite_path,
@@ -39,7 +40,7 @@ def to_dict(self) -> dict[str, object]:
         }
 
     @classmethod
-    def from_dict(cls, data: dict[str, object]) -> SuiteLease:
+    def from_dict(cls, data: dict[str, Any]) -> SuiteLease:
         return cls(
             suite_key=str(data.get("suite_key", "")),
             suite_path=str(data.get("suite_path", "")),
@@ -115,7 +116,7 @@ def _load_leases(self) -> dict[str, SuiteLease]:
             return {}
         try:
             payload = json.loads(self._state_file.read_text(encoding="utf-8"))
-        except Exception:
+        except Exception: # noqa
             return {}
 
         if not isinstance(payload, dict):
@@ -145,7 +146,8 @@ def _save_leases(self) -> None:
         content = json.dumps(payload, ensure_ascii=False, indent=2)
         base_tmp = self._state_file.with_suffix(".tmp")
 
-        for attempt, delay in enumerate((*_SAVE_RETRY_DELAYS_S, None), start=1):
+        delays: tuple[float | None, ...] = (*_SAVE_RETRY_DELAYS_S, None)
+        for attempt, delay in enumerate(delays, start=1):
             tmp_file = base_tmp.with_name(f"{base_tmp.stem}.{os.getpid()}.{random.randint(1000, 9999)}{base_tmp.suffix}")
             try:
                 tmp_file.write_text(content, encoding="utf-8")