229 lines
9.4 KiB
Python
229 lines
9.4 KiB
Python
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)
|