Move integration test into examples

.github/workflows/pipeline.yaml

@@ -32,6 +32,26 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           COVERALLS_FLAG_NAME: test-${{ matrix.python-version }}
 
+  integration-test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [ '3.8', '3.9', '3.10' ]
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v1
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          pip install --upgrade pip
+          pip install poetry==1.4.0
+          poetry install --all-extras --with test
+
+      - name: Execute sample pipeline
+        run: ./scripts/run_integration_tests.sh
+
   finish-coveralls:
     needs: test
     runs-on: ubuntu-latest

docs/runners/local.md

@@ -46,7 +46,7 @@ about this in the [installation](../guides/installation.md) guide.
         fondant run local <pipeline_ref> --auth-azure
         ```
 
-    You can also use the `--extra_volumes` argument to mount extra credentials or additional files.
+    You can also use the `--extra-volumes` argument to mount extra credentials or additional files.
     This volumes will be mounted to every component/service of the docker-compose spec.
 
 

examples/sample_pipeline/README.md

@@ -0,0 +1,13 @@
+# Sample pipeline
+
+This example is a simple sample pipeline which uses two reusable components
+(load_from_parquet, chunk_text), and a custom dummy component. The custom dummy component only
+returns the received dataframe. 
+
+The pipeline can be executed with the Fondant cli:
+
+```bash
+fondant run local pipeline.py
+```
+
+The automated integration test will use the `run.sh` script. 

examples/sample_pipeline/components/dummy_component/Dockerfile

@@ -0,0 +1,13 @@
+FROM --platform=linux/amd64 python:3.8-slim
+
+# install requirements
+COPY requirements.txt ./
+RUN pip3 install --no-cache-dir -r requirements.txt
+
+# Set the working directory to the component folder
+WORKDIR /component/src
+
+# Copy over src-files and spec of the component
+COPY src/ .
+
+ENTRYPOINT ["fondant", "execute", "main"]

tests/integration_tests/sample_pipeline_test/components/dummy_component/fondant_component.yaml → examples/sample_pipeline/components/dummy_component/fondant_component.yaml

@@ -1,7 +1,6 @@
 name: Dummy component
 description: Dummy component for testing custom components
-
-image: fndnt/dummy_component:dev
+image: dummy_component
 
 consumes:
   text_data:

examples/sample_pipeline/components/dummy_component/requirements.txt

@@ -0,0 +1 @@
+fondant[component]

tests/integration_tests/sample_pipeline_test/components/dummy_component/src/main.py → examples/sample_pipeline/components/dummy_component/src/main.py

@@ -16,9 +16,10 @@
 class DummyComponent(PandasTransformComponent):
     """Dummy component that returns the dataframe as it is."""
 
-    def __init__(self, *_):
+    def __init__(self, *_, **kwargs):
         pass
 
     def transform(self, dataframe: pd.DataFrame) -> pd.DataFrame:
         """Dummy component that returns the dataframe as it is."""
+        # raise RuntimeError
         return dataframe

examples/sample_pipeline/pipeline.py

@@ -0,0 +1,37 @@
+# This file contains a sample pipeline. Loading data from a parquet file,
+# using the load_from_parquet component, chain a custom dummy component, and use
+# the reusable chunking component
+import pyarrow as pa
+from pathlib import Path
+from fondant.pipeline import Pipeline
+
+BASE_PATH = Path("./.artifacts").resolve()
+BASE_PATH.mkdir(parents=True, exist_ok=True)
+
+# Define pipeline
+pipeline = Pipeline(name="dummy-pipeline", base_path=str(BASE_PATH))
+
+# Load from hub component
+load_component_column_mapping = {
+    "text": "text_data",
+}
+
+dataset = pipeline.read(
+    name_or_path="load_from_parquet",
+    arguments={
+        "dataset_uri": "/data/sample.parquet",
+        "column_name_mapping": load_component_column_mapping,
+        "n_rows_to_load": 5,
+    },
+    produces={"text_data": pa.string()},
+)
+
+dataset = dataset.apply(
+    name_or_path="./components/dummy_component",
+)
+
+dataset.apply(
+    name_or_path="chunk_text",
+    arguments={"chunk_size": 10, "chunk_overlap": 2},
+    consumes={"text": "text_data"},
+)

examples/sample_pipeline/run.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+# This script executes the sample pipeline in the example folder, checks the correct execution and
+# cleans up the directory again
+set -e
+
+
+# Setup teardown
+cleanup() {
+  rv=$?
+
+  # Try to remove .artifact folder
+  artifact_directory="./.artifacts"
+
+  if [ -d "$artifact_directory" ]; then
+    # Directory exists, remove it
+    rm -rf "$artifact_directory"
+  fi
+
+  exit $rv
+}
+
+trap cleanup EXIT
+
+# Bind local data directory to pipeline
+data_dir=$(readlink -f "data")
+
+# Run pipeline
+fondant run local pipeline.py \
+  --extra-volumes $data_dir:/data

scripts/run_integration_tests.sh

@@ -0,0 +1,34 @@
+#!/bin/bash
+# This script executes the sample pipeline in the example folder, checks the correct execution and
+# cleans up the directory again
+echo "Start integration tests execution ..."
+
+failed_tests=()
+
+# Find all run.sh scripts and execute them
+for test_script in ./examples/*/run.sh; do
+    test_name=$(basename "$(dirname "$test_script")")
+
+    echo "Running test: $test_name"
+
+    # Set working dir to the currect integration test
+    cd $(dirname "$test_script")
+
+    # Execute the run.sh script
+    bash ./run.sh
+
+    # Check the exit status
+    if [ $? -ne 0 ]; then
+        echo "Test $test_name failed!"
+        failed_tests+=("$test_name")
+    fi
+done
+
+echo "Tests completed"
+
+if [ ${#failed_tests[@]} -eq 0 ]; then
+    echo "All tests passed!"
+else
+    echo "Failed tests: ${failed_tests[@]}"
+    exit 1  # Indicate failure to cicd
+fi

src/fondant/pipeline/pipeline.py

@@ -491,7 +491,8 @@ def _validate_pipeline_definition(self, run_id: str):
                         msg = (
                             f"Component '{component_op.name}' is trying to invoke the field "
                             f"'{component_field_name}', which has not been defined or created "
-                            f"in the previous components."
+                            f"in the previous components. \n"
+                            f"Available field names: {list(manifest.fields.keys())}"
                         )
                         raise InvalidPipelineDefinition(
                             msg,

src/fondant/pipeline/runner.py

@@ -41,7 +41,7 @@ def _run(self, input_spec: str, *args, **kwargs):
         ]
 
         print("Starting pipeline run...")
-        subprocess.call(cmd)  # nosec
+        subprocess.check_call(cmd)  # nosec
         print("Finished pipeline run.")
 
     def run(

tests/pipeline/test_runner.py

@@ -23,7 +23,7 @@
 
 def test_docker_runner():
     """Test that the docker runner while mocking subprocess.call."""
-    with mock.patch("subprocess.call") as mock_call:
+    with mock.patch("subprocess.check_call") as mock_call:
         DockerRunner().run("some/path")
         mock_call.assert_called_once_with(
             [
@@ -41,7 +41,7 @@ def test_docker_runner():
 
 
 def test_docker_runner_from_pipeline():
-    with mock.patch("subprocess.call") as mock_call:
+    with mock.patch("subprocess.check_call") as mock_call:
         DockerRunner().run(PIPELINE)
         mock_call.assert_called_once_with(
             [

tests/test_cli.py

@@ -274,7 +274,7 @@ def test_local_run():
         extra_volumes=[],
         build_arg=[],
     )
-    with patch("subprocess.call") as mock_call:
+    with patch("subprocess.check_call") as mock_call:
         run_local(args)
         mock_call.assert_called_once_with(
             [
@@ -290,7 +290,7 @@ def test_local_run():
             ],
         )
 
-    with patch("subprocess.call") as mock_call:
+    with patch("subprocess.check_call") as mock_call:
         args1 = argparse.Namespace(
             local=True,
             vertex=False,
@@ -330,7 +330,7 @@ def test_local_run_cloud_credentials():
         with patch(
             "fondant.pipeline.compiler.DockerCompiler.compile",
         ) as mock_compiler, patch(
-            "subprocess.call",
+            "subprocess.check_call",
         ) as mock_runner:
             args = argparse.Namespace(
                 local=True,