chore: test refactor

Author: ChristoGrab

unit_tests/source_declarative_manifest/conftest.py

@@ -2,7 +2,10 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 #
 
+import sys
+import tempfile
 from pathlib import Path
+from typing import Generator
 
 import pytest
 import yaml
@@ -52,3 +55,49 @@ def valid_local_config_file():
 @pytest.fixture
 def invalid_local_config_file():
     return get_resource_path("invalid_local_pokeapi_config.json")
+
+
+# Sample component code for testing
+SAMPLE_COMPONENTS_PY_TEXT = """
+def sample_function() -> str:
+    return "Hello, World!"
+
+class SimpleClass:
+    def sample_method(self) -> str:
+        return sample_function()
+"""
+
+
+def verify_components_loaded() -> None:
+    """Verify that components were properly loaded."""
+    import components
+
+    assert hasattr(components, "sample_function")
+    assert components.sample_function() == "Hello, World!"
+
+    # Verify the components module is registered in sys.modules
+    assert "components" in sys.modules
+    assert "source_declarative_manifest.components" in sys.modules
+
+    # Verify they are the same module
+    assert sys.modules["components"] is sys.modules["source_declarative_manifest.components"]
+
+
+@pytest.fixture
+def components_file() -> Generator[str, None, None]:
+    """Create a temporary file with sample components code and clean up modules afterwards."""
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as temp_file:
+        temp_file.write(SAMPLE_COMPONENTS_PY_TEXT)
+        temp_file.flush()
+        file_path = temp_file.name
+
+    try:
+        yield file_path
+    finally:
+        # Clean up the modules
+        if "components" in sys.modules:
+            del sys.modules["components"]
+        if "source_declarative_manifest.components" in sys.modules:
+            del sys.modules["source_declarative_manifest.components"]
+        # Clean up the temporary file
+        Path(file_path).unlink(missing_ok=True)

unit_tests/source_declarative_manifest/test_source_declarative_w_custom_components.py

@@ -11,7 +11,7 @@
 from collections.abc import Callable, Mapping
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Any
+from typing import Any, Generator
 
 import pytest
 import yaml
@@ -36,15 +36,10 @@
     register_components_module_from_string,
 )
 from airbyte_cdk.utils.connector_paths import MANIFEST_YAML
-
-SAMPLE_COMPONENTS_PY_TEXT = """
-def sample_function() -> str:
-    return "Hello, World!"
-
-class SimpleClass:
-    def sample_method(self) -> str:
-        return sample_function()
-"""
+from unit_tests.source_declarative_manifest.conftest import (
+    SAMPLE_COMPONENTS_PY_TEXT,
+    verify_components_loaded,
+)
 
 
 def get_resource_path(file_name) -> str:
@@ -293,52 +288,21 @@ def _read_fn(*args, **kwargs):
             _read_fn()
 
 
-def test_register_components_from_file() -> None:
+def test_register_components_from_file(components_file: str) -> None:
     """Test that components can be properly loaded from a file."""
-    # Create a temporary file with the sample components
-    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as temp_file:
-        temp_file.write(SAMPLE_COMPONENTS_PY_TEXT)
-        temp_file.flush()
-        file_path = temp_file.name
-
-    try:
-        # Register the components
-        _register_components_from_file(file_path)
-
-        # Verify the components were loaded correctly
-        import components
-
-        assert hasattr(components, "sample_function")
-        assert components.sample_function() == "Hello, World!"
+    # Register the components
+    _register_components_from_file(components_file)
 
-        # Verify the components module is registered in sys.modules
-        assert "components" in sys.modules
-        assert "source_declarative_manifest.components" in sys.modules
+    # Verify the components were loaded correctly
+    verify_components_loaded()
 
-        # Verify they are the same module
-        assert sys.modules["components"] is sys.modules["source_declarative_manifest.components"]
 
-        # Clean up the modules
-        if "components" in sys.modules:
-            del sys.modules["components"]
-        if "source_declarative_manifest.components" in sys.modules:
-            del sys.modules["source_declarative_manifest.components"]
-    finally:
-        # Clean up the temporary file
-        Path(file_path).unlink(missing_ok=True)
-
-
-def test_parse_components_from_args(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_parse_components_from_args(monkeypatch: pytest.MonkeyPatch, components_file: str) -> None:
     """Test that components can be loaded from command line arguments."""
-    # Create a temporary file with sample components
-    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as temp_file:
-        temp_file.write(SAMPLE_COMPONENTS_PY_TEXT)
-        temp_file.flush()
-        file_path = temp_file.name
 
     # Mock the arguments
     class MockArgs:
-        components_path = file_path
+        components_path = components_file
 
     # Mock the parse_args function to return our mock args
     def mock_parse_args(*args: Any, **kwargs: Any) -> Any:
@@ -349,31 +313,11 @@ def mock_parse_args(*args: Any, **kwargs: Any) -> Any:
 
     monkeypatch.setattr(AirbyteEntrypoint, "parse_args", mock_parse_args)
 
-    try:
-        # Call the function with any args (they'll be ignored due to the mock)
-        result = _parse_components_from_args(["some", "args"])
-
-        # Verify result
-        assert result is True  # Should return True when successful
-
-        # Verify the components were loaded
-        import components
-
-        assert hasattr(components, "sample_function")
-        assert components.sample_function() == "Hello, World!"
-
-        # Verify both module names are registered
-        assert "components" in sys.modules
-        assert "source_declarative_manifest.components" in sys.modules
+    # Call the function with any args (they'll be ignored due to the mock)
+    result = _parse_components_from_args(["some", "args"])
 
-        # Verify they are the same module
-        assert sys.modules["components"] is sys.modules["source_declarative_manifest.components"]
+    # Verify result
+    assert result is True  # Should return True when successful
 
-        # Clean up the modules
-        if "components" in sys.modules:
-            del sys.modules["components"]
-        if "source_declarative_manifest.components" in sys.modules:
-            del sys.modules["source_declarative_manifest.components"]
-    finally:
-        # Clean up the temporary file
-        Path(file_path).unlink(missing_ok=True)
+    # Verify the components were loaded
+    verify_components_loaded()