(spec, arguments, opname, entity_map, with_txn_callback)
| 643 | |
| 644 | |
| 645 | def prepare_spec_arguments(spec, arguments, opname, entity_map, with_txn_callback): |
| 646 | for arg_name in list(arguments): |
| 647 | c2s = camel_to_snake(arg_name) |
| 648 | # Named "key" instead not fieldName. |
| 649 | if arg_name == "fieldName": |
| 650 | arguments["key"] = arguments.pop(arg_name) |
| 651 | # Aggregate uses "batchSize", while find uses batch_size. |
| 652 | elif (arg_name == "batchSize" or arg_name == "allowDiskUse") and opname == "aggregate": |
| 653 | continue |
| 654 | elif arg_name == "bypassDocumentValidation" and ( |
| 655 | opname == "aggregate" or "find_one_and" in opname |
| 656 | ): |
| 657 | continue |
| 658 | elif arg_name == "timeoutMode": |
| 659 | raise unittest.SkipTest("PyMongo does not support timeoutMode") |
| 660 | # Requires boolean returnDocument. |
| 661 | elif arg_name == "returnDocument": |
| 662 | arguments[c2s] = getattr(ReturnDocument, arguments.pop(arg_name).upper()) |
| 663 | elif "bulk_write" in opname and (c2s == "requests" or c2s == "models"): |
| 664 | # Parse each request into a bulk write model. |
| 665 | requests = [] |
| 666 | for request in arguments[c2s]: |
| 667 | if "name" in request: |
| 668 | # CRUD v2 format |
| 669 | bulk_model = camel_to_upper_camel(request["name"]) |
| 670 | bulk_class = getattr(operations, bulk_model) |
| 671 | bulk_arguments = camel_to_snake_args(request["arguments"]) |
| 672 | else: |
| 673 | # Unified test format |
| 674 | bulk_model, spec = next(iter(request.items())) |
| 675 | bulk_class = getattr(operations, camel_to_upper_camel(bulk_model)) |
| 676 | bulk_arguments = camel_to_snake_args(spec) |
| 677 | requests.append(bulk_class(**dict(bulk_arguments))) |
| 678 | arguments[c2s] = requests |
| 679 | elif arg_name == "session": |
| 680 | arguments["session"] = entity_map[arguments["session"]] |
| 681 | elif opname == "open_download_stream" and arg_name == "id": |
| 682 | arguments["file_id"] = arguments.pop(arg_name) |
| 683 | elif opname not in ("find", "find_one") and c2s == "max_time_ms": |
| 684 | # find is the only method that accepts snake_case max_time_ms. |
| 685 | # All other methods take kwargs which must use the server's |
| 686 | # camelCase maxTimeMS. See PYTHON-1855. |
| 687 | arguments["maxTimeMS"] = arguments.pop("max_time_ms") |
| 688 | elif opname == "with_transaction" and arg_name == "callback": |
| 689 | if "operations" in arguments[arg_name]: |
| 690 | # CRUD v2 format |
| 691 | callback_ops = arguments[arg_name]["operations"] |
| 692 | else: |
| 693 | # Unified test format |
| 694 | callback_ops = arguments[arg_name] |
| 695 | arguments["callback"] = lambda _: with_txn_callback(copy.deepcopy(callback_ops)) |
| 696 | elif opname == "drop_collection" and arg_name == "collection": |
| 697 | arguments["name_or_collection"] = arguments.pop(arg_name) |
| 698 | elif opname == "create_collection": |
| 699 | if arg_name == "collection": |
| 700 | arguments["name"] = arguments.pop(arg_name) |
| 701 | arguments["check_exists"] = False |
| 702 | # Any other arguments to create_collection are passed through |
no test coverage detected