admin
d583b32221
- 登录页和侧栏统一使用根目录 logo_square.png，并更新登录系统名称与副标题。 - 更新 Dashboard、项目库和工作区时间轴文案，移除底层时序视频图层说明。 - 演示视频项目显示名改为“演视LC视频序列”，启动时兼容迁移旧 Data_MyVideo_1 名称，恢复出厂设置使用新名。 - 调整侧栏用户管理入口为用户图标，底部当前用户入口为退出图标，并让退出提示不接收鼠标事件。 - 补充前端组件测试、后端演示重置测试和文档说明。
250 lines
11 KiB
Python
250 lines
11 KiB
Python
from models import Annotation, AuditLog, Frame, Mask, ProcessingTask, Project, Template, User
|
|
from routers.auth import create_access_token, hash_password
|
|
from statuses import PROJECT_STATUS_READY
|
|
|
|
|
|
def test_admin_user_management_and_audit_logs(client, db_session):
|
|
created = client.post("/api/admin/users", json={
|
|
"username": "doctor",
|
|
"password": "secret123",
|
|
"role": "annotator",
|
|
"is_active": True,
|
|
})
|
|
assert created.status_code == 201
|
|
user_id = created.json()["id"]
|
|
assert created.json()["role"] == "annotator"
|
|
|
|
updated = client.patch(f"/api/admin/users/{user_id}", json={
|
|
"password": "newsecret",
|
|
"is_active": False,
|
|
})
|
|
assert updated.status_code == 200
|
|
assert updated.json()["role"] == "annotator"
|
|
assert updated.json()["is_active"] == 0
|
|
|
|
users = client.get("/api/admin/users")
|
|
assert users.status_code == 200
|
|
assert any(user["username"] == "doctor" for user in users.json())
|
|
|
|
deleted = client.delete(f"/api/admin/users/{user_id}")
|
|
assert deleted.status_code == 204
|
|
|
|
logs = client.get("/api/admin/audit-logs")
|
|
assert logs.status_code == 200
|
|
actions = [log["action"] for log in logs.json()]
|
|
assert "admin.user_created" in actions
|
|
assert "admin.user_updated" in actions
|
|
assert "admin.user_deleted" in actions
|
|
|
|
|
|
def test_only_default_admin_role_is_supported(client, db_session):
|
|
extra_admin = client.post("/api/admin/users", json={
|
|
"username": "chief",
|
|
"password": "secret123",
|
|
"role": "admin",
|
|
"is_active": True,
|
|
})
|
|
assert extra_admin.status_code == 400
|
|
|
|
viewer = client.post("/api/admin/users", json={
|
|
"username": "observer",
|
|
"password": "secret123",
|
|
"role": "viewer",
|
|
"is_active": True,
|
|
})
|
|
assert viewer.status_code == 400
|
|
|
|
created = client.post("/api/admin/users", json={
|
|
"username": "doctor",
|
|
"password": "secret123",
|
|
"is_active": True,
|
|
})
|
|
assert created.status_code == 201
|
|
user_id = created.json()["id"]
|
|
assert created.json()["role"] == "annotator"
|
|
assert client.patch(f"/api/admin/users/{user_id}", json={"role": "admin"}).status_code == 400
|
|
assert client.patch(f"/api/admin/users/{user_id}", json={"role": "viewer"}).status_code == 400
|
|
|
|
admin_id = client.get("/api/auth/me").json()["id"]
|
|
assert client.patch(f"/api/admin/users/{admin_id}", json={"role": "annotator"}).status_code == 400
|
|
assert client.patch(f"/api/admin/users/{admin_id}", json={"username": "chief"}).status_code == 400
|
|
|
|
|
|
def test_admin_routes_require_admin_role(client, db_session):
|
|
user = User(username="doctor", password_hash=hash_password("secret123"), role="annotator", is_active=1)
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
db_session.refresh(user)
|
|
original_auth = client.headers["Authorization"]
|
|
client.headers.update({"Authorization": f"Bearer {create_access_token(user)}"})
|
|
try:
|
|
response = client.get("/api/admin/users")
|
|
assert response.status_code == 403
|
|
finally:
|
|
client.headers.update({"Authorization": original_auth})
|
|
|
|
|
|
def test_legacy_viewer_role_is_promoted_to_annotator(client, db_session):
|
|
project = client.post("/api/projects", json={"name": "Readonly Check"}).json()
|
|
user = User(username="readonly", password_hash=hash_password("secret123"), role="viewer", is_active=1)
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
db_session.refresh(user)
|
|
original_auth = client.headers["Authorization"]
|
|
client.headers.update({"Authorization": f"Bearer {create_access_token(user)}"})
|
|
try:
|
|
assert client.get("/api/projects").status_code == 200
|
|
assert client.post("/api/projects", json={"name": "Annotator Project"}).status_code == 201
|
|
assert client.patch(f"/api/projects/{project['id']}", json={"name": "Shared Edit"}).status_code == 200
|
|
assert client.get("/api/auth/me").json()["role"] == "annotator"
|
|
finally:
|
|
client.headers.update({"Authorization": original_auth})
|
|
|
|
|
|
def test_admin_cannot_delete_self_but_can_delete_project_author(client, db_session):
|
|
me = client.get("/api/auth/me").json()
|
|
assert client.delete(f"/api/admin/users/{me['id']}").status_code == 400
|
|
|
|
user = User(username="owner", password_hash=hash_password("secret123"), role="annotator", is_active=1)
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
db_session.refresh(user)
|
|
db_session.add(Project(name="Owned", owner_user_id=user.id))
|
|
db_session.commit()
|
|
|
|
response = client.delete(f"/api/admin/users/{user.id}")
|
|
assert response.status_code == 204
|
|
assert db_session.query(Project).filter(Project.name == "Owned").count() == 1
|
|
|
|
|
|
def test_demo_factory_reset_leaves_admin_and_parsed_demo_dicom(client, db_session, monkeypatch, tmp_path):
|
|
video_path = tmp_path / "Data_MyVideo_1.mp4"
|
|
video_path.write_bytes(b"demo-video")
|
|
monkeypatch.setattr("routers.admin.settings.demo_video_path", str(video_path))
|
|
dicom_dir = tmp_path / "dicom"
|
|
dicom_dir.mkdir()
|
|
for name in ["10.dcm", "2.dcm", "1.dcm"]:
|
|
(dicom_dir / name).write_bytes(name.encode())
|
|
monkeypatch.setattr("routers.admin.settings.demo_dicom_dir", str(dicom_dir))
|
|
|
|
parsed_frame_paths = []
|
|
for idx in range(3):
|
|
frame_path = tmp_path / f"frame_{idx:06d}.jpg"
|
|
frame_path.write_bytes(b"frame")
|
|
parsed_frame_paths.append(str(frame_path))
|
|
|
|
uploaded = []
|
|
monkeypatch.setattr("services.demo_media.upload_file", lambda object_name, data, content_type, length: uploaded.append({
|
|
"object_name": object_name,
|
|
"data": data,
|
|
"content_type": content_type,
|
|
"length": length,
|
|
}))
|
|
monkeypatch.setattr("services.demo_media.parse_dicom", lambda dicom_dir_arg, output_dir: parsed_frame_paths)
|
|
monkeypatch.setattr(
|
|
"services.demo_media.upload_frames_to_minio",
|
|
lambda frame_files, project_id: [f"projects/{project_id}/frames/frame_{idx:06d}.jpg" for idx, _ in enumerate(frame_files)],
|
|
)
|
|
|
|
extra_user = User(username="doctor", password_hash=hash_password("secret123"), role="annotator", is_active=1)
|
|
db_session.add(extra_user)
|
|
db_session.commit()
|
|
db_session.refresh(extra_user)
|
|
old_project = Project(name="Old", owner_user_id=extra_user.id, video_path="uploads/old.mp4")
|
|
db_session.add(old_project)
|
|
db_session.commit()
|
|
db_session.refresh(old_project)
|
|
frame = Frame(project_id=old_project.id, frame_index=0, image_url="frames/old.jpg")
|
|
db_session.add(frame)
|
|
task = ProcessingTask(task_type="parse_video", project_id=old_project.id)
|
|
private_template = Template(
|
|
name="Private",
|
|
description="private",
|
|
color="#fff",
|
|
z_index=1,
|
|
owner_user_id=extra_user.id,
|
|
)
|
|
system_template = Template(
|
|
name="头颈部CT分割",
|
|
description="头颈部CT分割",
|
|
color="#ef4444",
|
|
z_index=10,
|
|
owner_user_id=None,
|
|
mapping_rules={"classes": [{"name": "肿瘤/结节 (Tumor/Nodule)", "color": "#ff0000", "maskId": 1}], "rules": []},
|
|
)
|
|
db_session.add_all([task, private_template, system_template])
|
|
db_session.commit()
|
|
db_session.refresh(frame)
|
|
annotation = Annotation(project_id=old_project.id, frame_id=frame.id, mask_data={"label": "old"})
|
|
db_session.add(annotation)
|
|
db_session.commit()
|
|
db_session.refresh(annotation)
|
|
db_session.add(Mask(annotation_id=annotation.id, mask_url="masks/old.png"))
|
|
db_session.add(AuditLog(actor_user_id=extra_user.id, action="old.audit"))
|
|
db_session.commit()
|
|
|
|
response = client.post("/api/admin/demo-factory-reset", json={"confirmation": "RESET_DEMO_FACTORY"})
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["message"] == "演示环境已恢复出厂设置"
|
|
assert data["admin_user"]["username"] == "admin"
|
|
assert data["project"]["name"] == "演示DICOM序列"
|
|
assert data["project"]["status"] == PROJECT_STATUS_READY
|
|
assert data["project"]["source_type"] == "dicom"
|
|
assert data["project"]["frame_count"] == 3
|
|
assert data["project"]["video_path"] == f"uploads/{data['project']['id']}/dicom"
|
|
assert [project["name"] for project in data["projects"]] == ["演视LC视频序列", "演示DICOM序列"]
|
|
assert data["projects"][0]["status"] == "pending"
|
|
assert data["projects"][0]["source_type"] == "video"
|
|
assert data["projects"][0]["frame_count"] == 0
|
|
assert data["projects"][1]["status"] == PROJECT_STATUS_READY
|
|
assert data["projects"][1]["source_type"] == "dicom"
|
|
assert data["projects"][1]["frame_count"] == 3
|
|
assert [item["object_name"] for item in uploaded] == [
|
|
f"uploads/{data['projects'][0]['id']}/Data_MyVideo_1.mp4",
|
|
f"uploads/{data['project']['id']}/dicom/1.dcm",
|
|
f"uploads/{data['project']['id']}/dicom/2.dcm",
|
|
f"uploads/{data['project']['id']}/dicom/10.dcm",
|
|
]
|
|
assert [item["content_type"] for item in uploaded] == ["video/mp4", "application/dicom", "application/dicom", "application/dicom"]
|
|
|
|
assert [user.username for user in db_session.query(User).all()] == ["admin"]
|
|
assert db_session.query(Project).count() == 2
|
|
assert db_session.query(Frame).count() == 3
|
|
assert [frame.source_frame_number for frame in db_session.query(Frame).order_by(Frame.frame_index).all()] == [0, 1, 2]
|
|
assert db_session.query(Annotation).count() == 0
|
|
assert db_session.query(Mask).count() == 0
|
|
assert db_session.query(ProcessingTask).count() == 0
|
|
assert db_session.query(Template).filter(Template.owner_user_id.is_not(None)).count() == 0
|
|
preserved_templates = db_session.query(Template).filter(Template.owner_user_id.is_(None)).all()
|
|
templates_by_name = {template.name: template for template in preserved_templates}
|
|
assert set(templates_by_name) == {"腹腔镜胆囊切除术", "头颈部CT分割"}
|
|
head_neck_classes = templates_by_name["头颈部CT分割"].mapping_rules["classes"]
|
|
lap_classes = templates_by_name["腹腔镜胆囊切除术"].mapping_rules["classes"]
|
|
assert [item["name"] for item in head_neck_classes] == [
|
|
"肿瘤/结节",
|
|
"下颌骨",
|
|
"甲状腺",
|
|
"气管",
|
|
"颈椎",
|
|
"颈动脉",
|
|
"颈静脉",
|
|
"腮腺",
|
|
"下颌下腺",
|
|
"舌骨",
|
|
"待分类",
|
|
]
|
|
assert [item["maskId"] for item in head_neck_classes] == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 0]
|
|
assert len(lap_classes) == 36
|
|
assert lap_classes[-1]["name"] == "待分类"
|
|
assert lap_classes[-1]["maskId"] == 0
|
|
assert db_session.query(AuditLog).count() == 1
|
|
assert db_session.query(AuditLog).first().action == "admin.demo_factory_reset"
|
|
|
|
|
|
def test_demo_factory_reset_requires_exact_confirmation(client):
|
|
response = client.post("/api/admin/demo-factory-reset", json={"confirmation": "reset"})
|
|
|
|
assert response.status_code == 400
|