feat: 统一生产初始化流程并保护系统管理员

backend/tests/test_protected_admin.py

@@ -0,0 +1,42 @@
+import pytest
+
+from app.crud import user as user_crud
+from app.models.user import UserRole, UserStatus
+
+
+class _ScalarResult:
+    def __init__(self, value):
+        self._value = value
+
+    def scalar_one_or_none(self):
+        return self._value
+
+
+class _FakeSession:
+    def __init__(self, existing_admin=None):
+        self.existing_admin = existing_admin
+        self.added = []
+        self.commit_count = 0
+
+    async def execute(self, _stmt):
+        return _ScalarResult(self.existing_admin)
+
+    def add(self, obj):
+        self.added.append(obj)
+
+    async def commit(self):
+        self.commit_count += 1
+
+
+@pytest.mark.asyncio
+async def test_ensure_admin_exists_creates_protected_admin():
+    session = _FakeSession()
+
+    await user_crud.ensure_admin_exists(session)
+
+    assert len(session.added) == 1
+    admin = session.added[0]
+    assert admin.email == "admin@huapont.cn"
+    assert admin.role == UserRole.ADMIN
+    assert admin.status == UserStatus.ACTIVE
+    assert session.commit_count == 1

backend/tests/test_protected_admin_api.py

@@ -0,0 +1,107 @@
+import uuid
+
+import pytest
+from fastapi import HTTPException
+
+from app.api.v1 import users as users_api
+from app.models.user import User, UserRole, UserStatus
+from app.schemas.user import UserRead, UserUpdate
+
+
+def _make_user(
+    *,
+    email: str,
+    role: UserRole = UserRole.ADMIN,
+    status: UserStatus = UserStatus.ACTIVE,
+) -> User:
+    return User(
+        id=uuid.uuid4(),
+        email=email,
+        password_hash="hashed",
+        full_name="User",
+        department="SYSTEM",
+        role=role,
+        status=status,
+    )
+
+
+@pytest.mark.asyncio
+async def test_delete_user_blocks_protected_admin(monkeypatch):
+    current_user = _make_user(email="other-admin@huapont.cn")
+    protected_admin = _make_user(email="admin@huapont.cn")
+
+    async def fake_get_by_id(_db, _user_id):
+        return protected_admin
+
+    async def fake_count_active_admins(_db):
+        return 2
+
+    async def fake_user_has_memberships(_db, _user_id):
+        return False
+
+    async def fake_delete_user(_db, _user):
+        raise AssertionError("protected admin should not be deleted")
+
+    monkeypatch.setattr(users_api.user_crud, "get_by_id", fake_get_by_id)
+    monkeypatch.setattr(users_api.user_crud, "count_active_admins", fake_count_active_admins)
+    monkeypatch.setattr(users_api.member_crud, "user_has_memberships", fake_user_has_memberships)
+    monkeypatch.setattr(users_api.user_crud, "delete_user", fake_delete_user)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await users_api.delete_user(protected_admin.id, db=object(), current_user=current_user)
+
+    assert exc_info.value.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_update_user_blocks_protected_admin_email_change(monkeypatch):
+    protected_admin = _make_user(email="admin@huapont.cn")
+
+    async def fake_get_by_id(_db, _user_id):
+        return protected_admin
+
+    async def fake_count_active_admins(_db):
+        return 2
+
+    async def fake_update_user(_db, _user, _payload):
+        raise AssertionError("protected admin email should not be changed")
+
+    monkeypatch.setattr(users_api.user_crud, "get_by_id", fake_get_by_id)
+    monkeypatch.setattr(users_api.user_crud, "count_active_admins", fake_count_active_admins)
+    monkeypatch.setattr(users_api.user_crud, "update_user", fake_update_user)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await users_api.update_user(
+            protected_admin.id,
+            UserUpdate(email="renamed@huapont.cn"),
+            db=object(),
+            current_user=protected_admin,
+        )
+
+    assert exc_info.value.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_update_user_allows_protected_admin_password_change(monkeypatch):
+    protected_admin = _make_user(email="admin@huapont.cn")
+    captured_payload = {}
+
+    async def fake_get_by_id(_db, _user_id):
+        return protected_admin
+
+    async def fake_update_user(_db, _user, payload):
+        captured_payload["password"] = payload.password
+        return _user
+
+    monkeypatch.setattr(users_api.user_crud, "get_by_id", fake_get_by_id)
+    monkeypatch.setattr(users_api.user_crud, "update_user", fake_update_user)
+
+    result = await users_api.update_user(
+        protected_admin.id,
+        UserUpdate(password="Password123"),
+        db=object(),
+        current_user=protected_admin,
+    )
+
+    assert isinstance(result, UserRead | User)
+    assert captured_payload["password"] == "Password123"

backend/tests/test_study_delete.py

@@ -0,0 +1,31 @@
+import uuid
+
+import pytest
+
+from app.crud import study as study_crud
+
+
+class _FakeDb:
+    def __init__(self):
+        self.tables = []
+        self.committed = False
+
+    async def execute(self, stmt):
+        table = getattr(getattr(stmt, "table", None), "name", None)
+        if table:
+            self.tables.append(table)
+
+    async def commit(self):
+        self.committed = True
+
+
+@pytest.mark.asyncio
+async def test_delete_study_removes_contract_fee_payments_before_contract_fees():
+    db = _FakeDb()
+
+    await study_crud.delete(db, uuid.uuid4())
+
+    assert "contract_fee_payments" in db.tables
+    assert "contract_fees" in db.tables
+    assert db.tables.index("contract_fee_payments") < db.tables.index("contract_fees")
+    assert db.committed is True