mirror of
https://github.com/FuzzingLabs/fuzzforge_ai.git
synced 2026-02-13 11:12:46 +00:00
feat(test): add automated workflow testing framework
- Add test matrix configuration (.github/test-matrix.yaml) - Maps 8 workflows to workers, test projects, and parameters - Excludes LLM and OSS-Fuzz workflows - Defines fast, full, and platform test suites - Add workflow execution test script (scripts/test_workflows.py) - Executes workflows with parameter validation - Validates SARIF export and structure - Counts findings and measures execution time - Generates test summary reports - Add platform detection unit tests (cli/tests/test_platform_detection.py) - Tests platform detection (x86_64, aarch64, arm64) - Tests Dockerfile selection for multi-platform workers - Tests metadata.yaml parsing - Includes integration tests - Add GitHub Actions workflow (.github/workflows/test-workflows.yml) - Platform detection unit tests - Fast workflow tests (5 workflows on every PR) - Android platform-specific tests (AMD64 + ARM64) - Full workflow tests (on main/schedule) - Automatic log collection on failure - Add comprehensive testing documentation (docs/docs/development/testing.md) - Local testing guide - CI/CD testing explanation - Platform-specific testing guide - Debugging guide and best practices - Update test.yml with reference to new workflow tests - Remove tracked .fuzzforge/findings.db (already in .gitignore) Tested locally: - Single workflow test: python_sast (6.87s) ✅ - Fast test suite: 5/5 workflows passed ✅ - android_static_analysis (98.98s) ✅ - python_sast (6.78s) ✅ - secret_detection (38.04s) ✅ - gitleaks_detection (1.67s) ✅ - trufflehog_detection (1.64s) ✅
This commit is contained in:
tduhamel42
178
.github/test-matrix.yaml
vendored
Normal file
178
.github/test-matrix.yaml
vendored
Normal file
@@ -0,0 +1,178 @@
|
||||
# Test Matrix Configuration for Automated Workflow Testing
|
||||
#
|
||||
# This file defines which workflows to test, their required workers,
|
||||
# test projects, parameters, and expected outcomes.
|
||||
#
|
||||
# Excluded workflows:
|
||||
# - llm_analysis (requires LLM API keys)
|
||||
# - llm_secret_detection (requires LLM API keys)
|
||||
# - ossfuzz_campaign (requires OSS-Fuzz project configuration)
|
||||
|
||||
version: "1.0"
|
||||
|
||||
# Worker to Dockerfile mapping
|
||||
workers:
|
||||
android:
|
||||
dockerfiles:
|
||||
linux/amd64: "Dockerfile.amd64"
|
||||
linux/arm64: "Dockerfile.arm64"
|
||||
metadata: "workers/android/metadata.yaml"
|
||||
|
||||
python:
|
||||
dockerfiles:
|
||||
default: "Dockerfile"
|
||||
|
||||
rust:
|
||||
dockerfiles:
|
||||
default: "Dockerfile"
|
||||
|
||||
secrets:
|
||||
dockerfiles:
|
||||
default: "Dockerfile"
|
||||
|
||||
# Workflow test configurations
|
||||
workflows:
|
||||
# Android Static Analysis
|
||||
android_static_analysis:
|
||||
worker: android
|
||||
test_project: test_projects/android_test
|
||||
working_directory: test_projects/android_test
|
||||
parameters:
|
||||
apk_path: "BeetleBug.apk"
|
||||
timeout: 300
|
||||
platform_specific: true # Test on both amd64 and arm64
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [android, static-analysis, fast]
|
||||
|
||||
# Python SAST
|
||||
python_sast:
|
||||
worker: python
|
||||
test_project: test_projects/vulnerable_app
|
||||
working_directory: test_projects/vulnerable_app
|
||||
parameters: {}
|
||||
timeout: 180
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [python, sast, fast]
|
||||
|
||||
# Python Fuzzing (Atheris)
|
||||
atheris_fuzzing:
|
||||
worker: python
|
||||
test_project: test_projects/python_fuzz_waterfall
|
||||
working_directory: test_projects/python_fuzz_waterfall
|
||||
parameters:
|
||||
max_total_time: 30 # Short fuzzing run for testing
|
||||
artifact_prefix: "test-atheris"
|
||||
timeout: 120
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: false # May not find crashes in short run
|
||||
sarif_export: false
|
||||
tags: [python, fuzzing, slow]
|
||||
|
||||
# Rust Fuzzing (cargo-fuzz)
|
||||
cargo_fuzzing:
|
||||
worker: rust
|
||||
test_project: test_projects/rust_fuzz_test
|
||||
working_directory: test_projects/rust_fuzz_test
|
||||
parameters:
|
||||
max_total_time: 30 # Short fuzzing run for testing
|
||||
artifact_prefix: "test-cargo"
|
||||
timeout: 120
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: false # May not find crashes in short run
|
||||
sarif_export: false
|
||||
tags: [rust, fuzzing, slow]
|
||||
|
||||
# Secret Detection (combined)
|
||||
secret_detection:
|
||||
worker: secrets
|
||||
test_project: test_projects/secret_detection_benchmark
|
||||
working_directory: test_projects/secret_detection_benchmark
|
||||
parameters: {}
|
||||
timeout: 120
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [secrets, detection, fast]
|
||||
|
||||
# Gitleaks Detection
|
||||
gitleaks_detection:
|
||||
worker: secrets
|
||||
test_project: test_projects/secret_detection_benchmark
|
||||
working_directory: test_projects/secret_detection_benchmark
|
||||
parameters: {}
|
||||
timeout: 120
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [secrets, gitleaks, fast]
|
||||
|
||||
# TruffleHog Detection
|
||||
trufflehog_detection:
|
||||
worker: secrets
|
||||
test_project: test_projects/secret_detection_benchmark
|
||||
working_directory: test_projects/secret_detection_benchmark
|
||||
parameters: {}
|
||||
timeout: 120
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [secrets, trufflehog, fast]
|
||||
|
||||
# Security Assessment (composite workflow)
|
||||
security_assessment:
|
||||
worker: python # Uses multiple workers internally
|
||||
test_project: test_projects/vulnerable_app
|
||||
working_directory: test_projects/vulnerable_app
|
||||
parameters: {}
|
||||
timeout: 300
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [composite, security, slow]
|
||||
|
||||
# Test suites - groups of workflows for different scenarios
|
||||
test_suites:
|
||||
# Fast tests - run on every PR
|
||||
fast:
|
||||
workflows:
|
||||
- android_static_analysis
|
||||
- python_sast
|
||||
- secret_detection
|
||||
- gitleaks_detection
|
||||
- trufflehog_detection
|
||||
timeout: 900 # 15 minutes total
|
||||
|
||||
# Full tests - run on main/master
|
||||
full:
|
||||
workflows:
|
||||
- android_static_analysis
|
||||
- python_sast
|
||||
- atheris_fuzzing
|
||||
- cargo_fuzzing
|
||||
- secret_detection
|
||||
- gitleaks_detection
|
||||
- trufflehog_detection
|
||||
- security_assessment
|
||||
timeout: 1800 # 30 minutes total
|
||||
|
||||
# Platform-specific tests - test Dockerfile selection
|
||||
platform:
|
||||
workflows:
|
||||
- android_static_analysis
|
||||
- python_sast
|
||||
platforms:
|
||||
- linux/amd64
|
||||
- linux/arm64
|
||||
timeout: 600 # 10 minutes total
|
||||
319
.github/workflows/test-workflows.yml
vendored
Normal file
319
.github/workflows/test-workflows.yml
vendored
Normal file
@@ -0,0 +1,319 @@
|
||||
name: Workflow Integration Tests
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ main, master, dev, develop, test/** ]
|
||||
pull_request:
|
||||
branches: [ main, master, dev, develop ]
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
test_suite:
|
||||
description: 'Test suite to run'
|
||||
required: false
|
||||
default: 'fast'
|
||||
type: choice
|
||||
options:
|
||||
- fast
|
||||
- full
|
||||
- platform
|
||||
|
||||
jobs:
|
||||
#############################################################################
|
||||
# Platform Detection Unit Tests
|
||||
#############################################################################
|
||||
platform-detection-tests:
|
||||
name: Platform Detection Unit Tests
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Install dependencies
|
||||
working-directory: ./cli
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install pytest pytest-cov pyyaml
|
||||
pip install -e .
|
||||
|
||||
- name: Run platform detection tests
|
||||
working-directory: ./cli
|
||||
run: |
|
||||
pytest tests/test_platform_detection.py -v \
|
||||
--cov=src/fuzzforge_cli \
|
||||
--cov-report=term \
|
||||
--cov-report=xml
|
||||
|
||||
- name: Upload coverage
|
||||
uses: codecov/codecov-action@v4
|
||||
with:
|
||||
file: ./cli/coverage.xml
|
||||
flags: cli-platform-detection
|
||||
name: cli-platform-detection
|
||||
|
||||
#############################################################################
|
||||
# Fast Workflow Tests (AMD64 only)
|
||||
#############################################################################
|
||||
fast-workflow-tests:
|
||||
name: Fast Workflow Tests (AMD64)
|
||||
runs-on: ubuntu-latest
|
||||
needs: platform-detection-tests
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Install FuzzForge CLI
|
||||
working-directory: ./cli
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install -e .
|
||||
pip install pyyaml # Required by test script
|
||||
|
||||
- name: Copy environment template
|
||||
run: |
|
||||
mkdir -p volumes/env
|
||||
cp volumes/env/.env.template volumes/env/.env
|
||||
|
||||
- name: Start FuzzForge services
|
||||
run: |
|
||||
docker compose up -d
|
||||
echo "⏳ Waiting for services to be ready..."
|
||||
sleep 30
|
||||
|
||||
# Wait for backend to be healthy
|
||||
max_wait=60
|
||||
waited=0
|
||||
while [ $waited -lt $max_wait ]; do
|
||||
if docker ps --filter "name=fuzzforge-backend" --format "{{.Status}}" | grep -q "healthy"; then
|
||||
echo "✅ Backend is healthy"
|
||||
break
|
||||
fi
|
||||
echo "Waiting for backend... ($waited/$max_wait seconds)"
|
||||
sleep 5
|
||||
waited=$((waited + 5))
|
||||
done
|
||||
|
||||
- name: Run fast workflow tests
|
||||
run: |
|
||||
python scripts/test_workflows.py --suite fast --skip-service-start
|
||||
timeout-minutes: 20
|
||||
|
||||
- name: Collect logs on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "=== Docker container status ==="
|
||||
docker ps -a
|
||||
|
||||
echo "=== Backend logs ==="
|
||||
docker logs fuzzforge-backend --tail 100
|
||||
|
||||
echo "=== Worker logs ==="
|
||||
for worker in python secrets android; do
|
||||
if docker ps -a --format "{{.Names}}" | grep -q "fuzzforge-worker-$worker"; then
|
||||
echo "=== Worker: $worker ==="
|
||||
docker logs fuzzforge-worker-$worker --tail 50
|
||||
fi
|
||||
done
|
||||
|
||||
- name: Stop services
|
||||
if: always()
|
||||
run: docker compose down -v
|
||||
|
||||
#############################################################################
|
||||
# Platform-Specific Tests (Android Worker)
|
||||
#############################################################################
|
||||
android-platform-tests:
|
||||
name: Android Worker Platform Tests
|
||||
runs-on: ${{ matrix.os }}
|
||||
needs: platform-detection-tests
|
||||
strategy:
|
||||
matrix:
|
||||
include:
|
||||
- os: ubuntu-latest
|
||||
platform: linux/amd64
|
||||
arch: x86_64
|
||||
# ARM64 runner (uncomment when GitHub Actions ARM64 runners are available)
|
||||
# - os: ubuntu-24.04-arm
|
||||
# platform: linux/arm64
|
||||
# arch: aarch64
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Install FuzzForge CLI
|
||||
working-directory: ./cli
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install -e .
|
||||
pip install pyyaml
|
||||
|
||||
- name: Verify platform detection
|
||||
run: |
|
||||
echo "Expected platform: ${{ matrix.platform }}"
|
||||
echo "Expected arch: ${{ matrix.arch }}"
|
||||
echo "Actual arch: $(uname -m)"
|
||||
|
||||
# Verify platform matches
|
||||
if [ "$(uname -m)" != "${{ matrix.arch }}" ]; then
|
||||
echo "❌ Platform mismatch!"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
- name: Check Android worker Dockerfile selection
|
||||
run: |
|
||||
# Check which Dockerfile would be selected
|
||||
if [ "${{ matrix.platform }}" == "linux/amd64" ]; then
|
||||
expected_dockerfile="Dockerfile.amd64"
|
||||
else
|
||||
expected_dockerfile="Dockerfile.arm64"
|
||||
fi
|
||||
|
||||
echo "Expected Dockerfile: $expected_dockerfile"
|
||||
|
||||
# Verify the Dockerfile exists
|
||||
if [ ! -f "workers/android/$expected_dockerfile" ]; then
|
||||
echo "❌ Dockerfile not found: workers/android/$expected_dockerfile"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "✅ Dockerfile exists: $expected_dockerfile"
|
||||
|
||||
- name: Build Android worker for platform
|
||||
run: |
|
||||
echo "Building Android worker for platform: ${{ matrix.platform }}"
|
||||
docker compose build worker-android
|
||||
timeout-minutes: 15
|
||||
|
||||
- name: Copy environment template
|
||||
run: |
|
||||
mkdir -p volumes/env
|
||||
cp volumes/env/.env.template volumes/env/.env
|
||||
|
||||
- name: Start FuzzForge services
|
||||
run: |
|
||||
docker compose up -d
|
||||
sleep 30
|
||||
|
||||
- name: Run Android workflow test
|
||||
run: |
|
||||
python scripts/test_workflows.py \
|
||||
--workflow android_static_analysis \
|
||||
--platform ${{ matrix.platform }} \
|
||||
--skip-service-start
|
||||
timeout-minutes: 10
|
||||
|
||||
- name: Verify correct Dockerfile was used
|
||||
run: |
|
||||
# Check docker image labels or inspect to verify correct build
|
||||
docker inspect fuzzforge-worker-android | grep -i "dockerfile" || true
|
||||
|
||||
- name: Collect logs on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "=== Android worker logs ==="
|
||||
docker logs fuzzforge-worker-android --tail 100
|
||||
|
||||
- name: Stop services
|
||||
if: always()
|
||||
run: docker compose down -v
|
||||
|
||||
#############################################################################
|
||||
# Full Workflow Tests (on schedule or manual trigger)
|
||||
#############################################################################
|
||||
full-workflow-tests:
|
||||
name: Full Workflow Tests
|
||||
runs-on: ubuntu-latest
|
||||
needs: platform-detection-tests
|
||||
# Only run full tests on schedule, manual trigger, or main branch
|
||||
if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch' || github.ref == 'refs/heads/main' || github.ref == 'refs/heads/master'
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
|
||||
- name: Install FuzzForge CLI
|
||||
working-directory: ./cli
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install -e .
|
||||
pip install pyyaml
|
||||
|
||||
- name: Copy environment template
|
||||
run: |
|
||||
mkdir -p volumes/env
|
||||
cp volumes/env/.env.template volumes/env/.env
|
||||
|
||||
- name: Start FuzzForge services
|
||||
run: |
|
||||
docker compose up -d
|
||||
sleep 30
|
||||
|
||||
- name: Run full workflow tests
|
||||
run: |
|
||||
python scripts/test_workflows.py --suite full --skip-service-start
|
||||
timeout-minutes: 45
|
||||
|
||||
- name: Collect logs on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "=== Docker container status ==="
|
||||
docker ps -a
|
||||
|
||||
echo "=== All worker logs ==="
|
||||
for worker in python secrets rust android ossfuzz; do
|
||||
if docker ps -a --format "{{.Names}}" | grep -q "fuzzforge-worker-$worker"; then
|
||||
echo "=== Worker: $worker ==="
|
||||
docker logs fuzzforge-worker-$worker --tail 100
|
||||
fi
|
||||
done
|
||||
|
||||
- name: Stop services
|
||||
if: always()
|
||||
run: docker compose down -v
|
||||
|
||||
#############################################################################
|
||||
# Test Summary
|
||||
#############################################################################
|
||||
test-summary:
|
||||
name: Workflow Test Summary
|
||||
runs-on: ubuntu-latest
|
||||
needs: [platform-detection-tests, fast-workflow-tests, android-platform-tests]
|
||||
if: always()
|
||||
|
||||
steps:
|
||||
- name: Check test results
|
||||
run: |
|
||||
if [ "${{ needs.platform-detection-tests.result }}" != "success" ]; then
|
||||
echo "❌ Platform detection tests failed"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ "${{ needs.fast-workflow-tests.result }}" != "success" ]; then
|
||||
echo "❌ Fast workflow tests failed"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [ "${{ needs.android-platform-tests.result }}" != "success" ]; then
|
||||
echo "❌ Android platform tests failed"
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "✅ All workflow integration tests passed!"
|
||||
8
.github/workflows/test.yml
vendored
8
.github/workflows/test.yml
vendored
@@ -1,5 +1,13 @@
|
||||
name: Tests
|
||||
|
||||
# This workflow covers:
|
||||
# - Worker validation (Dockerfile and metadata checks)
|
||||
# - Docker image builds (only for modified workers)
|
||||
# - Python linting (ruff, mypy)
|
||||
# - Backend unit tests
|
||||
#
|
||||
# For end-to-end workflow integration tests, see: .github/workflows/test-workflows.yml
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [ main, master, dev, develop, feature/** ]
|
||||
|
||||
0
cli/tests/__init__.py
Normal file
0
cli/tests/__init__.py
Normal file
253
cli/tests/test_platform_detection.py
Normal file
253
cli/tests/test_platform_detection.py
Normal file
@@ -0,0 +1,253 @@
|
||||
"""
|
||||
Unit tests for platform detection and Dockerfile selection in WorkerManager.
|
||||
|
||||
These tests verify that the WorkerManager correctly detects the platform
|
||||
and selects the appropriate Dockerfile for workers with platform-specific
|
||||
configurations (e.g., Android worker with separate AMD64 and ARM64 Dockerfiles).
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch, mock_open
|
||||
import yaml
|
||||
|
||||
from fuzzforge_cli.worker_manager import WorkerManager
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def worker_manager():
|
||||
"""Create a WorkerManager instance for testing."""
|
||||
return WorkerManager()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_android_metadata():
|
||||
"""Mock metadata.yaml content for Android worker."""
|
||||
return """
|
||||
name: android
|
||||
version: "1.0.0"
|
||||
description: "Android application security testing worker"
|
||||
default_platform: linux/amd64
|
||||
|
||||
platforms:
|
||||
linux/amd64:
|
||||
dockerfile: Dockerfile.amd64
|
||||
description: "Full Android toolchain with MobSF support"
|
||||
supported_tools:
|
||||
- jadx
|
||||
- opengrep
|
||||
- mobsf
|
||||
- frida
|
||||
- androguard
|
||||
|
||||
linux/arm64:
|
||||
dockerfile: Dockerfile.arm64
|
||||
description: "Android toolchain without MobSF (ARM64/Apple Silicon compatible)"
|
||||
supported_tools:
|
||||
- jadx
|
||||
- opengrep
|
||||
- frida
|
||||
- androguard
|
||||
disabled_tools:
|
||||
mobsf: "Incompatible with Rosetta 2 emulation"
|
||||
"""
|
||||
|
||||
|
||||
class TestPlatformDetection:
|
||||
"""Test platform detection logic."""
|
||||
|
||||
def test_detect_platform_linux_x86_64(self, worker_manager):
|
||||
"""Test platform detection on Linux x86_64."""
|
||||
with patch('platform.machine', return_value='x86_64'), \
|
||||
patch('platform.system', return_value='Linux'):
|
||||
platform = worker_manager._detect_platform()
|
||||
assert platform == 'linux/amd64'
|
||||
|
||||
def test_detect_platform_linux_aarch64(self, worker_manager):
|
||||
"""Test platform detection on Linux aarch64."""
|
||||
with patch('platform.machine', return_value='aarch64'), \
|
||||
patch('platform.system', return_value='Linux'):
|
||||
platform = worker_manager._detect_platform()
|
||||
assert platform == 'linux/arm64'
|
||||
|
||||
def test_detect_platform_darwin_arm64(self, worker_manager):
|
||||
"""Test platform detection on macOS Apple Silicon."""
|
||||
with patch('platform.machine', return_value='arm64'), \
|
||||
patch('platform.system', return_value='Darwin'):
|
||||
platform = worker_manager._detect_platform()
|
||||
assert platform == 'linux/arm64'
|
||||
|
||||
def test_detect_platform_darwin_x86_64(self, worker_manager):
|
||||
"""Test platform detection on macOS Intel."""
|
||||
with patch('platform.machine', return_value='x86_64'), \
|
||||
patch('platform.system', return_value='Darwin'):
|
||||
platform = worker_manager._detect_platform()
|
||||
assert platform == 'linux/amd64'
|
||||
|
||||
|
||||
class TestDockerfileSelection:
|
||||
"""Test Dockerfile selection logic."""
|
||||
|
||||
def test_select_dockerfile_with_metadata_amd64(self, worker_manager, mock_android_metadata):
|
||||
"""Test Dockerfile selection for AMD64 platform with metadata."""
|
||||
with patch('platform.machine', return_value='x86_64'), \
|
||||
patch('platform.system', return_value='Linux'), \
|
||||
patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=mock_android_metadata)):
|
||||
|
||||
dockerfile = worker_manager._select_dockerfile('android')
|
||||
assert 'Dockerfile.amd64' in str(dockerfile)
|
||||
|
||||
def test_select_dockerfile_with_metadata_arm64(self, worker_manager, mock_android_metadata):
|
||||
"""Test Dockerfile selection for ARM64 platform with metadata."""
|
||||
with patch('platform.machine', return_value='arm64'), \
|
||||
patch('platform.system', return_value='Darwin'), \
|
||||
patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=mock_android_metadata)):
|
||||
|
||||
dockerfile = worker_manager._select_dockerfile('android')
|
||||
assert 'Dockerfile.arm64' in str(dockerfile)
|
||||
|
||||
def test_select_dockerfile_without_metadata(self, worker_manager):
|
||||
"""Test Dockerfile selection for worker without metadata (uses default Dockerfile)."""
|
||||
with patch('pathlib.Path.exists', return_value=False):
|
||||
dockerfile = worker_manager._select_dockerfile('python')
|
||||
assert str(dockerfile).endswith('Dockerfile')
|
||||
assert 'Dockerfile.amd64' not in str(dockerfile)
|
||||
assert 'Dockerfile.arm64' not in str(dockerfile)
|
||||
|
||||
def test_select_dockerfile_fallback_to_default(self, worker_manager):
|
||||
"""Test Dockerfile selection falls back to default platform when current platform not found."""
|
||||
# Metadata with only amd64 support
|
||||
limited_metadata = """
|
||||
name: test-worker
|
||||
default_platform: linux/amd64
|
||||
platforms:
|
||||
linux/amd64:
|
||||
dockerfile: Dockerfile.amd64
|
||||
"""
|
||||
with patch('platform.machine', return_value='arm64'), \
|
||||
patch('platform.system', return_value='Darwin'), \
|
||||
patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=limited_metadata)):
|
||||
|
||||
# Should fall back to default_platform (amd64) since arm64 is not defined
|
||||
dockerfile = worker_manager._select_dockerfile('test-worker')
|
||||
assert 'Dockerfile.amd64' in str(dockerfile)
|
||||
|
||||
|
||||
class TestMetadataParsing:
|
||||
"""Test metadata.yaml parsing and handling."""
|
||||
|
||||
def test_parse_valid_metadata(self, worker_manager, mock_android_metadata):
|
||||
"""Test parsing valid metadata.yaml."""
|
||||
with patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=mock_android_metadata)):
|
||||
|
||||
metadata_path = Path("workers/android/metadata.yaml")
|
||||
with open(metadata_path, 'r') as f:
|
||||
metadata = yaml.safe_load(f)
|
||||
|
||||
assert metadata['name'] == 'android'
|
||||
assert metadata['default_platform'] == 'linux/amd64'
|
||||
assert 'linux/amd64' in metadata['platforms']
|
||||
assert 'linux/arm64' in metadata['platforms']
|
||||
assert metadata['platforms']['linux/amd64']['dockerfile'] == 'Dockerfile.amd64'
|
||||
assert metadata['platforms']['linux/arm64']['dockerfile'] == 'Dockerfile.arm64'
|
||||
|
||||
def test_handle_missing_metadata(self, worker_manager):
|
||||
"""Test handling when metadata.yaml doesn't exist."""
|
||||
with patch('pathlib.Path.exists', return_value=False):
|
||||
# Should use default Dockerfile when metadata doesn't exist
|
||||
dockerfile = worker_manager._select_dockerfile('nonexistent-worker')
|
||||
assert str(dockerfile).endswith('Dockerfile')
|
||||
|
||||
def test_handle_malformed_metadata(self, worker_manager):
|
||||
"""Test handling malformed metadata.yaml."""
|
||||
malformed_yaml = "{ invalid: yaml: content:"
|
||||
|
||||
with patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=malformed_yaml)):
|
||||
|
||||
# Should fall back to default Dockerfile on YAML parse error
|
||||
dockerfile = worker_manager._select_dockerfile('broken-worker')
|
||||
assert str(dockerfile).endswith('Dockerfile')
|
||||
|
||||
|
||||
class TestWorkerStartWithPlatform:
|
||||
"""Test worker startup with platform-specific configuration."""
|
||||
|
||||
def test_start_android_worker_amd64(self, worker_manager, mock_android_metadata):
|
||||
"""Test starting Android worker on AMD64 platform."""
|
||||
with patch('platform.machine', return_value='x86_64'), \
|
||||
patch('platform.system', return_value='Linux'), \
|
||||
patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=mock_android_metadata)), \
|
||||
patch('subprocess.run') as mock_run:
|
||||
|
||||
mock_run.return_value = Mock(returncode=0)
|
||||
|
||||
# This would call _select_dockerfile internally
|
||||
dockerfile = worker_manager._select_dockerfile('android')
|
||||
assert 'Dockerfile.amd64' in str(dockerfile)
|
||||
|
||||
# Verify it would use MobSF-enabled image
|
||||
with open(Path("workers/android/metadata.yaml"), 'r') as f:
|
||||
metadata = yaml.safe_load(f)
|
||||
tools = metadata['platforms']['linux/amd64']['supported_tools']
|
||||
assert 'mobsf' in tools
|
||||
|
||||
def test_start_android_worker_arm64(self, worker_manager, mock_android_metadata):
|
||||
"""Test starting Android worker on ARM64 platform."""
|
||||
with patch('platform.machine', return_value='arm64'), \
|
||||
patch('platform.system', return_value='Darwin'), \
|
||||
patch('pathlib.Path.exists', return_value=True), \
|
||||
patch('builtins.open', mock_open(read_data=mock_android_metadata)), \
|
||||
patch('subprocess.run') as mock_run:
|
||||
|
||||
mock_run.return_value = Mock(returncode=0)
|
||||
|
||||
# This would call _select_dockerfile internally
|
||||
dockerfile = worker_manager._select_dockerfile('android')
|
||||
assert 'Dockerfile.arm64' in str(dockerfile)
|
||||
|
||||
# Verify MobSF is disabled on ARM64
|
||||
with open(Path("workers/android/metadata.yaml"), 'r') as f:
|
||||
metadata = yaml.safe_load(f)
|
||||
tools = metadata['platforms']['linux/arm64']['supported_tools']
|
||||
assert 'mobsf' not in tools
|
||||
assert 'mobsf' in metadata['platforms']['linux/arm64']['disabled_tools']
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestPlatformDetectionIntegration:
|
||||
"""Integration tests that verify actual platform detection."""
|
||||
|
||||
def test_current_platform_detection(self, worker_manager):
|
||||
"""Test that platform detection works on current platform."""
|
||||
platform = worker_manager._detect_platform()
|
||||
|
||||
# Should be one of the supported platforms
|
||||
assert platform in ['linux/amd64', 'linux/arm64']
|
||||
|
||||
# Should match the actual system
|
||||
import platform as sys_platform
|
||||
machine = sys_platform.machine()
|
||||
|
||||
if machine in ['x86_64', 'AMD64']:
|
||||
assert platform == 'linux/amd64'
|
||||
elif machine in ['aarch64', 'arm64']:
|
||||
assert platform == 'linux/arm64'
|
||||
|
||||
def test_android_metadata_exists(self):
|
||||
"""Test that Android worker metadata file exists."""
|
||||
metadata_path = Path(__file__).parent.parent.parent / "workers" / "android" / "metadata.yaml"
|
||||
assert metadata_path.exists(), "Android worker metadata.yaml should exist"
|
||||
|
||||
# Verify it's valid YAML
|
||||
with open(metadata_path, 'r') as f:
|
||||
metadata = yaml.safe_load(f)
|
||||
|
||||
assert 'platforms' in metadata
|
||||
assert 'linux/amd64' in metadata['platforms']
|
||||
assert 'linux/arm64' in metadata['platforms']
|
||||
558
docs/docs/development/testing.md
Normal file
558
docs/docs/development/testing.md
Normal file
@@ -0,0 +1,558 @@
|
||||
# Testing Guide
|
||||
|
||||
This guide explains FuzzForge's testing infrastructure, including unit tests, workflow integration tests, and platform-specific testing for multi-architecture support.
|
||||
|
||||
---
|
||||
|
||||
## Overview
|
||||
|
||||
FuzzForge has multiple layers of testing:
|
||||
|
||||
1. **Unit Tests** - Backend and CLI unit tests
|
||||
2. **Worker Validation** - Docker image and metadata validation
|
||||
3. **Platform Detection Tests** - Verify correct Dockerfile selection across platforms
|
||||
4. **Workflow Integration Tests** - End-to-end workflow execution validation
|
||||
5. **Multi-Platform Tests** - Verify platform-specific Docker images (AMD64 vs ARM64)
|
||||
|
||||
---
|
||||
|
||||
## Test Organization
|
||||
|
||||
```
|
||||
.github/
|
||||
├── workflows/
|
||||
│ ├── test.yml # Unit tests, linting, worker builds
|
||||
│ └── test-workflows.yml # Workflow integration tests
|
||||
├── test-matrix.yaml # Workflow test configuration
|
||||
└── scripts/
|
||||
└── validate-workers.sh # Worker validation script
|
||||
|
||||
cli/
|
||||
└── tests/
|
||||
└── test_platform_detection.py # Platform detection unit tests
|
||||
|
||||
backend/
|
||||
└── tests/
|
||||
├── unit/ # Backend unit tests
|
||||
└── integration/ # Backend integration tests (commented out)
|
||||
|
||||
scripts/
|
||||
└── test_workflows.py # Workflow execution test script
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Running Tests Locally
|
||||
|
||||
### Prerequisites
|
||||
|
||||
```bash
|
||||
# Start FuzzForge services
|
||||
docker compose up -d
|
||||
|
||||
# Install CLI in development mode
|
||||
cd cli
|
||||
pip install -e ".[dev]"
|
||||
pip install pytest pytest-cov pyyaml
|
||||
```
|
||||
|
||||
### Unit Tests
|
||||
|
||||
#### Backend Unit Tests
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
pytest tests/unit/ -v \
|
||||
--cov=toolbox/modules \
|
||||
--cov=src \
|
||||
--cov-report=html
|
||||
```
|
||||
|
||||
#### CLI Platform Detection Tests
|
||||
|
||||
```bash
|
||||
cd cli
|
||||
pytest tests/test_platform_detection.py -v
|
||||
```
|
||||
|
||||
### Workflow Integration Tests
|
||||
|
||||
#### Run Fast Test Suite
|
||||
|
||||
Tests a subset of fast-running workflows:
|
||||
|
||||
```bash
|
||||
python scripts/test_workflows.py --suite fast
|
||||
```
|
||||
|
||||
Workflows in fast suite:
|
||||
- `android_static_analysis`
|
||||
- `python_sast`
|
||||
- `secret_detection`
|
||||
- `gitleaks_detection`
|
||||
- `trufflehog_detection`
|
||||
|
||||
#### Run Full Test Suite
|
||||
|
||||
Tests all workflows (excludes LLM and OSS-Fuzz workflows):
|
||||
|
||||
```bash
|
||||
python scripts/test_workflows.py --suite full
|
||||
```
|
||||
|
||||
Additional workflows in full suite:
|
||||
- `atheris_fuzzing`
|
||||
- `cargo_fuzzing`
|
||||
- `security_assessment`
|
||||
|
||||
#### Run Single Workflow Test
|
||||
|
||||
```bash
|
||||
python scripts/test_workflows.py --workflow python_sast
|
||||
```
|
||||
|
||||
#### Test Platform-Specific Dockerfile
|
||||
|
||||
```bash
|
||||
python scripts/test_workflows.py \
|
||||
--workflow android_static_analysis \
|
||||
--platform linux/amd64
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Test Matrix Configuration
|
||||
|
||||
The test matrix (`.github/test-matrix.yaml`) defines:
|
||||
|
||||
- Workflow-to-worker mappings
|
||||
- Test projects for each workflow
|
||||
- Required parameters
|
||||
- Expected outcomes
|
||||
- Timeout values
|
||||
- Test suite groupings
|
||||
|
||||
### Example Configuration
|
||||
|
||||
```yaml
|
||||
workflows:
|
||||
python_sast:
|
||||
worker: python
|
||||
test_project: test_projects/vulnerable_app
|
||||
working_directory: test_projects/vulnerable_app
|
||||
parameters: {}
|
||||
timeout: 180
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [python, sast, fast]
|
||||
```
|
||||
|
||||
### Adding a New Workflow Test
|
||||
|
||||
1. Add workflow configuration to `.github/test-matrix.yaml`:
|
||||
|
||||
```yaml
|
||||
workflows:
|
||||
my_new_workflow:
|
||||
worker: python # Which worker runs this workflow
|
||||
test_project: test_projects/my_test
|
||||
working_directory: test_projects/my_test
|
||||
parameters:
|
||||
# Any required parameters
|
||||
severity: "high"
|
||||
timeout: 300
|
||||
expected:
|
||||
status: "COMPLETED"
|
||||
has_findings: true
|
||||
sarif_export: true
|
||||
tags: [python, custom, fast]
|
||||
```
|
||||
|
||||
2. Add to appropriate test suite:
|
||||
|
||||
```yaml
|
||||
test_suites:
|
||||
fast:
|
||||
workflows:
|
||||
- python_sast
|
||||
- my_new_workflow # Add here
|
||||
```
|
||||
|
||||
3. Ensure test project exists with appropriate test cases
|
||||
|
||||
---
|
||||
|
||||
## Platform-Specific Testing
|
||||
|
||||
### Why Platform-Specific Tests?
|
||||
|
||||
Some workers (like Android) have different capabilities on different platforms:
|
||||
|
||||
- **AMD64 (x86_64)**: Full toolchain including MobSF
|
||||
- **ARM64 (Apple Silicon)**: Limited toolchain (MobSF incompatible with Rosetta 2)
|
||||
|
||||
### Platform Detection
|
||||
|
||||
Platform detection happens in `cli/src/fuzzforge_cli/worker_manager.py`:
|
||||
|
||||
```python
|
||||
def _detect_platform(self) -> str:
|
||||
"""Detect current platform for Docker image selection."""
|
||||
machine = platform.machine()
|
||||
system = platform.system()
|
||||
|
||||
# Map to Docker platform identifiers
|
||||
if machine in ['x86_64', 'AMD64']:
|
||||
return 'linux/amd64'
|
||||
elif machine in ['aarch64', 'arm64']:
|
||||
return 'linux/arm64'
|
||||
else:
|
||||
return 'linux/amd64' # Default fallback
|
||||
```
|
||||
|
||||
### Dockerfile Selection
|
||||
|
||||
Workers with `metadata.yaml` can define platform-specific Dockerfiles:
|
||||
|
||||
```yaml
|
||||
# workers/android/metadata.yaml
|
||||
platforms:
|
||||
linux/amd64:
|
||||
dockerfile: Dockerfile.amd64
|
||||
description: "Full Android toolchain with MobSF support"
|
||||
|
||||
linux/arm64:
|
||||
dockerfile: Dockerfile.arm64
|
||||
description: "Android toolchain without MobSF"
|
||||
```
|
||||
|
||||
### Testing Platform Detection
|
||||
|
||||
```bash
|
||||
# Run platform detection unit tests
|
||||
cd cli
|
||||
pytest tests/test_platform_detection.py -v
|
||||
|
||||
# Test with mocked platforms
|
||||
pytest tests/test_platform_detection.py::TestPlatformDetection::test_detect_platform_linux_x86_64 -v
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## CI/CD Testing
|
||||
|
||||
### GitHub Actions Workflows
|
||||
|
||||
#### 1. Main Test Workflow (`.github/workflows/test.yml`)
|
||||
|
||||
Runs on every push and PR:
|
||||
|
||||
- **Worker Validation**: Validates Dockerfiles and metadata
|
||||
- **Docker Image Builds**: Builds only modified workers
|
||||
- **Linting**: Ruff and mypy checks
|
||||
- **Backend Unit Tests**: pytest on Python 3.11 and 3.12
|
||||
|
||||
#### 2. Workflow Integration Tests (`.github/workflows/test-workflows.yml`)
|
||||
|
||||
Runs end-to-end workflow tests:
|
||||
|
||||
- **Platform Detection Tests**: Unit tests for platform detection logic
|
||||
- **Fast Workflow Tests**: Quick smoke tests (runs on every PR)
|
||||
- **Android Platform Tests**: Verifies AMD64 and ARM64 Dockerfile selection
|
||||
- **Full Workflow Tests**: Comprehensive tests (runs on main/master or schedule)
|
||||
|
||||
### Test Triggers
|
||||
|
||||
```yaml
|
||||
# Runs on every push/PR
|
||||
on:
|
||||
push:
|
||||
branches: [ main, master, dev, develop, test/** ]
|
||||
pull_request:
|
||||
branches: [ main, master, dev, develop ]
|
||||
|
||||
# Manual trigger with test suite selection
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
test_suite:
|
||||
type: choice
|
||||
options:
|
||||
- fast
|
||||
- full
|
||||
- platform
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Debugging Test Failures
|
||||
|
||||
### Local Debugging
|
||||
|
||||
#### 1. Check Service Status
|
||||
|
||||
```bash
|
||||
docker ps
|
||||
docker logs fuzzforge-backend
|
||||
docker logs fuzzforge-worker-python
|
||||
```
|
||||
|
||||
#### 2. Run Workflow Manually
|
||||
|
||||
```bash
|
||||
cd test_projects/vulnerable_app
|
||||
ff workflow run python_sast . --wait --no-interactive
|
||||
```
|
||||
|
||||
#### 3. Check Findings
|
||||
|
||||
```bash
|
||||
ff findings list
|
||||
ff findings list python_sast-xxxxx --format json
|
||||
```
|
||||
|
||||
### CI Debugging
|
||||
|
||||
Test workflows automatically collect logs on failure:
|
||||
|
||||
```yaml
|
||||
- name: Collect logs on failure
|
||||
if: failure()
|
||||
run: |
|
||||
docker ps -a
|
||||
docker logs fuzzforge-backend --tail 100
|
||||
docker logs fuzzforge-worker-python --tail 50
|
||||
```
|
||||
|
||||
View logs in GitHub Actions:
|
||||
1. Go to failed workflow run
|
||||
2. Click on failed job
|
||||
3. Scroll to "Collect logs on failure" step
|
||||
|
||||
---
|
||||
|
||||
## Writing New Tests
|
||||
|
||||
### Adding a Backend Unit Test
|
||||
|
||||
```python
|
||||
# backend/tests/unit/test_my_feature.py
|
||||
import pytest
|
||||
from toolbox.modules.my_module import my_function
|
||||
|
||||
def test_my_function():
|
||||
result = my_function("input")
|
||||
assert result == "expected_output"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_function():
|
||||
result = await my_async_function()
|
||||
assert result is not None
|
||||
```
|
||||
|
||||
### Adding a CLI Unit Test
|
||||
|
||||
```python
|
||||
# cli/tests/test_my_feature.py
|
||||
import pytest
|
||||
from fuzzforge_cli.my_module import MyClass
|
||||
|
||||
@pytest.fixture
|
||||
def my_instance():
|
||||
return MyClass()
|
||||
|
||||
def test_my_method(my_instance):
|
||||
result = my_instance.my_method()
|
||||
assert result == expected_value
|
||||
```
|
||||
|
||||
### Adding a Platform Detection Test
|
||||
|
||||
```python
|
||||
# cli/tests/test_platform_detection.py
|
||||
from unittest.mock import patch
|
||||
|
||||
def test_detect_platform_linux_x86_64(worker_manager):
|
||||
with patch('platform.machine', return_value='x86_64'), \
|
||||
patch('platform.system', return_value='Linux'):
|
||||
platform = worker_manager._detect_platform()
|
||||
assert platform == 'linux/amd64'
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Test Coverage
|
||||
|
||||
### Viewing Coverage Reports
|
||||
|
||||
#### Backend Coverage
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
pytest tests/unit/ --cov=toolbox/modules --cov=src --cov-report=html
|
||||
open htmlcov/index.html
|
||||
```
|
||||
|
||||
#### CLI Coverage
|
||||
|
||||
```bash
|
||||
cd cli
|
||||
pytest tests/ --cov=src/fuzzforge_cli --cov-report=html
|
||||
open htmlcov/index.html
|
||||
```
|
||||
|
||||
### Coverage in CI
|
||||
|
||||
Coverage reports are automatically uploaded to Codecov:
|
||||
|
||||
- Backend: `codecov-backend`
|
||||
- CLI Platform Detection: `cli-platform-detection`
|
||||
|
||||
View at: https://codecov.io/gh/FuzzingLabs/fuzzforge_ai
|
||||
|
||||
---
|
||||
|
||||
## Test Best Practices
|
||||
|
||||
### 1. Fast Tests First
|
||||
|
||||
Order tests by execution time:
|
||||
- Unit tests (< 1s each)
|
||||
- Integration tests (< 10s each)
|
||||
- Workflow tests (< 5min each)
|
||||
|
||||
### 2. Use Test Fixtures
|
||||
|
||||
```python
|
||||
@pytest.fixture
|
||||
def temp_project(tmp_path):
|
||||
"""Create temporary test project."""
|
||||
project_dir = tmp_path / "test_project"
|
||||
project_dir.mkdir()
|
||||
# Setup project files
|
||||
return project_dir
|
||||
```
|
||||
|
||||
### 3. Mock External Dependencies
|
||||
|
||||
```python
|
||||
@patch('subprocess.run')
|
||||
def test_docker_command(mock_run):
|
||||
mock_run.return_value = Mock(returncode=0, stdout="success")
|
||||
result = run_docker_command()
|
||||
assert result == "success"
|
||||
```
|
||||
|
||||
### 4. Parametrize Similar Tests
|
||||
|
||||
```python
|
||||
@pytest.mark.parametrize("platform,expected", [
|
||||
("linux/amd64", "Dockerfile.amd64"),
|
||||
("linux/arm64", "Dockerfile.arm64"),
|
||||
])
|
||||
def test_dockerfile_selection(platform, expected):
|
||||
dockerfile = select_dockerfile(platform)
|
||||
assert expected in str(dockerfile)
|
||||
```
|
||||
|
||||
### 5. Tag Tests Appropriately
|
||||
|
||||
```python
|
||||
@pytest.mark.integration
|
||||
def test_full_workflow():
|
||||
# Integration test that requires services
|
||||
pass
|
||||
|
||||
@pytest.mark.slow
|
||||
def test_long_running_operation():
|
||||
# Test that takes > 10 seconds
|
||||
pass
|
||||
```
|
||||
|
||||
Run specific tags:
|
||||
```bash
|
||||
pytest -m "not slow" # Skip slow tests
|
||||
pytest -m integration # Only integration tests
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Continuous Improvement
|
||||
|
||||
### Adding Test Coverage
|
||||
|
||||
1. Identify untested code paths
|
||||
2. Write unit tests for core logic
|
||||
3. Add integration tests for end-to-end flows
|
||||
4. Update test matrix for new workflows
|
||||
|
||||
### Performance Optimization
|
||||
|
||||
1. Use test suites to group tests
|
||||
2. Run fast tests on every commit
|
||||
3. Run slow tests nightly or on main branch
|
||||
4. Parallelize independent tests
|
||||
|
||||
### Monitoring Test Health
|
||||
|
||||
1. Track test execution time trends
|
||||
2. Monitor flaky tests
|
||||
3. Keep coverage above 80%
|
||||
4. Review and update stale tests
|
||||
|
||||
---
|
||||
|
||||
## Related Documentation
|
||||
|
||||
- [Docker Setup](../how-to/docker-setup.md) - Worker management
|
||||
- [CLI Reference](../reference/cli-reference.md) - CLI commands
|
||||
- [Workflow Guide](../how-to/workflows.md) - Creating workflows
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Tests Timeout
|
||||
|
||||
**Symptom**: Workflow tests hang and timeout
|
||||
|
||||
**Solutions**:
|
||||
1. Check if services are running: `docker ps`
|
||||
2. Verify backend is healthy: `docker logs fuzzforge-backend`
|
||||
3. Increase timeout in test matrix
|
||||
4. Check for deadlocks in workflow code
|
||||
|
||||
### Worker Build Failures
|
||||
|
||||
**Symptom**: Docker image build fails in CI
|
||||
|
||||
**Solutions**:
|
||||
1. Test build locally: `docker compose build worker-python`
|
||||
2. Check Dockerfile syntax
|
||||
3. Verify base image is accessible
|
||||
4. Review build logs for specific errors
|
||||
|
||||
### Platform Detection Failures
|
||||
|
||||
**Symptom**: Wrong Dockerfile selected on ARM64
|
||||
|
||||
**Solutions**:
|
||||
1. Verify metadata.yaml syntax
|
||||
2. Check platform detection logic
|
||||
3. Test locally with: `python -c "import platform; print(platform.machine())"`
|
||||
4. Review WorkerManager._detect_platform() logic
|
||||
|
||||
### SARIF Export Validation Fails
|
||||
|
||||
**Symptom**: Workflow completes but SARIF validation fails
|
||||
|
||||
**Solutions**:
|
||||
1. Check SARIF file exists: `ls -la test-*.sarif`
|
||||
2. Validate JSON syntax: `jq . test-*.sarif`
|
||||
3. Verify SARIF schema: Must have `version` and `runs` fields
|
||||
4. Check workflow SARIF export logic
|
||||
|
||||
---
|
||||
|
||||
**Questions?** Open an issue or consult the [development discussions](https://github.com/FuzzingLabs/fuzzforge_ai/discussions).
|
||||
381
scripts/test_workflows.py
Executable file
381
scripts/test_workflows.py
Executable file
@@ -0,0 +1,381 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Automated workflow testing script for FuzzForge.
|
||||
|
||||
This script reads the test matrix configuration and executes workflows
|
||||
to validate end-to-end functionality, SARIF export, and platform-specific
|
||||
Dockerfile selection.
|
||||
|
||||
Usage:
|
||||
python scripts/test_workflows.py --suite fast
|
||||
python scripts/test_workflows.py --workflow python_sast
|
||||
python scripts/test_workflows.py --workflow android_static_analysis --platform linux/amd64
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
try:
|
||||
import yaml
|
||||
except ImportError:
|
||||
print("Error: PyYAML is required. Install with: pip install pyyaml")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkflowTestResult:
|
||||
"""Result of a workflow test execution."""
|
||||
workflow_name: str
|
||||
success: bool
|
||||
duration: float
|
||||
status: Optional[str] = None
|
||||
run_id: Optional[str] = None
|
||||
error: Optional[str] = None
|
||||
findings_count: Optional[int] = None
|
||||
sarif_exported: bool = False
|
||||
|
||||
|
||||
class WorkflowTester:
|
||||
"""Executes and validates FuzzForge workflows."""
|
||||
|
||||
def __init__(self, matrix_file: Path, root_dir: Path):
|
||||
self.matrix_file = matrix_file
|
||||
self.root_dir = root_dir
|
||||
self.matrix = self._load_matrix()
|
||||
self.results: List[WorkflowTestResult] = []
|
||||
|
||||
def _load_matrix(self) -> Dict[str, Any]:
|
||||
"""Load test matrix configuration."""
|
||||
with open(self.matrix_file, 'r') as f:
|
||||
return yaml.safe_load(f)
|
||||
|
||||
def check_services(self) -> bool:
|
||||
"""Check if FuzzForge services are running."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["docker", "ps", "--filter", "name=fuzzforge-backend", "--format", "{{.Status}}"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False
|
||||
)
|
||||
return "Up" in result.stdout
|
||||
except Exception as e:
|
||||
print(f"❌ Error checking services: {e}")
|
||||
return False
|
||||
|
||||
def start_services(self) -> bool:
|
||||
"""Start FuzzForge services if not running."""
|
||||
if self.check_services():
|
||||
print("✅ FuzzForge services already running")
|
||||
return True
|
||||
|
||||
print("🚀 Starting FuzzForge services...")
|
||||
try:
|
||||
subprocess.run(
|
||||
["docker", "compose", "up", "-d"],
|
||||
cwd=self.root_dir,
|
||||
check=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE
|
||||
)
|
||||
# Wait for services to be ready
|
||||
print("⏳ Waiting for services to be ready...")
|
||||
max_wait = 60
|
||||
waited = 0
|
||||
while waited < max_wait:
|
||||
if self.check_services():
|
||||
print("✅ Services ready")
|
||||
time.sleep(5) # Extra wait for full initialization
|
||||
return True
|
||||
time.sleep(2)
|
||||
waited += 2
|
||||
print(f"⚠️ Services did not become ready within {max_wait}s")
|
||||
return False
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f"❌ Failed to start services: {e}")
|
||||
return False
|
||||
|
||||
def execute_workflow(
|
||||
self,
|
||||
workflow_name: str,
|
||||
config: Dict[str, Any],
|
||||
platform: Optional[str] = None
|
||||
) -> WorkflowTestResult:
|
||||
"""Execute a single workflow and validate results."""
|
||||
start_time = time.time()
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Testing workflow: {workflow_name}")
|
||||
if platform:
|
||||
print(f"Platform: {platform}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
# Build command
|
||||
working_dir = self.root_dir / config['working_directory']
|
||||
cmd = [
|
||||
"ff", "workflow", "run",
|
||||
workflow_name,
|
||||
".",
|
||||
"--wait",
|
||||
"--no-interactive"
|
||||
]
|
||||
|
||||
# Add parameters
|
||||
params = config.get('parameters', {})
|
||||
for key, value in params.items():
|
||||
if isinstance(value, (str, int, float)):
|
||||
cmd.append(f"{key}={value}")
|
||||
|
||||
# Add SARIF export if expected
|
||||
sarif_file = None
|
||||
if config.get('expected', {}).get('sarif_export'):
|
||||
sarif_file = working_dir / f"test-{workflow_name}.sarif"
|
||||
cmd.extend(["--export-sarif", str(sarif_file)])
|
||||
|
||||
# Execute workflow
|
||||
print(f"Command: {' '.join(cmd)}")
|
||||
print(f"Working directory: {working_dir}")
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
cwd=working_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=config.get('timeout', 300)
|
||||
)
|
||||
|
||||
duration = time.time() - start_time
|
||||
print(f"\n⏱️ Duration: {duration:.2f}s")
|
||||
|
||||
# Parse output for run_id
|
||||
run_id = self._extract_run_id(result.stdout)
|
||||
|
||||
# Check if workflow completed successfully
|
||||
if result.returncode != 0:
|
||||
error_msg = result.stderr or result.stdout
|
||||
print(f"❌ Workflow failed with exit code {result.returncode}")
|
||||
print(f"Error: {error_msg[:500]}")
|
||||
return WorkflowTestResult(
|
||||
workflow_name=workflow_name,
|
||||
success=False,
|
||||
duration=duration,
|
||||
run_id=run_id,
|
||||
error=error_msg[:500]
|
||||
)
|
||||
|
||||
# Validate SARIF export
|
||||
sarif_exported = False
|
||||
if sarif_file and sarif_file.exists():
|
||||
sarif_exported = self._validate_sarif(sarif_file)
|
||||
print(f"✅ SARIF export validated" if sarif_exported else "⚠️ SARIF export invalid")
|
||||
|
||||
# Get findings count
|
||||
findings_count = self._count_findings(run_id) if run_id else None
|
||||
|
||||
print(f"✅ Workflow completed successfully")
|
||||
if findings_count is not None:
|
||||
print(f" Findings: {findings_count}")
|
||||
|
||||
return WorkflowTestResult(
|
||||
workflow_name=workflow_name,
|
||||
success=True,
|
||||
duration=duration,
|
||||
status="COMPLETED",
|
||||
run_id=run_id,
|
||||
findings_count=findings_count,
|
||||
sarif_exported=sarif_exported
|
||||
)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
duration = time.time() - start_time
|
||||
print(f"❌ Workflow timed out after {duration:.2f}s")
|
||||
return WorkflowTestResult(
|
||||
workflow_name=workflow_name,
|
||||
success=False,
|
||||
duration=duration,
|
||||
error=f"Timeout after {config.get('timeout')}s"
|
||||
)
|
||||
except Exception as e:
|
||||
duration = time.time() - start_time
|
||||
print(f"❌ Unexpected error: {e}")
|
||||
return WorkflowTestResult(
|
||||
workflow_name=workflow_name,
|
||||
success=False,
|
||||
duration=duration,
|
||||
error=str(e)
|
||||
)
|
||||
|
||||
def _extract_run_id(self, output: str) -> Optional[str]:
|
||||
"""Extract run_id from workflow output."""
|
||||
for line in output.split('\n'):
|
||||
if 'run_id' in line.lower() or 'execution id' in line.lower():
|
||||
# Try to extract the ID
|
||||
parts = line.split()
|
||||
for part in parts:
|
||||
if '-' in part and len(part) > 10:
|
||||
return part.strip(',:')
|
||||
return None
|
||||
|
||||
def _validate_sarif(self, sarif_file: Path) -> bool:
|
||||
"""Validate SARIF file structure."""
|
||||
try:
|
||||
with open(sarif_file, 'r') as f:
|
||||
sarif = json.load(f)
|
||||
# Basic SARIF validation
|
||||
return (
|
||||
'version' in sarif and
|
||||
'runs' in sarif and
|
||||
isinstance(sarif['runs'], list)
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"⚠️ SARIF validation error: {e}")
|
||||
return False
|
||||
|
||||
def _count_findings(self, run_id: str) -> Optional[int]:
|
||||
"""Count findings for a run."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["ff", "findings", "list", run_id, "--format", "json"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False
|
||||
)
|
||||
if result.returncode == 0:
|
||||
findings = json.loads(result.stdout)
|
||||
return len(findings) if isinstance(findings, list) else 0
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def run_suite(self, suite_name: str) -> bool:
|
||||
"""Run a predefined test suite."""
|
||||
suite = self.matrix.get('test_suites', {}).get(suite_name)
|
||||
if not suite:
|
||||
print(f"❌ Suite '{suite_name}' not found")
|
||||
return False
|
||||
|
||||
workflows = suite.get('workflows', [])
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Running test suite: {suite_name}")
|
||||
print(f"Workflows: {', '.join(workflows)}")
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
for workflow_name in workflows:
|
||||
config = self.matrix['workflows'].get(workflow_name)
|
||||
if not config:
|
||||
print(f"⚠️ Workflow '{workflow_name}' not found in matrix")
|
||||
continue
|
||||
|
||||
result = self.execute_workflow(workflow_name, config)
|
||||
self.results.append(result)
|
||||
|
||||
return self.print_summary()
|
||||
|
||||
def run_workflow(self, workflow_name: str, platform: Optional[str] = None) -> bool:
|
||||
"""Run a single workflow."""
|
||||
config = self.matrix['workflows'].get(workflow_name)
|
||||
if not config:
|
||||
print(f"❌ Workflow '{workflow_name}' not found")
|
||||
return False
|
||||
|
||||
result = self.execute_workflow(workflow_name, config, platform)
|
||||
self.results.append(result)
|
||||
|
||||
return result.success
|
||||
|
||||
def print_summary(self) -> bool:
|
||||
"""Print test summary."""
|
||||
print(f"\n\n{'='*60}")
|
||||
print("TEST SUMMARY")
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
total = len(self.results)
|
||||
passed = sum(1 for r in self.results if r.success)
|
||||
failed = total - passed
|
||||
|
||||
print(f"Total tests: {total}")
|
||||
print(f"Passed: {passed} ✅")
|
||||
print(f"Failed: {failed} ❌")
|
||||
print()
|
||||
|
||||
if failed > 0:
|
||||
print("Failed tests:")
|
||||
for result in self.results:
|
||||
if not result.success:
|
||||
print(f" - {result.workflow_name}")
|
||||
if result.error:
|
||||
print(f" Error: {result.error[:100]}")
|
||||
|
||||
print(f"\n{'='*60}\n")
|
||||
return failed == 0
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Test FuzzForge workflows")
|
||||
parser.add_argument(
|
||||
"--suite",
|
||||
choices=["fast", "full", "platform"],
|
||||
help="Test suite to run"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--workflow",
|
||||
help="Single workflow to test"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--platform",
|
||||
help="Platform for platform-specific testing (e.g., linux/amd64)"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--matrix",
|
||||
type=Path,
|
||||
default=Path(".github/test-matrix.yaml"),
|
||||
help="Path to test matrix file"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--skip-service-start",
|
||||
action="store_true",
|
||||
help="Skip starting services (assume already running)"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Determine root directory
|
||||
root_dir = Path(__file__).parent.parent
|
||||
|
||||
# Load tester
|
||||
matrix_file = root_dir / args.matrix
|
||||
if not matrix_file.exists():
|
||||
print(f"❌ Matrix file not found: {matrix_file}")
|
||||
sys.exit(1)
|
||||
|
||||
tester = WorkflowTester(matrix_file, root_dir)
|
||||
|
||||
# Start services if needed
|
||||
if not args.skip_service_start:
|
||||
if not tester.start_services():
|
||||
print("❌ Failed to start services")
|
||||
sys.exit(1)
|
||||
|
||||
# Run tests
|
||||
success = False
|
||||
if args.suite:
|
||||
success = tester.run_suite(args.suite)
|
||||
elif args.workflow:
|
||||
success = tester.run_workflow(args.workflow, args.platform)
|
||||
else:
|
||||
print("❌ Must specify --suite or --workflow")
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
sys.exit(0 if success else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
Reference in New Issue
Block a user