move tests over

Author: pedroslopez

airbyte_cdk/manifest_runner/main.py

@@ -1,32 +1,13 @@
 # Copyright (c) 2025 Airbyte, Inc., all rights reserved.
-"""Main entry point for the Airbyte CDK Manifest Runner server."""
-
-import sys
+"""Main entry point for the Airbyte Manifest Runner server."""
 
 import uvicorn
 
 
-def check_dependencies() -> bool:
-    """Check if all required dependencies are available."""
-    try:
-        import fastapi
-        import uvicorn
-
-        return True
-    except ImportError:
-        return False
-
-
 def run_server(
     host: str = "127.0.0.1", port: int = 8000, reload: bool = False, log_level: str = "info"
 ) -> None:
     """Run the FastAPI server."""
-    if not check_dependencies():
-        print("❌ Manifest runner dependencies not found. Please install with:")
-        print("  pip install airbyte-cdk[manifest-runner]")
-        print("  # or")
-        print("  poetry install --extras manifest-runner")
-        sys.exit(1)
 
     print(f"🚀 Starting Airbyte CDK Manifest Runner on {host}:{port}")
 

airbyte_cdk/manifest_runner/routers/manifest.py

@@ -86,4 +86,4 @@ def full_resolve(request: FullResolveRequest) -> ManifestResponse:
         streams.extend(generated_streams_list)
 
     manifest["streams"] = streams
-    return ManifestResponse(manifest)
+    return ManifestResponse(manifest=manifest)

poetry.lock

@@ -87,8 +87,7 @@ pytest = {version = "^7", optional = true }
 orjson = "^3.10.7"
 serpyco-rs = "^1.10.2"
 sqlalchemy = {version = "^2.0,!=2.0.36", optional = true }
-fastapi = { version = "^0.104.0", optional = true }
-uvicorn = { version = "^0.24.0", optional = true }
+fastapi = { version = ">=0.116.1", extras = ["standard"], optional = true }
 xmltodict = ">=0.13,<0.15"
 anyascii = "^0.3.2"
 whenever = "^0.6.16"
@@ -123,7 +122,7 @@ file-based = ["avro", "fastavro", "pyarrow", "unstructured", "pdf2image", "pdfmi
 vector-db-based = ["langchain", "openai", "cohere", "tiktoken"]
 sql = ["sqlalchemy"]
 dev = ["pytest"]
-manifest-runner = ["fastapi", "uvicorn"]
+manifest-runner = ["fastapi"]
 
 [tool.poetry.scripts]
 

pyproject.toml

@@ -0,0 +1 @@
+"""Tests for the manifest_runner package."""

unit_tests/manifest_runner/__init__.py

@@ -0,0 +1,142 @@
+from unittest.mock import Mock, patch
+
+import pytest
+
+from airbyte_cdk.manifest_runner.manifest_runner.runner import ManifestRunner
+
+
+class TestManifestRunner:
+    """Test cases for the ManifestRunner class."""
+
+    @pytest.fixture
+    def mock_source(self):
+        """Create a mock ManifestDeclarativeSource."""
+        return Mock()
+
+    @pytest.fixture
+    def manifest_runner(self, mock_source):
+        """Create a ManifestRunner instance with mocked source."""
+        return ManifestRunner(mock_source)
+
+    @pytest.fixture
+    def sample_config(self):
+        """Sample configuration for testing."""
+        return {"api_key": "test_key", "base_url": "https://api.example.com"}
+
+    @pytest.fixture
+    def sample_catalog(self):
+        """Sample configured catalog for testing."""
+        from airbyte_cdk.models.airbyte_protocol import (
+            AirbyteStream,
+            ConfiguredAirbyteCatalog,
+            ConfiguredAirbyteStream,
+            DestinationSyncMode,
+            SyncMode,
+        )
+
+        return ConfiguredAirbyteCatalog(
+            streams=[
+                ConfiguredAirbyteStream(
+                    stream=AirbyteStream(
+                        name="test_stream",
+                        json_schema={"type": "object"},
+                        supported_sync_modes=[SyncMode.full_refresh],
+                    ),
+                    sync_mode=SyncMode.full_refresh,
+                    destination_sync_mode=DestinationSyncMode.overwrite,
+                )
+            ]
+        )
+
+    @pytest.fixture
+    def sample_state(self):
+        """Sample state messages for testing."""
+        return []
+
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.runner.TestReader")
+    def test_test_read_success(
+        self, mock_test_reader_class, manifest_runner, sample_config, sample_catalog
+    ):
+        """Test successful test_read execution with various parameters and state messages."""
+        from airbyte_cdk.models.airbyte_protocol import (
+            AirbyteStateMessage,
+            AirbyteStateType,
+        )
+
+        # Mock the TestReader instance and its run_test_read method
+        mock_test_reader_instance = Mock()
+        mock_test_reader_class.return_value = mock_test_reader_instance
+
+        # Mock the StreamRead return value
+        mock_stream_read = Mock()
+        mock_test_reader_instance.run_test_read.return_value = mock_stream_read
+
+        # Test with state messages and various parameter values
+        state_messages = [
+            AirbyteStateMessage(
+                type=AirbyteStateType.STREAM,
+                stream={
+                    "stream_descriptor": {"name": "test_stream"},
+                    "stream_state": {"cursor": "2023-01-01"},
+                },
+            )
+        ]
+
+        record_limit = 50
+        page_limit = 3
+        slice_limit = 7
+
+        # Execute test_read
+        result = manifest_runner.test_read(
+            config=sample_config,
+            catalog=sample_catalog,
+            state=state_messages,
+            record_limit=record_limit,
+            page_limit=page_limit,
+            slice_limit=slice_limit,
+        )
+
+        # Verify TestReader was initialized with correct parameters
+        mock_test_reader_class.assert_called_once_with(
+            max_pages_per_slice=page_limit,
+            max_slices=slice_limit,
+            max_record_limit=record_limit,
+        )
+
+        # Verify run_test_read was called with correct parameters including state
+        mock_test_reader_instance.run_test_read.assert_called_once_with(
+            source=manifest_runner._source,
+            config=sample_config,
+            configured_catalog=sample_catalog,
+            state=state_messages,
+        )
+
+        # Verify the result is returned correctly
+        assert result == mock_stream_read
+
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.runner.TestReader")
+    def test_test_read_exception_handling(
+        self,
+        mock_test_reader_class,
+        manifest_runner,
+        sample_config,
+        sample_catalog,
+        sample_state,
+    ):
+        """Test that exceptions from TestReader are properly propagated."""
+        mock_test_reader_instance = Mock()
+        mock_test_reader_class.return_value = mock_test_reader_instance
+
+        # Make run_test_read raise an exception
+        mock_test_reader_instance.run_test_read.side_effect = Exception("Test error")
+
+        # Verify the exception is propagated
+        with pytest.raises(Exception, match="Test error"):
+            manifest_runner.test_read(
+                config=sample_config,
+                catalog=sample_catalog,
+                state=sample_state,
+                record_limit=100,
+                page_limit=5,
+                slice_limit=10,
+            )

unit_tests/manifest_runner/manifest_runner/__init__.py

@@ -0,0 +1,133 @@
+from unittest.mock import Mock, patch
+
+from airbyte_cdk.manifest_runner.manifest_runner.utils import (
+    SHOULD_MIGRATE_KEY,
+    SHOULD_NORMALIZE_KEY,
+    build_catalog,
+    build_source,
+)
+
+
+class TestManifestUtils:
+    """Test cases for the manifest_runner utils module."""
+
+    def test_build_catalog_creates_correct_structure(self):
+        """Test that build_catalog creates a properly structured ConfiguredAirbyteCatalog."""
+        stream_name = "test_stream"
+        catalog = build_catalog(stream_name)
+
+        # Verify catalog structure
+        assert len(catalog.streams) == 1
+
+        configured_stream = catalog.streams[0]
+        assert configured_stream.stream.name == stream_name
+        assert configured_stream.stream.json_schema == {}
+
+        # Verify sync modes
+        from airbyte_cdk.models.airbyte_protocol import DestinationSyncMode, SyncMode
+
+        assert SyncMode.full_refresh in configured_stream.stream.supported_sync_modes
+        assert SyncMode.incremental in configured_stream.stream.supported_sync_modes
+        assert configured_stream.sync_mode == SyncMode.incremental
+        assert configured_stream.destination_sync_mode == DestinationSyncMode.overwrite
+
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.utils.ManifestDeclarativeSource")
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.utils.ModelToComponentFactory")
+    def test_build_source_creates_manifest_declarative_source(
+        self, mock_component_factory_class, mock_source_class
+    ):
+        """Test that build_source creates a ManifestDeclarativeSource with correct parameters."""
+        # Setup mocks
+        mock_component_factory = Mock()
+        mock_component_factory_class.return_value = mock_component_factory
+        mock_source = Mock()
+        mock_source_class.return_value = mock_source
+
+        # Test with complex manifest and config structures
+        manifest = {
+            "version": "0.1.0",
+            "definitions": {"selector": {"extractor": {"field_path": ["data"]}}},
+            "streams": [
+                {
+                    "name": "users",
+                    "primary_key": "id",
+                    "retriever": {
+                        "requester": {
+                            "url_base": "https://api.example.com",
+                            "path": "/users",
+                        }
+                    },
+                }
+            ],
+            "check": {"stream_names": ["users"]},
+        }
+
+        config = {
+            "api_key": "sk-test-123",
+            "base_url": "https://api.example.com",
+            "timeout": 30,
+        }
+
+        # Call build_source
+        result = build_source(manifest, config)
+
+        # Verify ModelToComponentFactory was created with correct parameters
+        mock_component_factory_class.assert_called_once_with(
+            emit_connector_builder_messages=True,
+            limit_pages_fetched_per_slice=None,
+            limit_slices_fetched=None,
+            disable_retries=True,
+            disable_cache=True,
+        )
+
+        # Verify ManifestDeclarativeSource was created with correct parameters
+        mock_source_class.assert_called_once_with(
+            source_config=manifest,
+            config=config,
+            normalize_manifest=False,  # Default when flag not set
+            migrate_manifest=False,  # Default when flag not set
+            emit_connector_builder_messages=True,
+            component_factory=mock_component_factory,
+        )
+
+        assert result == mock_source
+
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.utils.ManifestDeclarativeSource")
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.utils.ModelToComponentFactory")
+    def test_build_source_with_normalize_flag(
+        self, mock_component_factory_class, mock_source_class
+    ):
+        """Test build_source when normalize flag is set."""
+        mock_component_factory = Mock()
+        mock_component_factory_class.return_value = mock_component_factory
+        mock_source = Mock()
+        mock_source_class.return_value = mock_source
+
+        manifest = {"streams": [{"name": "test_stream"}], SHOULD_NORMALIZE_KEY: True}
+        config = {"api_key": "test_key"}
+
+        build_source(manifest, config)
+
+        # Verify normalize_manifest is True
+        call_args = mock_source_class.call_args[1]
+        assert call_args["normalize_manifest"] is True
+        assert call_args["migrate_manifest"] is False
+
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.utils.ManifestDeclarativeSource")
+    @patch("airbyte_cdk.manifest_runner.manifest_runner.utils.ModelToComponentFactory")
+    def test_build_source_with_migrate_flag(self, mock_component_factory_class, mock_source_class):
+        """Test build_source when migrate flag is set."""
+        mock_component_factory = Mock()
+        mock_component_factory_class.return_value = mock_component_factory
+        mock_source = Mock()
+        mock_source_class.return_value = mock_source
+
+        manifest = {"streams": [{"name": "test_stream"}], SHOULD_MIGRATE_KEY: True}
+        config = {"api_key": "test_key"}
+
+        build_source(manifest, config)
+
+        # Verify migrate_manifest is True
+        call_args = mock_source_class.call_args[1]
+        assert call_args["normalize_manifest"] is False
+        assert call_args["migrate_manifest"] is True

unit_tests/manifest_runner/manifest_runner/test_runner.py

@@ -0,0 +1 @@
+"""Tests for the routers package."""