522 lines
18 KiB
Python
522 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import pytest
|
||
from PySide6.QtWidgets import QApplication, QFrame, QLabel, QPushButton, QVBoxLayout
|
||
|
||
from doctor_workstation.ui import diagnosis_media
|
||
from doctor_workstation.ui.diagnosis_drawer import RecordTable
|
||
from doctor_workstation.ui.diagnosis_media import (
|
||
InlineRecordingPlayer,
|
||
RecordingPlaybackCell,
|
||
normalize_recording_urls,
|
||
preferred_recording_url,
|
||
safe_http_url,
|
||
)
|
||
from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module
|
||
from doctor_workstation.ui.dialogs.diagnosis import (
|
||
DiagnosisDialog,
|
||
OrderDetailDrawer,
|
||
present_order_detail,
|
||
)
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def application() -> QApplication:
|
||
return QApplication.instance() or QApplication([])
|
||
|
||
|
||
class _Repository:
|
||
def get_prescription_order(self, order_id: int) -> dict[str, Any]:
|
||
return {"id": order_id}
|
||
|
||
def set_revisit_slot_start_offset(self, diagnosis_id: int, offset: int) -> dict[str, Any]:
|
||
return {"diagnosis_id": diagnosis_id, "revisit_slot_start_offset": offset}
|
||
|
||
def upload_call_recording(
|
||
self,
|
||
path: str,
|
||
diagnosis_id: int,
|
||
*,
|
||
call_record_id: int | None = None,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"path": path,
|
||
"diagnosis_id": diagnosis_id,
|
||
"call_record_id": call_record_id,
|
||
}
|
||
|
||
|
||
def _rich_order() -> dict[str, Any]:
|
||
return {
|
||
"id": 801,
|
||
"order_no": "RX-20260811-0801",
|
||
"diagnosis_id": 501,
|
||
"prescription_id": 601,
|
||
"amount": 428.5,
|
||
"linked_pay_paid_total": 300,
|
||
"refund_amount": 20,
|
||
"agency_collect_amount": 128.5,
|
||
"fulfillment_status": 5,
|
||
"prescription_audit_status": 1,
|
||
"payment_slip_audit_status": 1,
|
||
"doctor_name": "陈医生",
|
||
"assistant_name": "赵医助",
|
||
"creator_name": "赵医助",
|
||
"creator_account": "assistant.zhao",
|
||
"create_time": "2026-08-11 09:26",
|
||
"recipient_name": "林晓岚",
|
||
"recipient_phone": "18600004218",
|
||
"shipping_province": "河南省",
|
||
"shipping_city": "洛阳市",
|
||
"shipping_district": "洛龙区",
|
||
"shipping_address": "开元大道 88 号",
|
||
"is_follow_up": 1,
|
||
"medication_days": 14,
|
||
"service_channel": "线上复诊",
|
||
"service_package": ["调理服务", "复诊随访"],
|
||
"fee_type": 3,
|
||
"tracking_number": "SF164208110801",
|
||
"express_company": "sf",
|
||
"remark_assistant": "工作日下午送达",
|
||
"prescription_audit_remark": "辨证与用量已复核",
|
||
"payment_slip_audit_remark": "收款凭证已核验",
|
||
"prescription": {
|
||
"id": 601,
|
||
"sn": "RX601",
|
||
"patient_name": "林晓岚",
|
||
"gender_desc": "女",
|
||
"age": 34,
|
||
"phone": "18600004218",
|
||
"prescription_date": "2026-08-11",
|
||
"doctor_name": "陈医生",
|
||
"prescription_type": "饮片",
|
||
"clinical_diagnosis": "气阴两虚",
|
||
"dose_count": 14,
|
||
"dose_unit": "剂",
|
||
"usage_instruction": "水煎服",
|
||
"amount": 428.5,
|
||
"audit_status": 1,
|
||
"dosage_amount": 180,
|
||
"dosage_unit": "g",
|
||
"need_decoction": 1,
|
||
"times_per_day": 2,
|
||
"usage_days": 14,
|
||
"dietary_taboo": ["辛辣", "生冷"],
|
||
"void_status": 0,
|
||
},
|
||
"linked_pay_orders": [
|
||
{
|
||
"id": 9101,
|
||
"order_no": "PAY-9101",
|
||
"order_type_desc": "药品费用",
|
||
"amount": 300,
|
||
"status_desc": "已支付",
|
||
"creator_name": "赵医助",
|
||
"create_time": "2026-08-11 09:32",
|
||
}
|
||
],
|
||
"unlinked_pay_orders": [],
|
||
"logistics_trace": {
|
||
"state_text": "运输中",
|
||
"carrier_label": "顺丰速运",
|
||
"traces": [
|
||
{
|
||
"time": "2026-08-11 16:10",
|
||
"status": "运输中",
|
||
"context": "快件已离开洛阳集散中心",
|
||
},
|
||
{
|
||
"time": "2026-08-11 13:06",
|
||
"status": "已揽收",
|
||
"context": "顺丰速运已收取快件",
|
||
},
|
||
],
|
||
},
|
||
"logs": [
|
||
{
|
||
"id": 1,
|
||
"admin_name": "赵医助",
|
||
"action": "ship",
|
||
"summary": "确认发货并填写顺丰运单",
|
||
"create_time": "2026-08-11 13:08",
|
||
}
|
||
],
|
||
}
|
||
|
||
|
||
def test_order_offset_copy_tooltip_preview_and_list_columns_are_exact(
|
||
application: QApplication,
|
||
) -> None:
|
||
dialog = DiagnosisDialog(_Repository(), permissions=["*"])
|
||
label = dialog.findChild(QLabel, "DiagnosisOrderOffsetLabel")
|
||
assert label is not None
|
||
assert label.text() == "复诊统计起始偏移"
|
||
assert label.toolTip() == diagnosis_module._ORDER_OFFSET_HELP
|
||
assert dialog.order_offset_help.toolTip() == diagnosis_module._ORDER_OFFSET_HELP
|
||
assert dialog.order_offset_save.text() == "保存"
|
||
|
||
dialog._editable = True
|
||
dialog._can_offset = True
|
||
dialog._saved_order_offset = 0
|
||
dialog.order_offset.setValue(2)
|
||
assert dialog.order_offset_preview.text() == "第 1 笔实单计为三诊"
|
||
assert dialog.order_offset_save.isEnabled()
|
||
|
||
dialog._fill_orders(
|
||
[
|
||
{
|
||
"id": 801,
|
||
"order_no": "RX-801",
|
||
"global_visit_seq": 4,
|
||
"counts_for_revisit_rate": 0,
|
||
"amount": 286,
|
||
"fulfillment_status": 2,
|
||
}
|
||
]
|
||
)
|
||
table = dialog._table_registry["orders"][1]
|
||
assert table.horizontalHeaderItem(0).text() == "订单编号"
|
||
assert table.item(0, 1).text() == "4诊"
|
||
assert table.item(0, 2).text() == "否"
|
||
assert table.item(0, 3).text() == "¥286.00"
|
||
assert table.item(0, 6).text() == "待发货"
|
||
dialog.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_order_detail_is_eighty_percent_readonly_drawer_with_real_sections(
|
||
application: QApplication,
|
||
) -> None:
|
||
host = DiagnosisDialog(_Repository(), permissions=["*"])
|
||
host.resize(1200, 760)
|
||
host.show()
|
||
drawer = host._build_order_detail_dialog(_rich_order(), 801)
|
||
drawer.show()
|
||
application.processEvents()
|
||
|
||
assert isinstance(drawer, OrderDetailDrawer)
|
||
assert drawer.size() == host.size()
|
||
assert abs(drawer.drawer_panel.width() - round(host.width() * 0.8)) <= 1
|
||
assert drawer.drawer_panel.property("readonly") is True
|
||
labels = [label.text() for label in drawer.findChildren(QLabel)]
|
||
for section in ("金额概览", "处方详情", "收款记录", "履约与收货信息", "物流轨迹", "操作日志"):
|
||
assert section in labels
|
||
assert "¥428.50" in labels
|
||
assert "¥300.00" in labels
|
||
assert "确认发货并填写顺丰运单" in labels
|
||
assert "快件已离开洛阳集散中心" in labels
|
||
linked = drawer.findChild(RecordTable, "DiagnosisOrderLinkedPayments")
|
||
assert linked is not None and linked.rowCount() == 1
|
||
|
||
drawer.close()
|
||
host.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_present_order_detail_shared_entry_matches_admin_drawer_sections(
|
||
application: QApplication,
|
||
) -> None:
|
||
host = QFrame()
|
||
host.resize(1100, 720)
|
||
host.show()
|
||
application.processEvents()
|
||
drawer = present_order_detail(
|
||
host,
|
||
_rich_order(),
|
||
order_id=801,
|
||
permissions=["tcm.prescriptionOrder/logs", "tcm.prescriptionOrder/detail"],
|
||
exec_=False,
|
||
)
|
||
application.processEvents()
|
||
assert isinstance(drawer, OrderDetailDrawer)
|
||
labels = [label.text() for label in drawer.findChildren(QLabel)]
|
||
for section in ("金额概览", "处方详情", "收款记录", "履约与收货信息", "物流轨迹", "操作日志"):
|
||
assert section in labels
|
||
assert "RX-20260811-0801" in " ".join(labels)
|
||
drawer.close()
|
||
host.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_order_detail_missing_fields_use_explicit_empty_states_without_fake_zero(
|
||
application: QApplication,
|
||
) -> None:
|
||
host = DiagnosisDialog(_Repository(), permissions=["*"])
|
||
host.resize(1000, 680)
|
||
drawer = host._build_order_detail_dialog({"id": 809}, 809)
|
||
drawer.show()
|
||
application.processEvents()
|
||
|
||
labels = [label.text() for label in drawer.findChildren(QLabel)]
|
||
amount_values = [
|
||
label.text()
|
||
for label in drawer.findChildren(QLabel)
|
||
if label.property("orderAmountValue") is True
|
||
]
|
||
assert amount_values == ["—", "—", "—", "—", "—"]
|
||
assert "¥0.00" not in labels
|
||
assert "无处方数据（详情接口未返回 prescription）" in labels
|
||
assert "详情接口未返回关联收款记录字段" in labels
|
||
assert "详情接口未返回未关联收款记录字段" in labels
|
||
assert "订单详情未返回快递单号，暂无物流轨迹" in labels
|
||
assert "详情接口未返回操作日志数据" in labels
|
||
|
||
drawer.close()
|
||
host.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_order_logs_remain_permission_gated(application: QApplication) -> None:
|
||
host = DiagnosisDialog(
|
||
_Repository(),
|
||
permissions=["tcm.prescriptionOrder/detail"],
|
||
)
|
||
drawer = host._build_order_detail_dialog(_rich_order(), 801)
|
||
denied = next(
|
||
label for label in drawer.findChildren(QLabel) if label.property("permissionDenied") is True
|
||
)
|
||
assert denied.text() == "当前账号无操作日志查看权限"
|
||
assert "确认发货并填写顺丰运单" not in [label.text() for label in drawer.findChildren(QLabel)]
|
||
drawer.close()
|
||
host.close()
|
||
|
||
|
||
def test_recording_preference_inline_height_alternates_and_safe_external_open(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
urls = [
|
||
"https://bucket.cos.ap-shanghai.myqcloud.com/replay/index.m3u8",
|
||
"https://media.example.invalid/replay.mp4",
|
||
"https://media.example.invalid/replay.webm",
|
||
"file:///C:/private/replay.mp4",
|
||
"https://media.example.invalid/replay.mp4",
|
||
]
|
||
assert normalize_recording_urls(urls) == urls[:-1]
|
||
assert preferred_recording_url(urls) == "https://media.example.invalid/replay.mp4"
|
||
assert safe_http_url("file:///C:/private/replay.mp4") is None
|
||
|
||
opened: list[str] = []
|
||
monkeypatch.setattr(
|
||
diagnosis_media.QDesktopServices,
|
||
"openUrl",
|
||
lambda url: opened.append(url.toString()) or True,
|
||
)
|
||
cell = RecordingPlaybackCell(urls, record_id=48)
|
||
cell.show()
|
||
application.processEvents()
|
||
player = cell.findChild(InlineRecordingPlayer, "DiagnosisInlineRecordingPlayer")
|
||
assert player is not None
|
||
assert player.target == "https://media.example.invalid/replay.mp4"
|
||
assert player.maximumHeight() == 180
|
||
assert player.property("maximumPlaybackHeight") == 180
|
||
assert player._source_attached is False
|
||
assert player.player is None
|
||
assert player.audio_output is None
|
||
assert player.video is None
|
||
alternate_buttons = cell.findChildren(QPushButton, "DiagnosisRecordingAlternateLink")
|
||
assert [button.text() for button in alternate_buttons] == [
|
||
"COS HLS 1",
|
||
"链接 2",
|
||
"MP4 3",
|
||
]
|
||
assert alternate_buttons[-1].isEnabled() is False
|
||
alternate_buttons[0].click()
|
||
assert opened == [urls[0]]
|
||
|
||
empty = RecordingPlaybackCell([], record_id=49)
|
||
empty_state = empty.findChild(QLabel, "DiagnosisEmptyState")
|
||
assert empty_state is not None and empty_state.text() == "暂无录制回放"
|
||
invalid = RecordingPlaybackCell(["file:///C:/private/replay.mp4"], record_id=50)
|
||
invalid_state = invalid.findChild(QLabel, "DiagnosisUnsupportedState")
|
||
assert invalid_state is not None and "回放地址无效" in invalid_state.text()
|
||
empty.close()
|
||
invalid.close()
|
||
cell.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_recording_rows_do_not_eagerly_create_native_players(
|
||
application: QApplication,
|
||
) -> None:
|
||
cells = [
|
||
RecordingPlaybackCell(
|
||
[f"https://media.example.invalid/replay-{index}.mp4"],
|
||
record_id=index,
|
||
)
|
||
for index in range(40)
|
||
]
|
||
application.processEvents()
|
||
|
||
inline_players = [cell.inline_player for cell in cells]
|
||
assert all(player is not None for player in inline_players)
|
||
assert all(player.player is None for player in inline_players if player is not None)
|
||
assert all(player.audio_output is None for player in inline_players if player is not None)
|
||
assert all(player.video is None for player in inline_players if player is not None)
|
||
|
||
for cell in cells:
|
||
cell.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_video_table_embeds_player_and_preserves_row_bound_upload(
|
||
application: QApplication,
|
||
) -> None:
|
||
dialog = DiagnosisDialog(_Repository(), permissions=["*"])
|
||
dialog._editable = True
|
||
dialog._can_video_upload = True
|
||
dialog._diagnosis_id = 501
|
||
dialog._tab_generations["video"] = 7
|
||
uploaded: list[int | None] = []
|
||
dialog._upload_call_recording = lambda call_record_id=None: uploaded.append(call_record_id) # type: ignore[method-assign]
|
||
dialog._fill_video(
|
||
[
|
||
{
|
||
"id": 48,
|
||
"recording_urls_list": [
|
||
"https://media.example.invalid/replay.mp4",
|
||
"https://media.example.invalid/replay-backup.m3u8",
|
||
],
|
||
"call_type": 2,
|
||
"status": 2,
|
||
"recording_status_text": "录制完成",
|
||
},
|
||
{
|
||
"id": 47,
|
||
"recording_urls_list": [],
|
||
"call_type": 1,
|
||
"status": 3,
|
||
"recording_status_text": "暂无录制",
|
||
},
|
||
]
|
||
)
|
||
|
||
preview = QFrame()
|
||
preview_layout = QVBoxLayout(preview)
|
||
page = dialog._tab_pages["video"]
|
||
page.setParent(preview)
|
||
page.show()
|
||
preview_layout.addWidget(page)
|
||
preview.resize(1200, 560)
|
||
preview.show()
|
||
application.processEvents()
|
||
|
||
table = dialog._table_registry["video"][1]
|
||
playback = table.cellWidget(0, 0)
|
||
assert isinstance(playback, RecordingPlaybackCell)
|
||
assert playback.property("callRecordId") == 48
|
||
player = playback.findChild(InlineRecordingPlayer)
|
||
assert player is not None
|
||
surface = player.findChild(QFrame, "DiagnosisInlineRecordingSurface")
|
||
external = player.findChild(QPushButton, "DiagnosisInlineRecordingExternal")
|
||
fallback = player.findChild(QPushButton, "DiagnosisInlineRecordingFallback")
|
||
assert surface is not None and external is not None and fallback is not None
|
||
|
||
assert table.rowHeight(0) >= playback.required_table_row_height()
|
||
assert playback.height() >= playback.minimumSizeHint().height()
|
||
assert 158 <= player.height() <= 180
|
||
assert surface.height() >= 122
|
||
assert player.position.width() >= 50
|
||
assert player.position.height() >= 12
|
||
assert player.time_label.height() >= 15
|
||
for control in (player.play_button, external, fallback):
|
||
assert control.isVisibleTo(player)
|
||
assert control.height() >= 24
|
||
for control in (player.play_button, player.position, player.time_label, external, fallback):
|
||
assert player.rect().contains(control.geometry())
|
||
|
||
# Guard the actual rendered pixels: the previous regression produced only
|
||
# a 16 px dark strip despite the class-level maximumHeight declaration.
|
||
playback_image = playback.grab().toImage()
|
||
dark_rows = []
|
||
for y in range(playback_image.height()):
|
||
dark_pixels = sum(
|
||
1
|
||
for x in range(playback_image.width())
|
||
if max(
|
||
playback_image.pixelColor(x, y).red(),
|
||
playback_image.pixelColor(x, y).green(),
|
||
playback_image.pixelColor(x, y).blue(),
|
||
)
|
||
<= 55
|
||
)
|
||
if dark_pixels >= round(playback_image.width() * 0.65):
|
||
dark_rows.append(y)
|
||
assert dark_rows and dark_rows[-1] - dark_rows[0] + 1 >= 120
|
||
|
||
assert table.item(1, 0).text() == "暂无录制回放"
|
||
assert dialog._recording_players == []
|
||
upload_host = table.cellWidget(0, 8)
|
||
assert upload_host is not None
|
||
upload = upload_host.findChild(QPushButton, "DiagnosisVideoRowUpload")
|
||
assert upload is not None
|
||
assert upload.property("callRecordId") == 48
|
||
assert table.item(0, 8).text() == ""
|
||
assert abs(upload_host.rect().center().y() - upload.geometry().center().y()) <= 1
|
||
upload.click()
|
||
assert uploaded == [48]
|
||
preview.close()
|
||
dialog.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_inline_player_rejects_stale_owner_generation(application: QApplication) -> None:
|
||
class _Owner:
|
||
_tab_generations = {"video": 4}
|
||
|
||
owner = _Owner()
|
||
player = InlineRecordingPlayer(
|
||
"https://media.example.invalid/replay.mp4",
|
||
render_owner=owner,
|
||
owner_generation=4,
|
||
)
|
||
owner._tab_generations["video"] = 5
|
||
if player.player is not None:
|
||
assert player._attach_source() is False
|
||
assert player.play_button.isEnabled() is False
|
||
assert "已刷新" in player.placeholder.text()
|
||
player.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_drawer_and_inline_player_render_non_empty_images(
|
||
application: QApplication,
|
||
tmp_path: Any,
|
||
) -> None:
|
||
host = DiagnosisDialog(_Repository(), permissions=["*"])
|
||
host.resize(1100, 720)
|
||
host.show()
|
||
drawer = host._build_order_detail_dialog(_rich_order(), 801)
|
||
drawer.show()
|
||
application.processEvents()
|
||
order_image = drawer.grab().toImage()
|
||
order_path = tmp_path / "order.png"
|
||
assert order_image.width() == 1100 and order_image.height() == 720
|
||
assert order_image.save(str(order_path), "PNG")
|
||
assert order_path.stat().st_size > 10_000
|
||
|
||
cell = RecordingPlaybackCell(
|
||
[
|
||
"https://media.example.invalid/replay.mp4",
|
||
"https://media.example.invalid/replay.m3u8",
|
||
],
|
||
record_id=48,
|
||
)
|
||
cell.resize(520, 235)
|
||
cell.show()
|
||
application.processEvents()
|
||
video_image = cell.grab().toImage()
|
||
video_path = tmp_path / "video.png"
|
||
assert video_image.save(str(video_path), "PNG")
|
||
assert video_path.stat().st_size > 2_000
|
||
|
||
cell.close()
|
||
drawer.close()
|
||
host.close()
|
||
application.processEvents()
|