def test_moving_average(self):
     avg = MovingAverage([10])
     self.assertEqual(10, avg.get())
     avg2 = avg.clone_with(20)
     self.assertEqual(15, avg2.get())
     avg3 = avg2.clone_with(30)
     self.assertEqual(20, avg3.get())
     avg4 = avg3.clone_with(-100)
     self.assertEqual((10 + 20 + 30 - 100) / 4., avg4.get())
     avg5 = avg4.clone_with(17)
     self.assertEqual((10 + 20 + 30 - 100 + 17) / 5., avg5.get())
     avg6 = avg5.clone_with(43)
     self.assertEqual((20 + 30 - 100 + 17 + 43) / 5., avg6.get())
     avg7 = avg6.clone_with(-1111)
     self.assertEqual((30 - 100 + 17 + 43 - 1111) / 5., avg7.get())

    def run_scenario(self):
        moving_average = MovingAverage()

        if scenario_def['avg_rtt_ms'] != "NULL":
            moving_average.add_sample(scenario_def['avg_rtt_ms'])

        if scenario_def['new_rtt_ms'] != "NULL":
            moving_average.add_sample(scenario_def['new_rtt_ms'])

        self.assertAlmostEqual(moving_average.get(),
                               scenario_def['new_avg_rtt'])

    def __init__(
            self,
            server_description,
            topology,
            pool,
            topology_settings):
        """Class to monitor a MongoDB server on a background thread.

        Pass an initial ServerDescription, a Topology, a Pool, and
        TopologySettings.

        The Topology is weakly referenced. The Pool must be exclusive to this
        Monitor.
        """
        self._server_description = server_description
        self._pool = pool
        self._settings = topology_settings
        self._avg_round_trip_time = MovingAverage()
        self._listeners = self._settings._pool_options.event_listeners
        pub = self._listeners is not None
        self._publish = pub and self._listeners.enabled_for_server_heartbeat

        # We strongly reference the executor and it weakly references us via
        # this closure. When the monitor is freed, stop the executor soon.
        def target():
            monitor = self_ref()
            if monitor is None:
                return False  # Stop the executor.
            Monitor._run(monitor)
            return True

        executor = periodic_executor.PeriodicExecutor(
            interval=self._settings.heartbeat_frequency,
            min_interval=common.MIN_HEARTBEAT_INTERVAL,
            target=target,
            name="pymongo_server_monitor_thread")

        self._executor = executor

        # Avoid cycles. When self or topology is freed, stop executor soon.
        self_ref = weakref.ref(self, executor.close)
        self._topology = weakref.proxy(topology, executor.close)

    def __init__(
            self,
            server_description,
            topology,
            pool,
            topology_settings):
        """Class to monitor a MongoDB server on a background thread.

        Pass an initial ServerDescription, a Topology, a Pool, and
        TopologySettings.

        The Topology is weakly referenced. The Pool must be exclusive to this
        Monitor.
        """
        self._server_description = server_description

        # A weakref callback, takes ref to the freed topology as its parameter.
        def close(dummy):
            self.close()

        self._topology = weakref.proxy(topology, close)
        self._pool = pool
        self._settings = topology_settings
        self._avg_round_trip_time = MovingAverage()

        # We strongly reference the executor and it weakly references us via
        # this closure. When the monitor is freed, stop the executor soon.
        self_ref = weakref.ref(self, close)

        def target():
            monitor = self_ref()
            if monitor is None:
                return False  # Stop the executor.
            Monitor._run(monitor)
            return True

        self._executor = periodic_executor.PeriodicExecutor(
            condition_class=self._settings.condition_class,
            interval=common.HEARTBEAT_FREQUENCY,
            min_interval=common.MIN_HEARTBEAT_INTERVAL,
            target=target)

 def test_5_sample_moving_average(self):
     avg = MovingAverage(5)
     self.assertEqual(None, avg.get())
     avg.update(10)
     self.assertEqual(10, avg.get())
     avg.update(20)
     self.assertEqual(15, avg.get())
     avg.update(30)
     self.assertEqual(20, avg.get())
     avg.update(-100)
     self.assertEqual((10 + 20 + 30 - 100) / 4, avg.get())
     avg.update(17)
     self.assertEqual((10 + 20 + 30 - 100 + 17) / 5., avg.get())
     avg.update(43)
     self.assertEqual((20 + 30 - 100 + 17 + 43) / 5., avg.get())
     avg.update(-1111)
     self.assertEqual((30 - 100 + 17 + 43 - 1111) / 5., avg.get())

 def test_2_sample_moving_average(self):
     avg = MovingAverage(2)
     self.assertEqual(None, avg.get())
     avg.update(10)
     self.assertEqual(10, avg.get())
     avg.update(20)
     self.assertEqual(15, avg.get())
     avg.update(30)
     self.assertEqual(25, avg.get())
     avg.update(-100)
     self.assertEqual(-35, avg.get())

 def test_trivial_moving_average(self):
     avg = MovingAverage(1)
     self.assertEqual(None, avg.get())
     avg.update(10)
     self.assertEqual(10, avg.get())
     avg.update(20)
     self.assertEqual(20, avg.get())
     avg.update(0)
     self.assertEqual(0, avg.get())

 def test_empty_moving_average(self):
     avg = MovingAverage(0)
     self.assertEqual(None, avg.get())
     avg.update(10)
     self.assertEqual(None, avg.get())

class Monitor(object):
    def __init__(
            self,
            server_description,
            topology,
            pool,
            topology_settings):
        """Class to monitor a MongoDB server on a background thread.

        Pass an initial ServerDescription, a Topology, a Pool, and
        TopologySettings.

        The Topology is weakly referenced. The Pool must be exclusive to this
        Monitor.
        """
        self._server_description = server_description
        self._pool = pool
        self._settings = topology_settings
        self._avg_round_trip_time = MovingAverage()

        # We strongly reference the executor and it weakly references us via
        # this closure. When the monitor is freed, stop the executor soon.
        def target():
            monitor = self_ref()
            if monitor is None:
                return False  # Stop the executor.
            Monitor._run(monitor)
            return True

        executor = periodic_executor.PeriodicExecutor(
            interval=common.HEARTBEAT_FREQUENCY,
            min_interval=common.MIN_HEARTBEAT_INTERVAL,
            target=target,
            name="pymongo_server_monitor_thread")

        self._executor = executor
        
        # Avoid cycles. When self or topology is freed, stop executor soon.
        self_ref = weakref.ref(self, executor.close)
        self._topology = weakref.proxy(topology, executor.close)

    def open(self):
        """Start monitoring, or restart after a fork.

        Multiple calls have no effect.
        """
        self._executor.open()

    def close(self):
        """Close and stop monitoring.

        open() restarts the monitor after closing.
        """
        self._executor.close()

        # Increment the pool_id and maybe close the socket. If the executor
        # thread has the socket checked out, it will be closed when checked in.
        self._pool.reset()

    def join(self, timeout=None):
        self._executor.join(timeout)

    def request_check(self):
        """If the monitor is sleeping, wake and check the server soon."""
        self._executor.wake()

    def _run(self):
        try:
            self._server_description = self._check_with_retry()
            self._topology.on_change(self._server_description)
        except ReferenceError:
            # Topology was garbage-collected.
            self.close()

    def _check_with_retry(self):
        """Call ismaster once or twice. Reset server's pool on error.

        Returns a ServerDescription.
        """
        # According to the spec, if an ismaster call fails we reset the
        # server's pool. If a server was once connected, change its type
        # to Unknown only after retrying once.
        address = self._server_description.address
        retry = self._server_description.server_type != SERVER_TYPE.Unknown

        try:
            return self._check_once()
        except ReferenceError:
            raise
        except Exception as error:
            self._topology.reset_pool(address)
            default = ServerDescription(address, error=error)
            if not retry:
                self._avg_round_trip_time.reset()
                # Server type defaults to Unknown.
                return default

            # Try a second and final time. If it fails return original error.
            try:
                return self._check_once()
#.........这里部分代码省略.........

 def test_moving_average(self):
     avg = MovingAverage()
     self.assertIsNone(avg.get())
     avg.add_sample(10)
     self.assertAlmostEqual(10, avg.get())
     avg.add_sample(20)
     self.assertAlmostEqual(12, avg.get())
     avg.add_sample(30)
     self.assertAlmostEqual(15.6, avg.get())

class Monitor(object):
    def __init__(
            self,
            server_description,
            topology,
            pool,
            topology_settings):
        """Class to monitor a MongoDB server on a background thread.

        Pass an initial ServerDescription, a Topology, a Pool, and
        TopologySettings.

        The Topology is weakly referenced. The Pool must be exclusive to this
        Monitor.
        """
        self._server_description = server_description
        self._pool = pool
        self._settings = topology_settings
        self._avg_round_trip_time = MovingAverage()
        self._listeners = self._settings._pool_options.event_listeners
        pub = self._listeners is not None
        self._publish = pub and self._listeners.enabled_for_server_heartbeat

        # We strongly reference the executor and it weakly references us via
        # this closure. When the monitor is freed, stop the executor soon.
        def target():
            monitor = self_ref()
            if monitor is None:
                return False  # Stop the executor.
            Monitor._run(monitor)
            return True

        executor = periodic_executor.PeriodicExecutor(
            interval=self._settings.heartbeat_frequency,
            min_interval=common.MIN_HEARTBEAT_INTERVAL,
            target=target,
            name="pymongo_server_monitor_thread")

        self._executor = executor

        # Avoid cycles. When self or topology is freed, stop executor soon.
        self_ref = weakref.ref(self, executor.close)
        self._topology = weakref.proxy(topology, executor.close)

    def open(self):
        """Start monitoring, or restart after a fork.

        Multiple calls have no effect.
        """
        self._executor.open()

    def close(self):
        """Close and stop monitoring.

        open() restarts the monitor after closing.
        """
        self._executor.close()

        # Increment the pool_id and maybe close the socket. If the executor
        # thread has the socket checked out, it will be closed when checked in.
        self._pool.reset()

    def join(self, timeout=None):
        self._executor.join(timeout)

    def request_check(self):
        """If the monitor is sleeping, wake and check the server soon."""
        self._executor.wake()

    def _run(self):
        try:
            self._server_description = self._check_with_retry()
            self._topology.on_change(self._server_description)
        except ReferenceError:
            # Topology was garbage-collected.
            self.close()

    def _check_with_retry(self):
        """Call ismaster once or twice. Reset server's pool on error.

        Returns a ServerDescription.
        """
        # According to the spec, if an ismaster call fails we reset the
        # server's pool. If a server was once connected, change its type
        # to Unknown only after retrying once.
        address = self._server_description.address
        retry = True
        metadata = None
        if self._server_description.server_type == SERVER_TYPE.Unknown:
            retry = False
            metadata = self._pool.opts.metadata

        start = _time()
        try:
            # If the server type is unknown, send metadata with first check.
            return self._check_once(metadata=metadata)
        except ReferenceError:
            raise
        except Exception as error:
            error_time = _time() - start
#.........这里部分代码省略.........