refactor API

Author: stremovsky

databunkerpro/api.py

@@ -350,13 +350,16 @@ def get_user(
         self,
         mode: str,
         identity: str,
+        version: Optional[int] = None,
         request_metadata: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
         """Get user information from DatabunkerPro."""
         data = {
             "mode": mode,
             "identity": identity,
         }
+        if version is not None:
+            data["version"] = version
         return self._make_request("UserGet", data, request_metadata)
 
     def update_user(
@@ -445,6 +448,34 @@ def request_user_deletion(
         }
         return self._make_request("UserDeleteRequest", data, request_metadata)
 
+    def search_user(
+        self,
+        mode: str,
+        identity: str,
+        unlock_uuid: str,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Search for a user with unlock UUID."""
+        data = {
+            "mode": mode,
+            "identity": identity,
+            "unlockuuid": unlock_uuid,
+        }
+        return self._make_request("UserSearch", data, request_metadata)
+
+    def list_user_versions(
+        self,
+        mode: str,
+        identity: str,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """List all versions of a user record."""
+        data = {
+            "mode": mode,
+            "identity": identity,
+        }
+        return self._make_request("UserListVersions", data, request_metadata)
+
     # User Authentication
     def prelogin_user(
         self,
@@ -650,6 +681,21 @@ def list_app_names(
         """List all application names in the system."""
         return self._make_request("AppdataListAppNames", None, request_metadata)
 
+    def list_app_data_versions(
+        self,
+        mode: str,
+        identity: str,
+        appname: str,
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """List all versions of application data for a user."""
+        data = {
+            "mode": mode,
+            "identity": identity,
+            "appname": appname,
+        }
+        return self._make_request("AppdataListVersions", data, request_metadata)
+
     # Legal Basis Management
     def create_legal_basis(
         self,
@@ -1344,19 +1390,32 @@ def bulk_list_unlock(
         return self._make_request("BulkListUnlock", None, request_metadata)
 
     def bulk_list_users(
+        self,
+        unlock_uuid: str,
+        users: List[Dict[str, str]],
+        request_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """List specific users in a bulk operation."""
+        data = {
+            "unlockuuid": unlock_uuid,
+            "users": users,
+        }
+        return self._make_request("BulkListUsers", data, request_metadata)
+
+    def bulk_list_all_users(
         self,
         unlock_uuid: str,
         offset: int = 0,
         limit: int = 10,
         request_metadata: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
-        """List users in a bulk operation."""
+        """List all users in a bulk operation with pagination."""
         data = {
             "unlockuuid": unlock_uuid,
             "offset": offset,
             "limit": limit,
         }
-        return self._make_request("BulkListUsers", data, request_metadata)
+        return self._make_request("BulkListAllUsers", data, request_metadata)
 
     def bulk_list_group_users(
         self,

tests/test_api.py

@@ -162,11 +162,14 @@ def test_bulk_operations_workflow(self):
         """Test complete bulk operations workflow: create users, unlock bulk, and fetch all users."""
         # Step 1: Create 10 users using bulk creation
         users_data = []
+        created_emails = []
         for i in range(10):
+            email = f"bulktest{random.randint(1000, 999999)}@example.com"
+            created_emails.append(email)
             users_data.append(
                 {
                     "profile": {
-                        "email": f"bulktest{random.randint(1000, 999999)}@example.com",
+                        "email": email,
                         "name": f"Bulk Test User {i+1}",
                         "phone": str(random.randint(1000000, 9999999)),
                     }
@@ -183,8 +186,12 @@ def test_bulk_operations_workflow(self):
         self.assertIn("created", create_result)
         self.assertEqual(len(create_result["created"]), 10)
 
-        # Store created user tokens for verification
+        # Store created user tokens and emails for verification
         created_tokens = [user["token"] for user in create_result["created"]]
+        created_user_identities = [
+            {"mode": "email", "identity": user["profile"]["email"]} 
+            for user in create_result["created"]
+        ]
 
         # Step 2: Initiate bulk list unlock operation
         unlock_result = self.api.bulk_list_unlock()
@@ -196,9 +203,28 @@ def test_bulk_operations_workflow(self):
 
         unlock_uuid = unlock_result["unlockuuid"]
 
-        # Step 3: Fetch all users using bulk list operation
-        # First, get the total count by fetching with a large limit
-        bulk_users_result = self.api.bulk_list_users(unlock_uuid, offset=0, limit=100)
+        # Step 3: Test bulk_list_users() with subset of created users
+        # Select first 5 users for subset testing
+        subset_users = created_user_identities[:5]
+        subset_result = self.api.bulk_list_users(unlock_uuid, subset_users)
+        
+        # Verify subset fetch was successful
+        self.assertIsInstance(subset_result, dict)
+        self.assertEqual(subset_result.get("status"), "ok")
+        self.assertIn("rows", subset_result)
+        
+        # Verify we got the expected number of users in subset
+        subset_users_returned = subset_result["rows"]
+        self.assertLessEqual(len(subset_users_returned), 5, "Subset should not exceed requested users")
+        
+        # Verify subset users are from our created set
+        subset_emails = [user.get("profile", {}).get("email") for user in subset_users_returned]
+        for email in subset_emails:
+            if email:  # Only check if email exists
+                self.assertIn(email, created_emails, f"Email {email} should be from our created users")
+
+        # Step 4: Fetch all users using bulk list operation
+        bulk_users_result = self.api.bulk_list_all_users(unlock_uuid, offset=0, limit=100)
 
         # Verify bulk fetch was successful
         self.assertIsInstance(bulk_users_result, dict)
@@ -225,13 +251,13 @@ def test_bulk_operations_workflow(self):
             f"Expected to find 10 created users in bulk results, but found {found_created_users}",
         )
 
-        # Step 4: Test pagination by fetching users in smaller batches
+        # Step 5: Test pagination by fetching users in smaller batches
         paginated_users = []
         offset = 0
         limit = 5
 
         while True:
-            page_result = self.api.bulk_list_users(
+            page_result = self.api.bulk_list_all_users(
                 unlock_uuid, offset=offset, limit=limit
             )
             self.assertEqual(page_result.get("status"), "ok")
@@ -264,6 +290,91 @@ def test_bulk_operations_workflow(self):
 
         return created_tokens
 
+    def test_bulk_list_users_subset(self):
+        """Test bulk_list_users() method to fetch a specific subset of user records."""
+        # Step 1: Create 5 users for testing
+        users_data = []
+        test_emails = []
+        for i in range(5):
+            email = f"subsettest{random.randint(1000, 999999)}@example.com"
+            test_emails.append(email)
+            users_data.append(
+                {
+                    "profile": {
+                        "email": email,
+                        "name": f"Subset Test User {i+1}",
+                        "phone": str(random.randint(1000000, 9999999)),
+                    }
+                }
+            )
+
+        # Create users in bulk
+        create_result = self.api.create_users_bulk(users_data)
+        self.assertEqual(create_result.get("status"), "ok")
+        self.assertEqual(len(create_result["created"]), 5)
+
+        # Store created user data
+        created_tokens = [user["token"] for user in create_result["created"]]
+        created_user_identities = [
+            {"mode": "email", "identity": user["profile"]["email"]} 
+            for user in create_result["created"]
+        ]
+
+        # Step 2: Initiate bulk list unlock operation
+        unlock_result = self.api.bulk_list_unlock()
+        self.assertEqual(unlock_result.get("status"), "ok")
+        unlock_uuid = unlock_result["unlockuuid"]
+
+        # Step 3: Test bulk_list_users with all created users
+        all_users_result = self.api.bulk_list_users(unlock_uuid, created_user_identities)
+        
+        # Verify the result
+        self.assertEqual(all_users_result.get("status"), "ok")
+        self.assertIn("rows", all_users_result)
+        
+        returned_users = all_users_result["rows"]
+        self.assertLessEqual(len(returned_users), 5, "Should not return more users than requested")
+        
+        # Verify all returned users are from our created set
+        returned_emails = [user.get("profile", {}).get("email") for user in returned_users]
+        for email in returned_emails:
+            if email:
+                self.assertIn(email, test_emails, f"Email {email} should be from our test users")
+
+        # Step 4: Test bulk_list_users with subset (first 3 users)
+        subset_identities = created_user_identities[:3]
+        subset_result = self.api.bulk_list_users(unlock_uuid, subset_identities)
+        
+        # Verify subset result
+        self.assertEqual(subset_result.get("status"), "ok")
+        self.assertIn("rows", subset_result)
+        
+        subset_users = subset_result["rows"]
+        self.assertLessEqual(len(subset_users), 3, "Subset should not exceed 3 users")
+        
+        # Verify subset users are from our created set
+        subset_emails = [user.get("profile", {}).get("email") for user in subset_users]
+        for email in subset_emails:
+            if email:
+                self.assertIn(email, test_emails, f"Email {email} should be from our test users")
+
+        # Step 5: Test with empty users list
+        empty_result = self.api.bulk_list_users(unlock_uuid, [])
+        self.assertEqual(empty_result.get("status"), "ok")
+        self.assertIn("rows", empty_result)
+        self.assertEqual(len(empty_result["rows"]), 0, "Empty users list should return empty result")
+
+        # Clean up: Delete the created users
+        for token in created_tokens:
+            try:
+                delete_result = self.api.delete_user("token", token)
+                if delete_result.get("status") != "ok":
+                    print(f"Warning: Failed to delete user with token {token}")
+            except Exception as e:
+                print(f"Warning: Exception during cleanup for token {token}: {str(e)}")
+
+        return created_tokens
+
 
 if __name__ == "__main__":
     unittest.main()