Dbt producer compatibility test

Author: pawelmarut

.github/workflows/producer_dbt.yml

@@ -74,7 +74,6 @@ jobs:
           pip install dbt-core==${{ inputs.dbt_release }}
           pip install dbt-postgres
           pip install openlineage-dbt==${{ inputs.ol_release }}
-          pip install -r producer/dbt/test_runner/requirements.txt
 
       - name: Set producer output event dir
         if: ${{ steps.init.outputs.scenarios }}
@@ -93,14 +92,9 @@ jobs:
           for scenario in "${scenarios[@]}"
           do
               echo "Running dbt scenario: $scenario"
-
-              if ! python3 producer/dbt/test_runner/cli.py run-scenario \
-                --scenario "$scenario" \
-                --output-dir "${{ steps.set-producer-output.outputs.event_dir }}"
-              then
-                echo "Error: dbt scenario failed: $scenario"
-                exit 1
-              fi
+          
+              mkdir -p "${{ steps.set-producer-output.outputs.event_dir }}/$scenario"
+              bash producer/dbt/scenarios/$scenario/test/run.sh "${{ steps.set-producer-output.outputs.event_dir }}/$scenario"
 
               echo "Finished running scenario: $scenario"
           done

.gitignore

@@ -170,25 +170,6 @@ ignored/
 bin/
 
 # OpenLineage event files generated during local testing
-openlineage_events.json
-openlineage_events.jsonl
-*/openlineage_events.json
-*/openlineage_events.jsonl
-**/events/openlineage_events.json
-**/events/openlineage_events.jsonl
-
-# Test output files (keep directory structure, ignore contents)
-producer/dbt/test_output/*
-!producer/dbt/test_output/.gitkeep
-
-# Auto-generated report files (generated by CI/CD)
-*_producer_report.json
-*_consumer_report.json
-generated-files/report.json
-
-# Virtual environments  
-venv/
-test_venv/
-*/venv/
-*/test_venv/
-**/test_venv/
+**/output/*
+**/test/openlineage.yml
+generated-files/report.json

producer/dbt/run_dbt_tests.sh

@@ -51,10 +51,10 @@ if [[ -z "$OPENLINEAGE_DIRECTORY" ]]; then
     usage
 fi
 
-OL_SPEC_DIRECTORIES=$OPENLINEAGE_DIRECTORY/spec/,$OPENLINEAGE_DIRECTORY/spec/facets/,$OPENLINEAGE_DIRECTORY/spec/registry/gcp/dataproc/facets,$OPENLINEAGE_DIRECTORY/spec/registry/gcp/lineage/facets
+OL_SPEC_DIRECTORIES=$OPENLINEAGE_DIRECTORY/spec/,$OPENLINEAGE_DIRECTORY/spec/facets/
 
 # fail if scenarios are not defined in scenario directory
-[[ $(ls scenarios | wc -l) -gt 0 ]] || { echo >&2 "NO SCENARIOS DEFINED IN scenarios"; exit 1; }
+[[ $(find scenarios | wc -l) -gt 0 ]] || { echo >&2 "NO SCENARIOS DEFINED IN scenarios"; exit 1; }
 
 mkdir -p "$PRODUCER_OUTPUT_EVENTS_DIR"
 
@@ -73,208 +73,56 @@ echo "==========================================================================
 #
 ################################################################################
 
-echo "Setting up test environment..."
-
-# Get script directory for relative paths
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-cd "$SCRIPT_DIR"
-
-# Check if Python test runner exists
-if [[ ! -f "test_runner/cli.py" ]]; then
-    echo "Error: Python test runner not found at test_runner/cli.py"
-    exit 1
-fi
-
 # Check if scenario directory exists
 if [[ ! -d "scenarios" ]]; then
     echo "Error: scenarios directory not found"
     exit 1
 fi
 
+#install python dependencies
+python -m pip install --upgrade pip
+
+if [ -f ./runner/requirements.txt ]; then
+  pip install -r ./runner/requirements.txt
+fi
+
 ################################################################################
 #
 # RUN dbt PRODUCER TESTS
 #
 ################################################################################
 
 echo "Running dbt producer tests..."
-
-# Set up Python environment
-export PYTHONPATH="$SCRIPT_DIR/test_runner:$PYTHONPATH"
+POSIX_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+BASE_DIR="$(cygpath -m "$POSIX_DIR")"
 
 # Run tests for each scenario
-TOTAL_SCENARIOS=0
-PASSED_SCENARIOS=0
-FAILED_SCENARIOS=0
-
 echo "Discovering test scenarios..."
 for scenario_dir in scenarios/*/; do
     if [[ -d "$scenario_dir" && -f "${scenario_dir}config.json" ]]; then
         SCENARIO_NAME=$(basename "$scenario_dir")
         echo "Found scenario: $SCENARIO_NAME"
-        TOTAL_SCENARIOS=$((TOTAL_SCENARIOS + 1))
-        
-        echo "----------------------------------------"
-        echo "Running scenario: $SCENARIO_NAME"
-        echo "----------------------------------------"
-        
-        # Run the atomic tests for this scenario
-        echo "Step 1: Running atomic tests..."
-        if python3 test_runner/cli.py run-atomic --base-path "." --verbose; then
-            echo "✅ Atomic tests passed for $SCENARIO_NAME"
-            
-            # Run OpenLineage event validation if events exist
-            echo "Step 2: Validating OpenLineage events..."
-            EVENTS_FILE="events/openlineage_events.jsonl"
-            if [[ -f "$EVENTS_FILE" ]]; then
-                echo "📋 Validating events from: $EVENTS_FILE"
-                echo "📋 Against spec version: $OPENLINEAGE_RELEASE"
-                if python3 test_runner/cli.py validate-events --events-file "$EVENTS_FILE" --spec-dir "$OPENLINEAGE_DIRECTORY/spec"; then
-                    echo "✅ Event validation passed for $SCENARIO_NAME (spec: $OPENLINEAGE_RELEASE)"
-                    PASSED_SCENARIOS=$((PASSED_SCENARIOS + 1))
-                else
-                    echo "❌ Event validation failed for $SCENARIO_NAME (spec: $OPENLINEAGE_RELEASE)"
-                    FAILED_SCENARIOS=$((FAILED_SCENARIOS + 1))
-                fi
-            else
-                echo "⚠️  No OpenLineage events found at $EVENTS_FILE, skipping validation for $SCENARIO_NAME"
-                PASSED_SCENARIOS=$((PASSED_SCENARIOS + 1))
-            fi
-        else
-            echo "❌ Atomic tests failed for $SCENARIO_NAME"
-            FAILED_SCENARIOS=$((FAILED_SCENARIOS + 1))
-        fi
-        
-        echo ""
+
+        mkdir -p "$PRODUCER_OUTPUT_EVENTS_DIR/$SCENARIO_NAME"
+        "$scenario_dir"test/run.sh "$BASE_DIR/$PRODUCER_OUTPUT_EVENTS_DIR/$SCENARIO_NAME"
+
+        echo "Scenario $SCENARIO_NAME completed"
     fi
 done
 
-################################################################################
-#
-# GENERATE REPORT
-#
-################################################################################
-
-echo "=============================================================================="
-echo "                         TEST RESULTS                                        "
-echo "=============================================================================="
-echo "Total scenarios: $TOTAL_SCENARIOS"
-echo "Passed scenarios: $PASSED_SCENARIOS"
-echo "Failed scenarios: $FAILED_SCENARIOS"
-echo "OpenLineage Spec Version: $OPENLINEAGE_RELEASE"
-echo "Events File: events/openlineage_events.jsonl"
-echo "Report File: $REPORT_PATH"
-echo "=============================================================================="
-echo "Failed scenarios: $FAILED_SCENARIOS"
-echo "=============================================================================="
+echo "EVENT VALIDATION FOR SPEC VERSION $OPENLINEAGE_RELEASE"
 
 # Generate JSON report
 REPORT_DIR=$(dirname "$REPORT_PATH")
 mkdir -p "$REPORT_DIR"
 
-cat > "$REPORT_PATH" << EOF
-{
-  "producer": "dbt",
-  "openlineage_release": "$OPENLINEAGE_RELEASE",
-  "test_execution_time": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
-  "total_scenarios": $TOTAL_SCENARIOS,
-  "passed_scenarios": $PASSED_SCENARIOS,
-  "failed_scenarios": $FAILED_SCENARIOS,
-  "success_rate": $(echo "scale=2; $PASSED_SCENARIOS * 100 / $TOTAL_SCENARIOS" | bc -l 2>/dev/null || echo "0"),
-  "output_events_directory": "$PRODUCER_OUTPUT_EVENTS_DIR",
-  "scenarios": []
-}
-EOF
-
-echo "Report generated: $REPORT_PATH"
-
-################################################################################
-#
-# CLEANUP AND EXIT
-#
-################################################################################
-
-echo "Cleaning up temporary files..."
-
-# Exit with appropriate code
-if [[ $FAILED_SCENARIOS -eq 0 ]]; then
-    echo "🎉 All tests passed!"
-    exit 0
-else
-    echo "❌ Some tests failed. Check the output above for details."
-    exit 1
-fi
-EOF
-
-# Create staging models
-cat > dbt_project/models/staging/stg_customers.sql << EOF
-SELECT 
-    customer_id,
-    UPPER(name) as customer_name,
-    LOWER(email) as email,
-    signup_date,
-    status
-FROM {{ ref('customers') }}
-WHERE status = 'active'
-EOF
-
-cat > dbt_project/models/staging/stg_orders.sql << EOF
-SELECT 
-    order_id,
-    customer_id,
-    product,
-    amount,
-    order_date
-FROM {{ ref('orders') }}
-EOF
-
-# Create mart model
-mkdir -p dbt_project/models/marts
-cat > dbt_project/models/marts/customer_orders.sql << EOF
-SELECT 
-    c.customer_id,
-    c.customer_name,
-    COUNT(o.order_id) as total_orders,
-    SUM(o.amount) as total_spent
-FROM {{ ref('stg_customers') }} c
-LEFT JOIN {{ ref('stg_orders') }} o 
-    ON c.customer_id = o.customer_id
-GROUP BY c.customer_id, c.customer_name
-EOF
-
-echo "Running dbt with OpenLineage..."
-cd dbt_project
-
-# Install dependencies and run dbt
-dbt deps --no-version-check || echo "No packages to install"
-dbt seed --no-version-check
-dbt run --no-version-check
-
-cd ..
-
-echo "dbt execution completed. Checking for generated events..."
-
-# Check the events file
-if [[ -f "events/openlineage_events.jsonl" ]]; then
-    event_count=$(wc -l < "events/openlineage_events.jsonl")
-    echo "Generated $event_count OpenLineage events"
-    echo "Events saved to: events/openlineage_events.jsonl"
-else
-    echo "Warning: No OpenLineage events file generated at events/openlineage_events.jsonl"
-    echo "Creating minimal event file for testing..."
-    mkdir -p "events"
-    echo '{"eventType": "COMPLETE", "eventTime": "2023-01-01T00:00:00Z", "run": {"runId": "test-run-id"}, "job": {"namespace": "dbt://local", "name": "test-job"}, "inputs": [], "outputs": [], "schemaURL": "https://openlineage.io/spec/'$OPENLINEAGE_RELEASE'/OpenLineage.json#/$defs/RunEvent"}' > "events/openlineage_events.jsonl"
-fi
-
-echo "EVENT VALIDATION FOR SPEC VERSION $OPENLINEAGE_RELEASE"
-
 pip install -r ../../scripts/requirements.txt
 
 python ../../scripts/validate_ol_events.py \
---event_base_dir="events" \
---spec_dirs="$OL_SPEC_DIRECTORIES" \
+--event_base_dir="$PRODUCER_OUTPUT_EVENTS_DIR" \
+--spec_base_dir="$OL_SPEC_DIRECTORIES" \
 --target="$REPORT_PATH" \
---component="dbt_producer" \
+--component="scenarios" \
 --producer_dir=. \
 --openlineage_version="$OPENLINEAGE_RELEASE"
 

producer/dbt/runner/models/marts/customer_analytics.sql

@@ -1,5 +1,3 @@
-{{ config(materialized='table') }}
-
 select
     c.customer_id,
     c.customer_name,

producer/dbt/runner/models/staging/stg_customers.sql

@@ -1,5 +1,3 @@
-{{ config(materialized='table') }}
-
 select
     customer_id,
     name as customer_name,

producer/dbt/runner/models/staging/stg_orders.sql

@@ -1,5 +1,3 @@
-{{ config(materialized='table') }}
-
 select
     order_id,
     customer_id,

producer/dbt/runner/openlineage_test.duckdb

@@ -1,7 +1,7 @@
 openlineage_compatibility_test:
-  target: dev
+  target: postgres
   outputs:
-    dev:
+    postgres:
       type: postgres
       host: "{{ env_var('DBT_POSTGRES_HOST', 'localhost') }}"
       port: "{{ env_var('DBT_POSTGRES_PORT', '5432') | as_number }}"

producer/dbt/runner/profiles.yml

@@ -0,0 +1,9 @@
+# OpenLineage dbt Producer Test Dependencies
+# Install: pip install -r requirements.txt
+
+# dbt dependencies
+dbt-core>=1.5.0
+dbt-postgres>=1.5.0
+
+# OpenLineage integration (if available)
+openlineage-dbt>=0.28.0

producer/dbt/runner/requirements.txt

@@ -0,0 +1,20 @@
+version: "3.9"
+
+name: csv_to_postgres
+
+services:
+  postgres:
+    image: postgres:15-alpine
+    container_name: postgres15
+    restart: always
+    environment:
+      POSTGRES_USER: testuser
+      POSTGRES_PASSWORD: testpass
+      POSTGRES_DB: dbt_test
+    ports:
+      - "5432:5432"
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+
+volumes:
+  postgres_data: