PYTHON-5663 Fix unified test discovery

test/asynchronous/test_auth_oidc.py

@@ -30,7 +30,7 @@
 
 sys.path[0:0] = [""]
 
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.utils_shared import EventListener, OvertCommandListener
 
 from bson import SON
@@ -54,14 +54,13 @@
 _IS_SYNC = False
 
 ROOT = Path(__file__).parent.parent.resolve()
-TEST_PATH = ROOT / "auth" / "unified"
 ENVIRON = os.environ.get("OIDC_ENV", "test")
 DOMAIN = os.environ.get("OIDC_DOMAIN", "")
 TOKEN_DIR = os.environ.get("OIDC_TOKEN_DIR", "")
 TOKEN_FILE = os.environ.get("OIDC_TOKEN_FILE", "")
 
 # Generate unified tests.
-globals().update(generate_test_classes(str(TEST_PATH), module=__name__))
+globals().update(generate_test_classes(get_test_path("auth", "unified"), module=__name__))
 
 pytestmark = pytest.mark.auth_oidc
 

test/asynchronous/test_auth_spec.py

@@ -27,16 +27,15 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 from pymongo import AsyncMongoClient
 from pymongo.auth_oidc_shared import OIDCCallback
 
 pytestmark = pytest.mark.auth
 
 _IS_SYNC = False
-
-_TEST_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "auth")
+_TEST_PATH = get_test_path("auth")
 
 
 class TestAuthSpec(AsyncPyMongoTestCase):

test/asynchronous/test_change_stream.py

@@ -35,7 +35,7 @@
     async_client_context,
     unittest,
 )
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.utils_shared import (
     AllowListEventListener,
     EventListener,
@@ -1143,12 +1143,9 @@ def asyncTearDown(self):
         self.listener.reset()
 
 
-_TEST_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "change_streams")
-
-
 globals().update(
     generate_test_classes(
-        os.path.join(_TEST_PATH, "unified"),
+        get_test_path("change_streams", "unified"),
         module=__name__,
     )
 )

test/asynchronous/test_client_metadata.py

@@ -19,7 +19,7 @@
 import time
 import unittest
 from test.asynchronous import AsyncIntegrationTest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.utils_shared import CMAPListener
 from typing import Any, Optional
 
@@ -40,16 +40,8 @@
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "handshake", "unified")
-else:
-    _TEST_PATH = os.path.join(
-        pathlib.Path(__file__).resolve().parent.parent, "handshake", "unified"
-    )
-
 # Generate unified tests.
-globals().update(generate_test_classes(_TEST_PATH, module=__name__))
+globals().update(generate_test_classes(get_test_path("handshake", "unified"), module=__name__))
 
 
 def _get_handshake_driver_info(request):

test/asynchronous/test_collection_management.py

@@ -22,20 +22,12 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "collection_management")
-else:
-    _TEST_PATH = os.path.join(
-        pathlib.Path(__file__).resolve().parent.parent, "collection_management"
-    )
-
 # Generate unified tests.
-globals().update(generate_test_classes(_TEST_PATH, module=__name__))
+globals().update(generate_test_classes(get_test_path("collection_management"), module=__name__))
 
 if __name__ == "__main__":
     unittest.main()

test/asynchronous/test_command_logging.py

@@ -22,20 +22,13 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "command_logging")
-else:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "command_logging")
-
-
 globals().update(
     generate_test_classes(
-        _TEST_PATH,
+        get_test_path("command_logging"),
         module=__name__,
     )
 )

test/asynchronous/test_command_monitoring.py

@@ -22,20 +22,13 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "command_monitoring")
-else:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "command_monitoring")
-
-
 globals().update(
     generate_test_classes(
-        _TEST_PATH,
+        get_test_path("command_monitoring"),
         module=__name__,
     )
 )

test/asynchronous/test_connection_logging.py

@@ -22,20 +22,13 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "connection_logging")
-else:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "connection_logging")
-
-
 globals().update(
     generate_test_classes(
-        _TEST_PATH,
+        get_test_path("connection_logging"),
         module=__name__,
     )
 )

test/asynchronous/test_crud_unified.py

@@ -22,18 +22,12 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "crud", "unified")
-else:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "crud", "unified")
-
 # Generate unified tests.
-globals().update(generate_test_classes(_TEST_PATH, module=__name__))
+globals().update(generate_test_classes(get_test_path("crud", "unified"), module=__name__))
 
 if __name__ == "__main__":
     unittest.main()

test/asynchronous/test_csot.py

@@ -22,7 +22,7 @@
 sys.path[0:0] = [""]
 
 from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.asynchronous.utils import flaky
 
 import pymongo
@@ -31,14 +31,8 @@
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    TEST_PATH = os.path.join(Path(__file__).resolve().parent, "csot")
-else:
-    TEST_PATH = os.path.join(Path(__file__).resolve().parent.parent, "csot")
-
 # Generate unified tests.
-globals().update(generate_test_classes(TEST_PATH, module=__name__))
+globals().update(generate_test_classes(get_test_path("csot"), module=__name__))
 
 
 class TestCSOT(AsyncIntegrationTest):

test/asynchronous/test_discovery_and_monitoring.py

@@ -40,7 +40,7 @@
     unittest,
 )
 from test.asynchronous.pymongo_mocks import DummyMonitor
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.asynchronous.utils import (
     async_get_pool,
 )
@@ -76,14 +76,7 @@
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    SDAM_PATH = os.path.join(Path(__file__).resolve().parent, "discovery_and_monitoring")
-else:
-    SDAM_PATH = os.path.join(
-        Path(__file__).resolve().parent.parent,
-        "discovery_and_monitoring",
-    )
+SDAM_PATH = get_test_path("discovery_and_monitoring")
 
 
 async def create_mock_topology(uri, monitor_class=DummyMonitor):

test/asynchronous/test_encryption.py

@@ -54,7 +54,7 @@
     unittest,
 )
 from test.asynchronous.test_bulk import AsyncBulkTestBase
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.asynchronous.utils_spec_runner import AsyncSpecRunner
 from test.helpers_shared import (
     ALL_KMS_PROVIDERS,
@@ -275,11 +275,7 @@ def unmanaged_create_client_encryption(
 
 
 # Location of JSON test files.
-if _IS_SYNC:
-    BASE = os.path.join(pathlib.Path(__file__).resolve().parent, "client-side-encryption")
-else:
-    BASE = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "client-side-encryption")
-
+BASE = get_test_path("client-side-encryption")
 SPEC_PATH = os.path.join(BASE, "spec")
 
 OPTS = CodecOptions()

test/asynchronous/test_gridfs_spec.py

@@ -22,18 +22,12 @@
 sys.path[0:0] = [""]
 
 from test import unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    TEST_PATH = os.path.join(Path(__file__).resolve().parent, "gridfs")
-else:
-    TEST_PATH = os.path.join(Path(__file__).resolve().parent.parent, "gridfs")
-
 # Generate unified tests.
-globals().update(generate_test_classes(TEST_PATH, module=__name__))
+globals().update(generate_test_classes(get_test_path("gridfs"), module=__name__))
 
 if __name__ == "__main__":
     unittest.main()

test/asynchronous/test_index_management.py

@@ -28,7 +28,7 @@
 sys.path[0:0] = [""]
 
 from test.asynchronous import AsyncIntegrationTest, AsyncPyMongoTestCase, unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.utils_shared import AllowListEventListener, OvertCommandListener
 
 from pymongo.errors import OperationFailure
@@ -40,12 +40,6 @@
 
 pytestmark = pytest.mark.search_index
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "index_management")
-else:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "index_management")
-
 _NAME = "test-search-index"
 
 
@@ -370,7 +364,7 @@ async def test_case_7(self):
 
 globals().update(
     generate_test_classes(
-        _TEST_PATH,
+        get_test_path("index_management"),
         module=__name__,
     )
 )

test/asynchronous/test_load_balancer.py

@@ -30,7 +30,7 @@
 sys.path[0:0] = [""]
 
 from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.utils_shared import (
     async_wait_until,
     create_async_event,
@@ -40,14 +40,8 @@
 
 pytestmark = pytest.mark.load_balancer
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "load_balancer")
-else:
-    _TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent.parent, "load_balancer")
-
 # Generate unified tests.
-globals().update(generate_test_classes(_TEST_PATH, module=__name__))
+globals().update(generate_test_classes(get_test_path("load_balancer"), module=__name__))
 
 
 class TestLB(AsyncIntegrationTest):

test/asynchronous/test_read_write_concern_spec.py

@@ -24,7 +24,7 @@
 sys.path[0:0] = [""]
 
 from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
-from test.asynchronous.unified_format import generate_test_classes
+from test.asynchronous.unified_format import generate_test_classes, get_test_path
 from test.utils_shared import OvertCommandListener
 
 from pymongo import DESCENDING
@@ -42,11 +42,7 @@
 
 _IS_SYNC = False
 
-# Location of JSON test specifications.
-if _IS_SYNC:
-    TEST_PATH = os.path.join(Path(__file__).resolve().parent, "read_write_concern")
-else:
-    TEST_PATH = os.path.join(Path(__file__).resolve().parent.parent, "read_write_concern")
+TEST_PATH = get_test_path("read_write_concern")
 
 
 class TestReadWriteConcernSpec(AsyncIntegrationTest):