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)