diff --git a/client/src/cmbuyer_client/app.py b/client/src/cmbuyer_client/app.py index 2c291a9..d86b02a 100644 --- a/client/src/cmbuyer_client/app.py +++ b/client/src/cmbuyer_client/app.py @@ -5,8 +5,9 @@ from __future__ import annotations import sys from collections.abc import Sequence +from .core.errors import StateError from .logging_policy import configure_application_logger -from .runtime import RuntimePaths +from .runtime import LocalStateRuntime, RuntimePaths def select_application_argv(argv: Sequence[str] | None) -> list[str]: @@ -30,8 +31,7 @@ def main(argv: Sequence[str] | None = None) -> int: return 1 try: - from PySide6.QtCore import Qt - from PySide6.QtWidgets import QApplication, QLabel, QMainWindow + from PySide6.QtWidgets import QApplication except ImportError: logger.error("缺少 PySide6，无法启动桌面界面。") print("无法启动采购工具：缺少 PySide6。请先安装 requirements.txt 中的依赖。", file=sys.stderr) @@ -39,19 +39,47 @@ def main(argv: Sequence[str] | None = None) -> int: application = QApplication.instance() or QApplication(select_application_argv(argv)) application.setApplicationName("采购工具") + runtime: LocalStateRuntime | None = None + coordinator = None + try: + runtime = LocalStateRuntime.open(paths) + try: + summary = runtime.store.load_profile_summary("default") + except StateError as error: + if error.reason != "profile_not_found": + raise + summary = None - window = QMainWindow() - window.setWindowTitle("采购工具") - window.setAccessibleName("采购工具") - window.setMinimumSize(420, 240) - window.resize(560, 320) + from .polling.coordinator import PollingCoordinator + from .ui.main_window import PurchaseToolWindow - message = QLabel("应用骨架已初始化。\n采购执行功能尚未启用。") - message.setAlignment(Qt.AlignmentFlag.AlignCenter) - message.setWordWrap(True) - message.setAccessibleName("当前状态") - window.setCentralWidget(message) - - logger.info("应用已启动；采购执行功能尚未启用。") - window.show() - return application.exec() + settings = None if summary is None else summary.settings + has_token = False if summary is None else summary.has_stored_device_token + coordinator = PollingCoordinator( + profile_id="default", + store=runtime.store, + gateway_factory=None, + consumer=None, + profile_settings=settings, + poll_interval_seconds=15 if settings is None else settings.poll_interval_seconds, + failure_threshold=3 if settings is None else settings.failure_threshold, + ) + window = PurchaseToolWindow( + store=runtime.store, + coordinator=coordinator, + profile_settings=settings, + has_stored_device_token=has_token, + ) + logger.info("应用已启动；单趟执行能力尚未接入，真实领取保持禁用。") + window.show() + return application.exec() + except (OSError, RuntimeError, StateError): + logger.error("无法打开采购工具本地安全状态。") + print("无法启动采购工具：本地安全状态不可用。", file=sys.stderr) + return 3 + finally: + worker_stopped = True + if coordinator is not None: + worker_stopped = coordinator.shutdown() + if runtime is not None and worker_stopped: + runtime.close() diff --git a/client/src/cmbuyer_client/polling/__init__.py b/client/src/cmbuyer_client/polling/__init__.py new file mode 100644 index 0000000..97e5e31 --- /dev/null +++ b/client/src/cmbuyer_client/polling/__init__.py @@ -0,0 +1,5 @@ +"""采购工具轮询会话协调器。""" + +from .coordinator import ClaimedTaskView, PollingCoordinator, PollingState, RecoveryStatus, StartReadiness + +__all__ = ["ClaimedTaskView", "PollingCoordinator", "PollingState", "RecoveryStatus", "StartReadiness"] diff --git a/client/src/cmbuyer_client/polling/coordinator.py b/client/src/cmbuyer_client/polling/coordinator.py new file mode 100644 index 0000000..875b5f9 --- /dev/null +++ b/client/src/cmbuyer_client/polling/coordinator.py @@ -0,0 +1,586 @@ +"""在 Qt 事件循环中协调可恢复的领取会话。""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from enum import Enum +from typing import Protocol + +from PySide6.QtCore import QObject, QThread, QTimer, Signal, Slot + +from cmbuyer_client.core.errors import ( + AmbiguousRemoteError, + ClientError, + CredentialRemoteError, + ManualRemoteError, + ProtocolRemoteError, + StateError, +) +from cmbuyer_client.core.models import ClaimedTask +from cmbuyer_client.localstate.models import PollingSession, ProfileSettings, RecoverySnapshot +from cmbuyer_client.logging_policy import redact_text + + +SAFE_AMBIGUOUS_REASONS = frozenset( + ("http_result_unknown", "server_result_unknown", "truncated_response") +) + + +class PollingState(str, Enum): + STOPPED = "STOPPED" + STARTING = "STARTING" + BLOCKED = "BLOCKED" + RECOVERING = "RECOVERING" + WAITING = "WAITING" + CLAIMING = "CLAIMING" + ACTIVE = "ACTIVE" + RECOVERY_REQUIRED = "RECOVERY_REQUIRED" + + +class PollingStore(Protocol): + def recovery_snapshot(self, profile_id: str) -> RecoverySnapshot: ... + + def start_or_resume_polling(self, profile_id: str) -> PollingSession: ... + + def request_stop(self, profile_id: str) -> PollingSession: ... + + +class ClaimGateway(Protocol): + def claim_next(self, profile_id: str) -> ClaimedTask | None: ... + + +class ExecutionConsumer(Protocol): + def accept_claim(self, claimed: ClaimedTask) -> None: ... + + +@dataclass(frozen=True) +class StartReadiness: + """由后续已取证执行能力注入；T-304 自己不探测网络或设备。""" + + ready: bool + reason: str + + +@dataclass(frozen=True) +class ClaimedTaskView: + """允许发往 UI 的最小投影，刻意不包含 authorization/claim token。""" + + task_id: str + title: str + status: str = "已领取" + + @classmethod + def from_claim(cls, claimed: ClaimedTask) -> "ClaimedTaskView": + return cls(task_id=claimed.task.id, title=redact_text(claimed.task.title)) + + +@dataclass(frozen=True) +class RecoveryStatus: + """可进入 UI 的恢复摘要；不携带 task snapshot、claim token 或密文。""" + + has_open_session: bool + session_accept_new: bool + has_pending_claim: bool + has_active_claim: bool + has_pending_active_work: bool + + @classmethod + def from_snapshot(cls, snapshot: RecoverySnapshot) -> "RecoveryStatus": + return cls( + has_open_session=snapshot.session is not None, + session_accept_new=bool(snapshot.session and snapshot.session.accept_new), + has_pending_claim=snapshot.pending_claim is not None, + has_active_claim=snapshot.active_claim is not None, + has_pending_active_work=bool(snapshot.pending_renew or snapshot.pending_evidence), + ) + + +@dataclass(frozen=True) +class _BootstrapResult: + recovery: RecoveryStatus + session: PollingSession | None + normalized_stop: PollingSession | None = None + + +class _PollingWorker(QObject): + bootstrap_finished = Signal(object) + claim_finished = Signal(object) + stop_finished = Signal(object) + failed = Signal(str, object) + + def __init__(self, store: PollingStore) -> None: + super().__init__() + self._store = store + self._gateway: ClaimGateway | None = None + + @Slot(object) + def configure_gateway(self, gateway: object) -> None: + if not hasattr(gateway, "claim_next"): + self.failed.emit("configure", RuntimeError("invalid_claim_gateway")) + return + self._gateway = gateway # type: ignore[assignment] + + @Slot(str) + def bootstrap(self, profile_id: str) -> None: + try: + snapshot = self._store.recovery_snapshot(profile_id) + recovery = RecoveryStatus.from_snapshot(snapshot) + if snapshot.session is not None and snapshot.session.accept_new: + stopped = self._store.request_stop(profile_id) + self.bootstrap_finished.emit(_BootstrapResult(recovery, None, stopped)) + return + if snapshot.active_claim is not None: + self.bootstrap_finished.emit(_BootstrapResult(recovery, None)) + return + session = self._store.start_or_resume_polling(profile_id) + self.bootstrap_finished.emit(_BootstrapResult(recovery, session)) + except Exception as error: + self.failed.emit("bootstrap", error) + + @Slot(str) + def inspect_restart(self, profile_id: str) -> None: + try: + snapshot = self._store.recovery_snapshot(profile_id) + recovery = RecoveryStatus.from_snapshot(snapshot) + stopped = None + if snapshot.session is not None and snapshot.session.accept_new: + stopped = self._store.request_stop(profile_id) + self.bootstrap_finished.emit(_BootstrapResult(recovery, None, stopped)) + except Exception as error: + self.failed.emit("inspect", error) + + @Slot(str) + def claim(self, profile_id: str) -> None: + try: + # DurableClientGateway 在返回前已经提交 EMPTY 或 active claim；UI 不能 + # 以 generation 过期为由丢弃这个业务结果。 + if self._gateway is None: + raise RuntimeError("claim_gateway_not_configured") + self.claim_finished.emit(self._gateway.claim_next(profile_id)) + except Exception as error: + self.failed.emit("claim", error) + + @Slot(str) + def stop(self, profile_id: str) -> None: + try: + self.stop_finished.emit(self._store.request_stop(profile_id)) + except Exception as error: + self.failed.emit("stop", error) + + +class PollingCoordinator(QObject): + """把计时、阻塞 I/O 和可见状态收敛到一个会话边界。""" + + state_changed = Signal(object, str, int) + claim_visible = Signal(object) + recovery_status_changed = Signal(object) + configuration_freeze_changed = Signal(bool) + settled = Signal() + _configure_gateway_requested = Signal(object) + _bootstrap_requested = Signal(str) + _inspect_requested = Signal(str) + _claim_requested = Signal(str) + _stop_requested_signal = Signal(str) + + def __init__( + self, + *, + profile_id: str, + store: PollingStore | None, + gateway_factory: Callable[[ProfileSettings], ClaimGateway] | None, + consumer: ExecutionConsumer | None, + profile_settings: ProfileSettings | None = None, + readiness: StartReadiness | None = None, + poll_interval_seconds: int = 15, + failure_threshold: int = 3, + timer_interval_ms: int | None = None, + parent: QObject | None = None, + ) -> None: + super().__init__(parent) + if not 5 <= poll_interval_seconds <= 300: + raise ValueError("invalid_poll_interval") + if not 1 <= failure_threshold <= 10: + raise ValueError("invalid_failure_threshold") + self.profile_id = profile_id + self._store = store + self._gateway_factory = gateway_factory + self._gateway: ClaimGateway | None = None + self._consumer = consumer + self._profile_settings = profile_settings + self._frozen_profile: ProfileSettings | None = None + self._readiness = readiness + self.recovery_status: RecoveryStatus | None = None + self._timer_interval_override = timer_interval_ms + self._interval_ms = timer_interval_ms or ( + profile_settings.poll_interval_seconds * 1000 if profile_settings is not None else poll_interval_seconds * 1000 + ) + if self._interval_ms <= 0: + raise ValueError("invalid_timer_interval") + self._failure_threshold = ( + profile_settings.failure_threshold if profile_settings is not None else failure_threshold + ) + self._consecutive_failures = 0 + self._operation: str | None = None + self._stop_requested = False + self._epoch = 0 + self._scheduled_epoch: int | None = None + self._post_stop_state = PollingState.STOPPED + self._post_stop_reason = "轮询已停止。" + self._thread: QThread | None = None + self._worker: _PollingWorker | None = None + + self._timer = QTimer(self) + self._timer.setSingleShot(True) + self._timer.timeout.connect(self._on_timer_timeout) + + if consumer is None: + self.state = PollingState.BLOCKED + self.reason = "单趟执行能力尚未接入，不能领取真实任务。" + elif store is None or gateway_factory is None: + self.state = PollingState.BLOCKED + self.reason = "轮询依赖未完整注入，不能领取真实任务。" + elif profile_settings is None: + self.state = PollingState.BLOCKED + self.reason = "尚未保存完整配置，不能开始轮询。" + elif readiness is None or not readiness.ready: + self.state = PollingState.BLOCKED + self.reason = "执行就绪条件未满足，不能开始轮询。" if readiness is None else readiness.reason + else: + self.state = PollingState.STOPPED + self.reason = "轮询已停止。" + + if store is not None: + self._thread = QThread(self) + self._worker = _PollingWorker(store) + self._worker.moveToThread(self._thread) + self._configure_gateway_requested.connect(self._worker.configure_gateway) + self._inspect_requested.connect(self._worker.inspect_restart) + self._bootstrap_requested.connect(self._worker.bootstrap) + self._claim_requested.connect(self._worker.claim) + self._stop_requested_signal.connect(self._worker.stop) + self._worker.bootstrap_finished.connect(self._on_bootstrap_finished) + self._worker.claim_finished.connect(self._on_claim_finished) + self._worker.stop_finished.connect(self._on_stop_finished) + self._worker.failed.connect(self._on_worker_failed) + self._thread.start() + self._operation = "inspect" + self._set_state(PollingState.STARTING, "正在读取重启恢复状态并关闭遗留自动领取许可…") + self._inspect_requested.emit(self.profile_id) + + @property + def can_start(self) -> bool: + return ( + self._consumer is not None + and self._store is not None + and self._gateway_factory is not None + and self._profile_settings is not None + and self._readiness is not None + and self._readiness.ready + and self.state == PollingState.STOPPED + and self._operation is None + ) + + @property + def consecutive_failures(self) -> int: + return self._consecutive_failures + + @property + def operation_in_flight(self) -> bool: + return self._operation is not None + + def update_profile_settings(self, settings: ProfileSettings) -> None: + if self._operation is None and self.state in (PollingState.STOPPED, PollingState.BLOCKED): + self._profile_settings = settings + self._refresh_idle_gate() + + def update_readiness(self, readiness: StartReadiness) -> None: + self._readiness = readiness + self._refresh_idle_gate() + + def _refresh_idle_gate(self) -> None: + if self._operation is not None or self.state == PollingState.RECOVERY_REQUIRED: + return + if self._consumer is None: + self._set_state(PollingState.BLOCKED, "单趟执行能力尚未接入，不能领取真实任务。") + elif self._gateway_factory is None or self._store is None: + self._set_state(PollingState.BLOCKED, "轮询依赖未完整注入，不能领取真实任务。") + elif self._profile_settings is None: + self._set_state(PollingState.BLOCKED, "尚未保存完整配置，不能开始轮询。") + elif self._readiness is None or not self._readiness.ready: + reason = "执行就绪条件未满足，不能开始轮询。" if self._readiness is None else self._readiness.reason + self._set_state(PollingState.BLOCKED, reason) + else: + self._set_state(PollingState.STOPPED, "轮询已停止。") + + def start(self) -> None: + # 这道门禁必须早于任何 store/gateway 调用；独立应用没有 consumer， + # 即使调用方绕过禁用按钮直接调用本方法也保持零 HTTP。 + if self._consumer is None: + self._set_state(PollingState.BLOCKED, "单趟执行能力尚未接入，不能领取真实任务。") + return + if self._store is None or self._gateway_factory is None or self._worker is None: + self._set_state(PollingState.BLOCKED, "轮询依赖未完整注入，不能领取真实任务。") + return + if self._profile_settings is None: + self._set_state(PollingState.BLOCKED, "尚未保存完整配置，不能开始轮询。") + return + if self._readiness is None or not self._readiness.ready: + reason = "执行就绪条件未满足，不能开始轮询。" if self._readiness is None else self._readiness.reason + self._set_state(PollingState.BLOCKED, reason) + return + if not self.can_start: + return + frozen = self._profile_settings + try: + gateway = self._gateway_factory(frozen) + except Exception: + self._set_state(PollingState.BLOCKED, "领取网关无法按本次冻结配置建立，不能开始轮询。") + return + if gateway is None or not hasattr(gateway, "claim_next"): + self._set_state(PollingState.BLOCKED, "领取网关未完整注入，不能开始轮询。") + return + self._frozen_profile = frozen + self._interval_ms = self._timer_interval_override or frozen.poll_interval_seconds * 1000 + self._failure_threshold = frozen.failure_threshold + self._gateway = gateway + self._configure_gateway_requested.emit(gateway) + self._epoch += 1 + self._scheduled_epoch = None + self._timer.stop() + self._stop_requested = False + self._post_stop_state = PollingState.STOPPED + self._consecutive_failures = 0 + self._operation = "bootstrap" + self._set_state(PollingState.STARTING, "正在读取本地恢复状态并建立轮询会话…") + self._bootstrap_requested.emit(self.profile_id) + + def stop(self) -> None: + if self.state == PollingState.STOPPED and self._operation is None: + return + if self._store is None or self._worker is None: + return + # epoch/latch 双保险：stopEvent 先让 timer 队列中已经排队的 timeout + # 失效，再处理持久 stop；之后只有显式 Start 才会获得新 epoch。 + self._epoch += 1 + self._scheduled_epoch = None + self._stop_requested = True + self._timer.stop() + if self._operation in ("inspect", "bootstrap", "claim", "stop"): + if self._operation == "claim": + self._set_state(PollingState.CLAIMING, "正在等待本次有界领取返回；不会取消或重发请求。") + return + self._request_stop(PollingState.STOPPED, "轮询已停止；只阻止下一次领取。") + + def shutdown(self, wait_ms: int = 5000) -> bool: + """只结束空闲 worker；飞行中 I/O 必须由事件循环等待 settled。""" + + self._timer.stop() + if self._operation is not None: + return False + if self._thread is not None and self._thread.isRunning(): + self._thread.quit() + return self._thread.wait(wait_ms) + return True + + @Slot(object) + def _on_bootstrap_finished(self, raw: object) -> None: + completed_operation = self._operation + self._operation = None + result = raw + if not isinstance(result, _BootstrapResult): + self._block("本地恢复结果无效，轮询已阻止。") + return + self.recovery_status = result.recovery + self.recovery_status_changed.emit(result.recovery) + self.configuration_freeze_changed.emit( + result.recovery.has_pending_claim or result.recovery.has_active_claim + ) + if completed_operation == "inspect": + if result.recovery.has_active_claim: + self._set_state(PollingState.RECOVERY_REQUIRED, "遗留会话已停止；必须先安全恢复当前任务。") + elif result.recovery.has_pending_claim: + self._set_state(PollingState.STOPPED, "遗留领取请求已停止；显式开始后只使用原幂等键恢复。") + else: + self._refresh_idle_gate() + self.settled.emit() + return + if result.normalized_stop is not None: + if result.recovery.has_active_claim: + self._set_state(PollingState.RECOVERY_REQUIRED, "遗留会话已停止；必须先安全恢复当前任务。") + else: + self._set_state(PollingState.STOPPED, "遗留会话已停止；请再次显式开始轮询。") + self.settled.emit() + return + if result.recovery.has_active_claim: + self._set_state( + PollingState.RECOVERY_REQUIRED, + "检测到未关闭的采购任务，必须先完成安全恢复，不能领取新任务。", + ) + self.settled.emit() + return + if self._stop_requested: + self._request_stop(PollingState.STOPPED, "轮询已停止；未发起领取请求。") + return + if result.recovery.has_pending_claim: + self._set_state(PollingState.RECOVERING, "正在使用原幂等键恢复结果不明的领取请求…") + else: + self._set_state(PollingState.WAITING, "轮询会话已启动，正在等待领取。") + self._schedule_claim(0) + + def _schedule_claim(self, delay_ms: int) -> None: + self._scheduled_epoch = self._epoch + self._timer.start(delay_ms) + + @Slot() + def _on_timer_timeout(self) -> None: + scheduled_epoch = self._scheduled_epoch + self._scheduled_epoch = None + if scheduled_epoch != self._epoch: + return + self._begin_claim(scheduled_epoch) + + def _begin_claim(self, dispatch_epoch: int) -> None: + if dispatch_epoch != self._epoch: + return + if self._operation is not None or self._stop_requested: + return + if self.state not in (PollingState.WAITING, PollingState.RECOVERING): + return + self._operation = "claim" + self.configuration_freeze_changed.emit(True) + self._set_state(PollingState.CLAIMING, "正在领取已授权任务…") + self._claim_requested.emit(self.profile_id) + + @Slot(object) + def _on_claim_finished(self, claimed: object) -> None: + self._operation = None + self._consecutive_failures = 0 + if claimed is not None and not isinstance(claimed, ClaimedTask): + self._request_stop(PollingState.BLOCKED, "领取结果类型无效，轮询已阻止。") + return + if isinstance(claimed, ClaimedTask): + self.claim_visible.emit(ClaimedTaskView.from_claim(claimed)) + if self._stop_requested: + self._request_stop( + PollingState.RECOVERY_REQUIRED, + "停止期间领取已落库；必须先安全恢复该任务，不能领取下一条。", + ) + return + try: + consumer = self._consumer + if consumer is None: + raise RuntimeError("execution_consumer_missing") + consumer.accept_claim(claimed) + except Exception: + self._request_stop( + PollingState.RECOVERY_REQUIRED, + "执行 consumer 未接收已落库任务；必须安全恢复，不能重新领取。", + ) + return + self._set_state(PollingState.ACTIVE, "任务已安全领取并交给单趟执行能力。") + return + self.configuration_freeze_changed.emit(False) + if self._stop_requested: + self._request_stop(PollingState.STOPPED, "轮询已停止；本次没有可领取任务。") + return + self._set_state(PollingState.WAITING, "暂无已授权任务，等待下一次轮询。") + self._schedule_claim(self._interval_ms) + + def _request_stop(self, target: PollingState, reason: str) -> None: + if self._operation == "stop": + return + self._timer.stop() + self._post_stop_state = target + self._post_stop_reason = reason + self._operation = "stop" + self._stop_requested_signal.emit(self.profile_id) + + @Slot(object) + def _on_stop_finished(self, session: object) -> None: + self._operation = None + if not isinstance(session, PollingSession) or session.accept_new: + self._block("停止状态未能持久化，轮询已阻止。") + return + self._set_state(self._post_stop_state, self._post_stop_reason) + self.settled.emit() + + @Slot(str, object) + def _on_worker_failed(self, operation: str, error: object) -> None: + self._operation = None + if operation == "inspect": + if isinstance(error, StateError) and error.reason == "profile_not_found": + self.recovery_status = RecoveryStatus(False, False, False, False, False) + self.recovery_status_changed.emit(self.recovery_status) + self._refresh_idle_gate() + else: + self._block("本地恢复状态无法安全读取；已停止且不能领取任务。") + self.settled.emit() + return + if ( + operation == "claim" + and isinstance(error, AmbiguousRemoteError) + and error.reason in SAFE_AMBIGUOUS_REASONS + ): + self._consecutive_failures += 1 + if self._stop_requested: + self._request_stop(PollingState.STOPPED, "轮询已停止；结果不明的原领取请求已保留。") + elif self._consecutive_failures >= self._failure_threshold: + self._request_stop( + PollingState.BLOCKED, + "连续领取失败达到阈值；原幂等请求已保留，需排查后重新开始。", + ) + else: + self._set_state( + PollingState.RECOVERING, + "领取结果不明；等待使用相同幂等键恢复，不会创建新请求。", + ) + self._schedule_claim(self._interval_ms) + return + + if operation == "claim" and isinstance(error, AmbiguousRemoteError): + self.configuration_freeze_changed.emit(True) + self._request_stop( + PollingState.BLOCKED, + "领取响应无法证明可安全定时恢复；原槽已保留，需显式开始后同键恢复。", + ) + return + + if operation == "stop": + self._block("停止状态无法安全落库，轮询已阻止；未清除任何恢复事实。") + return + + if isinstance(error, CredentialRemoteError): + reason = "设备凭据无效或已撤销；修复凭据后再手工开始。" + elif isinstance(error, ProtocolRemoteError): + reason = "服务响应与固定协议不兼容；已停止普通重试。" + elif isinstance(error, ManualRemoteError): + reason = "服务端要求人工处理；已停止普通重试。" + elif isinstance(error, ClientError): + reason = "本地安全状态无法推进；已停止普通重试。" + else: + reason = "轮询发生未分类错误；已失败闭合。" + + if operation == "claim": + # DurableClientGateway 已把协议错误和 409 人工冲突标成 terminal， + # 二者没有 pending/active；凭据或本地错误则可能保留 pending，继续冻结。 + self.configuration_freeze_changed.emit( + not isinstance(error, (ProtocolRemoteError, ManualRemoteError)) + ) + + if operation in ("bootstrap", "configure"): + self._block(reason) + else: + self._request_stop(PollingState.BLOCKED, reason) + + def _block(self, reason: str) -> None: + self._timer.stop() + self._scheduled_epoch = None + self._set_state(PollingState.BLOCKED, reason) + if self._operation is None: + self.settled.emit() + + def _set_state(self, state: PollingState, reason: str) -> None: + self.state = state + self.reason = reason + self.state_changed.emit(state, reason, self._consecutive_failures) diff --git a/client/src/cmbuyer_client/ui/__init__.py b/client/src/cmbuyer_client/ui/__init__.py new file mode 100644 index 0000000..e05b9c9 --- /dev/null +++ b/client/src/cmbuyer_client/ui/__init__.py @@ -0,0 +1,5 @@ +"""采购工具原生 Qt Widgets 界面。""" + +from .main_window import PurchaseToolWindow + +__all__ = ["PurchaseToolWindow"] diff --git a/client/src/cmbuyer_client/ui/execution.py b/client/src/cmbuyer_client/ui/execution.py new file mode 100644 index 0000000..c7e15f7 --- /dev/null +++ b/client/src/cmbuyer_client/ui/execution.py @@ -0,0 +1,419 @@ +"""采购执行 Tab：状态、当前任务、滚动日志和历史记录主从视图。""" + +from __future__ import annotations + +from PySide6.QtCore import QModelIndex, QSize, Qt, Signal, Slot +from PySide6.QtGui import QAction, QKeySequence, QShortcut +from PySide6.QtWidgets import ( + QAbstractItemView, + QFrame, + QGroupBox, + QHBoxLayout, + QLabel, + QMenu, + QPlainTextEdit, + QPushButton, + QSizePolicy, + QSplitter, + QStackedWidget, + QTableView, + QToolButton, + QVBoxLayout, + QWidget, +) + +from cmbuyer_client.polling.coordinator import ClaimedTaskView, PollingCoordinator, PollingState +from cmbuyer_client.logging_policy import redact_text + +from .records import PurchaseRecord, PurchaseRecordModel, PurchaseRecordProvider + + +class _LogView(QPlainTextEdit): + def __init__(self, parent: QWidget | None = None) -> None: + super().__init__(parent) + self.setObjectName("rollingLog") + self.setReadOnly(True) + self.setPlaceholderText("轮询启动后将在这里显示脱敏日志。") + self.setAccessibleName("滚动日志") + + def append_event(self, text: str) -> None: + bar = self.verticalScrollBar() + follow = bar.value() >= bar.maximum() - 2 + self.appendPlainText(redact_text(text)) + if follow: + bar.setValue(bar.maximum()) + + +class _RecordTableView(QTableView): + """把双击与 Enter 收敛为唯一 activation 信号，避免平台重复发命令。""" + + record_activated = Signal(object) + + def mouseDoubleClickEvent(self, event) -> None: + index = self.indexAt(event.position().toPoint()) + if index.isValid(): + self.setCurrentIndex(index.siblingAtColumn(0)) + self.record_activated.emit(index) + event.accept() + return + super().mouseDoubleClickEvent(event) + + def keyPressEvent(self, event) -> None: + if event.key() in (Qt.Key.Key_Return, Qt.Key.Key_Enter) and self.currentIndex().isValid(): + self.record_activated.emit(self.currentIndex()) + event.accept() + return + super().keyPressEvent(event) + + +class ExecutionPage(QWidget): + LIVE_PAGE = 0 + DETAIL_PAGE = 1 + COMPACT_DETAIL_WIDTH = 760 + + def __init__( + self, + coordinator: PollingCoordinator, + record_provider: PurchaseRecordProvider | None = None, + parent: QWidget | None = None, + ) -> None: + super().__init__(parent) + self.setObjectName("executionPage") + self._coordinator = coordinator + self._selected_record_id: str | None = None + self._saved_scroll = 0 + + outer = QVBoxLayout(self) + outer.setContentsMargins(12, 12, 12, 12) + + status_row = QHBoxLayout() + self.service_status = self._status("采购服务", "待首次真实领取验证") + self.device_status = self._status("ADB", "待后续执行能力验证") + self.app_status = self._status("拼多多版本", "待后续执行能力验证") + self.session_status = self._status("会话", "已停止") + for widget in (self.service_status, self.device_status, self.app_status, self.session_status): + status_row.addWidget(widget) + status_row.addStretch(1) + + self.poll_action = QAction("开始轮询", self) + self.poll_action.setObjectName("pollAction") + self.poll_action.triggered.connect(self._toggle_polling) + self.addAction(self.poll_action) + self.poll_button = QPushButton() + self.poll_button.setObjectName("pollButton") + self.poll_button.clicked.connect(self.poll_action.trigger) + status_row.addWidget(self.poll_button) + outer.addLayout(status_row) + + self.banner = QLabel() + self.banner.setObjectName("sessionBanner") + self.banner.setWordWrap(True) + self.banner.setAccessibleName("轮询会话状态") + self.banner.setFrameShape(QFrame.Shape.StyledPanel) + outer.addWidget(self.banner) + + self.body_splitter = QSplitter(Qt.Orientation.Horizontal) + self.body_splitter.setObjectName("executionSplitter") + self.body_splitter.setChildrenCollapsible(False) + self.left_stack = QStackedWidget() + self.left_stack.setObjectName("leftWorkspace") + self.left_stack.addWidget(self._build_live_page()) + self.left_stack.addWidget(self._build_detail_page()) + self.body_splitter.addWidget(self.left_stack) + self.body_splitter.addWidget(self._build_records_page()) + self.body_splitter.setStretchFactor(0, 2) + self.body_splitter.setStretchFactor(1, 1) + self.body_splitter.setSizes([760, 380]) + outer.addWidget(self.body_splitter, 1) + + self.view_record_action = QAction("查看所选记录", self) + self.view_record_action.setObjectName("viewSelectedRecord") + self.view_record_action.setEnabled(False) + self.view_record_action.triggered.connect(self.open_selected_record) + self.addAction(self.view_record_action) + self.view_record_button.setDefaultAction(self.view_record_action) + + self.return_action = QAction("返回当前任务", self) + self.return_action.setObjectName("returnToCurrentTask") + self.return_action.setEnabled(False) + self.return_action.triggered.connect(self.return_to_live) + self.addAction(self.return_action) + self.return_button.setDefaultAction(self.return_action) + self.escape_shortcut = QShortcut(QKeySequence(Qt.Key.Key_Escape), self) + self.escape_shortcut.setContext(Qt.ShortcutContext.WidgetWithChildrenShortcut) + self.escape_shortcut.activated.connect(self._escape) + + self.record_view.clicked.connect(self._on_record_selected) + self.record_view.record_activated.connect(self._open_index) + self.record_view.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) + self.record_view.customContextMenuRequested.connect(self._show_record_menu) + self.record_view.selectionModel().currentChanged.connect(self._on_current_changed) + coordinator.state_changed.connect(self._on_polling_state) + coordinator.claim_visible.connect(self._show_claimed_task) + self._on_polling_state(coordinator.state, coordinator.reason, coordinator.consecutive_failures) + if record_provider is not None: + # T-304 没有公共历史仓库；只接受调用方准备好的 View DTO 快照， + # 独立应用不注入 provider，模型保持真实空态。 + self.set_records(record_provider.snapshot()) + + @staticmethod + def _status(name: str, value: str) -> QLabel: + label = QLabel(f"{name}\n{value}") + label.setFrameShape(QFrame.Shape.StyledPanel) + label.setMinimumWidth(118) + label.setAccessibleName(name) + return label + + def _build_live_page(self) -> QWidget: + page = QWidget() + layout = QVBoxLayout(page) + task_group = QGroupBox("当前任务") + task_layout = QHBoxLayout(task_group) + self.current_task_text = QLabel("当前没有任务。\n启动后只领取已授权任务。") + self.current_task_text.setObjectName("currentTaskText") + self.current_task_text.setWordWrap(True) + self.current_task_text.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignLeft) + self.current_task_text.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + self.current_image = QLabel("暂无可信商品图片") + self.current_image.setObjectName("currentTaskImage") + self.current_image.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.current_image.setFrameShape(QFrame.Shape.StyledPanel) + self.current_image.setMinimumSize(QSize(180, 120)) + self.current_image.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + task_layout.addWidget(self.current_task_text, 2) + task_layout.addWidget(self.current_image, 1) + layout.addWidget(task_group, 1) + + log_group = QGroupBox("滚动日志") + log_layout = QVBoxLayout(log_group) + self.log_view = _LogView() + log_layout.addWidget(self.log_view) + layout.addWidget(log_group, 2) + return page + + def _build_detail_page(self) -> QWidget: + page = QWidget() + page.setObjectName("recordDetailPage") + layout = QVBoxLayout(page) + header = QHBoxLayout() + self.detail_title = QLabel("采购记录详情") + self.detail_title.setObjectName("recordDetailTitle") + self.detail_title.setStyleSheet("font-size: 18px; font-weight: 600;") + header.addWidget(self.detail_title) + header.addStretch(1) + self.return_button = QToolButton() + self.return_button.setObjectName("returnCurrentTaskButton") + header.addWidget(self.return_button) + layout.addLayout(header) + + top = QSplitter(Qt.Orientation.Horizontal) + self.detail_original = QPlainTextEdit() + self.detail_original.setObjectName("recordOriginalText") + self.detail_original.setReadOnly(True) + self.detail_original.setPlaceholderText("没有可显示的原始文字。") + self.detail_image = QLabel("没有可显示的可信图片") + self.detail_image.setObjectName("recordImage") + self.detail_image.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.detail_image.setFrameShape(QFrame.Shape.StyledPanel) + top.addWidget(self.detail_original) + top.addWidget(self.detail_image) + top.setStretchFactor(0, 2) + top.setStretchFactor(1, 1) + layout.addWidget(top, 2) + + result_group = QGroupBox("采购结果") + result_layout = QVBoxLayout(result_group) + self.detail_result = QPlainTextEdit() + self.detail_result.setObjectName("recordResult") + self.detail_result.setReadOnly(True) + self.detail_result.setPlaceholderText("暂无采购结果。") + result_layout.addWidget(self.detail_result) + layout.addWidget(result_group, 1) + return page + + def _build_records_page(self) -> QWidget: + page = QGroupBox("采购记录") + page.setObjectName("recordsPanel") + self.records_panel = page + layout = QVBoxLayout(page) + header = QHBoxLayout() + self.records_summary = QLabel("暂无记录") + header.addWidget(self.records_summary) + header.addStretch(1) + self.view_record_button = QToolButton() + self.view_record_button.setObjectName("viewSelectedRecordButton") + header.addWidget(self.view_record_button) + layout.addLayout(header) + self.record_model = PurchaseRecordModel(parent=self) + self.record_view = _RecordTableView() + self.record_view.setObjectName("purchaseRecordTable") + self.record_view.setModel(self.record_model) + self.record_view.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) + self.record_view.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) + self.record_view.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) + self.record_view.setAlternatingRowColors(True) + self.record_view.setSortingEnabled(False) + self.record_view.horizontalHeader().setStretchLastSection(False) + self.record_view.horizontalHeader().setSectionResizeMode(0, self.record_view.horizontalHeader().ResizeMode.Stretch) + self.record_view.horizontalHeader().setSectionResizeMode(1, self.record_view.horizontalHeader().ResizeMode.ResizeToContents) + self.record_view.verticalHeader().setVisible(False) + layout.addWidget(self.record_view) + return page + + def set_records(self, records: list[PurchaseRecord]) -> None: + selected_id = self._selected_record_id + self.record_model.set_records(records) + self.records_summary.setText(f"共 {len(records)} 条" if records else "暂无记录") + if selected_id is not None: + row = self.record_model.row_for_id(selected_id) + if row >= 0: + self.record_view.setCurrentIndex(self.record_model.index(row, 0)) + if self.left_stack.currentIndex() == self.DETAIL_PAGE: + self._render_record(self.record_model.record_at(row)) + return + self._selected_record_id = None + self.view_record_action.setEnabled(False) + if self.left_stack.currentIndex() == self.DETAIL_PAGE: + self.return_to_live() + + @Slot(object, str, int) + def _on_polling_state(self, state: object, reason: str, failures: int) -> None: + polling_state = state if isinstance(state, PollingState) else PollingState.BLOCKED + self.session_status.setText(f"会话\n{self._state_text(polling_state)}") + suffix = f"（连续失败 {failures} 次）" if failures else "" + self.banner.setText(reason + suffix) + running = polling_state in ( + PollingState.STARTING, + PollingState.RECOVERING, + PollingState.WAITING, + PollingState.CLAIMING, + PollingState.ACTIVE, + ) + self.poll_action.setText("停止轮询" if running else "开始轮询") + self.poll_action.setEnabled(running or self._coordinator.can_start) + self.poll_button.setText(self.poll_action.text()) + self.poll_button.setEnabled(self.poll_action.isEnabled()) + self.poll_button.setToolTip("" if self.poll_action.isEnabled() else reason) + + @staticmethod + def _state_text(state: PollingState) -> str: + return { + PollingState.STOPPED: "已停止", + PollingState.STARTING: "启动中", + PollingState.BLOCKED: "已阻止", + PollingState.RECOVERING: "安全恢复", + PollingState.WAITING: "等待领取", + PollingState.CLAIMING: "正在领取", + PollingState.ACTIVE: "任务执行中", + PollingState.RECOVERY_REQUIRED: "待安全恢复", + }[state] + + @Slot() + def _toggle_polling(self) -> None: + if self._coordinator.state in ( + PollingState.STARTING, + PollingState.RECOVERING, + PollingState.WAITING, + PollingState.CLAIMING, + PollingState.ACTIVE, + ): + self._coordinator.stop() + else: + self._coordinator.start() + + @Slot(QModelIndex) + def _on_record_selected(self, index: QModelIndex) -> None: + record = self.record_model.record_at(index.row()) + if record is None: + return + self._selected_record_id = record.record_id + self.view_record_action.setEnabled(True) + if self.left_stack.currentIndex() == self.DETAIL_PAGE: + self._saved_scroll = self.record_view.verticalScrollBar().value() + self._render_record(record) + + @Slot(object) + def _show_claimed_task(self, raw: object) -> None: + if not isinstance(raw, ClaimedTaskView): + return + self.current_task_text.setText( + f"标题：{raw.title}\n任务 ID：{raw.task_id}\n状态：{raw.status}" + ) + self.log_view.append_event(f"已安全领取任务 {raw.task_id}；等待单趟执行能力处理。") + + @Slot(QModelIndex, QModelIndex) + def _on_current_changed(self, current: QModelIndex, previous: QModelIndex) -> None: + del previous + if current.isValid(): + self._on_record_selected(current) + + @Slot(QModelIndex) + def _open_index(self, index: QModelIndex) -> None: + if index.isValid(): + self.record_view.setCurrentIndex(index.siblingAtColumn(0)) + self._on_record_selected(index) + self.view_record_action.trigger() + + @Slot() + def open_selected_record(self) -> None: + if self._selected_record_id is None: + return + row = self.record_model.row_for_id(self._selected_record_id) + record = self.record_model.record_at(row) + if record is None: + return + self._saved_scroll = self.record_view.verticalScrollBar().value() + self._render_record(record) + self.left_stack.setCurrentIndex(self.DETAIL_PAGE) + self.return_action.setEnabled(True) + self._apply_compact_detail() + self.return_button.setFocus() + + def _render_record(self, record: PurchaseRecord | None) -> None: + if record is None: + self.detail_title.setText("记录不存在") + self.detail_original.clear() + self.detail_result.clear() + self.detail_image.setText("没有可显示的可信图片") + return + self.detail_title.setText(record.title) + self.detail_original.setPlainText(record.original_text) + self.detail_result.setPlainText(record.result_text) + self.detail_image.setText(record.image_description or "没有可显示的可信图片") + + @Slot() + def return_to_live(self) -> None: + if self.left_stack.currentIndex() != self.DETAIL_PAGE: + return + self.left_stack.setCurrentIndex(self.LIVE_PAGE) + self.return_action.setEnabled(False) + self.records_panel.setVisible(True) + row = self.record_model.row_for_id(self._selected_record_id or "") + if row >= 0: + self.record_view.setCurrentIndex(self.record_model.index(row, 0)) + self.record_view.verticalScrollBar().setValue(self._saved_scroll) + self.record_view.setFocus() + + @Slot() + def _escape(self) -> None: + # Qt popup/menu 优先消费 Esc；只有详情态的页面级 shortcut 会执行返回。 + if self.left_stack.currentIndex() == self.DETAIL_PAGE: + self.return_action.trigger() + + @Slot(object) + def _show_record_menu(self, point: object) -> None: + index = self.record_view.indexAt(point) + if index.isValid(): + self.record_view.setCurrentIndex(index.siblingAtColumn(0)) + self._on_record_selected(index) + menu = QMenu(self.record_view) + menu.addAction(self.view_record_action) + menu.exec(self.record_view.viewport().mapToGlobal(point)) + + def resizeEvent(self, event) -> None: + super().resizeEvent(event) + self._apply_compact_detail() + + def _apply_compact_detail(self) -> None: + compact_detail = self.width() < self.COMPACT_DETAIL_WIDTH and self.left_stack.currentIndex() == self.DETAIL_PAGE + self.records_panel.setVisible(not compact_detail) diff --git a/client/src/cmbuyer_client/ui/main_window.py b/client/src/cmbuyer_client/ui/main_window.py new file mode 100644 index 0000000..3bb94ed --- /dev/null +++ b/client/src/cmbuyer_client/ui/main_window.py @@ -0,0 +1,95 @@ +"""采购工具固定双 Tab 原生窗口。""" + +from __future__ import annotations + +from PySide6.QtCore import QTimer, Qt, Slot +from PySide6.QtGui import QCloseEvent +from PySide6.QtWidgets import QMainWindow, QTabWidget + +from cmbuyer_client.localstate.models import ProfileSettings +from cmbuyer_client.polling.coordinator import PollingCoordinator, PollingState, RecoveryStatus + +from .execution import ExecutionPage +from .records import PurchaseRecordProvider +from .settings import ProfileStore, SettingsPage + + +class PurchaseToolWindow(QMainWindow): + EXECUTION_PAGE_ID = "purchase-execution" + SETTINGS_PAGE_ID = "settings" + + def __init__( + self, + *, + store: ProfileStore, + coordinator: PollingCoordinator, + profile_settings: ProfileSettings | None, + has_stored_device_token: bool, + record_provider: PurchaseRecordProvider | None = None, + parent=None, + ) -> None: + super().__init__(parent) + self.setWindowTitle("采购工具") + self.setAccessibleName("采购工具") + self.setMinimumSize(720, 520) + self.resize(1180, 760) + self._coordinator = coordinator + self._close_pending = False + + recovery = coordinator.recovery_status + frozen = recovery is None or recovery.has_pending_claim or recovery.has_active_claim + self.tabs = QTabWidget() + self.tabs.setObjectName("mainTabs") + self.tabs.setTabsClosable(False) + self.tabs.setMovable(False) + self.execution_page = ExecutionPage(coordinator, record_provider) + self.execution_page.setProperty("pageId", self.EXECUTION_PAGE_ID) + self.settings_page = SettingsPage( + store, + settings=profile_settings, + has_stored_device_token=has_stored_device_token, + identity_frozen=frozen, + ) + self.settings_page.setProperty("pageId", self.SETTINGS_PAGE_ID) + self.tabs.addTab(self.execution_page, "采购执行") + self.tabs.addTab(self.settings_page, "配置") + self.tabs.setCurrentWidget(self.execution_page) + self.setCentralWidget(self.tabs) + + self.settings_page.settings_saved.connect( + lambda settings, has_token: self._coordinator.update_profile_settings(settings) + ) + coordinator.recovery_status_changed.connect(self._on_recovery_status) + coordinator.configuration_freeze_changed.connect(self.settings_page.set_identity_frozen) + coordinator.settled.connect(self._finish_pending_close) + + @Slot(object) + def _on_recovery_status(self, raw: object) -> None: + if not isinstance(raw, RecoveryStatus): + self.settings_page.set_identity_frozen(True) + return + self.settings_page.set_identity_frozen(raw.has_pending_claim or raw.has_active_claim) + + def closeEvent(self, event: QCloseEvent) -> None: + running = self._coordinator.state in ( + PollingState.STARTING, + PollingState.RECOVERING, + PollingState.WAITING, + PollingState.CLAIMING, + PollingState.ACTIVE, + ) or self._coordinator.operation_in_flight + if running: + # 不 terminate 飞行中的 QThread。先提升 stop latch，等待 HTTP 自身 + # 超时和 DurableClientGateway 落库，再由 settled 重试关闭。 + self._close_pending = True + self._coordinator.stop() + event.ignore() + return + event.accept() + + @Slot() + def _finish_pending_close(self) -> None: + if not self._close_pending: + return + self._close_pending = False + QTimer.singleShot(0, self.close) diff --git a/client/src/cmbuyer_client/ui/records.py b/client/src/cmbuyer_client/ui/records.py new file mode 100644 index 0000000..e7fa036 --- /dev/null +++ b/client/src/cmbuyer_client/ui/records.py @@ -0,0 +1,95 @@ +"""采购记录的只读 Qt Model/View 数据源。""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Protocol + +from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt + +from cmbuyer_client.logging_policy import redact_text +from cmbuyer_client.core.errors import ValidationError +from cmbuyer_client.core.validation import rfc3339_z_nanoseconds + + +@dataclass(frozen=True) +class PurchaseRecord: + record_id: str + title: str + status: str + created_at: str + original_text: str = "" + result_text: str = "" + image_description: str = "" + created_at_nanoseconds: int = field(init=False, repr=False) + + def __post_init__(self) -> None: + # 记录 provider 只能注入可显示摘要；最终 UI 边界仍统一脱敏，避免 + # consumer bug 把 Bearer/裸 token 放入 model、详情或可见日志。 + for field in ("title", "status", "original_text", "result_text", "image_description"): + object.__setattr__(self, field, redact_text(getattr(self, field))) + if not isinstance(self.created_at, str): + raise ValueError("noncanonical_record_timestamp") + fraction = self.created_at[20:-1] if len(self.created_at) > 20 and self.created_at.endswith("Z") else "" + if fraction and fraction.endswith("0"): + raise ValueError("noncanonical_record_timestamp") + try: + timestamp = rfc3339_z_nanoseconds(self.created_at) + except ValidationError: + raise ValueError("noncanonical_record_timestamp") from None + object.__setattr__(self, "created_at_nanoseconds", timestamp) + + +class PurchaseRecordProvider(Protocol): + """只返回已准备好的无秘密 View DTO；不得在 GUI 线程查询 SQLite/HTTP。""" + + def snapshot(self) -> list[PurchaseRecord]: ... + + +class PurchaseRecordModel(QAbstractTableModel): + RECORD_ID_ROLE = int(Qt.ItemDataRole.UserRole) + 1 + + def __init__(self, records: list[PurchaseRecord] | None = None, parent=None) -> None: + super().__init__(parent) + self._records: list[PurchaseRecord] = [] + self.set_records(records or []) + + def rowCount(self, parent: QModelIndex = QModelIndex()) -> int: + return 0 if parent.isValid() else len(self._records) + + def columnCount(self, parent: QModelIndex = QModelIndex()) -> int: + return 0 if parent.isValid() else 2 + + def data(self, index: QModelIndex, role: int = int(Qt.ItemDataRole.DisplayRole)): + if not index.isValid() or not 0 <= index.row() < len(self._records): + return None + record = self._records[index.row()] + if role == int(Qt.ItemDataRole.DisplayRole): + return record.title if index.column() == 0 else record.status + if role == self.RECORD_ID_ROLE: + return record.record_id + if role == int(Qt.ItemDataRole.ToolTipRole): + return f"{record.title}\n{record.created_at}" + if role == int(Qt.ItemDataRole.TextAlignmentRole) and index.column() == 1: + return int(Qt.AlignmentFlag.AlignCenter) + return None + + def headerData(self, section: int, orientation: Qt.Orientation, role: int = int(Qt.ItemDataRole.DisplayRole)): + if role != int(Qt.ItemDataRole.DisplayRole) or orientation != Qt.Orientation.Horizontal: + return None + return ("标题", "状态")[section] if 0 <= section < 2 else None + + def set_records(self, records: list[PurchaseRecord]) -> None: + self.beginResetModel() + self._records = sorted( + records, + key=lambda item: (item.created_at_nanoseconds, item.record_id), + reverse=True, + ) + self.endResetModel() + + def record_at(self, row: int) -> PurchaseRecord | None: + return self._records[row] if 0 <= row < len(self._records) else None + + def row_for_id(self, record_id: str) -> int: + return next((row for row, item in enumerate(self._records) if item.record_id == record_id), -1) diff --git a/client/src/cmbuyer_client/ui/settings.py b/client/src/cmbuyer_client/ui/settings.py new file mode 100644 index 0000000..dfcb121 --- /dev/null +++ b/client/src/cmbuyer_client/ui/settings.py @@ -0,0 +1,247 @@ +"""只做本地校验和显式保存的配置页。""" + +from __future__ import annotations + +from pathlib import Path +from typing import Protocol + +from PySide6.QtCore import QRegularExpression, Qt, Signal, Slot +from PySide6.QtGui import QAction, QKeySequence, QRegularExpressionValidator +from PySide6.QtWidgets import ( + QComboBox, + QFormLayout, + QLabel, + QLineEdit, + QPushButton, + QScrollArea, + QSpinBox, + QVBoxLayout, + QWidget, +) + +from cmbuyer_client.core.models import SecretToken +from cmbuyer_client.core.errors import ValidationError +from cmbuyer_client.core.validation import require_uuid4 +from cmbuyer_client.localstate.models import LOOPBACK_SERVICE_URL, ProfileSettings + + +class ProfileStore(Protocol): + def save_profile(self, settings: ProfileSettings, token: SecretToken | None) -> None: ... + + +class SettingsPage(QScrollArea): + settings_saved = Signal(object, bool) + VALIDATION_HINT = "服务身份将在首次真实领取时验证；设备与 App 状态由后续已取证执行能力验证。" + + def __init__( + self, + store: ProfileStore, + *, + profile_id: str = "default", + settings: ProfileSettings | None = None, + has_stored_device_token: bool = False, + identity_frozen: bool = False, + parent: QWidget | None = None, + ) -> None: + super().__init__(parent) + self.setObjectName("settingsPage") + self.setWidgetResizable(True) + self._store = store + self._profile_id = profile_id + self._has_stored_device_token = has_stored_device_token + self._loaded_settings = settings + self._identity_frozen = identity_frozen + + content = QWidget() + outer = QVBoxLayout(content) + title = QLabel("配置") + title.setObjectName("settingsTitle") + title.setStyleSheet("font-size: 20px; font-weight: 600;") + outer.addWidget(title) + + form = QFormLayout() + form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.ExpandingFieldsGrow) + form.setLabelAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) + outer.addLayout(form) + + self.service_url = QLineEdit(LOOPBACK_SERVICE_URL) + self.service_url.setReadOnly(True) + self.service_url.setObjectName("serviceUrl") + form.addRow("采购服务 URL", self.service_url) + + self.device_id = QLineEdit() + self.device_id.setObjectName("deviceId") + self.device_id.setPlaceholderText("小写 UUIDv4") + form.addRow("设备 UUID", self.device_id) + + self.device_token = QLineEdit() + self.device_token.setObjectName("deviceToken") + self.device_token.setEchoMode(QLineEdit.EchoMode.Password) + self.device_token.setMaxLength(64) + self.device_token.setPlaceholderText("首次必填；已有凭据时留空表示保留") + self.device_token.setValidator(QRegularExpressionValidator(QRegularExpression("[0-9a-f]{0,64}"), self)) + form.addRow("设备 token", self.device_token) + + self.token_status = QLabel() + self.token_status.setObjectName("tokenStatus") + form.addRow("凭据状态", self.token_status) + + self.adb_path = QLineEdit() + self.adb_path.setObjectName("adbPath") + form.addRow("ADB 路径", self.adb_path) + + self.adb_serial = QLineEdit() + self.adb_serial.setObjectName("adbSerial") + form.addRow("设备 serial", self.adb_serial) + + self.transport = QComboBox() + self.transport.setObjectName("transport") + self.transport.addItem("USB", "usb") + self.transport.addItem("WiFi", "wifi") + form.addRow("连接方式", self.transport) + + self.poll_interval = self._spin(5, 300, 15, "pollInterval", " 秒") + form.addRow("轮询间隔", self.poll_interval) + self.failure_threshold = self._spin(1, 10, 3, "failureThreshold", " 次") + form.addRow("连续失败停止阈值", self.failure_threshold) + self.http_timeout = self._spin(1, 120, 10, "httpTimeout", " 秒") + form.addRow("HTTP 超时", self.http_timeout) + self.step_timeout = self._spin(5, 300, 45, "stepTimeout", " 秒") + form.addRow("真机步骤超时", self.step_timeout) + + self.validation_hint = QLabel(self.VALIDATION_HINT) + self.validation_hint.setObjectName("deferredValidationHint") + self.validation_hint.setWordWrap(True) + outer.addWidget(self.validation_hint) + + self.feedback = QLabel() + self.feedback.setObjectName("settingsFeedback") + self.feedback.setWordWrap(True) + self.feedback.setAccessibleName("配置保存状态") + outer.addWidget(self.feedback) + + self.save_action = QAction("保存配置", self) + self.save_action.setShortcut(QKeySequence.StandardKey.Save) + self.save_action.triggered.connect(self.save) + self.addAction(self.save_action) + self.save_button = QPushButton("保存配置") + self.save_button.setObjectName("saveSettings") + self.save_button.clicked.connect(self.save_action.trigger) + outer.addWidget(self.save_button, 0, Qt.AlignmentFlag.AlignRight) + outer.addStretch(1) + self.setWidget(content) + + if settings is not None: + self._load(settings) + self._update_token_status() + self.set_identity_frozen(identity_frozen) + + @staticmethod + def _spin(minimum: int, maximum: int, value: int, name: str, suffix: str) -> QSpinBox: + field = QSpinBox() + field.setObjectName(name) + field.setRange(minimum, maximum) + field.setValue(value) + field.setSuffix(suffix) + return field + + def _load(self, settings: ProfileSettings) -> None: + self.device_id.setText(settings.device_id) + self.adb_path.setText(settings.adb_path) + self.adb_serial.setText(settings.adb_serial) + self.transport.setCurrentIndex(max(0, self.transport.findData(settings.transport))) + self.poll_interval.setValue(settings.poll_interval_seconds) + self.failure_threshold.setValue(settings.failure_threshold) + self.http_timeout.setValue(settings.http_timeout_seconds) + self.step_timeout.setValue(settings.step_timeout_seconds) + + def set_identity_frozen(self, frozen: bool) -> None: + self._identity_frozen = frozen + for field in ( + self.device_id, + self.adb_path, + self.adb_serial, + self.transport, + self.poll_interval, + self.failure_threshold, + self.http_timeout, + self.step_timeout, + ): + field.setEnabled(not frozen) + if frozen: + self.feedback.setText("存在待恢复或执行中的领取；服务与设备身份参数已冻结。") + elif self.feedback.text().startswith("存在待恢复或执行中的领取"): + self.feedback.clear() + + @Slot() + def save(self) -> None: + self.feedback.clear() + token_text = self.device_token.text().strip() + device_id = self.device_id.text().strip() + try: + require_uuid4(device_id, "invalid_device_id") + except (TypeError, ValueError, ValidationError): + self._validation_error(self.device_id, "设备 UUID 必须是小写 UUIDv4。") + return + if not self._has_stored_device_token and not token_text: + self._validation_error(self.device_token, "首次保存必须填写设备 token。") + return + if token_text and (len(token_text) != 64 or any(character not in "0123456789abcdef" for character in token_text)): + self._validation_error(self.device_token, "设备 token 必须为 64 位小写十六进制。") + return + try: + if self._identity_frozen: + if self._loaded_settings is None: + self._validation_error(self.device_id, "冻结配置缺少原始设置，不能保存。") + return + settings = self._loaded_settings + else: + adb_text = self.adb_path.text().strip() + if not adb_text: + self._validation_error(self.adb_path, "请填写 ADB 路径。") + return + adb_file = Path(adb_text).expanduser() + if not adb_file.is_file(): + self._validation_error(self.adb_path, "ADB 路径必须指向本机已存在的文件。") + return + adb_file = adb_file.resolve(strict=True) + serial = self.adb_serial.text().strip() + if not serial: + self._validation_error(self.adb_serial, "请填写设备 serial。") + return + settings = ProfileSettings( + profile_id=self._profile_id, + service_url=LOOPBACK_SERVICE_URL, + device_id=device_id, + adb_path=str(adb_file), + adb_serial=serial, + transport=str(self.transport.currentData()), + poll_interval_seconds=self.poll_interval.value(), + failure_threshold=self.failure_threshold.value(), + http_timeout_seconds=self.http_timeout.value(), + step_timeout_seconds=self.step_timeout.value(), + ) + token = SecretToken(token_text) if token_text else None + except (TypeError, ValueError): + self._validation_error(self.device_id, "配置格式不正确，请检查设备 UUID 与各项参数。") + return + try: + # None 明确表示保留 T-303 中已有的 DPAPI 密文，绝不是清除凭据。 + self._store.save_profile(settings, token) + except Exception: + self.feedback.setText("配置保存失败，本地安全存储未更新。") + (self.device_token if token_text else self.device_id).setFocus() + return + self._has_stored_device_token = True + self._loaded_settings = settings + self.device_token.clear() + self._update_token_status() + self.feedback.setText("配置已保存。本地校验不代表服务、设备或 App 已就绪。") + self.settings_saved.emit(settings, True) + + def _validation_error(self, field: QWidget, message: str) -> None: + self.feedback.setText(message) + field.setFocus() + + def _update_token_status(self) -> None: + self.token_status.setText("已保存" if self._has_stored_device_token else "未保存") diff --git a/client/tests/polling/__init__.py b/client/tests/polling/__init__.py new file mode 100644 index 0000000..2c52318 --- /dev/null +++ b/client/tests/polling/__init__.py @@ -0,0 +1 @@ +"""轮询协调器测试。""" diff --git a/client/tests/polling/test_coordinator.py b/client/tests/polling/test_coordinator.py new file mode 100644 index 0000000..74d0527 --- /dev/null +++ b/client/tests/polling/test_coordinator.py @@ -0,0 +1,349 @@ +from __future__ import annotations + +import os +import threading +import unittest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication + +from cmbuyer_client.core.errors import ( + AmbiguousRemoteError, + CredentialRemoteError, + ManualRemoteError, + ProtocolRemoteError, + StateError, +) +from cmbuyer_client.core.models import ClaimRequest, ClaimedTask +from cmbuyer_client.localstate.models import PollingSession, ProfileSettings, RecoverySnapshot +from cmbuyer_client.polling.coordinator import ( + ClaimedTaskView, + PollingCoordinator, + PollingState, + StartReadiness, +) +from tests.core.test_models import claim_wire + + +PROFILE = "default" +SESSION_ID = "13c9f507-7473-4fa6-8d71-8786c34c6301" +REQUEST_ID = "23c9f507-7473-4fa6-8d71-8786c34c6301" +SENTINEL_TOKEN = "e" * 64 + + +def profile(http_timeout: int = 10, failure_threshold: int = 3) -> ProfileSettings: + return ProfileSettings( + PROFILE, + "http://127.0.0.1:8080", + "f3c9f507-7473-4fa6-8d71-8786c34c6301", + "D:/Portable/adb/adb.exe", + "device-serial", + "usb", + failure_threshold=failure_threshold, + http_timeout_seconds=http_timeout, + ) + + +def claimed_task() -> ClaimedTask: + wire = claim_wire() + wire["attempt"]["claim_token"] = SENTINEL_TOKEN + return ClaimedTask.from_wire(wire) + + +def snapshot( + *, + accept_new: bool | None = None, + pending: bool = False, + active: bool = False, +) -> RecoverySnapshot: + session = None if accept_new is None else PollingSession(PROFILE, SESSION_ID, accept_new) + request = ClaimRequest(SESSION_ID, REQUEST_ID) if pending else None + return RecoverySnapshot(session, request, claimed_task() if active else None, None, ()) + + +class FakeStore: + def __init__(self, current: RecoverySnapshot) -> None: + self.current = current + self.recovery_calls = 0 + self.start_calls = 0 + self.stop_calls = 0 + + def recovery_snapshot(self, profile_id: str) -> RecoverySnapshot: + self.recovery_calls += 1 + return self.current + + def start_or_resume_polling(self, profile_id: str) -> PollingSession: + self.start_calls += 1 + session = self.current.session or PollingSession(PROFILE, SESSION_ID, True) + session = PollingSession(PROFILE, session.session_id, True) + self.current = RecoverySnapshot( + session, + self.current.pending_claim, + self.current.active_claim, + self.current.pending_renew, + self.current.pending_evidence, + ) + return session + + def request_stop(self, profile_id: str) -> PollingSession: + self.stop_calls += 1 + if self.current.session is None: + raise StateError("polling_session_not_found") + session = PollingSession(PROFILE, self.current.session.session_id, False) + self.current = RecoverySnapshot( + session, + self.current.pending_claim, + self.current.active_claim, + self.current.pending_renew, + self.current.pending_evidence, + ) + return session + + +class FakeGateway: + def __init__(self, outcomes: list[object] | None = None, gate: threading.Event | None = None) -> None: + self.outcomes = list(outcomes or [None]) + self.gate = gate + self.calls = 0 + + def claim_next(self, profile_id: str): + self.calls += 1 + if self.gate is not None: + self.gate.wait(2) + outcome = self.outcomes.pop(0) if self.outcomes else None + if isinstance(outcome, Exception): + raise outcome + return outcome + + +class FakeConsumer: + def __init__(self) -> None: + self.claims: list[ClaimedTask] = [] + + def accept_claim(self, claimed: ClaimedTask) -> None: + self.claims.append(claimed) + + +def wait_until(predicate, timeout_ms: int = 2000) -> None: + elapsed = 0 + while not predicate() and elapsed < timeout_ms: + QTest.qWait(10) + elapsed += 10 + if not predicate(): + raise AssertionError("condition_not_reached") + + +class PollingCoordinatorTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.app = QApplication.instance() or QApplication([]) + + def make_coordinator( + self, + store: FakeStore, + gateway: FakeGateway | None, + consumer: FakeConsumer | None, + *, + readiness: StartReadiness | None = StartReadiness(True, "ready"), + settings: ProfileSettings | None = None, + threshold: int = 3, + interval_ms: int = 5, + ) -> tuple[PollingCoordinator, list[ProfileSettings]]: + frozen: list[ProfileSettings] = [] + + def factory(value: ProfileSettings): + frozen.append(value) + assert gateway is not None + return gateway + + coordinator = PollingCoordinator( + profile_id=PROFILE, + store=store, + gateway_factory=None if gateway is None else factory, + consumer=consumer, + profile_settings=settings or profile(failure_threshold=threshold), + readiness=readiness, + failure_threshold=threshold, + timer_interval_ms=interval_ms, + ) + self.addCleanup(lambda: self.assertTrue(coordinator.shutdown())) + wait_until(lambda: not coordinator.operation_in_flight) + return coordinator, frozen + + def test_restart_normalizes_waiting_pending_and_active_without_http(self) -> None: + for name, current, expected in ( + ("waiting", snapshot(accept_new=True), PollingState.STOPPED), + ("pending", snapshot(accept_new=True, pending=True), PollingState.STOPPED), + ("active", snapshot(accept_new=True, active=True), PollingState.RECOVERY_REQUIRED), + ): + with self.subTest(name=name): + store = FakeStore(current) + gateway = FakeGateway() + consumer = FakeConsumer() + coordinator, frozen = self.make_coordinator(store, gateway, consumer) + self.assertEqual(coordinator.state, expected) + self.assertFalse(store.current.session.accept_new) + self.assertEqual(store.stop_calls, 1) + self.assertEqual(gateway.calls, 0) + self.assertEqual(frozen, []) + + def test_missing_consumer_direct_start_is_zero_gateway_and_zero_session_start(self) -> None: + store = FakeStore(snapshot()) + gateway = FakeGateway() + coordinator, frozen = self.make_coordinator(store, gateway, None) + coordinator.start() + QTest.qWait(20) + self.assertEqual(coordinator.state, PollingState.BLOCKED) + self.assertIn("尚未接入", coordinator.reason) + self.assertEqual(store.start_calls, 0) + self.assertEqual(gateway.calls, 0) + self.assertEqual(frozen, []) + + def test_readiness_is_rechecked_inside_start_handler(self) -> None: + store = FakeStore(snapshot()) + gateway = FakeGateway() + coordinator, frozen = self.make_coordinator( + store, + gateway, + FakeConsumer(), + readiness=StartReadiness(False, "设备尚未就绪"), + ) + coordinator.start() + self.assertEqual(coordinator.state, PollingState.BLOCKED) + self.assertEqual(store.start_calls, 0) + self.assertEqual(gateway.calls, 0) + self.assertEqual(frozen, []) + + def test_empty_waits_then_stop_persists_accept_new_false(self) -> None: + store = FakeStore(snapshot()) + gateway = FakeGateway([None]) + coordinator, frozen = self.make_coordinator(store, gateway, FakeConsumer(), interval_ms=1000) + coordinator.start() + wait_until(lambda: coordinator.state == PollingState.WAITING and gateway.calls == 1) + self.assertEqual(frozen[0].http_timeout_seconds, 10) + coordinator.stop() + wait_until(lambda: coordinator.state == PollingState.STOPPED and not coordinator.operation_in_flight) + self.assertFalse(store.current.session.accept_new) + self.assertEqual(gateway.calls, 1) + + def test_explicit_start_that_observes_stale_accept_new_stops_without_http(self) -> None: + store = FakeStore(snapshot()) + gateway = FakeGateway([None]) + coordinator, frozen = self.make_coordinator(store, gateway, FakeConsumer()) + store.current = snapshot(accept_new=True) + coordinator.start() + wait_until(lambda: coordinator.state == PollingState.STOPPED and not coordinator.operation_in_flight) + self.assertFalse(store.current.session.accept_new) + self.assertEqual(gateway.calls, 0) + self.assertEqual(len(frozen), 1) + self.assertIn("再次显式开始", coordinator.reason) + + def test_stop_during_claim_commits_then_requires_recovery_without_consumer_delivery(self) -> None: + gate = threading.Event() + store = FakeStore(snapshot()) + gateway = FakeGateway([claimed_task()], gate) + consumer = FakeConsumer() + coordinator, _ = self.make_coordinator(store, gateway, consumer) + visible: list[object] = [] + coordinator.claim_visible.connect(visible.append) + coordinator.start() + wait_until(lambda: coordinator.state == PollingState.CLAIMING) + coordinator._begin_claim(coordinator._epoch) + self.assertEqual(gateway.calls, 1) + coordinator.stop() + old_epoch = coordinator._epoch - 1 + coordinator._begin_claim(old_epoch) + gate.set() + wait_until(lambda: coordinator.state == PollingState.RECOVERY_REQUIRED and not coordinator.operation_in_flight) + self.assertEqual(consumer.claims, []) + self.assertEqual(gateway.calls, 1) + self.assertFalse(store.current.session.accept_new) + self.assertEqual(len(visible), 1) + self.assertIsInstance(visible[0], ClaimedTaskView) + self.assertNotIn(SENTINEL_TOKEN, repr(visible[0])) + + def test_only_reason_whitelist_gets_automatic_same_gateway_retry(self) -> None: + store = FakeStore(snapshot(accept_new=False, pending=True)) + gateway = FakeGateway( + [AmbiguousRemoteError("http_result_unknown"), AmbiguousRemoteError("truncated_response")] + ) + coordinator, _ = self.make_coordinator(store, gateway, FakeConsumer(), threshold=2) + coordinator.start() + wait_until(lambda: coordinator.state == PollingState.BLOCKED and not coordinator.operation_in_flight) + self.assertEqual(gateway.calls, 2) + self.assertEqual(store.current.pending_claim.claim_request_id, REQUEST_ID) + self.assertFalse(store.current.session.accept_new) + + def test_schema_ambiguity_blocks_after_one_call_and_preserves_pending(self) -> None: + for reason in ( + "unknown_success_status", + "invalid_claim_success_response", + "invalid_claim_lease", + "ambiguous_response_framing", + "invalid_content_length", + "response_too_large", + ): + with self.subTest(reason=reason): + store = FakeStore(snapshot(accept_new=False, pending=True)) + gateway = FakeGateway([AmbiguousRemoteError(reason)]) + coordinator, _ = self.make_coordinator(store, gateway, FakeConsumer()) + coordinator.start() + wait_until(lambda: coordinator.state == PollingState.BLOCKED and not coordinator.operation_in_flight) + self.assertEqual(gateway.calls, 1) + self.assertEqual(store.current.pending_claim.claim_request_id, REQUEST_ID) + + def test_credential_manual_and_local_failures_do_not_enter_timer_retry(self) -> None: + for error, expected_frozen in ( + (CredentialRemoteError("invalid_device_credentials"), True), + (ManualRemoteError("claim_requires_manual"), False), + (ProtocolRemoteError("invalid_protocol"), False), + (StateError("localstate_integrity_failed"), True), + ): + with self.subTest(error=error.reason): + store = FakeStore(snapshot(accept_new=False, pending=True)) + gateway = FakeGateway([error]) + coordinator, _ = self.make_coordinator(store, gateway, FakeConsumer()) + freeze_events: list[bool] = [] + coordinator.configuration_freeze_changed.connect(freeze_events.append) + coordinator.start() + wait_until(lambda: coordinator.state == PollingState.BLOCKED and not coordinator.operation_in_flight) + QTest.qWait(30) + self.assertEqual(gateway.calls, 1) + self.assertFalse(store.current.session.accept_new) + self.assertEqual(freeze_events[-1], expected_frozen) + + def test_stop_latched_bootstrap_active_emits_settled_for_pending_close(self) -> None: + store = FakeStore(snapshot()) + gateway = FakeGateway() + coordinator, _ = self.make_coordinator(store, gateway, FakeConsumer()) + store.current = snapshot(accept_new=False, active=True) + settled: list[bool] = [] + coordinator.settled.connect(lambda: settled.append(True)) + coordinator.start() + coordinator.stop() + wait_until(lambda: coordinator.state == PollingState.RECOVERY_REQUIRED) + self.assertTrue(settled) + self.assertEqual(gateway.calls, 0) + + def test_each_explicit_start_freezes_profile_for_gateway_factory(self) -> None: + store = FakeStore(snapshot()) + gateway = FakeGateway([None]) + coordinator, frozen = self.make_coordinator(store, gateway, FakeConsumer(), interval_ms=1000) + changed = profile(http_timeout=27) + coordinator.update_profile_settings(changed) + coordinator.start() + wait_until(lambda: gateway.calls == 1) + self.assertEqual(frozen, [changed]) + coordinator.stop() + wait_until(lambda: not coordinator.operation_in_flight) + + def test_claim_view_redacts_sentinel_even_if_title_contains_it(self) -> None: + wire = claim_wire() + wire["task"]["title"] = "标题 " + SENTINEL_TOKEN + wire["attempt"]["claim_token"] = SENTINEL_TOKEN + claimed = ClaimedTask.from_wire(wire) + view = ClaimedTaskView.from_claim(claimed) + self.assertNotIn(SENTINEL_TOKEN, repr(view)) + self.assertIn("已隐藏", view.title) diff --git a/client/tests/test_app.py b/client/tests/test_app.py index 13d2de7..5f441a6 100644 --- a/client/tests/test_app.py +++ b/client/tests/test_app.py @@ -22,3 +22,36 @@ class ApplicationArgumentsTests(unittest.TestCase): def test_none_uses_process_arguments(self) -> None: with mock.patch("cmbuyer_client.app.sys.argv", ["process-name", "--process-option"]): self.assertEqual(["process-name", "--process-option"], select_application_argv(None)) + + def test_standalone_entry_does_not_construct_http_or_device_capabilities(self) -> None: + source = (CLIENT_ROOT / "src" / "cmbuyer_client" / "app.py").read_text(encoding="utf-8") + for forbidden in ( + "HttpTaskSource", + "HttpTransport", + "DurableClientGateway", + "import uiautomator2", + ".device", + ".pdd", + ): + with self.subTest(forbidden=forbidden): + self.assertNotIn(forbidden, source) + self.assertIn("gateway_factory=None", source) + self.assertIn("consumer=None", source) + + def test_t304_ui_has_no_hidden_connection_probe_or_irreversible_capability_import(self) -> None: + paths = [ + *(CLIENT_ROOT / "src" / "cmbuyer_client" / "ui").glob("*.py"), + *(CLIENT_ROOT / "src" / "cmbuyer_client" / "polling").glob("*.py"), + ] + source = "\n".join(path.read_text(encoding="utf-8") for path in paths) + for forbidden in ( + "uiautomator2", + "cmbuyer_client.pdd", + "cmbuyer_client.device", + "submit_order", + "payment", + "http.client", + "subprocess", + ): + with self.subTest(forbidden=forbidden): + self.assertNotIn(forbidden, source) diff --git a/client/tests/ui/__init__.py b/client/tests/ui/__init__.py new file mode 100644 index 0000000..9d6f78a --- /dev/null +++ b/client/tests/ui/__init__.py @@ -0,0 +1 @@ +"""原生 Qt Widgets UI 测试。""" diff --git a/client/tests/ui/test_execution.py b/client/tests/ui/test_execution.py new file mode 100644 index 0000000..eca4759 --- /dev/null +++ b/client/tests/ui/test_execution.py @@ -0,0 +1,228 @@ +from __future__ import annotations + +import os +import tempfile +import unittest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from PySide6.QtCore import QObject, Qt, Signal +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication + +from cmbuyer_client.polling.coordinator import PollingState, RecoveryStatus +from cmbuyer_client.ui.execution import ExecutionPage +from cmbuyer_client.ui.main_window import PurchaseToolWindow +from cmbuyer_client.ui.records import PurchaseRecord + + +class FakeCoordinator(QObject): + state_changed = Signal(object, str, int) + claim_visible = Signal(object) + settled = Signal() + recovery_status_changed = Signal(object) + configuration_freeze_changed = Signal(bool) + + def __init__(self) -> None: + super().__init__() + self.state = PollingState.BLOCKED + self.reason = "单趟执行能力尚未接入，不能领取真实任务。" + self.consecutive_failures = 0 + self.can_start = False + self.operation_in_flight = False + self.recovery_status = RecoveryStatus(False, False, False, False, False) + self.starts = 0 + self.stops = 0 + + def start(self) -> None: + self.starts += 1 + + def stop(self) -> None: + self.stops += 1 + + def update_profile_settings(self, settings) -> None: + pass + + +class FakeStore: + def __init__(self) -> None: + self.calls = 0 + + def save_profile(self, settings, token) -> None: + self.calls += 1 + + +def records() -> list[PurchaseRecord]: + return [ + PurchaseRecord("old", "旧记录", "失败", "2026-08-04T09:00:00Z", "旧文字", "旧结果"), + PurchaseRecord("new", "新记录", "待付款", "2026-08-04T10:00:00Z", "新文字", "新结果"), + ] + + +class ExecutionPageTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.app = QApplication.instance() or QApplication([]) + + def setUp(self) -> None: + self.coordinator = FakeCoordinator() + self.page = ExecutionPage(self.coordinator) + self.page.resize(1100, 700) + self.page.show() + QTest.qWait(20) + + def tearDown(self) -> None: + self.page.close() + + def test_standalone_start_is_disabled_with_nearby_reason_and_empty_real_model(self) -> None: + self.assertFalse(self.page.poll_button.isEnabled()) + self.assertIn("尚未接入", self.page.banner.text()) + self.assertEqual(self.page.record_model.rowCount(), 0) + self.assertEqual(self.page.record_model.columnCount(), 2) + self.assertEqual(self.coordinator.starts, 0) + + def test_records_are_newest_first_and_open_via_action_then_escape_restores_focus(self) -> None: + self.page.set_records(records()) + self.assertEqual(self.page.record_model.data(self.page.record_model.index(0, 0)), "新记录") + index = self.page.record_model.index(0, 0) + self.page.record_view.setCurrentIndex(index) + self.page.view_record_action.trigger() + self.assertEqual(self.page.left_stack.currentIndex(), self.page.DETAIL_PAGE) + self.assertEqual(self.page.detail_title.text(), "新记录") + QTest.keyClick(self.page.return_button, Qt.Key.Key_Escape) + QTest.qWait(20) + self.assertEqual(self.page.left_stack.currentIndex(), self.page.LIVE_PAGE) + self.assertEqual(self.page.record_view.currentIndex().row(), 0) + self.assertTrue(self.page.record_view.hasFocus()) + + def test_detail_selection_updates_in_place_without_modal(self) -> None: + self.page.set_records(records()) + self.page.record_view.setCurrentIndex(self.page.record_model.index(0, 0)) + self.page.open_selected_record() + second = self.page.record_model.index(1, 0) + self.page.record_view.setCurrentIndex(second) + self.page._on_record_selected(second) + self.assertEqual(self.page.detail_title.text(), "旧记录") + self.assertEqual(self.page.detail_result.toPlainText(), "旧结果") + + def test_double_click_enter_and_visible_action_share_inline_detail_command(self) -> None: + self.page.set_records(records()) + index = self.page.record_model.index(0, 0) + self.page.record_view.setCurrentIndex(index) + rect = self.page.record_view.visualRect(index) + QTest.mouseDClick(self.page.record_view.viewport(), Qt.MouseButton.LeftButton, pos=rect.center()) + self.assertEqual(self.page.left_stack.currentIndex(), self.page.DETAIL_PAGE) + self.assertIsNone(QApplication.activeModalWidget()) + self.page.return_to_live() + self.page.record_view.setFocus() + QTest.keyClick(self.page.record_view, Qt.Key.Key_Return) + self.assertEqual(self.page.left_stack.currentIndex(), self.page.DETAIL_PAGE) + self.page.return_to_live() + self.page.view_record_action.trigger() + self.assertEqual(self.page.left_stack.currentIndex(), self.page.DETAIL_PAGE) + + def test_responsive_resize_keeps_model_selection_and_inline_detail(self) -> None: + self.page.set_records(records()) + model = self.page.record_model + self.page.record_view.setCurrentIndex(model.index(0, 0)) + self.page.open_selected_record() + for width in (700, 900, 1200): + self.page.resize(width, 700) + QTest.qWait(10) + self.assertIs(self.page.record_view.model(), model) + self.assertEqual(self.page.record_view.currentIndex().row(), 0) + self.assertEqual(self.page.left_stack.currentIndex(), self.page.DETAIL_PAGE) + + def test_record_model_detail_and_log_redact_bare_token_sentinel(self) -> None: + token = "e" * 64 + record = PurchaseRecord("secret", "标题 " + token, "失败", "2026-08-04T10:00:00Z", token, token) + self.page.set_records([record]) + self.page.record_view.setCurrentIndex(self.page.record_model.index(0, 0)) + self.page.open_selected_record() + self.page.log_view.append_event("Bearer " + token) + visible = "\n".join( + ( + str(self.page.record_model.data(self.page.record_model.index(0, 0))), + self.page.detail_original.toPlainText(), + self.page.detail_result.toPlainText(), + self.page.log_view.toPlainText(), + ) + ) + self.assertNotIn(token, visible) + self.assertIn("已隐藏", visible) + + +class MainWindowTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.app = QApplication.instance() or QApplication([]) + + def test_fixed_default_tabs_and_close_do_not_start_or_stop_blocked_session(self) -> None: + coordinator = FakeCoordinator() + window = PurchaseToolWindow( + store=FakeStore(), + coordinator=coordinator, + profile_settings=None, + has_stored_device_token=False, + ) + window.show() + QTest.qWait(10) + self.assertEqual(window.tabs.count(), 2) + self.assertEqual([window.tabs.tabText(i) for i in range(2)], ["采购执行", "配置"]) + self.assertIs(window.tabs.currentWidget(), window.execution_page) + self.assertFalse(window.tabs.tabsClosable()) + window.close() + self.assertEqual(coordinator.starts, 0) + self.assertEqual(coordinator.stops, 0) + + def test_close_waits_for_inflight_settlement_instead_of_terminating(self) -> None: + coordinator = FakeCoordinator() + coordinator.state = PollingState.CLAIMING + coordinator.operation_in_flight = True + window = PurchaseToolWindow( + store=FakeStore(), + coordinator=coordinator, + profile_settings=None, + has_stored_device_token=False, + ) + window.show() + self.assertFalse(window.close()) + self.assertTrue(window.isVisible()) + self.assertEqual(coordinator.stops, 1) + coordinator.operation_in_flight = False + coordinator.state = PollingState.STOPPED + coordinator.settled.emit() + QTest.qWait(20) + self.assertFalse(window.isVisible()) + + def test_recovery_status_freezes_and_unfreezes_non_token_settings(self) -> None: + coordinator = FakeCoordinator() + window = PurchaseToolWindow( + store=FakeStore(), + coordinator=coordinator, + profile_settings=None, + has_stored_device_token=False, + ) + coordinator.recovery_status_changed.emit(RecoveryStatus(True, False, True, False, False)) + self.assertFalse(window.settings_page.device_id.isEnabled()) + self.assertTrue(window.settings_page.device_token.isEnabled()) + coordinator.recovery_status_changed.emit(RecoveryStatus(False, False, False, False, False)) + self.assertTrue(window.settings_page.device_id.isEnabled()) + window.close() + + +class PurchaseRecordTimestampTests(unittest.TestCase): + def test_records_sort_by_real_rfc3339_nanoseconds_not_raw_text(self) -> None: + precise = [ + PurchaseRecord("later", "稍后", "完成", "2026-08-04T10:00:00.9Z"), + PurchaseRecord("earlier", "稍早", "完成", "2026-08-04T10:00:00.11Z"), + ] + from cmbuyer_client.ui.records import PurchaseRecordModel + + model = PurchaseRecordModel(precise) + self.assertEqual(model.record_at(0).record_id, "later") + + def test_record_timestamp_rejects_offset_and_noncanonical_trailing_zero(self) -> None: + for timestamp in ("2026-08-04T10:00:00+08:00", "2026-08-04T10:00:00.10Z"): + with self.subTest(timestamp=timestamp), self.assertRaises(ValueError): + PurchaseRecord("id", "标题", "完成", timestamp) diff --git a/client/tests/ui/test_settings.py b/client/tests/ui/test_settings.py new file mode 100644 index 0000000..29bb7f7 --- /dev/null +++ b/client/tests/ui/test_settings.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import os +from pathlib import Path +import tempfile +import unittest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication, QPushButton + +from cmbuyer_client.localstate.models import ProfileSettings +from cmbuyer_client.ui.settings import SettingsPage + + +DEVICE_ID = "f3c9f507-7473-4fa6-8d71-8786c34c6301" +TOKEN = "b" * 64 + + +class FakeStore: + def __init__(self, error: Exception | None = None) -> None: + self.calls: list[tuple[ProfileSettings, object]] = [] + self.error = error + + def save_profile(self, settings: ProfileSettings, token: object) -> None: + self.calls.append((settings, token)) + if self.error is not None: + raise self.error + + +class SettingsPageTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.app = QApplication.instance() or QApplication([]) + + def setUp(self) -> None: + self.directory = tempfile.TemporaryDirectory() + self.adb = Path(self.directory.name) / "adb.exe" + self.adb.touch() + + def tearDown(self) -> None: + self.directory.cleanup() + + def fill(self, page: SettingsPage) -> None: + page.device_id.setText(DEVICE_ID) + page.adb_path.setText(str(self.adb)) + page.adb_serial.setText("device-serial") + + def existing_settings(self) -> ProfileSettings: + return ProfileSettings( + "default", + "http://127.0.0.1:8080", + DEVICE_ID, + str(self.adb), + "device-serial", + "usb", + ) + + def test_first_save_requires_token_and_performs_zero_store_write(self) -> None: + store = FakeStore() + page = SettingsPage(store) + page.show() + QTest.qWait(10) + self.fill(page) + page.save() + self.assertEqual(store.calls, []) + self.assertTrue(page.device_token.hasFocus()) + self.assertIn("首次保存", page.feedback.text()) + + def test_visual_order_focuses_invalid_device_before_missing_token_or_adb(self) -> None: + store = FakeStore() + page = SettingsPage(store) + page.show() + QTest.qWait(10) + page.device_id.setText("not-a-uuid") + page.save() + self.assertEqual(store.calls, []) + self.assertTrue(page.device_id.hasFocus()) + self.assertIn("UUIDv4", page.feedback.text()) + + def test_existing_blank_token_passes_none_and_success_never_refills(self) -> None: + store = FakeStore() + page = SettingsPage(store, settings=self.existing_settings(), has_stored_device_token=True) + page.save() + self.assertEqual(len(store.calls), 1) + self.assertIsNone(store.calls[0][1]) + self.assertEqual(page.device_token.text(), "") + self.assertEqual(page.token_status.text(), "已保存") + + def test_nonempty_token_replaces_once_then_is_cleared(self) -> None: + store = FakeStore() + page = SettingsPage(store, settings=self.existing_settings(), has_stored_device_token=True) + page.device_token.setText(TOKEN) + page.save() + self.assertEqual(len(store.calls), 1) + self.assertEqual(store.calls[0][1].value, TOKEN) + self.assertEqual(page.device_token.text(), "") + self.assertNotIn(TOKEN, page.feedback.text()) + self.assertNotIn(TOKEN, page.token_status.text()) + + def test_failed_save_preserves_token_and_focuses_secret_field_without_echo(self) -> None: + store = FakeStore(RuntimeError("must-not-appear-" + TOKEN)) + page = SettingsPage(store, settings=self.existing_settings(), has_stored_device_token=True) + page.show() + QTest.qWait(10) + page.device_token.setText(TOKEN) + page.save() + self.assertEqual(page.device_token.text(), TOKEN) + self.assertTrue(page.device_token.hasFocus()) + self.assertNotIn(TOKEN, page.feedback.text()) + + def test_pending_or_active_freezes_every_non_token_field_and_preserves_exact_settings(self) -> None: + original = self.existing_settings() + store = FakeStore() + page = SettingsPage( + store, + settings=original, + has_stored_device_token=True, + identity_frozen=True, + ) + for field in ( + page.device_id, + page.adb_path, + page.adb_serial, + page.transport, + page.poll_interval, + page.failure_threshold, + page.http_timeout, + page.step_timeout, + ): + self.assertFalse(field.isEnabled()) + self.assertTrue(page.device_token.isEnabled()) + page.device_token.setText(TOKEN) + page.save() + self.assertEqual(store.calls[0][0], original) + + def test_page_has_no_connection_probe_command_and_explains_deferred_validation(self) -> None: + page = SettingsPage(FakeStore()) + button_texts = [button.text() for button in page.findChildren(QPushButton)] + self.assertEqual(button_texts, ["保存配置"]) + self.assertIn("首次真实领取", page.validation_hint.text()) + self.assertEqual(page.service_url.text(), "http://127.0.0.1:8080") + self.assertTrue(page.service_url.isReadOnly())