Pre_Seg_Server/backend/tests/test_admin.py

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()
    assert [template.name for template in preserved_templates] == ["头颈部CT分割"]
    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