Compare response from bulk.execute() to expected response.
(self, expected, actual)
| 49 | self.coll_w0 = self.coll.with_options(write_concern=WriteConcern(w=0)) |
| 50 | |
| 51 | def assertEqualResponse(self, expected, actual): |
| 52 | """Compare response from bulk.execute() to expected response.""" |
| 53 | for key, value in expected.items(): |
| 54 | if key == "nModified": |
| 55 | self.assertEqual(value, actual["nModified"]) |
| 56 | elif key == "upserted": |
| 57 | expected_upserts = value |
| 58 | actual_upserts = actual["upserted"] |
| 59 | self.assertEqual( |
| 60 | len(expected_upserts), |
| 61 | len(actual_upserts), |
| 62 | 'Expected %d elements in "upserted", got %d' |
| 63 | % (len(expected_upserts), len(actual_upserts)), |
| 64 | ) |
| 65 | |
| 66 | for e, a in zip(expected_upserts, actual_upserts): |
| 67 | self.assertEqualUpsert(e, a) |
| 68 | |
| 69 | elif key == "writeErrors": |
| 70 | expected_errors = value |
| 71 | actual_errors = actual["writeErrors"] |
| 72 | self.assertEqual( |
| 73 | len(expected_errors), |
| 74 | len(actual_errors), |
| 75 | 'Expected %d elements in "writeErrors", got %d' |
| 76 | % (len(expected_errors), len(actual_errors)), |
| 77 | ) |
| 78 | |
| 79 | for e, a in zip(expected_errors, actual_errors): |
| 80 | self.assertEqualWriteError(e, a) |
| 81 | |
| 82 | else: |
| 83 | self.assertEqual( |
| 84 | actual.get(key), |
| 85 | value, |
| 86 | f"{key!r} value of {actual.get(key)!r} does not match expected {value!r}", |
| 87 | ) |
| 88 | |
| 89 | def assertEqualUpsert(self, expected, actual): |
| 90 | """Compare bulk.execute()['upserts'] to expected value. |
no test coverage detected