(self, exception, spec)
| 646 | self.skipTest("PyMongo does not support timeoutMode") |
| 647 | |
| 648 | def process_error(self, exception, spec): |
| 649 | if isinstance(exception, unittest.SkipTest): |
| 650 | raise |
| 651 | is_error = spec.get("isError") |
| 652 | is_client_error = spec.get("isClientError") |
| 653 | is_timeout_error = spec.get("isTimeoutError") |
| 654 | error_contains = spec.get("errorContains") |
| 655 | error_code = spec.get("errorCode") |
| 656 | error_code_name = spec.get("errorCodeName") |
| 657 | error_labels_contain = spec.get("errorLabelsContain") |
| 658 | error_labels_omit = spec.get("errorLabelsOmit") |
| 659 | expect_result = spec.get("expectResult") |
| 660 | error_response = spec.get("errorResponse") |
| 661 | if error_response: |
| 662 | if isinstance(exception, ClientBulkWriteException): |
| 663 | self.match_evaluator.match_result(error_response, exception.error.details) |
| 664 | else: |
| 665 | self.match_evaluator.match_result(error_response, exception.details) |
| 666 | |
| 667 | if is_error: |
| 668 | # already satisfied because exception was raised |
| 669 | pass |
| 670 | |
| 671 | if is_client_error: |
| 672 | if isinstance(exception, ClientBulkWriteException): |
| 673 | error = exception.error |
| 674 | else: |
| 675 | error = exception |
| 676 | # Connection errors are considered client errors. |
| 677 | if isinstance(error, ConnectionFailure): |
| 678 | self.assertNotIsInstance(error, NotPrimaryError) |
| 679 | elif isinstance(error, CorruptGridFile): |
| 680 | pass |
| 681 | elif isinstance(error, (InvalidOperation, ConfigurationError, EncryptionError, NoFile)): |
| 682 | pass |
| 683 | else: |
| 684 | self.assertNotIsInstance(error, PyMongoError) |
| 685 | |
| 686 | if is_timeout_error: |
| 687 | self.assertIsInstance(exception, PyMongoError) |
| 688 | if not exception.timeout: |
| 689 | # Re-raise the exception for better diagnostics. |
| 690 | raise exception |
| 691 | |
| 692 | if error_contains: |
| 693 | if isinstance(exception, BulkWriteError): |
| 694 | errmsg = str(exception.details).lower() |
| 695 | elif isinstance(exception, ClientBulkWriteException): |
| 696 | errmsg = str(exception.details).lower() |
| 697 | else: |
| 698 | errmsg = str(exception).lower() |
| 699 | self.assertIn(error_contains.lower(), errmsg) |
| 700 | |
| 701 | if error_code: |
| 702 | if isinstance(exception, ClientBulkWriteException): |
| 703 | self.assertEqual(error_code, exception.error.details.get("code")) |
| 704 | else: |
| 705 | self.assertEqual(error_code, exception.details.get("code")) |
no test coverage detected