Function _run_scenario

test/asynchronous/test_sdam_monitoring_spec.py:191–251

(self)

            await _run_scenario(self)

    async def _run_scenario(self):
        class NoopMonitor(Monitor):
            """Override the _run method to do nothing."""

            async def _run(self):
                await asyncio.sleep(0.05)

        m = AsyncMongoClient(
            host=scenario_def["uri"],
            port=27017,
            event_listeners=[self.all_listener],
            _monitor_class=NoopMonitor,
        )
        topology = await m._get_topology()

        try:
            for phase in scenario_def["phases"]:
                for source, response in phase.get("responses", []):
                    source_address = clean_node(source)
                    await topology.on_change(
                        ServerDescription(
                            address=source_address, hello=Hello(response), round_trip_time=0
                        )
                    )

                expected_results = phase["outcome"]["events"]
                expected_len = len(expected_results)
                await async_wait_until(
                    lambda: len(self.all_listener.results) >= expected_len,
                    "publish all events",
                    timeout=15,
                )

                # Wait some time to catch possible lagging extra events.
                await async_wait_until(lambda: topology._events.empty(), "publish lagging events")

                i = 0
                while i < expected_len:
                    result = (
                        self.all_listener.results[i] if len(self.all_listener.results) > i else None
                    )
                    # The order of ServerOpening/ClosedEvents doesn't matter
                    if isinstance(
                        result, (monitoring.ServerOpeningEvent, monitoring.ServerClosedEvent)
                    ):
                        i, passed, message = compare_multiple_events(
                            i, expected_results, self.all_listener.results
                        )
                        self.assertTrue(passed, message)
                    else:
                        self.assertTrue(*compare_events(expected_results[i], result))
                        i += 1

                # Assert no extra events.
                extra_events = self.all_listener.results[expected_len:]
                if extra_events:
                    self.fail(f"Extra events {extra_events!r}")