(self, expectation, actual)
| 609 | self.test.assertEqual(getattr(actual, field), expected) |
| 610 | |
| 611 | def match_event(self, expectation, actual): |
| 612 | name, spec = next(iter(expectation.items())) |
| 613 | if name == "commandStartedEvent": |
| 614 | self.test.assertIsInstance(actual, CommandStartedEvent) |
| 615 | elif name == "commandSucceededEvent": |
| 616 | self.test.assertIsInstance(actual, CommandSucceededEvent) |
| 617 | elif name == "commandFailedEvent": |
| 618 | self.test.assertIsInstance(actual, CommandFailedEvent) |
| 619 | elif name == "poolCreatedEvent": |
| 620 | self.test.assertIsInstance(actual, PoolCreatedEvent) |
| 621 | elif name == "poolReadyEvent": |
| 622 | self.test.assertIsInstance(actual, PoolReadyEvent) |
| 623 | elif name == "poolClearedEvent": |
| 624 | self.test.assertIsInstance(actual, PoolClearedEvent) |
| 625 | self.test.assertIsInstance(actual.interrupt_connections, bool) |
| 626 | elif name == "poolClosedEvent": |
| 627 | self.test.assertIsInstance(actual, PoolClosedEvent) |
| 628 | elif name == "connectionCreatedEvent": |
| 629 | self.test.assertIsInstance(actual, ConnectionCreatedEvent) |
| 630 | elif name == "connectionReadyEvent": |
| 631 | self.test.assertIsInstance(actual, ConnectionReadyEvent) |
| 632 | elif name == "connectionClosedEvent": |
| 633 | self.test.assertIsInstance(actual, ConnectionClosedEvent) |
| 634 | elif name == "connectionCheckOutStartedEvent": |
| 635 | self.test.assertIsInstance(actual, ConnectionCheckOutStartedEvent) |
| 636 | elif name == "connectionCheckOutFailedEvent": |
| 637 | self.test.assertIsInstance(actual, ConnectionCheckOutFailedEvent) |
| 638 | elif name == "connectionCheckedOutEvent": |
| 639 | self.test.assertIsInstance(actual, ConnectionCheckedOutEvent) |
| 640 | elif name == "connectionCheckedInEvent": |
| 641 | self.test.assertIsInstance(actual, ConnectionCheckedInEvent) |
| 642 | elif name == "serverDescriptionChangedEvent": |
| 643 | self.test.assertIsInstance(actual, ServerDescriptionChangedEvent) |
| 644 | elif name == "serverHeartbeatStartedEvent": |
| 645 | self.test.assertIsInstance(actual, ServerHeartbeatStartedEvent) |
| 646 | elif name == "serverHeartbeatSucceededEvent": |
| 647 | self.test.assertIsInstance(actual, ServerHeartbeatSucceededEvent) |
| 648 | elif name == "serverHeartbeatFailedEvent": |
| 649 | self.test.assertIsInstance(actual, ServerHeartbeatFailedEvent) |
| 650 | elif name == "topologyDescriptionChangedEvent": |
| 651 | self.test.assertIsInstance(actual, TopologyDescriptionChangedEvent) |
| 652 | elif name == "topologyOpeningEvent": |
| 653 | self.test.assertIsInstance(actual, TopologyOpenedEvent) |
| 654 | elif name == "topologyClosedEvent": |
| 655 | self.test.assertIsInstance(actual, TopologyClosedEvent) |
| 656 | else: |
| 657 | raise Exception(f"Unsupported event type {name}") |
| 658 | |
| 659 | self.match_event_fields(actual, spec) |
| 660 | |
| 661 | |
| 662 | def coerce_result(opname, result): |
no test coverage detected