(expected_dict, actual)
| 76 | |
| 77 | |
| 78 | def compare_events(expected_dict, actual): |
| 79 | if not expected_dict: |
| 80 | return False, "Error: Bad expected value in YAML test" |
| 81 | if not actual: |
| 82 | return False, "Error: Event published was None" |
| 83 | |
| 84 | expected_type, expected = list(expected_dict.items())[0] |
| 85 | |
| 86 | if expected_type == "server_opening_event": |
| 87 | if not isinstance(actual, monitoring.ServerOpeningEvent): |
| 88 | return False, "Expected ServerOpeningEvent, got %s" % (actual.__class__) |
| 89 | if expected["address"] != "{}:{}".format(*actual.server_address): |
| 90 | return ( |
| 91 | False, |
| 92 | "ServerOpeningEvent published with wrong address (expected" " {}, got {}".format( |
| 93 | expected["address"], actual.server_address |
| 94 | ), |
| 95 | ) |
| 96 | |
| 97 | elif expected_type == "server_description_changed_event": |
| 98 | if not isinstance(actual, monitoring.ServerDescriptionChangedEvent): |
| 99 | return (False, "Expected ServerDescriptionChangedEvent, got %s" % (actual.__class__)) |
| 100 | if expected["address"] != "{}:{}".format(*actual.server_address): |
| 101 | return ( |
| 102 | False, |
| 103 | "ServerDescriptionChangedEvent has wrong address" " (expected {}, got {}".format( |
| 104 | expected["address"], actual.server_address |
| 105 | ), |
| 106 | ) |
| 107 | |
| 108 | if not compare_server_descriptions(expected["newDescription"], actual.new_description): |
| 109 | return (False, "New ServerDescription incorrect in ServerDescriptionChangedEvent") |
| 110 | if not compare_server_descriptions( |
| 111 | expected["previousDescription"], actual.previous_description |
| 112 | ): |
| 113 | return ( |
| 114 | False, |
| 115 | "Previous ServerDescription incorrect in ServerDescriptionChangedEvent", |
| 116 | ) |
| 117 | |
| 118 | elif expected_type == "server_closed_event": |
| 119 | if not isinstance(actual, monitoring.ServerClosedEvent): |
| 120 | return False, "Expected ServerClosedEvent, got %s" % (actual.__class__) |
| 121 | if expected["address"] != "{}:{}".format(*actual.server_address): |
| 122 | return ( |
| 123 | False, |
| 124 | "ServerClosedEvent published with wrong address" " (expected {}, got {}".format( |
| 125 | expected["address"], actual.server_address |
| 126 | ), |
| 127 | ) |
| 128 | |
| 129 | elif expected_type == "topology_opening_event": |
| 130 | if not isinstance(actual, monitoring.TopologyOpenedEvent): |
| 131 | return False, "Expected TopologyOpenedEvent, got %s" % (actual.__class__) |
| 132 | |
| 133 | elif expected_type == "topology_description_changed_event": |
| 134 | if not isinstance(actual, monitoring.TopologyDescriptionChangedEvent): |
| 135 | return ( |
no test coverage detected