Files
Pre_Seg_Server/backend/tests/test_projects.py
admin 481ffa5b67 完善项目导入、模板与分割工作区交互
- 增强 DICOM/视频项目导入与演示数据:DICOM 按文件名自然顺序处理,导入后展示上传与解析任务进度,恢复演示出厂设置保留演示视频和演示 DICOM 项目,并补充 demo media seed 逻辑。

- 完善项目管理:项目支持重命名、删除、复制,删除使用站内确认弹窗,复制支持新项目重置和全内容复制,DICOM 项目不显示生成帧入口。

- 完善 GT Mask 与导出链路:只支持 8-bit maskid 图导入,非法/全背景图明确拒绝,尺寸自动适配,高精度 polygon 回显;统一导出默认当前帧,GT_label 使用 uint8 和真实 maskid,待分类 maskid 0 与背景一致。

- 完善分割工作区交互:新增画笔和橡皮擦并支持尺寸控制,移除创建点/线段入口,工具栏按类别分隔,AI 智能分割使用明确 AI 图标,取消黄色 seed point,清空/删除传播 mask 后同步清理空帧时间轴状态。

- 完善传播与时间轴:自动传播使用 SAM 2.1 权重任务,参考帧无遮罩时提示,传播历史按同一蓝色系递进变暗,删除/清空传播链时保留人工或独立 AI 标注来源。

- 完善模板库:新增头颈部 CT 分割默认模板,所有模板保留 maskid 0 待分类,支持鼠标复制模板、拖拽层级、JSON 批量导入预览、删除 label 和站内删除确认。

- 完善用户与高风险确认:用户改密码、删除用户、恢复演示出厂设置和清空人工/AI 标注帧均改为站内确认交互,避免浏览器原生 prompt/confirm。

- 补充前后端测试与文档:更新项目、模板、GT 导入、导出、传播、DICOM、用户管理等测试,并同步 README、AGENTS 和 doc 下实现/契约/测试计划文档。
2026-05-03 17:11:59 +08:00

217 lines
8.7 KiB
Python

from models import Annotation, Frame, Mask, ProcessingTask, Project, User
from routers.auth import create_access_token, hash_password
def test_project_crud_and_frames(client, monkeypatch):
monkeypatch.setattr("routers.projects.get_presigned_url", lambda key, expires=3600: f"http://storage/{key}")
created = client.post("/api/projects", json={
"name": "Demo",
"description": "desc",
"thumbnail_url": "thumb.jpg",
"parse_fps": 12,
})
assert created.status_code == 201
project_id = created.json()["id"]
frame = client.post(f"/api/projects/{project_id}/frames", json={
"project_id": project_id,
"frame_index": 0,
"image_url": "frames/0.jpg",
"width": 640,
"height": 360,
})
assert frame.status_code == 201
frame_id = frame.json()["id"]
listing = client.get("/api/projects")
assert listing.status_code == 200
assert listing.json()[0]["frame_count"] == 1
assert listing.json()[0]["thumbnail_url"] == "http://storage/thumb.jpg"
frames = client.get(f"/api/projects/{project_id}/frames")
assert frames.status_code == 200
assert frames.json()[0]["image_url"] == "http://storage/frames/0.jpg"
single_frame = client.get(f"/api/projects/{project_id}/frames/{frame_id}")
assert single_frame.status_code == 200
assert single_frame.json()["frame_index"] == 0
updated = client.patch(f"/api/projects/{project_id}", json={"name": "Renamed", "status": "ready"})
assert updated.status_code == 200
assert updated.json()["name"] == "Renamed"
assert updated.json()["status"] == "ready"
empty_name = client.patch(f"/api/projects/{project_id}", json={"name": " "})
assert empty_name.status_code == 400
deleted = client.delete(f"/api/projects/{project_id}")
assert deleted.status_code == 204
assert client.get(f"/api/projects/{project_id}").status_code == 404
def test_delete_project_cascades_related_records(client, db_session):
created = client.post("/api/projects", json={"name": "Cascade Delete"})
assert created.status_code == 201
project_id = created.json()["id"]
frame = client.post(f"/api/projects/{project_id}/frames", json={
"project_id": project_id,
"frame_index": 0,
"image_url": "frames/0.jpg",
"width": 640,
"height": 360,
})
assert frame.status_code == 201
frame_id = frame.json()["id"]
annotation = client.post("/api/ai/annotate", json={
"project_id": project_id,
"frame_id": frame_id,
"mask_data": {"polygons": [[[0.1, 0.1], [0.2, 0.1], [0.2, 0.2]]]},
})
assert annotation.status_code == 201
annotation_id = annotation.json()["id"]
db_session.add(Mask(annotation_id=annotation_id, mask_url="masks/1.png"))
db_session.add(ProcessingTask(task_type="parse_video", project_id=project_id, status="queued"))
db_session.commit()
deleted = client.delete(f"/api/projects/{project_id}")
assert deleted.status_code == 204
assert db_session.query(Project).filter(Project.id == project_id).count() == 0
assert db_session.query(Frame).filter(Frame.project_id == project_id).count() == 0
assert db_session.query(Annotation).filter(Annotation.project_id == project_id).count() == 0
assert db_session.query(Mask).count() == 0
assert db_session.query(ProcessingTask).filter(ProcessingTask.project_id == project_id).count() == 0
def test_copy_project_reset_copies_frame_sequence_without_annotations(client, db_session):
created = client.post("/api/projects", json={
"name": "Reset Source",
"description": "desc",
"video_path": "uploads/source.mp4",
"thumbnail_url": "thumb.jpg",
"status": "ready",
"parse_fps": 12,
})
assert created.status_code == 201
project_id = created.json()["id"]
frame = client.post(f"/api/projects/{project_id}/frames", json={
"project_id": project_id,
"frame_index": 0,
"image_url": "frames/source/frame_000000.jpg",
"width": 640,
"height": 360,
"timestamp_ms": 0,
"source_frame_number": 0,
})
assert frame.status_code == 201
annotation = client.post("/api/ai/annotate", json={
"project_id": project_id,
"frame_id": frame.json()["id"],
"mask_data": {"label": "Tumor", "polygons": [[[0.1, 0.1], [0.2, 0.1], [0.2, 0.2]]]},
})
assert annotation.status_code == 201
copied = client.post(f"/api/projects/{project_id}/copy", json={"mode": "reset"})
assert copied.status_code == 201
copied_body = copied.json()
assert copied_body["name"] == "Reset Source 副本"
assert copied_body["frame_count"] == 1
assert copied_body["video_path"] == "uploads/source.mp4"
assert copied_body["parse_fps"] == 12
copied_frames = db_session.query(Frame).filter(Frame.project_id == copied_body["id"]).all()
assert len(copied_frames) == 1
assert copied_frames[0].image_url == "frames/source/frame_000000.jpg"
assert db_session.query(Annotation).filter(Annotation.project_id == copied_body["id"]).count() == 0
def test_copy_project_full_copies_annotations_and_mask_metadata(client, db_session):
created = client.post("/api/projects", json={
"name": "Full Source",
"status": "ready",
})
assert created.status_code == 201
project_id = created.json()["id"]
frame = client.post(f"/api/projects/{project_id}/frames", json={
"project_id": project_id,
"frame_index": 0,
"image_url": "frames/source/frame_000000.jpg",
"width": 640,
"height": 360,
})
assert frame.status_code == 201
frame_id = frame.json()["id"]
annotation = client.post("/api/ai/annotate", json={
"project_id": project_id,
"frame_id": frame_id,
"mask_data": {"label": "Tumor", "polygons": [[[0.1, 0.1], [0.2, 0.1], [0.2, 0.2]]]},
"points": [[0.1, 0.1]],
"bbox": [0.1, 0.1, 0.1, 0.1],
})
assert annotation.status_code == 201
annotation_id = annotation.json()["id"]
db_session.add(Mask(annotation_id=annotation_id, mask_url="masks/source.png", format="png"))
db_session.commit()
copied = client.post(f"/api/projects/{project_id}/copy", json={"mode": "full"})
assert copied.status_code == 201
copied_body = copied.json()
copied_frames = db_session.query(Frame).filter(Frame.project_id == copied_body["id"]).all()
copied_annotations = db_session.query(Annotation).filter(Annotation.project_id == copied_body["id"]).all()
assert copied_body["name"] == "Full Source 副本"
assert len(copied_frames) == 1
assert len(copied_annotations) == 1
assert copied_annotations[0].id != annotation_id
assert copied_annotations[0].frame_id == copied_frames[0].id
assert copied_annotations[0].mask_data["label"] == "Tumor"
assert copied_annotations[0].bbox == [0.1, 0.1, 0.1, 0.1]
assert copied_annotations[0].masks[0].mask_url == "masks/source.png"
def test_project_and_frame_404s(client):
assert client.get("/api/projects/999").status_code == 404
assert client.patch("/api/projects/999", json={"name": "x"}).status_code == 404
assert client.delete("/api/projects/999").status_code == 404
assert client.post("/api/projects/999/copy", json={"mode": "reset"}).status_code == 404
assert client.post("/api/projects/999/frames", json={
"project_id": 999,
"frame_index": 0,
"image_url": "missing.jpg",
}).status_code == 404
assert client.get("/api/projects/999/frames").status_code == 404
assert client.get("/api/projects/999/frames/1").status_code == 404
def test_projects_are_scoped_to_authenticated_owner(client, db_session):
owner_project = client.post("/api/projects", json={"name": "Owner Project"}).json()
other_user = User(
username="other",
password_hash=hash_password("pass"),
role="annotator",
is_active=1,
)
db_session.add(other_user)
db_session.commit()
db_session.refresh(other_user)
other_project = Project(name="Other Project", owner_user_id=other_user.id)
db_session.add(other_project)
db_session.commit()
db_session.refresh(other_project)
listing = client.get("/api/projects")
assert [project["id"] for project in listing.json()] == [owner_project["id"]]
assert client.get(f"/api/projects/{other_project.id}").status_code == 404
original_auth = client.headers["Authorization"]
client.headers.update({"Authorization": f"Bearer {create_access_token(other_user)}"})
try:
other_listing = client.get("/api/projects")
assert [project["id"] for project in other_listing.json()] == [other_project.id]
assert client.get(f"/api/projects/{owner_project['id']}").status_code == 404
finally:
client.headers.update({"Authorization": original_auth})