From 14f626cb26c7fe3873fbee660fe05d35b1c10dfd Mon Sep 17 00:00:00 2001 From: "A. Jesse Jiryu Davis" Date: Sun, 20 Nov 2016 20:39:06 -0500 Subject: [PATCH] Factor Max Staleness and Server Selection tests --- test/test_max_staleness.py | 194 +---------------------- test/test_server_selection.py | 208 +------------------------ test/utils_selection_tests.py | 279 ++++++++++++++++++++++++++++++++++ 3 files changed, 287 insertions(+), 394 deletions(-) create mode 100644 test/utils_selection_tests.py diff --git a/test/test_max_staleness.py b/test/test_max_staleness.py index 64c851cae..8f7f69f1e 100644 --- a/test/test_max_staleness.py +++ b/test/test_max_staleness.py @@ -14,26 +14,20 @@ """Test maxStalenessSeconds support.""" -import datetime import os -import time import sys +import time import warnings sys.path[0:0] = [""] -from bson import json_util -from pymongo import MongoClient, read_preferences -from pymongo.common import clean_node, HEARTBEAT_FREQUENCY -from pymongo.errors import ConfigurationError, ConnectionFailure -from pymongo.ismaster import IsMaster -from pymongo.server_description import ServerDescription +from pymongo import MongoClient +from pymongo.errors import ConfigurationError from pymongo.server_selectors import writable_server_selector -from pymongo.settings import TopologySettings -from pymongo.topology import Topology from test import client_context, unittest from test.utils import rs_or_single_client +from test.utils_selection_tests import create_selection_tests # Location of JSON test specifications. _TEST_PATH = os.path.join( @@ -41,188 +35,10 @@ _TEST_PATH = os.path.join( 'max_staleness') -class MockSocketInfo(object): - def close(self): - pass - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - pass - - -class MockPool(object): - def __init__(self, *args, **kwargs): - pass - - def reset(self): - pass - - -class MockMonitor(object): - def __init__(self, server_description, topology, pool, topology_settings): - pass - - def open(self): - pass - - def request_check(self): - pass - - def close(self): - pass - - -def get_addresses(server_list): - seeds = [] - hosts = [] - for server in server_list: - seeds.append(clean_node(server['address'])) - hosts.append(server['address']) - return seeds, hosts - - -def make_last_write_date(server): - epoch = datetime.datetime.utcfromtimestamp(0) - millis = server.get('lastWrite', {}).get('lastWriteDate') - if millis: - diff = ((millis % 1000) + 1000) % 1000 - seconds = (millis - diff) / 1000 - micros = diff * 1000 - return epoch + datetime.timedelta( - seconds=seconds, microseconds=micros) - else: - # "Unknown" server. - return epoch - - -def make_server_description(server, hosts): - """Make ServerDescription from server info from JSON file.""" - server_type = server['type'] - if server_type == "Unknown": - return ServerDescription(clean_node(server['address']), IsMaster({})) - - ismaster_response = {'ok': True, 'hosts': hosts} - if server_type != "Standalone" and server_type != "Mongos": - ismaster_response['setName'] = "rs" - - if server_type == "RSPrimary": - ismaster_response['ismaster'] = True - elif server_type == "RSSecondary": - ismaster_response['secondary'] = True - elif server_type == "Mongos": - ismaster_response['msg'] = 'isdbgrid' - - ismaster_response['lastWrite'] = { - 'lastWriteDate': make_last_write_date(server) - } - - for field in 'maxWireVersion', 'tags', 'idleWritePeriodMillis': - if field in server: - ismaster_response[field] = server[field] - - # Sets _last_update_time to now. - sd = ServerDescription(clean_node(server['address']), - IsMaster(ismaster_response), - round_trip_time=server['avg_rtt_ms']) - - sd._last_update_time = server['lastUpdateTime'] / 1000.0 # ms to sec. - return sd - - -class TestAllScenarios(unittest.TestCase): +class TestAllScenarios(create_selection_tests(_TEST_PATH)): pass -def create_test(scenario_def): - def run_scenario(self): - if 'heartbeatFrequencyMS' in scenario_def: - frequency = int(scenario_def['heartbeatFrequencyMS']) / 1000.0 - else: - frequency = HEARTBEAT_FREQUENCY - - # Initialize topologies. - seeds, hosts = get_addresses( - scenario_def['topology_description']['servers']) - - topology = Topology( - TopologySettings(seeds=seeds, - monitor_class=MockMonitor, - pool_class=MockPool, - heartbeat_frequency=frequency)) - - # Update topologies with server descriptions. - for server in scenario_def['topology_description']['servers']: - server_description = make_server_description(server, hosts) - topology.on_change(server_description) - - # Create server selector. - # Make first letter lowercase to match read_pref's modes. - pref_def = scenario_def['read_preference'] - mode_string = pref_def.get('mode', 'primary') - mode_string = mode_string[:1].lower() + mode_string[1:] - mode = read_preferences.read_pref_mode_from_name(mode_string) - max_staleness = pref_def.get('maxStalenessSeconds', -1) - tag_sets = pref_def.get('tag_sets') - - if scenario_def.get('error'): - with self.assertRaises(ConfigurationError): - # Error can be raised when making Read Pref or selecting. - pref = read_preferences.make_read_preference( - mode, tag_sets=tag_sets, max_staleness=max_staleness) - - topology.select_server(pref) - return - - expected_addrs = set([ - server['address'] for server in scenario_def['in_latency_window']]) - - # Select servers. - pref = read_preferences.make_read_preference( - mode, tag_sets=tag_sets, max_staleness=max_staleness) - - if not expected_addrs: - with self.assertRaises(ConnectionFailure): - topology.select_servers(pref, server_selection_timeout=0) - return - - servers = topology.select_servers(pref, server_selection_timeout=0) - actual_addrs = set(['%s:%d' % s.description.address for s in servers]) - - for unexpected in actual_addrs - expected_addrs: - self.fail("'%s' shouldn't have been selected, but was" % unexpected) - - for unselected in expected_addrs - actual_addrs: - self.fail("'%s' should have been selected, but wasn't" % unselected) - - return run_scenario - - -def create_tests(): - for dirpath, _, filenames in os.walk(_TEST_PATH): - dirname = os.path.split(dirpath) - dirname = os.path.split(dirname[-2])[-1] + '_' + dirname[-1] - - for filename in filenames: - if not filename.endswith('.json'): - continue - - with open(os.path.join(dirpath, filename)) as scenario_stream: - scenario_def = json_util.loads(scenario_stream.read()) - - # Construct test from scenario. - new_test = create_test(scenario_def) - test_name = 'test_%s_%s' % ( - dirname, os.path.splitext(filename)[0]) - - new_test.__name__ = test_name - setattr(TestAllScenarios, new_test.__name__, new_test) - - -create_tests() - - class TestMaxStaleness(unittest.TestCase): def test_max_staleness(self): client = MongoClient() diff --git a/test/test_server_selection.py b/test/test_server_selection.py index 0565a6b3d..b54eea4f3 100644 --- a/test/test_server_selection.py +++ b/test/test_server_selection.py @@ -12,23 +12,15 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Test the topology module.""" +"""Test the topology module's Server Selection Spec implementation.""" -import json import os import sys sys.path[0:0] = [""] -from pymongo import read_preferences -from pymongo.common import clean_node -from pymongo.errors import AutoReconnect -from pymongo.ismaster import IsMaster -from pymongo.server_description import ServerDescription -from pymongo.settings import TopologySettings -from pymongo.server_selectors import writable_server_selector -from pymongo.topology import Topology from test import unittest +from test.utils_selection_tests import create_selection_tests # Location of JSON test specifications. @@ -37,202 +29,8 @@ _TEST_PATH = os.path.join( os.path.join('server_selection', 'server_selection')) -class MockSocketInfo(object): - def close(self): - pass - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - pass - - -class MockPool(object): - def __init__(self, *args, **kwargs): - pass - - def reset(self): - pass - - def remove_stale_sockets(self): - pass - - -class MockMonitor(object): - def __init__(self, server_description, topology, pool, topology_settings): - pass - - def open(self): - pass - - def request_check(self): - pass - - def close(self): - pass - - -def get_addresses(server_list): - seeds = [] - hosts = [] - for server in server_list: - seeds.append(clean_node(server['address'])) - hosts.append(server['address']) - return seeds, hosts - - -def make_server_description(server, hosts): - """Make ServerDescription from server info from JSON file.""" - ismaster_response = {} - ismaster_response['tags'] = server['tags'] - ismaster_response['ok'] = True - ismaster_response['hosts'] = hosts - - server_type = server['type'] - - if server_type != "Standalone" and server_type != "Mongos": - ismaster_response['setName'] = True - if server_type == "RSPrimary": - ismaster_response['ismaster'] = True - elif server_type == "RSSecondary": - ismaster_response['secondary'] = True - elif server_type == "Mongos": - ismaster_response['msg'] = 'isdbgrid' - - return ServerDescription(clean_node(server['address']), - IsMaster(ismaster_response), - # round_trip_time is in seconds. - round_trip_time=server['avg_rtt_ms'] / 1000.0) - - -class TestAllScenarios(unittest.TestCase): +class TestAllScenarios(create_selection_tests(_TEST_PATH)): pass - -def create_test(scenario_def): - def run_scenario(self): - - # Initialize topologies. - seeds, hosts = get_addresses( - scenario_def['topology_description']['servers']) - - # "Eligible servers" is defined in the server selection spec as - # the set of servers matching both the ReadPreference's mode - # and tag sets. - top_latency = Topology( - TopologySettings(seeds=seeds, monitor_class=MockMonitor, - pool_class=MockPool)) - # "In latency window" is defined in the server selection - # spec as the subset of suitable_servers that falls within the - # allowable latency window. - top_suitable = Topology( - TopologySettings(seeds=seeds, local_threshold_ms=1000000, - monitor_class=MockMonitor, - pool_class=MockPool)) - - # Update topologies with server descriptions. - for server in scenario_def['topology_description']['servers']: - server_description = make_server_description(server, hosts) - top_suitable.on_change(server_description) - top_latency.on_change(server_description) - - # Create server selector. - if scenario_def["operation"] == "write": - instance = writable_server_selector - else: - # Make first letter lowercase to match read_pref's modes. - mode_string = scenario_def['read_preference']['mode'] - if mode_string: - mode_string = mode_string[:1].lower() + mode_string[1:] - - mode = read_preferences.read_pref_mode_from_name(mode_string) - tag_sets = None - if scenario_def['read_preference']['tag_sets'][0]: - tag_sets = scenario_def['read_preference']['tag_sets'] - instance = read_preferences.make_read_preference(mode, tag_sets) - - # Select servers. - if not scenario_def['suitable_servers']: - self.assertRaises(AutoReconnect, top_suitable.select_server, - instance, - server_selection_timeout=0) - return - - if not scenario_def['in_latency_window']: - self.assertRaises(AutoReconnect, top_latency.select_server, - instance, - server_selection_timeout=0) - return - - actual_suitable_s = top_suitable.select_servers(instance, - server_selection_timeout=0) - actual_latency_s = top_latency.select_servers(instance, - server_selection_timeout=0) - - expected_suitable_servers = {} - for server in scenario_def['suitable_servers']: - server_description = make_server_description(server, hosts) - expected_suitable_servers[server['address']] = server_description - - actual_suitable_servers = {} - for s in actual_suitable_s: - actual_suitable_servers["%s:%d" % (s.description.address[0], - s.description.address[1])] = s.description - - self.assertEqual(len(actual_suitable_servers), - len(expected_suitable_servers)) - for k, actual in actual_suitable_servers.items(): - expected = expected_suitable_servers[k] - self.assertEqual(expected.address, actual.address) - self.assertEqual(expected.server_type, actual.server_type) - self.assertEqual(expected.round_trip_time, actual.round_trip_time) - self.assertEqual(expected.tags, actual.tags) - self.assertEqual(expected.all_hosts, actual.all_hosts) - - expected_latency_servers = {} - for server in scenario_def['in_latency_window']: - server_description = make_server_description(server, hosts) - expected_latency_servers[server['address']] = server_description - - actual_latency_servers = {} - for s in actual_latency_s: - actual_latency_servers["%s:%d" % - (s.description.address[0], - s.description.address[1])] = s.description - - self.assertEqual(len(actual_latency_servers), - len(expected_latency_servers)) - for k, actual in actual_latency_servers.items(): - expected = expected_latency_servers[k] - self.assertEqual(expected.address, actual.address) - self.assertEqual(expected.server_type, actual.server_type) - self.assertEqual(expected.round_trip_time, actual.round_trip_time) - self.assertEqual(expected.tags, actual.tags) - self.assertEqual(expected.all_hosts, actual.all_hosts) - - return run_scenario - - -def create_tests(): - for dirpath, _, filenames in os.walk(_TEST_PATH): - dirname = os.path.split(dirpath) - dirname = os.path.split(dirname[-2])[-1] + '_' + dirname[-1] - - for filename in filenames: - with open(os.path.join(dirpath, filename)) as scenario_stream: - scenario_def = json.load(scenario_stream) - - # Construct test from scenario. - new_test = create_test(scenario_def) - test_name = 'test_%s_%s' % ( - dirname, os.path.splitext(filename)[0]) - - new_test.__name__ = test_name - setattr(TestAllScenarios, new_test.__name__, new_test) - - -create_tests() - if __name__ == "__main__": unittest.main() diff --git a/test/utils_selection_tests.py b/test/utils_selection_tests.py new file mode 100644 index 000000000..ac627c28f --- /dev/null +++ b/test/utils_selection_tests.py @@ -0,0 +1,279 @@ +# Copyright 2015-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. + +"""Utilities for testing Server Selection and Max Staleness.""" + +import datetime +import os +import sys + +sys.path[0:0] = [""] + +from bson import json_util +from pymongo import read_preferences +from pymongo.common import clean_node, HEARTBEAT_FREQUENCY +from pymongo.errors import AutoReconnect, ConfigurationError +from pymongo.ismaster import IsMaster +from pymongo.server_description import ServerDescription +from pymongo.settings import TopologySettings +from pymongo.server_selectors import writable_server_selector +from pymongo.topology import Topology +from test import unittest + + +class MockSocketInfo(object): + def close(self): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + pass + + +class MockPool(object): + def __init__(self, *args, **kwargs): + pass + + def reset(self): + pass + + def remove_stale_sockets(self): + pass + + +class MockMonitor(object): + def __init__(self, server_description, topology, pool, topology_settings): + pass + + def open(self): + pass + + def request_check(self): + pass + + def close(self): + pass + + +def get_addresses(server_list): + seeds = [] + hosts = [] + for server in server_list: + seeds.append(clean_node(server['address'])) + hosts.append(server['address']) + return seeds, hosts + + +def make_last_write_date(server): + epoch = datetime.datetime.utcfromtimestamp(0) + millis = server.get('lastWrite', {}).get('lastWriteDate') + if millis: + diff = ((millis % 1000) + 1000) % 1000 + seconds = (millis - diff) / 1000 + micros = diff * 1000 + return epoch + datetime.timedelta( + seconds=seconds, microseconds=micros) + else: + # "Unknown" server. + return epoch + + +def make_server_description(server, hosts): + """Make a ServerDescription from server info in a JSON test.""" + server_type = server['type'] + if server_type == "Unknown": + return ServerDescription(clean_node(server['address']), IsMaster({})) + + ismaster_response = {'ok': True, 'hosts': hosts} + if server_type != "Standalone" and server_type != "Mongos": + ismaster_response['setName'] = "rs" + + if server_type == "RSPrimary": + ismaster_response['ismaster'] = True + elif server_type == "RSSecondary": + ismaster_response['secondary'] = True + elif server_type == "Mongos": + ismaster_response['msg'] = 'isdbgrid' + + ismaster_response['lastWrite'] = { + 'lastWriteDate': make_last_write_date(server) + } + + for field in 'maxWireVersion', 'tags', 'idleWritePeriodMillis': + if field in server: + ismaster_response[field] = server[field] + + # Sets _last_update_time to now. + sd = ServerDescription(clean_node(server['address']), + IsMaster(ismaster_response), + round_trip_time=server['avg_rtt_ms'] / 1000.0) + + if 'lastUpdateTime' in server: + sd._last_update_time = server['lastUpdateTime'] / 1000.0 # ms to sec. + + return sd + + +def get_topology_type_name(scenario_def): + td = scenario_def['topology_description'] + name = td['type'] + if name == 'Unknown': + # PyMongo never starts a topology in type Unknown. + return 'Sharded' if len(td['servers']) > 1 else 'Single' + else: + return name + + +def create_test(scenario_def): + def run_scenario(self): + # Initialize topologies. + if 'heartbeatFrequencyMS' in scenario_def: + frequency = int(scenario_def['heartbeatFrequencyMS']) / 1000.0 + else: + frequency = HEARTBEAT_FREQUENCY + + settings = dict( + monitor_class=MockMonitor, + heartbeat_frequency=frequency, + pool_class=MockPool) + + settings['seeds'], hosts = get_addresses( + scenario_def['topology_description']['servers']) + + # "Eligible servers" is defined in the server selection spec as + # the set of servers matching both the ReadPreference's mode + # and tag sets. + top_latency = Topology(TopologySettings(**settings)) + + # "In latency window" is defined in the server selection + # spec as the subset of suitable_servers that falls within the + # allowable latency window. + settings['local_threshold_ms'] = 1000000 + top_suitable = Topology(TopologySettings(**settings)) + + # Update topologies with server descriptions. + for server in scenario_def['topology_description']['servers']: + server_description = make_server_description(server, hosts) + top_suitable.on_change(server_description) + top_latency.on_change(server_description) + + # Create server selector. + if scenario_def.get("operation") == "write": + pref = writable_server_selector + else: + # Make first letter lowercase to match read_pref's modes. + pref_def = scenario_def['read_preference'] + mode_string = pref_def.get('mode', 'primary') + mode_string = mode_string[:1].lower() + mode_string[1:] + mode = read_preferences.read_pref_mode_from_name(mode_string) + max_staleness = pref_def.get('maxStalenessSeconds', -1) + tag_sets = pref_def.get('tag_sets') + + if scenario_def.get('error'): + with self.assertRaises(ConfigurationError): + # Error can be raised when making Read Pref or selecting. + pref = read_preferences.make_read_preference( + mode, tag_sets=tag_sets, max_staleness=max_staleness) + + top_latency.select_server(pref) + return + + pref = read_preferences.make_read_preference( + mode, tag_sets=tag_sets, max_staleness=max_staleness) + + # Select servers. + if not scenario_def.get('suitable_servers'): + with self.assertRaises(AutoReconnect): + top_suitable.select_server(pref, server_selection_timeout=0) + + return + + if not scenario_def['in_latency_window']: + with self.assertRaises(AutoReconnect): + top_latency.select_server(pref, server_selection_timeout=0) + + return + + actual_suitable_s = top_suitable.select_servers( + pref, server_selection_timeout=0) + actual_latency_s = top_latency.select_servers( + pref, server_selection_timeout=0) + + expected_suitable_servers = {} + for server in scenario_def['suitable_servers']: + server_description = make_server_description(server, hosts) + expected_suitable_servers[server['address']] = server_description + + actual_suitable_servers = {} + for s in actual_suitable_s: + actual_suitable_servers["%s:%d" % (s.description.address[0], + s.description.address[1])] = s.description + + self.assertEqual(len(actual_suitable_servers), + len(expected_suitable_servers)) + for k, actual in actual_suitable_servers.items(): + expected = expected_suitable_servers[k] + self.assertEqual(expected.address, actual.address) + self.assertEqual(expected.server_type, actual.server_type) + self.assertEqual(expected.round_trip_time, actual.round_trip_time) + self.assertEqual(expected.tags, actual.tags) + self.assertEqual(expected.all_hosts, actual.all_hosts) + + expected_latency_servers = {} + for server in scenario_def['in_latency_window']: + server_description = make_server_description(server, hosts) + expected_latency_servers[server['address']] = server_description + + actual_latency_servers = {} + for s in actual_latency_s: + actual_latency_servers["%s:%d" % + (s.description.address[0], + s.description.address[1])] = s.description + + self.assertEqual(len(actual_latency_servers), + len(expected_latency_servers)) + for k, actual in actual_latency_servers.items(): + expected = expected_latency_servers[k] + self.assertEqual(expected.address, actual.address) + self.assertEqual(expected.server_type, actual.server_type) + self.assertEqual(expected.round_trip_time, actual.round_trip_time) + self.assertEqual(expected.tags, actual.tags) + self.assertEqual(expected.all_hosts, actual.all_hosts) + + return run_scenario + + +def create_selection_tests(test_dir): + class TestAllScenarios(unittest.TestCase): + pass + + for dirpath, _, filenames in os.walk(test_dir): + dirname = os.path.split(dirpath) + dirname = os.path.split(dirname[-2])[-1] + '_' + dirname[-1] + + for filename in filenames: + with open(os.path.join(dirpath, filename)) as scenario_stream: + scenario_def = json_util.loads(scenario_stream.read()) + + # Construct test from scenario. + new_test = create_test(scenario_def) + test_name = 'test_%s_%s' % ( + dirname, os.path.splitext(filename)[0]) + + new_test.__name__ = test_name + setattr(TestAllScenarios, new_test.__name__, new_test) + + return TestAllScenarios