admin
739953bc13
- 侧边栏 Logo 改为导入根目录 logo_square.png，favicon 也切换为 /logo_square.png，并让前端服务显式提供该根目录图片。 - 头颈部CT分割默认模板分类名改为纯中文，去掉括号英文翻译，颜色和 maskid 保持用户给定顺序。 - 增加旧版头颈部CT英文括号 label 的窄迁移，启动 seed 时自动把旧默认系统模板更新为纯中文默认。 - 更新前端 Logo 测试、后端默认模板和恢复出厂设置测试，覆盖纯中文分类和根目录方形 Logo。 - 更新 AGENTS、README、前端审计、需求冻结和测试计划文档，记录根目录 Logo 和头颈部CT纯中文默认分类。
216 lines
9.4 KiB
Python
216 lines
9.4 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"]
|
|
|
|
updated = client.patch(f"/api/admin/users/{user_id}", json={
|
|
"role": "viewer",
|
|
"password": "newsecret",
|
|
"is_active": False,
|
|
})
|
|
assert updated.status_code == 200
|
|
assert updated.json()["role"] == "viewer"
|
|
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_admin_routes_require_admin_role(client, db_session):
|
|
user = User(username="viewer", 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:
|
|
response = client.get("/api/admin/users")
|
|
assert response.status_code == 403
|
|
finally:
|
|
client.headers.update({"Authorization": original_auth})
|
|
|
|
|
|
def test_viewer_role_is_read_only_for_business_mutations(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": "Nope"}).status_code == 403
|
|
assert client.patch(f"/api/projects/{project['id']}", json={"name": "Nope"}).status_code == 403
|
|
assert client.post("/api/ai/annotate", json={"project_id": project["id"]}).status_code == 403
|
|
finally:
|
|
client.headers.update({"Authorization": original_auth})
|
|
|
|
|
|
def test_admin_cannot_delete_self_or_user_with_projects(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 == 409
|
|
|
|
|
|
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"]] == ["Data_MyVideo_1", "演示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
|