zhouchengcheng/ctms
1
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity
Files
f6e5a4c74403d635bee9aecabebcac8ca572a831
ctms/backend/tests/test_visit_schedule.py
T
Cheng Zhou a2add43fa5 feat(subject): 参与者状态依赖筛选日期并支持后续编辑 
- 未填写筛选日期的参与者不再默认进入 SCREENING，状态显示为占位符。
- SubjectUpdate 不再开放 status 入参，改由后端依据日期推导，并新增
  screening_date 字段允许 PM/CRA 后续补录筛选日期。
- 参与者管理页与详情页同步：详情页支持编辑筛选日期，列表页与详情页
  在没有筛选日期时显示占位符而非旧 enum 值。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 12:36:33 +08:00

284 lines
11 KiB
Python
Raw Blame History

from datetime import date
from types import SimpleNamespace
from app.crud.subject import derive_subject_status, should_generate_visits_after_subject_update
from app.crud.visit import (
build_early_termination_visit_changes,
build_visit_schedule_dates,
get_last_planned_visit_window_start_date,
sort_visits_for_display,
validate_early_termination_date,
)
from app.schemas.study import StudyUpdate
def test_build_visit_schedule_dates_uses_per_visit_windows():
baseline_date = date(2026, 5, 8)
visits = build_visit_schedule_dates(
[
{
"visit_code": "基线访视",
"baseline_offset_days": 0,
"window_before_days": 0,
"window_after_days": 0,
},
{
"visit_code": "V1",
"baseline_offset_days": 7,
"window_before_days": 2,
"window_after_days": 2,
},
{
"visit_code": "V2",
"baseline_offset_days": 15,
"window_before_days": 2,
"window_after_days": 2,
},
{
"visit_code": "V3",
"baseline_offset_days": 23,
"window_before_days": 3,
"window_after_days": 3,
},
],
baseline_date,
)
assert [
(visit["visit_code"], visit["baseline_offset_days"], visit["planned_date"], visit["window_start"], visit["window_end"])
for visit in visits
] == [
("基线访视", 0, date(2026, 5, 8), date(2026, 5, 8), date(2026, 5, 8)),
("V1", 7, date(2026, 5, 15), date(2026, 5, 13), date(2026, 5, 17)),
("V2", 15, date(2026, 5, 23), date(2026, 5, 21), date(2026, 5, 25)),
("V3", 23, date(2026, 5, 31), date(2026, 5, 28), date(2026, 6, 3)),
]
def test_build_visit_schedule_dates_keeps_visit_structure_without_baseline_date():
visits = build_visit_schedule_dates(
[
{
"visit_code": "基线访视",
"baseline_offset_days": 0,
"window_before_days": 0,
"window_after_days": 0,
},
{
"visit_code": "V1",
"baseline_offset_days": 7,
"window_before_days": 2,
"window_after_days": 2,
},
],
None,
)
assert [
(visit["visit_code"], visit["planned_date"], visit["window_start"], visit["window_end"])
for visit in visits
] == [
("基线访视", None, None, None),
("V1", None, None, None),
]
def test_study_update_allows_multiple_visits_with_same_baseline_offset():
payload = StudyUpdate(
visit_schedule=[
{
"visit_code": "筛选访视",
"baseline_offset_days": 0,
"window_before_days": 0,
"window_after_days": 0,
},
{
"visit_code": "基线访视",
"baseline_offset_days": 0,
"window_before_days": 0,
"window_after_days": 0,
},
]
)
assert [item.visit_code for item in payload.visit_schedule or []] == ["筛选访视", "基线访视"]
def test_sort_visits_for_display_follows_configured_visit_order_without_dates():
visits = [
SimpleNamespace(visit_code="V1", planned_date=None),
SimpleNamespace(visit_code="V2", planned_date=None),
SimpleNamespace(visit_code="基线访视", planned_date=None),
]
visit_schedule = [
{"visit_code": "基线访视"},
{"visit_code": "V1"},
{"visit_code": "V2"},
]
assert [visit.visit_code for visit in sort_visits_for_display(visits, visit_schedule)] == ["基线访视", "V1", "V2"]
def test_sort_visits_for_display_does_not_infer_business_order():
visits = [
SimpleNamespace(visit_code="基线访视", planned_date=None),
SimpleNamespace(visit_code="V1", planned_date=None),
SimpleNamespace(visit_code="V2", planned_date=None),
]
visit_schedule = [
{"visit_code": "V1"},
{"visit_code": "V2"},
{"visit_code": "基线访视"},
]
assert [visit.visit_code for visit in sort_visits_for_display(visits, visit_schedule)] == ["V1", "V2", "基线访视"]
def test_sort_visits_places_early_termination_after_last_actual_visit():
visits = [
SimpleNamespace(visit_code="筛选访视", planned_date=date(2026, 5, 3), actual_date=date(2026, 5, 3)),
SimpleNamespace(visit_code="基线访视", planned_date=date(2026, 5, 3), actual_date=date(2026, 5, 3)),
SimpleNamespace(visit_code="V1", planned_date=date(2026, 5, 10), actual_date=None),
SimpleNamespace(visit_code="V2", planned_date=date(2026, 5, 17), actual_date=None),
SimpleNamespace(visit_code="提前终止", planned_date=None, actual_date=date(2026, 5, 13)),
]
visit_schedule = [
{"visit_code": "筛选访视"},
{"visit_code": "基线访视"},
{"visit_code": "V1"},
{"visit_code": "V2"},
]
assert [visit.visit_code for visit in sort_visits_for_display(visits, visit_schedule)] == [
"筛选访视",
"基线访视",
"提前终止",
"V1",
"V2",
]
def test_should_generate_visits_when_baseline_date_is_set_or_changed():
assert should_generate_visits_after_subject_update(
previous_baseline_date=None,
next_baseline_date=date(2026, 5, 8),
)
assert should_generate_visits_after_subject_update(
previous_baseline_date=date(2026, 5, 8),
next_baseline_date=date(2026, 5, 9),
)
assert not should_generate_visits_after_subject_update(
previous_baseline_date=None,
next_baseline_date=None,
)
def test_build_early_termination_visit_changes_adds_event_and_cancels_future_planned_visits():
visits = [
SimpleNamespace(visit_code="筛选访视", planned_date=date(2026, 5, 1), actual_date=date(2026, 5, 1), status="DONE"),
SimpleNamespace(visit_code="基线访视", planned_date=date(2026, 5, 1), actual_date=date(2026, 5, 1), status="DONE"),
SimpleNamespace(visit_code="V0", planned_date=date(2026, 5, 5), actual_date=None, status="LOST"),
SimpleNamespace(
visit_code="V1",
planned_date=date(2026, 5, 6),
window_start=date(2026, 5, 4),
window_end=date(2026, 5, 8),
actual_date=None,
status="PLANNED",
),
SimpleNamespace(
visit_code="V2",
planned_date=date(2026, 5, 15),
window_start=date(2026, 5, 12),
window_end=date(2026, 5, 18),
actual_date=None,
status="PLANNED",
),
]
changes = build_early_termination_visit_changes(
visits,
termination_date=date(2026, 5, 6),
reason="不良事件退出",
)
assert changes.event_visit_code == "提前终止"
assert changes.event_actual_date == date(2026, 5, 6)
assert changes.event_notes == "不良事件退出"
assert [visit.visit_code for visit in changes.visits_to_cancel] == ["V1", "V2"]
def test_validate_early_termination_date_requires_date_before_last_visit_window_start():
visits = [
SimpleNamespace(visit_code="筛选访视", planned_date=date(2026, 5, 1), window_start=date(2026, 5, 1)),
SimpleNamespace(visit_code="基线访视", planned_date=date(2026, 5, 1), window_start=date(2026, 5, 1)),
SimpleNamespace(visit_code="V1", planned_date=date(2026, 5, 8), window_start=date(2026, 5, 6)),
SimpleNamespace(visit_code="V2", planned_date=date(2026, 5, 15), window_start=date(2026, 5, 12)),
SimpleNamespace(visit_code="提前终止", planned_date=None, window_start=None),
]
assert get_last_planned_visit_window_start_date(visits) == date(2026, 5, 12)
validate_early_termination_date(date(2026, 5, 11), visits)
try:
validate_early_termination_date(date(2026, 5, 12), visits)
except ValueError as exc:
assert "提前终止日期必须早于方案最后一个计划访视窗口开始日" in str(exc)
else:
raise AssertionError("same-day final visit window start should not be accepted as early termination")
def test_derive_subject_status_marks_dropped_when_early_termination_has_actual_date():
subject = SimpleNamespace(screening_date=date(2026, 5, 1), enrollment_date=date(2026, 5, 3))
visits = [
SimpleNamespace(visit_code="V1", planned_date=date(2026, 5, 10), actual_date=date(2026, 5, 10), status="DONE", created_at=None),
SimpleNamespace(visit_code="提前终止", planned_date=None, actual_date=date(2026, 5, 12), status="DONE", created_at=None),
]
assert derive_subject_status(subject, visits, [{"visit_code": "V1"}]) == "DROPPED"
def test_derive_subject_status_marks_completed_when_last_standard_visit_has_actual_date():
subject = SimpleNamespace(screening_date=date(2026, 5, 1), enrollment_date=date(2026, 5, 3))
visits = [
SimpleNamespace(visit_code="V1", planned_date=date(2026, 5, 10), actual_date=date(2026, 5, 10), status="DONE", created_at=None),
SimpleNamespace(visit_code="V2", planned_date=date(2026, 5, 17), actual_date=date(2026, 5, 17), status="DONE", created_at=None),
]
assert derive_subject_status(subject, visits, [{"visit_code": "V1"}, {"visit_code": "V2"}]) == "COMPLETED"
def test_derive_subject_status_rolls_back_to_enrolled_when_followup_after_standard_end_is_unfinished():
subject = SimpleNamespace(screening_date=date(2026, 5, 1), enrollment_date=date(2026, 5, 3))
visits = [
SimpleNamespace(visit_code="V1", planned_date=date(2026, 5, 10), actual_date=date(2026, 5, 10), status="DONE", created_at=None),
SimpleNamespace(visit_code="V2", planned_date=date(2026, 5, 17), actual_date=date(2026, 5, 17), status="DONE", created_at=None),
SimpleNamespace(visit_code="安全性随访", planned_date=date(2026, 5, 24), actual_date=None, status="PLANNED", created_at=None),
]
assert derive_subject_status(subject, visits, [{"visit_code": "V1"}, {"visit_code": "V2"}]) == "ENROLLED"
def test_derive_subject_status_uses_completed_standard_visit_when_extra_visit_is_before_standard_end():
subject = SimpleNamespace(screening_date=date(2026, 5, 1), enrollment_date=date(2026, 5, 3))
visits = [
SimpleNamespace(visit_code="V1", planned_date=date(2026, 5, 10), actual_date=date(2026, 5, 10), status="DONE", created_at=None),
SimpleNamespace(visit_code="安全性随访", planned_date=date(2026, 5, 12), actual_date=None, status="PLANNED", created_at=None),
SimpleNamespace(visit_code="V2", planned_date=date(2026, 5, 17), actual_date=date(2026, 5, 17), status="DONE", created_at=None),
]
assert derive_subject_status(subject, visits, [{"visit_code": "V1"}, {"visit_code": "V2"}]) == "COMPLETED"
def test_derive_subject_status_marks_screening_until_enrollment_date_is_filled():
subject = SimpleNamespace(screening_date=date(2026, 5, 1), enrollment_date=None)
assert derive_subject_status(subject, [], []) == "SCREENING"
def test_derive_subject_status_is_empty_before_screening_date_is_filled():
subject = SimpleNamespace(screening_date=None, enrollment_date=None)
assert derive_subject_status(subject, [], []) == ""
Reference in New Issue View Git Blame Copy Permalink