airbyte_cdk.test.utils.manifest_only_fixtures

 1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 2
 3
 4import importlib.util
 5from pathlib import Path
 6from types import ModuleType
 7
 8import pytest
 9
10# The following fixtures are used to load a manifest-only connector's components module and manifest file.
11# They can be accessed from any test file in the connector's unit_tests directory by importing them as follows:
12
13# from airbyte_cdk.test.utils.manifest_only_fixtures import components_module, connector_dir, manifest_path
14
15# individual components can then be referenced as: components_module.<CustomComponentClass>
16
17
18@pytest.fixture(scope="session")
19def connector_dir(request: pytest.FixtureRequest) -> Path:
20    """Return the connector's root directory."""
21
22    current_dir = Path(request.config.invocation_params.dir)
23
24    # If the tests are run locally from the connector's unit_tests directory, return the parent (connector) directory
25    if current_dir.name == "unit_tests":
26        return current_dir.parent
27    # In CI, the tests are run from the connector directory itself
28    return current_dir
29
30
31@pytest.fixture(scope="session")
32def components_module(connector_dir: Path) -> ModuleType | None:
33    """Load and return the components module from the connector directory.
34
35    This assumes the components module is located at <connector_dir>/components.py.
36    """
37    components_path = connector_dir / "components.py"
38    if not components_path.exists():
39        return None
40
41    components_spec = importlib.util.spec_from_file_location("components", components_path)
42    if components_spec is None:
43        return None
44
45    components_module = importlib.util.module_from_spec(components_spec)
46    if components_spec.loader is None:
47        return None
48
49    components_spec.loader.exec_module(components_module)
50    return components_module
51
52
53@pytest.fixture(scope="session")
54def manifest_path(connector_dir: Path) -> Path:
55    """Return the path to the connector's manifest file."""
56    path = connector_dir / "manifest.yaml"
57    if not path.exists():
58        raise FileNotFoundError(f"Manifest file not found at {path}")
59    return path