admin
2a2e6b9b6c
- 删除项目库右上角独立新建项目入口，保留导入视频/DICOM 自动建项目流程 - 视频项目支持已生成帧后的重新生成帧入口，并提示会清空旧帧、标注和 mask - 后端重新拆帧任务开始前清理旧帧、旧标注和旧 mask，避免重复帧序列 - 项目帧列表接口默认返回完整帧序列，避免工作区总帧数被 1000 条默认 limit 截断 - 增加可选 docker-compose.gpu.yml，并补充 Docker 使用本机 GPU 的前提和启动说明 - 更新项目库、API 映射、恢复演示文案、后端媒体/项目测试和前端文档
245 lines
10 KiB
Python
245 lines
10 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_list_project_frames_returns_more_than_default_timeline_window(client, db_session, monkeypatch):
|
|
monkeypatch.setattr("routers.projects.get_presigned_url", lambda key, expires=3600: f"http://storage/{key}")
|
|
created = client.post("/api/projects", json={"name": "Long Sequence", "status": "ready"})
|
|
assert created.status_code == 201
|
|
project_id = created.json()["id"]
|
|
|
|
db_session.add_all([
|
|
Frame(project_id=project_id, frame_index=index, image_url=f"frames/{index}.jpg")
|
|
for index in range(1001)
|
|
])
|
|
db_session.commit()
|
|
|
|
frames = client.get(f"/api/projects/{project_id}/frames")
|
|
assert frames.status_code == 200
|
|
body = frames.json()
|
|
assert len(body) == 1001
|
|
assert body[-1]["frame_index"] == 1000
|
|
|
|
limited = client.get(f"/api/projects/{project_id}/frames?limit=2")
|
|
assert limited.status_code == 200
|
|
assert [frame["frame_index"] for frame in limited.json()] == [0, 1]
|
|
|
|
|
|
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_shared_between_authenticated_users(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"], other_project.id}
|
|
assert client.get(f"/api/projects/{other_project.id}").status_code == 200
|
|
|
|
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()} == {owner_project["id"], other_project.id}
|
|
assert client.get(f"/api/projects/{owner_project['id']}").status_code == 200
|
|
renamed = client.patch(f"/api/projects/{owner_project['id']}", json={"name": "Edited By Other"})
|
|
assert renamed.status_code == 200
|
|
assert renamed.json()["name"] == "Edited By Other"
|
|
finally:
|
|
client.headers.update({"Authorization": original_auth})
|
|
|
|
assert client.get(f"/api/projects/{owner_project['id']}").json()["name"] == "Edited By Other"
|