move test_spec to connector base, clean up imports

Author: aaronsteers

airbyte_cdk/test/standard_tests/connector_base.py

@@ -137,7 +137,39 @@ def create_docker_connector(
             docker_image=docker_image,
         )
 
-    # Test Definitions
+    # Test Definitions (Generic for all connectors)
+
+    def test_spec(
+        self,
+        *,
+        use_docker_image: str | bool,
+    ) -> None:
+        """Standard test for `spec`.
+
+        This test does not require a `scenario` input, since `spec`
+        does not require any inputs.
+
+        We assume `spec` should always succeed and it should always generate
+        a valid `SPEC` message.
+
+        Note: the parsing of messages by type also implicitly validates that
+        the generated `SPEC` message is valid JSON.
+        """
+        scenario = ConnectorTestScenario()  # Empty scenario, empty config
+        result = run_test_job(
+            verb="spec",
+            test_scenario=scenario,
+            connector=self.create_connector(
+                scenario=scenario,
+                use_docker_image=use_docker_image,
+            ),
+        )
+        # If an error occurs, it will be raised above.
+
+        assert len(result.spec_messages) == 1, (
+            "Expected exactly 1 spec message but got {len(result.spec_messages)}",
+            result.errors,
+        )
 
     def test_check(
         self,

airbyte_cdk/test/standard_tests/source_base.py

@@ -12,7 +12,7 @@
     SyncMode,
     Type,
 )
-from airbyte_cdk.test import entrypoint_wrapper
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
 from airbyte_cdk.test.standard_tests._job_runner import run_test_job
 from airbyte_cdk.test.standard_tests.connector_base import (
     ConnectorTestSuiteBase,
@@ -41,7 +41,7 @@ def test_check(
         This test is designed to validate the connector's ability to establish a connection
         and return its status with the expected message type.
         """
-        result: entrypoint_wrapper.EntrypointOutput = run_test_job(
+        result: EntrypointOutput = run_test_job(
             connector=self.create_connector(
                 scenario=scenario,
                 use_docker_image=use_docker_image,
@@ -74,30 +74,6 @@ def test_discover(
             test_scenario=scenario,
         )
 
-    def test_spec(self) -> None:
-        """Standard test for `spec`.
-
-        This test does not require a `scenario` input, since `spec`
-        does not require any inputs.
-
-        We assume `spec` should always succeed and it should always generate
-        a valid `SPEC` message.
-
-        Note: the parsing of messages by type also implicitly validates that
-        the generated `SPEC` message is valid JSON.
-        """
-        result = run_test_job(
-            verb="spec",
-            test_scenario=None,
-            connector=self.create_connector(scenario=None),
-        )
-        # If an error occurs, it will be raised above.
-
-        assert len(result.spec_messages) == 1, (
-            "Expected exactly 1 spec message but got {len(result.spec_messages)}",
-            result.errors,
-        )
-
     def test_basic_read(
         self,
         scenario: ConnectorTestScenario,
@@ -111,7 +87,7 @@ def test_basic_read(
         obtain the catalog of streams, and then it runs a `read` job to fetch
         records from those streams.
         """
-        discover_result = run_test_job(
+        discover_result: EntrypointOutput = run_test_job(
             self.create_connector(
                 scenario,
                 use_docker_image=use_docker_image,
@@ -133,7 +109,7 @@ def test_basic_read(
                 for stream in discover_result.catalog.catalog.streams  # type: ignore [reportOptionalMemberAccess, union-attr]
             ]
         )
-        result = run_test_job(
+        result: EntrypointOutput = run_test_job(
             self.create_connector(
                 scenario,
                 use_docker_image=use_docker_image,
@@ -174,7 +150,7 @@ def test_fail_read_with_bad_catalog(
         )
         # Set expected status to "failed" to ensure the test fails if the connector.
         scenario.status = "failed"
-        result: entrypoint_wrapper.EntrypointOutput = run_test_job(
+        result: EntrypointOutput = run_test_job(
             self.create_connector(
                 scenario,
                 use_docker_image=use_docker_image,