PYTHON-5530 Reduce usage of legacy test runner (#2642)

blink1073 · web-flow · commit ae88b5a08f6d · 2025-12-10T13:40:24.000-06:00
diff --git a/test/asynchronous/test_encryption.py b/test/asynchronous/test_encryption.py
@@ -33,7 +33,6 @@
 from test.asynchronous import AsyncIntegrationTest, AsyncPyMongoTestCase, async_client_context
 from test.asynchronous.test_bulk import AsyncBulkTestBase
 from test.asynchronous.utils import flaky
-from test.asynchronous.utils_spec_runner import AsyncSpecRunner, AsyncSpecTestCreator
 from threading import Thread
 from typing import Any, Dict, Mapping, Optional
 
@@ -55,7 +54,6 @@
 )
 from test.asynchronous.test_bulk import AsyncBulkTestBase
 from test.asynchronous.unified_format import generate_test_classes
-from test.asynchronous.utils_spec_runner import AsyncSpecRunner
 from test.helpers_shared import (
     ALL_KMS_PROVIDERS,
     AWS_CREDS,
@@ -624,125 +622,6 @@ async def test_with_statement(self):
 }
 
 
-class AsyncTestSpec(AsyncSpecRunner):
-    @classmethod
-    @unittest.skipUnless(_HAVE_PYMONGOCRYPT, "pymongocrypt is not installed")
-    async def _setup_class(cls):
-        await super()._setup_class()
-
-    def parse_auto_encrypt_opts(self, opts):
-        """Parse clientOptions.autoEncryptOpts."""
-        opts = camel_to_snake_args(opts)
-        kms_providers = opts["kms_providers"]
-        if "aws" in kms_providers:
-            kms_providers["aws"] = AWS_CREDS
-            if not any(AWS_CREDS.values()):
-                self.skipTest("AWS environment credentials are not set")
-        if "awsTemporary" in kms_providers:
-            kms_providers["aws"] = AWS_TEMP_CREDS
-            del kms_providers["awsTemporary"]
-            if not any(AWS_TEMP_CREDS.values()):
-                self.skipTest("AWS Temp environment credentials are not set")
-        if "awsTemporaryNoSessionToken" in kms_providers:
-            kms_providers["aws"] = AWS_TEMP_NO_SESSION_CREDS
-            del kms_providers["awsTemporaryNoSessionToken"]
-            if not any(AWS_TEMP_NO_SESSION_CREDS.values()):
-                self.skipTest("AWS Temp environment credentials are not set")
-        if "azure" in kms_providers:
-            kms_providers["azure"] = AZURE_CREDS
-            if not any(AZURE_CREDS.values()):
-                self.skipTest("Azure environment credentials are not set")
-        if "gcp" in kms_providers:
-            kms_providers["gcp"] = GCP_CREDS
-            if not any(AZURE_CREDS.values()):
-                self.skipTest("GCP environment credentials are not set")
-        if "kmip" in kms_providers:
-            kms_providers["kmip"] = KMIP_CREDS
-            opts["kms_tls_options"] = DEFAULT_KMS_TLS
-        if "key_vault_namespace" not in opts:
-            opts["key_vault_namespace"] = "keyvault.datakeys"
-        if "extra_options" in opts:
-            opts.update(camel_to_snake_args(opts.pop("extra_options")))
-
-        opts = dict(opts)
-        return AutoEncryptionOpts(**opts)
-
-    def parse_client_options(self, opts):
-        """Override clientOptions parsing to support autoEncryptOpts."""
-        encrypt_opts = opts.pop("autoEncryptOpts", None)
-        if encrypt_opts:
-            opts["auto_encryption_opts"] = self.parse_auto_encrypt_opts(encrypt_opts)
-
-        return super().parse_client_options(opts)
-
-    def get_object_name(self, op):
-        """Default object is collection."""
-        return op.get("object", "collection")
-
-    def maybe_skip_scenario(self, test):
-        super().maybe_skip_scenario(test)
-        desc = test["description"].lower()
-        if (
-            "timeoutms applied to listcollections to get collection schema" in desc
-            and sys.platform in ("win32", "darwin")
-        ):
-            self.skipTest("PYTHON-3706 flaky test on Windows/macOS")
-        if "type=symbol" in desc:
-            self.skipTest("PyMongo does not support the symbol type")
-        if "timeoutms applied to listcollections to get collection schema" in desc and not _IS_SYNC:
-            self.skipTest("PYTHON-4844 flaky test on async")
-
-    async def setup_scenario(self, scenario_def):
-        """Override a test's setup."""
-        key_vault_data = scenario_def["key_vault_data"]
-        encrypted_fields = scenario_def["encrypted_fields"]
-        json_schema = scenario_def["json_schema"]
-        data = scenario_def["data"]
-        coll = async_client_context.client.get_database("keyvault", codec_options=OPTS)["datakeys"]
-        await coll.delete_many({})
-        if key_vault_data:
-            await coll.insert_many(key_vault_data)
-
-        db_name = self.get_scenario_db_name(scenario_def)
-        coll_name = self.get_scenario_coll_name(scenario_def)
-        db = async_client_context.client.get_database(db_name, codec_options=OPTS)
-        await db.drop_collection(coll_name, encrypted_fields=encrypted_fields)
-        wc = WriteConcern(w="majority")
-        kwargs: Dict[str, Any] = {}
-        if json_schema:
-            kwargs["validator"] = {"$jsonSchema": json_schema}
-            kwargs["codec_options"] = OPTS
-        if not data:
-            kwargs["write_concern"] = wc
-        if encrypted_fields:
-            kwargs["encryptedFields"] = encrypted_fields
-        await db.create_collection(coll_name, **kwargs)
-        coll = db[coll_name]
-        if data:
-            # Load data.
-            await coll.with_options(write_concern=wc).insert_many(scenario_def["data"])
-
-    def allowable_errors(self, op):
-        """Override expected error classes."""
-        errors = super().allowable_errors(op)
-        # An updateOne test expects encryption to error when no $ operator
-        # appears but pymongo raises a client side ValueError in this case.
-        if op["name"] == "updateOne":
-            errors += (ValueError,)
-        return errors
-
-
-def create_test(scenario_def, test, name):
-    @async_client_context.require_test_commands
-    async def run_scenario(self):
-        await self.run_scenario(scenario_def, test)
-
-    return run_scenario
-
-
-test_creator = AsyncSpecTestCreator(create_test, AsyncTestSpec, os.path.join(SPEC_PATH, "legacy"))
-test_creator.create_tests()
-
 if _HAVE_PYMONGOCRYPT:
     globals().update(
         generate_test_classes(
diff --git a/test/asynchronous/test_transactions.py b/test/asynchronous/test_transactions.py
@@ -18,7 +18,6 @@
 import asyncio
 import sys
 from io import BytesIO
-from test.asynchronous.utils_spec_runner import AsyncSpecRunner
 
 from gridfs.asynchronous.grid_file import AsyncGridFS, AsyncGridFSBucket
 from pymongo.asynchronous.pool import PoolState
@@ -61,15 +60,8 @@
 UNPIN_TEST_MAX_ATTEMPTS = 50
 
 
-class AsyncTransactionsBase(AsyncSpecRunner):
-    def maybe_skip_scenario(self, test):
-        super().maybe_skip_scenario(test)
-        if (
-            "secondary" in self.id()
-            and not async_client_context.is_mongos
-            and not async_client_context.has_secondaries
-        ):
-            raise unittest.SkipTest("No secondaries")
+class AsyncTransactionsBase(AsyncIntegrationTest):
+    pass
 
 
 class TestTransactions(AsyncTransactionsBase):
diff --git a/test/test_encryption.py b/test/test_encryption.py
@@ -33,7 +33,6 @@
 from test import IntegrationTest, PyMongoTestCase, client_context
 from test.test_bulk import BulkTestBase
 from test.utils import flaky
-from test.utils_spec_runner import SpecRunner, SpecTestCreator
 from threading import Thread
 from typing import Any, Dict, Mapping, Optional
 
@@ -75,7 +74,6 @@
     is_greenthread_patched,
     wait_until,
 )
-from test.utils_spec_runner import SpecRunner
 
 from bson import BSON, DatetimeMS, Decimal128, encode, json_util
 from bson.binary import UUID_SUBTYPE, Binary, UuidRepresentation
@@ -622,125 +620,6 @@ def test_with_statement(self):
 }
 
 
-class TestSpec(SpecRunner):
-    @classmethod
-    @unittest.skipUnless(_HAVE_PYMONGOCRYPT, "pymongocrypt is not installed")
-    def _setup_class(cls):
-        super()._setup_class()
-
-    def parse_auto_encrypt_opts(self, opts):
-        """Parse clientOptions.autoEncryptOpts."""
-        opts = camel_to_snake_args(opts)
-        kms_providers = opts["kms_providers"]
-        if "aws" in kms_providers:
-            kms_providers["aws"] = AWS_CREDS
-            if not any(AWS_CREDS.values()):
-                self.skipTest("AWS environment credentials are not set")
-        if "awsTemporary" in kms_providers:
-            kms_providers["aws"] = AWS_TEMP_CREDS
-            del kms_providers["awsTemporary"]
-            if not any(AWS_TEMP_CREDS.values()):
-                self.skipTest("AWS Temp environment credentials are not set")
-        if "awsTemporaryNoSessionToken" in kms_providers:
-            kms_providers["aws"] = AWS_TEMP_NO_SESSION_CREDS
-            del kms_providers["awsTemporaryNoSessionToken"]
-            if not any(AWS_TEMP_NO_SESSION_CREDS.values()):
-                self.skipTest("AWS Temp environment credentials are not set")
-        if "azure" in kms_providers:
-            kms_providers["azure"] = AZURE_CREDS
-            if not any(AZURE_CREDS.values()):
-                self.skipTest("Azure environment credentials are not set")
-        if "gcp" in kms_providers:
-            kms_providers["gcp"] = GCP_CREDS
-            if not any(AZURE_CREDS.values()):
-                self.skipTest("GCP environment credentials are not set")
-        if "kmip" in kms_providers:
-            kms_providers["kmip"] = KMIP_CREDS
-            opts["kms_tls_options"] = DEFAULT_KMS_TLS
-        if "key_vault_namespace" not in opts:
-            opts["key_vault_namespace"] = "keyvault.datakeys"
-        if "extra_options" in opts:
-            opts.update(camel_to_snake_args(opts.pop("extra_options")))
-
-        opts = dict(opts)
-        return AutoEncryptionOpts(**opts)
-
-    def parse_client_options(self, opts):
-        """Override clientOptions parsing to support autoEncryptOpts."""
-        encrypt_opts = opts.pop("autoEncryptOpts", None)
-        if encrypt_opts:
-            opts["auto_encryption_opts"] = self.parse_auto_encrypt_opts(encrypt_opts)
-
-        return super().parse_client_options(opts)
-
-    def get_object_name(self, op):
-        """Default object is collection."""
-        return op.get("object", "collection")
-
-    def maybe_skip_scenario(self, test):
-        super().maybe_skip_scenario(test)
-        desc = test["description"].lower()
-        if (
-            "timeoutms applied to listcollections to get collection schema" in desc
-            and sys.platform in ("win32", "darwin")
-        ):
-            self.skipTest("PYTHON-3706 flaky test on Windows/macOS")
-        if "type=symbol" in desc:
-            self.skipTest("PyMongo does not support the symbol type")
-        if "timeoutms applied to listcollections to get collection schema" in desc and not _IS_SYNC:
-            self.skipTest("PYTHON-4844 flaky test on async")
-
-    def setup_scenario(self, scenario_def):
-        """Override a test's setup."""
-        key_vault_data = scenario_def["key_vault_data"]
-        encrypted_fields = scenario_def["encrypted_fields"]
-        json_schema = scenario_def["json_schema"]
-        data = scenario_def["data"]
-        coll = client_context.client.get_database("keyvault", codec_options=OPTS)["datakeys"]
-        coll.delete_many({})
-        if key_vault_data:
-            coll.insert_many(key_vault_data)
-
-        db_name = self.get_scenario_db_name(scenario_def)
-        coll_name = self.get_scenario_coll_name(scenario_def)
-        db = client_context.client.get_database(db_name, codec_options=OPTS)
-        db.drop_collection(coll_name, encrypted_fields=encrypted_fields)
-        wc = WriteConcern(w="majority")
-        kwargs: Dict[str, Any] = {}
-        if json_schema:
-            kwargs["validator"] = {"$jsonSchema": json_schema}
-            kwargs["codec_options"] = OPTS
-        if not data:
-            kwargs["write_concern"] = wc
-        if encrypted_fields:
-            kwargs["encryptedFields"] = encrypted_fields
-        db.create_collection(coll_name, **kwargs)
-        coll = db[coll_name]
-        if data:
-            # Load data.
-            coll.with_options(write_concern=wc).insert_many(scenario_def["data"])
-
-    def allowable_errors(self, op):
-        """Override expected error classes."""
-        errors = super().allowable_errors(op)
-        # An updateOne test expects encryption to error when no $ operator
-        # appears but pymongo raises a client side ValueError in this case.
-        if op["name"] == "updateOne":
-            errors += (ValueError,)
-        return errors
-
-
-def create_test(scenario_def, test, name):
-    @client_context.require_test_commands
-    def run_scenario(self):
-        self.run_scenario(scenario_def, test)
-
-    return run_scenario
-
-
-test_creator = SpecTestCreator(create_test, TestSpec, os.path.join(SPEC_PATH, "legacy"))
-test_creator.create_tests()
-
 if _HAVE_PYMONGOCRYPT:
     globals().update(
         generate_test_classes(
diff --git a/test/test_transactions.py b/test/test_transactions.py
@@ -18,7 +18,6 @@
 import asyncio
 import sys
 from io import BytesIO
-from test.utils_spec_runner import SpecRunner
 
 from gridfs.synchronous.grid_file import GridFS, GridFSBucket
 from pymongo.server_selectors import writable_server_selector
@@ -61,15 +60,8 @@
 UNPIN_TEST_MAX_ATTEMPTS = 50
 
 
-class TransactionsBase(SpecRunner):
-    def maybe_skip_scenario(self, test):
-        super().maybe_skip_scenario(test)
-        if (
-            "secondary" in self.id()
-            and not client_context.is_mongos
-            and not client_context.has_secondaries
-        ):
-            raise unittest.SkipTest("No secondaries")
+class TransactionsBase(IntegrationTest):
+    pass
 
 
 class TestTransactions(TransactionsBase):
