289 lines
12 KiB
Python
289 lines
12 KiB
Python
# Copyright 2016 MongoDB, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Run the sdam monitoring spec tests."""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import weakref
|
|
|
|
sys.path[0:0] = [""]
|
|
|
|
from bson.json_util import object_hook
|
|
from pymongo import monitoring
|
|
from pymongo import periodic_executor
|
|
from pymongo.ismaster import IsMaster
|
|
from pymongo.monitor import Monitor
|
|
from pymongo.read_preferences import MovingAverage
|
|
from pymongo.server_description import ServerDescription
|
|
from pymongo.server_type import SERVER_TYPE
|
|
from pymongo.topology import TOPOLOGY_TYPE
|
|
from test import unittest, client_context, client_knobs
|
|
from test.utils import (ServerAndTopologyEventListener,
|
|
single_client,
|
|
wait_until)
|
|
|
|
# Location of JSON test specifications.
|
|
_TEST_PATH = os.path.join(
|
|
os.path.dirname(os.path.realpath(__file__)),
|
|
'sdam_monitoring')
|
|
|
|
|
|
def compare_server_descriptions(expected, actual):
|
|
if ((not expected['address'] == "%s:%s" % actual.address) or
|
|
(not SERVER_TYPE.__getattribute__(expected['type']) ==
|
|
actual.server_type)):
|
|
return False
|
|
expected_hosts = set(
|
|
expected['arbiters'] + expected['passives'] + expected['hosts'])
|
|
return expected_hosts == set("%s:%s" % s for s in actual.all_hosts)
|
|
|
|
|
|
def compare_topology_descriptions(expected, actual):
|
|
if not (TOPOLOGY_TYPE.__getattribute__(
|
|
expected['topologyType']) == actual.topology_type):
|
|
return False
|
|
expected = expected['servers']
|
|
actual = actual.server_descriptions()
|
|
if len(expected) != len(actual):
|
|
return False
|
|
for exp_server in expected:
|
|
for address, actual_server in actual.items():
|
|
if compare_server_descriptions(exp_server, actual_server):
|
|
break
|
|
else:
|
|
return False
|
|
return True
|
|
|
|
|
|
def compare_events(expected_dict, actual):
|
|
if not expected_dict:
|
|
return False, "Error: Bad expected value in YAML test"
|
|
if not actual:
|
|
return False, "Error: Event published was None"
|
|
|
|
expected_type, expected = list(expected_dict.items())[0]
|
|
|
|
if expected_type == "server_opening_event":
|
|
if not isinstance(actual, monitoring.ServerOpeningEvent):
|
|
return False, "Expected ServerOpeningEvent, got %s" % (
|
|
actual.__class__)
|
|
if not expected['address'] == "%s:%s" % actual.server_address:
|
|
return (False,
|
|
"ServerOpeningEvent published with wrong address (expected"
|
|
" %s, got %s" % (expected['address'],
|
|
actual.server_address))
|
|
|
|
elif expected_type == "server_description_changed_event":
|
|
|
|
if not isinstance(actual, monitoring.ServerDescriptionChangedEvent):
|
|
return (False,
|
|
"Expected ServerDescriptionChangedEvent, got %s" % (
|
|
actual.__class__))
|
|
if not expected['address'] == "%s:%s" % actual.server_address:
|
|
return (False, "ServerDescriptionChangedEvent has wrong address"
|
|
" (expected %s, got %s" % (expected['address'],
|
|
actual.server_address))
|
|
|
|
if not compare_server_descriptions(
|
|
expected['newDescription'], actual.new_description):
|
|
return (False, "New ServerDescription incorrect in"
|
|
" ServerDescriptionChangedEvent")
|
|
if not compare_server_descriptions(expected['previousDescription'],
|
|
actual.previous_description):
|
|
return (False, "Previous ServerDescription incorrect in"
|
|
" ServerDescriptionChangedEvent")
|
|
|
|
elif expected_type == "server_closed_event":
|
|
if not isinstance(actual, monitoring.ServerClosedEvent):
|
|
return False, "Expected ServerClosedEvent, got %s" % (
|
|
actual.__class__)
|
|
if not expected['address'] == "%s:%s" % actual.server_address:
|
|
return (False, "ServerClosedEvent published with wrong address"
|
|
" (expected %s, got %s" % (expected['address'],
|
|
actual.server_address))
|
|
|
|
elif expected_type == "topology_opening_event":
|
|
if not isinstance(actual, monitoring.TopologyOpenedEvent):
|
|
return False, "Expected TopologyOpeningEvent, got %s" % (
|
|
actual.__class__)
|
|
|
|
elif expected_type == "topology_description_changed_event":
|
|
if not isinstance(actual, monitoring.TopologyDescriptionChangedEvent):
|
|
return (False, "Expected TopologyDescriptionChangedEvent,"
|
|
" got %s" % (actual.__class__))
|
|
if not compare_topology_descriptions(expected['newDescription'],
|
|
actual.new_description):
|
|
return (False, "New TopologyDescription incorrect in "
|
|
"TopologyDescriptionChangedEvent")
|
|
if not compare_topology_descriptions(
|
|
expected['previousDescription'],
|
|
actual.previous_description):
|
|
return (False, "Previous TopologyDescription incorrect in"
|
|
" TopologyDescriptionChangedEvent")
|
|
|
|
elif expected_type == "topology_closed_event":
|
|
if not isinstance(actual, monitoring.TopologyClosedEvent):
|
|
return False, "Expected TopologyClosedEvent, got %s" % (
|
|
actual.__class__)
|
|
|
|
else:
|
|
return False, "Incorrect event: expected %s, actual %s" % (
|
|
expected_type, actual)
|
|
|
|
return True, ""
|
|
|
|
|
|
def compare_multiple_events(i, expected_results, actual_results):
|
|
events_in_a_row = []
|
|
j = i
|
|
while(j < len(expected_results) and isinstance(
|
|
actual_results[j],
|
|
actual_results[i].__class__)):
|
|
events_in_a_row.append(actual_results[j])
|
|
j += 1
|
|
message = ''
|
|
for event in events_in_a_row:
|
|
for k in range(i, j):
|
|
passed, message = compare_events(expected_results[k], event)
|
|
if passed:
|
|
expected_results[k] = None
|
|
break
|
|
else:
|
|
return i, False, message
|
|
return j, True, ''
|
|
|
|
|
|
class TestAllScenarios(unittest.TestCase):
|
|
|
|
@classmethod
|
|
@client_context.require_connection
|
|
def setUp(cls):
|
|
cls.all_listener = ServerAndTopologyEventListener()
|
|
cls.saved_listeners = monitoring._LISTENERS
|
|
monitoring._LISTENERS = monitoring._Listeners([], [], [], [])
|
|
|
|
@classmethod
|
|
def tearDown(cls):
|
|
monitoring._LISTENERS = cls.saved_listeners
|
|
|
|
|
|
def create_test(scenario_def):
|
|
def run_scenario(self):
|
|
responses = (r for r in scenario_def['phases'][0]['responses'])
|
|
|
|
with client_knobs(events_queue_frequency=0.1):
|
|
class MockMonitor(Monitor):
|
|
def __init__(self, server_description, topology, pool,
|
|
topology_settings):
|
|
"""Have to copy entire constructor from Monitor so that we
|
|
can override _run and change the periodic executor's
|
|
interval."""
|
|
|
|
self._server_description = server_description
|
|
self._pool = pool
|
|
self._settings = topology_settings
|
|
self._avg_round_trip_time = MovingAverage()
|
|
options = self._settings._pool_options
|
|
self._listeners = options.event_listeners
|
|
self._publish = self._listeners is not None
|
|
|
|
def target():
|
|
monitor = self_ref()
|
|
if monitor is None:
|
|
return False
|
|
MockMonitor._run(monitor) # Change target to subclass
|
|
return True
|
|
|
|
# Shorten interval
|
|
executor = periodic_executor.PeriodicExecutor(
|
|
interval=0.1,
|
|
min_interval=0.1,
|
|
target=target,
|
|
name="pymongo_server_monitor_thread")
|
|
self._executor = executor
|
|
self_ref = weakref.ref(self, executor.close)
|
|
self._topology = weakref.proxy(topology, executor.close)
|
|
|
|
def _run(self):
|
|
try:
|
|
if self._server_description.address != ('a', 27017):
|
|
# Because PyMongo doesn't keep information about
|
|
# the order of addresses, we might accidentally
|
|
# start a MockMonitor on the wrong server first,
|
|
# so we need to only mock responses for the server
|
|
# the test's response is supposed to come from.
|
|
return
|
|
response = next(responses)[1]
|
|
isMaster = IsMaster(response)
|
|
self._server_description = ServerDescription(
|
|
address=self._server_description.address,
|
|
ismaster=isMaster)
|
|
self._topology.on_change(self._server_description)
|
|
except (ReferenceError, StopIteration):
|
|
# Topology was garbage-collected.
|
|
self.close()
|
|
|
|
m = single_client(h=scenario_def['uri'], p=27017,
|
|
event_listeners=(self.all_listener,),
|
|
_monitor_class=MockMonitor)
|
|
|
|
expected_results = scenario_def['phases'][0]['outcome']['events']
|
|
|
|
expected_len = len(expected_results)
|
|
wait_until(lambda: len(self.all_listener.results) >= expected_len,
|
|
"publish all events", timeout=15)
|
|
|
|
try:
|
|
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) or
|
|
isinstance(result,
|
|
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
|
|
|
|
finally:
|
|
m.close()
|
|
return run_scenario
|
|
|
|
|
|
def create_tests():
|
|
for dirpath, _, filenames in os.walk(_TEST_PATH):
|
|
for filename in filenames:
|
|
with open(os.path.join(dirpath, filename)) as scenario_stream:
|
|
scenario_def = json.load(
|
|
scenario_stream, object_hook=object_hook)
|
|
# Construct test from scenario.
|
|
new_test = create_test(scenario_def)
|
|
test_name = 'test_%s' % (os.path.splitext(filename)[0],)
|
|
new_test.__name__ = test_name
|
|
setattr(TestAllScenarios, new_test.__name__, new_test)
|
|
|
|
|
|
create_tests()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|