test bulk creation

Author: stremovsky

databunkerpro/api.py

@@ -107,6 +107,52 @@ def create_user(
             data.update(options)
         return self._make_request("UserCreate", "POST", data, request_metadata)
 
+    def create_users_bulk(
+        self,
+        records: List[Dict[str, Any]],
+        options: Optional[Dict[str, Any]] = None,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Create multiple users in bulk with their profiles and group information.
+
+        Args:
+            records (List[Dict[str, Any]]): List of user records to create, each containing:
+                - profile (Dict[str, Any]): User profile data
+                - groupname (str | int, optional): Group name or ID
+                - groupid (int, optional): Group ID
+                - rolename (str | int, optional): Role name or ID
+                - roleid (int, optional): Role ID
+            options (Dict[str, Any], optional): Global options for all users:
+                - finaltime (str, optional): Global expiration time for all users
+                - slidingtime (str, optional): Global sliding time period for all users
+            request_metadata (Dict[str, Any], optional): Additional metadata for the request
+
+        Returns:
+            Dict[str, Any]: The API response containing information about created users
+
+        Example:
+            >>> api.create_users_bulk([
+            ...     {
+            ...         "profile": {"email": "user1@example.com", "name": "User One"},
+            ...         "groupname": "premium",
+            ...         "rolename": "admin"
+            ...     },
+            ...     {
+            ...         "profile": {"email": "user2@example.com", "name": "User Two"},
+            ...         "groupid": 123,
+            ...         "rolename": "team-member"
+            ...     }
+            ... ], {
+            ...     "finaltime": "2024-12-31",
+            ...     "slidingtime": "30d"
+            ... })
+        """
+        data = {"records": records}
+        if options:
+            data.update(options)
+        return self._make_request("UserCreateBulk", "POST", data, request_metadata)
+
     def get_user(
         self,
         mode: str,

tests/test_api.py

@@ -73,6 +73,46 @@ def test_create_user(self):
         self.assertIn("token", result)
         return user_data["email"]
 
+    def test_create_users_bulk(self):
+        """Test bulk user creation."""
+        # Create test data for multiple users
+        users_data = [
+            {
+                "profile": {
+                    "email": f"test{random.randint(1000, 999999)}@example.com",
+                    "name": f"Test User {random.randint(1000, 999999)}",
+                    "phone": str(random.randint(1000, 999999)),
+                },
+                # "groupname": "test-group",
+                # "rolename": "test-role"
+            },
+            {
+                "profile": {
+                    "email": f"test{random.randint(1000, 999999)}@example.com",
+                    "name": f"Test User {random.randint(1000, 999999)}",
+                    "phone": str(random.randint(1000, 999999)),
+                },
+                # "groupid": 1,
+                # "roleid": 1
+            }
+        ]
+        # Test bulk creation with global options
+        options = {
+            "finaltime": "1y",
+            "slidingtime": "30d"
+        }
+        result = self.api.create_users_bulk(users_data, options)
+        self.assertIsInstance(result, dict)
+        self.assertEqual(result.get("status"), "ok")
+        self.assertIn("created", result)
+        self.assertEqual(len(result["created"]), len(users_data))
+        # Verify each created user
+        for user_record in result["created"]:
+            self.assertIn("token", user_record)
+            self.assertIn("profile", user_record)
+            self.assertEqual(user_record["profile"]["email"], user_record["profile"]["email"])
+        return [user["profile"]["email"] for user in users_data]
+
     def test_get_user(self):
         """Test user retrieval."""
         email = self.test_create_user()