Enable running all integration tests against Kind in the CI

.github/workflows/test.yml

@@ -1,7 +1,7 @@
 name: test
 on:
   push:
-    branches: [ "main" ]
+    branches: [ "main", "issue-81-enable-integration-tests" ]
   pull_request_target:
     branches: ["main"]
 
@@ -68,7 +68,7 @@ jobs:
           path: .coverage
           include-hidden-files: true
 
-  Run-Integration-Tests:
+  Run-GKE-Integration-Tests:
     needs: Authorize
     runs-on: ubuntu-latest
     strategy:
@@ -134,26 +134,109 @@ jobs:
 
       - name: Run integration tests
         run: |
-          hatch run tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }}:test-integration
+          hatch run tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }}:test-integration-gke
         env:
-          RAY_SPEC_FILENAME: "ray-gke.yaml"
+          RAY_SPEC_FILENAME: "ray-linux.yaml"
+          USE_GKE: 1
 
       - name: Upload coverage to Github
         uses: actions/upload-artifact@v4
         with:
           name: coverage-integration-test-${{ matrix.python-version }}-${{ matrix.airflow-version }}
           path: .coverage
           include-hidden-files: true
+
       - name: Delete GKE cluster
         if: always()
         run: |
           gcloud container clusters delete ${{ steps.cluster-name.outputs.name }} --zone us-central1-a --quiet
 
+  Run-Kind-Integration-Tests:
+    needs: Authorize
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [ "3.11" ]
+        airflow-version: [ "2.9" ]
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          ref: ${{ github.event.pull_request.head.sha || github.ref }}
+      - uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cache/pip
+            .local/share/hatch/
+          key: coverage-integration-kubernetes-test-${{ runner.os }}-${{ matrix.python-version }}-${{ matrix.airflow-version }}-${{ hashFiles('pyproject.toml') }}-${{ hashFiles('cosmos/__init__.py') }}
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Create KinD cluster
+        uses: container-tools/kind-action@v1
+        #with:
+        #  config: dev/kind-config.yaml
+
+      - name: Install MetalLB
+        run: |
+          kubectl apply -f https://raw.githubusercontent.com/metallb/metallb/v0.13.10/config/manifests/metallb-native.yaml
+
+      - name: Wait for MetalLB to be ready
+        run: |
+          echo "Waiting for MetalLB controller and webhook to be ready..."
+          kubectl get pods -n metallb-system --show-labels
+          kubectl wait --namespace metallb-system --for=condition=Ready pod --selector=app=metallb,component=controller --timeout=120s
+          kubectl wait --namespace metallb-system --for=condition=Ready pod --selector=app=metallb,component=speaker --timeout=120s
+
+      - name: Configure IPAddressPool
+        run: |
+          cat <<EOF | kubectl apply -f -
+          apiVersion: metallb.io/v1beta1
+          kind: IPAddressPool
+          metadata:
+            name: kind-pool
+            namespace: metallb-system
+          spec:
+            addresses:
+            - 172.18.255.1-172.18.255.254
+          ---
+          apiVersion: metallb.io/v1beta1
+          kind: L2Advertisement
+          metadata:
+            name: kind-l2-advertisement
+            namespace: metallb-system
+          spec: {}
+          EOF
+
+      - name: Install packages and dependencies
+        run: |
+          python -m pip install uv
+          uv pip install --system hatch
+          hatch -e tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }} run pip freeze
+
+      - name: Run integration tests in Kind
+        run: |
+          hatch run tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }}:test-integration-kind
+        env:
+          RAY_SPEC_FILENAME: "ray-linux.yaml"
+          AIRFLOW_HOME: /home/runner/work/astro-provider-ray/astro-provider-ray/
+          KUBECONFIG: /home/runner/.kube/config
+
+      - name: Upload coverage to Github
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-integration-kubernetes-test-${{ matrix.python-version }}-${{ matrix.airflow-version }}
+          path: .coverage
+          include-hidden-files: true
+
   Code-Coverage:
     if: github.event.action != 'labeled'
     needs:
       - Run-Unit-Tests
-      - Run-Integration-Tests
+      - Run-GKE-Integration-Tests
+      - Run-Kind-Integration-Tests
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3

dev/dags/ray_single_operator.py

@@ -6,7 +6,7 @@
 from ray_provider.operators import SubmitRayJob
 
 CONN_ID = "ray_conn"
-RAY_SPEC = Path(__file__).parent / "scripts/ray.yaml"
+RAY_SPEC = Path(__file__).parent / "scripts/ray-macos.yaml"
 FOLDER_PATH = Path(__file__).parent / "ray_scripts"
 RAY_RUNTIME_ENV = {"working_dir": str(FOLDER_PATH)}
 

dev/dags/ray_taskflow_example.py

@@ -7,7 +7,7 @@
 from ray_provider.decorators import ray
 
 CONN_ID = "ray_conn"
-RAY_SPEC_FILENAME = os.getenv("RAY_SPEC_FILENAME", "ray.yaml")
+RAY_SPEC_FILENAME = os.getenv("RAY_SPEC_FILENAME", "ray-macos.yaml")
 RAY_SPEC = Path(__file__).parent / "scripts" / RAY_SPEC_FILENAME
 
 FOLDER_PATH = Path(__file__).parent / "ray_scripts"

dev/dags/setup-teardown.py

@@ -6,7 +6,7 @@
 from ray_provider.operators import DeleteRayCluster, SetupRayCluster, SubmitRayJob
 
 CONN_ID = "ray_conn"
-RAY_SPEC = Path(__file__).parent / "scripts/ray.yaml"
+RAY_SPEC = Path(__file__).parent / "scripts/ray-macos.yaml"
 FOLDER_PATH = Path(__file__).parent / "ray_scripts"
 
 with DAG(

dev/tests/dags/test_dag_example.py

@@ -71,4 +71,4 @@ def test_dag_retries(dag_id, dag, fileloc):
     """
     test if a DAG has retries set
     """
-    assert dag.default_args.get("retries", None) >= 2, f"{dag_id} in {fileloc} must have task retries >= 2."
+    assert dag.default_args.get("retries", 2) >= 2, f"{dag_id} in {fileloc} must have task retries >= 2."

pyproject.toml

@@ -94,7 +94,8 @@ airflow = ["2.7", "2.8", "2.9"]
 freeze = "pip freeze"
 test = 'sh scripts/test/unit_test.sh'
 test-cov = 'sh scripts/test/unit_cov.sh'
-test-integration = 'sh scripts/test/integration_test.sh'
+test-integration-gke = 'sh scripts/test/integration_gke_test.sh'
+test-integration-kind = 'sh scripts/test/integration_kind_test.sh'
 static-check = "pre-commit run --all-files"
 
 [tool.pytest.ini_options]

ray_provider/hooks.py

@@ -318,7 +318,7 @@ def _wait_for_load_balancer(
         service_name: str,
         namespace: str = "default",
         max_retries: int = 30,
-        retry_interval: int = 40,
+        retry_interval: int = 15,
     ) -> dict[str, Any]:
         """
         Wait for the LoadBalancer to be ready and return its details.

scripts/test/integration_test.sh → scripts/test/integration_gke_test.sh

@@ -1,8 +1,8 @@
 pytest -vv \
+    -k "ray_taskflow_example" \
     --cov=ray_provider \
     --cov-report=term-missing \
     --cov-report=xml \
-    --durations=0 \
     -m integration \
     -s \
     --log-cli-level=DEBUG

scripts/test/integration_kind_test.sh

@@ -0,0 +1,8 @@
+pytest -vv \
+    -k "not ray_taskflow_example_existing_cluster" \
+    --cov=ray_provider \
+    --cov-report=term-missing \
+    --cov-report=xml \
+    -m integration \
+    -s \
+    --log-cli-level=DEBUG

tests/test_dag_example.py

@@ -41,6 +41,7 @@ def setup_airflow_db():
             conn_id=conn_id,
             conn_type="ray",
             extra={
+                "address": "http://0.0.0.0:8265/",
                 "kube_config_path": os.environ.get("KUBECONFIG"),
                 "namespace": "ray",
                 "cluster_context": None,  # Set to None as we don't know how to get cluster context for a kind cluster
@@ -59,13 +60,5 @@ def setup_airflow_db():
 def test_dag_runs(setup_airflow_db, dag_id, dag, fileloc):
     print(f"Testing DAG: {dag_id}, located at: {fileloc}")
     assert dag is not None, f"DAG {dag_id} not found!"
-
-    if dag_id != "Ray_Taskflow_Example":
-        pytest.skip(f"Currently untested {dag_id}")
-        return
-
-    try:
-        dr = dag.test()
-        assert dr.state == "success"
-    except Exception as e:
-        pytest.fail(f"Error running DAG {dag_id}: {e}")
+    dr = dag.test()
+    assert dr.state == "success"