fix: harden user management operations

tests/test_user_deletion.py

@@ -0,0 +1,99 @@
+import os
+import unittest
+from types import SimpleNamespace
+
+os.environ.setdefault("DATABASE_URL", "sqlite://")
+os.environ.setdefault("APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=")
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import sessionmaker
+
+from web.backend.database import Base
+from web.backend.models import Account, AuditLog, LoginTask, User
+from web.backend.routers.users import delete_user, list_users, rename_user
+from web.backend.schemas import UserRename
+
+
+class UserDeletionTests(unittest.TestCase):
+    def setUp(self):
+        self.engine = create_engine("sqlite://")
+
+        @event.listens_for(self.engine, "connect")
+        def enable_foreign_keys(connection, _):
+            connection.execute("PRAGMA foreign_keys=ON")
+
+        Base.metadata.create_all(self.engine)
+        self.session = sessionmaker(bind=self.engine)()
+        self.admin = User(username="admin", password_hash="hash", role="super_admin")
+        self.user = User(username="support", password_hash="hash", role="support")
+        self.session.add_all([self.admin, self.user])
+        self.session.commit()
+        self.session.add(Account(
+            username="douyu-user",
+            password="password",
+            email="mail@example.com",
+            email_password="mail-password",
+            assigned_to=self.user.id,
+        ))
+        self.session.commit()
+        account = self.session.query(Account).one()
+        self.session.add(LoginTask(
+            batch_id="batch-1",
+            account_id=account.id,
+            created_by=self.user.id,
+        ))
+        self.session.commit()
+
+    def tearDown(self):
+        self.session.close()
+        Base.metadata.drop_all(self.engine)
+        self.engine.dispose()
+
+    def test_delete_user_soft_deletes_and_releases_assigned_accounts(self):
+        result = delete_user(
+            self.user.id,
+            db=self.session,
+            current=SimpleNamespace(id=self.admin.id, username=self.admin.username),
+        )
+
+        deleted = self.session.get(User, self.user.id)
+        account = self.session.query(Account).one()
+        self.assertTrue(result["success"])
+        self.assertEqual(result["released_accounts"], 1)
+        self.assertFalse(deleted.is_active)
+        self.assertIsNotNone(deleted.deleted_at)
+        self.assertEqual(deleted.deleted_username, "support")
+        self.assertTrue(deleted.username.startswith(f"deleted-{deleted.id}-"))
+        self.assertIsNone(account.assigned_to)
+        self.assertEqual(self.session.query(AuditLog).one().target, "support")
+        self.assertEqual(self.session.query(LoginTask).one().created_by, deleted.id)
+        self.assertEqual([item.id for item in list_users(db=self.session)], [self.admin.id])
+
+    def test_cannot_delete_current_user(self):
+        with self.assertRaisesRegex(Exception, "不能删除当前登录用户"):
+            delete_user(
+                self.user.id,
+                db=self.session,
+                current=SimpleNamespace(id=self.user.id, username=self.user.username),
+            )
+
+    def test_rename_preserves_user_id_and_related_data(self):
+        result = rename_user(
+            self.user.id,
+            UserRename(username="renamed-support"),
+            db=self.session,
+            current=SimpleNamespace(id=self.admin.id, username=self.admin.username),
+        )
+
+        renamed = self.session.get(User, self.user.id)
+        account = self.session.query(Account).one()
+        task = self.session.query(LoginTask).one()
+        self.assertEqual(result.username, "renamed-support")
+        self.assertEqual(renamed.id, self.user.id)
+        self.assertEqual(account.assigned_to, self.user.id)
+        self.assertEqual(task.created_by, self.user.id)
+        self.assertEqual(self.session.query(AuditLog).one().target, "support -> renamed-support")
+
+
+if __name__ == "__main__":
+    unittest.main()