🧪 Data Quality & ML Test Automation Framework

Test Automation for Data Engineering & ML Pipelines

🎯 Overview

This framework provides comprehensive test automation for data quality, ML model validation, and API correctness in enterprise data pipelines. It bridges Data Engineering, Machine Learning, and Quality Assurance to ensure end-to-end trust in data-driven systems.

Key Capabilities

✅ Raw Data Quality: Schema validation, missing value detection, volume anomaly detection
✅ Warehouse Integrity: Referential integrity, fact metrics, negative value detection
✅ ML Output Validation: YOLO confidence thresholds, class distribution, detection presence
✅ Drift & Anomaly Detection: Evidently AI-powered drift monitoring
✅ API Correctness: FastAPI endpoint validation, response schema checks
✅ End-to-End Trust: Full pipeline validation (Raw → Warehouse → ML → API)

🏗️ Architecture

Test Stack

Layer	Tool	Justification
Test Runner	pytest	Industry standard, scalable, CI-friendly
Data Quality	Great Expectations	Best-in-class for data validation & drift
ML Testing	Evidently AI	Drift, data quality & ML monitoring
API Testing	pytest + httpx	Async-friendly, FastAPI-native
DB Validation	SQLAlchemy + pytest	Schema & metric assertions
Reporting	pytest-html + Allure	Portfolio-grade reports
CI Ready	GitHub Actions	Recruiter-recognized

Project Structure

data-quality-ml-test-framework/
├── README.md
├── requirements.txt
├── pytest.ini
├── .env
│
├── config/
│   ├── db_config.py          # Database connection management
│   ├── api_config.py         # API client configuration
│   └── expectations_config.yml  # Great Expectations config
│
├── data_quality/
│   ├── raw_data_tests/       # Raw data validation
│   │   ├── test_schema_validation.py
│   │   ├── test_missing_values.py
│   │   └── test_volume_anomalies.py
│   │
│   └── warehouse_tests/      # Warehouse integrity
│       ├── test_referential_integrity.py
│       ├── test_fact_metrics.py
│       └── test_negative_values.py
│
├── ml_tests/                 # ML validation
│   ├── test_yolo_outputs.py
│   ├── test_confidence_thresholds.py
│   ├── test_class_distribution.py
│   └── drift/
│       └── test_image_feature_drift.py
│
├── api_tests/                # API endpoint tests
│   ├── test_health.py
│   ├── test_top_products.py
│   ├── test_channel_activity.py
│   ├── test_visual_content.py
│   └── test_search.py
│
├── monitoring/
│   ├── evidently_report.py   # Evidently AI drift reports
│   └── data_drift_dashboard.html
│
├── tests/
│   └── test_e2e_trust.py     # End-to-end pipeline validation
│
├── reports/                  # Test reports & coverage
│   ├── test_results.html
│   └── allure-results/
│
└── ci/
    └── github-actions.yml     # CI/CD pipeline

🚀 Quick Start

Prerequisites

Python 3.11+
PostgreSQL 12+ (for warehouse tests)
Access to FastAPI service (for API tests)

Installation

Clone the repository

git clone https://github.com/yourusername/data-quality-ml-test-framework.git
cd data-quality-ml-test-framework

Create virtual environment

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

Install dependencies
```
pip install -r requirements.txt
```

Configure environment

cp .env .env
# Edit .env with your database and API credentials

Configuration

Edit .env file with your settings:

# Database Configuration
DB_HOST=localhost
DB_PORT=5432
DB_NAME=telegram_data
DB_USER=postgres
DB_PASSWORD=your_password

# API Configuration
API_BASE_URL=http://localhost:8000
API_TIMEOUT=30

# Test Configuration
VOLUME_ANOMALY_THRESHOLD=0.5      # 50% change triggers alert
CONFIDENCE_THRESHOLD=0.25         # Minimum YOLO confidence
MIN_DETECTION_PERCENTAGE=0.7      # 70% images must have detections

📊 Test Execution

Run All Tests

pytest -v --html=reports/test_results.html

Run by Category

# Data quality tests only
pytest data_quality/ -v -m data_quality

# ML tests only
pytest ml_tests/ -v -m ml

# API tests only
pytest api_tests/ -v -m api

# E2E tests
pytest tests/ -v -m e2e

# Drift tests
pytest ml_tests/drift/ -v -m drift

Generate Reports

# HTML report
pytest --html=reports/test_results.html --self-contained-html

# Coverage report
pytest --cov=. --cov-report=html:reports/coverage

# Allure report
pytest --alluredir=reports/allure-results
allure serve reports/allure-results

🧪 Test Categories

1. Data Quality Tests

Raw Data Validation

Schema Validation: Ensures required columns exist and have correct types
Missing Values: Detects null/empty values in critical fields
Volume Anomalies: Identifies daily message count spikes/drops (>50% threshold)

Example Assertions:

message_id, channel_name, message_date must not be null
No negative views or forwards
No future-dated messages
Daily volume change < configurable threshold

Warehouse Integrity

Referential Integrity: Validates foreign key relationships
Fact Metrics: Ensures aggregations match raw data
Negative Values: Detects invalid negative counts

2. ML Testing

YOLO Output Validation

Confidence Thresholds: All detections ≥ 0.25 confidence
Valid Class Labels: Categories ∈ {promotional, product_display, lifestyle, other}
Detection Presence: At least 70% of images have ≥1 detection
Class Distribution: No single class dominates >90%

Drift Detection

Category Distribution Drift: Detects >10% change in category distribution
Confidence Score Drift: Monitors average confidence changes
Evidently AI Reports: HTML dashboard for drift visualization

3. API Testing

Endpoint Validation

Health Checks: API availability and status
Response Schema: Validates JSON structure
Business Logic: Ensures API responses match warehouse data
Pagination: Validates limit/offset parameters

Example Test:

def test_visual_content_stats(client):
    response = client.get("/api/reports/visual-content")
    assert response.status_code == 200
    data = response.json()
    assert "image_categories" in data
    assert data["image_categories"]["promotional"] >= 0

4. End-to-End Trust Test

Validates complete data flow:

Raw JSON → PostgreSQL → dbt marts → YOLO → API

Checks:

Row count consistency across layers
Metric alignment (warehouse vs API)
ML enrichment coverage
Data integrity throughout pipeline

📈 Monitoring & Reporting

Evidently AI Drift Dashboard

Generate drift reports:

python monitoring/evidently_report.py

This creates monitoring/data_drift_dashboard.html with:

Image category distribution drift
Confidence score drift
Detection count drift
Data quality metrics

Test Reports

HTML Report: reports/test_results.html (self-contained)
Coverage Report: reports/coverage/index.html
Allure Report: reports/allure-results/ (requires Allure CLI)

🔄 CI/CD Integration

GitHub Actions

The framework includes a complete CI/CD pipeline (.github/workflows/test.yml):

Triggers: Push, PR, scheduled (daily), manual
Services: PostgreSQL container for testing
Steps:
1. Install dependencies
2. Run data quality tests
3. Run ML tests
4. Run API tests
5. Run E2E tests
6. Generate reports
7. Upload artifacts

Local CI Simulation

# Run full test suite (simulates CI)
pytest -v --cov=. --cov-report=html --html=reports/test_results.html

🎓 Portfolio Value

This framework demonstrates:

Enterprise Test Architecture: Scalable, maintainable test structure
Data Engineering Expertise: SQLAlchemy, Great Expectations, warehouse validation
ML Testing Proficiency: Model validation, drift detection, confidence thresholds
API Testing Skills: FastAPI, httpx, schema validation
CI/CD Best Practices: GitHub Actions, automated reporting
Fintech-Grade Anomaly Detection: Applied to medical/telegram data

Key Differentiators

✅ Bridges Data + ML + QA: Rare combination of skills
✅ Production-Ready: Error handling, skip logic, comprehensive assertions
✅ Portfolio-Grade: Professional structure, documentation, CI/CD
✅ Recruiter-Recognized: Industry-standard tools (pytest, Great Expectations, Evidently)

🛠️ Customization

Adding New Tests

Data Quality Test

# data_quality/raw_data_tests/test_custom.py
@pytest.mark.data_quality
def test_custom_validation(db_session):
    # Your test logic

ML Test

# ml_tests/test_custom_ml.py
@pytest.mark.ml
def test_custom_ml_validation(db_session):
    # Your test logic

API Test

# api_tests/test_custom_api.py
@pytest.mark.api
def test_custom_endpoint(client):
    # Your test logic

Configuring Thresholds

Edit .env:

VOLUME_ANOMALY_THRESHOLD=0.5      # Adjust anomaly sensitivity
CONFIDENCE_THRESHOLD=0.25         # Adjust ML confidence threshold
MIN_DETECTION_PERCENTAGE=0.7      # Adjust detection coverage requirement

📝 Test Examples

Data Quality Test

def test_no_negative_views(db_session):
    """Ensure views count is never negative."""
    query = text("""
        SELECT COUNT(*) as negative_count
        FROM fact_messages
        WHERE views < 0
    """)
    result = db_session.execute(query).fetchone()
    assert result.negative_count == 0

ML Test

def test_confidence_scores_above_threshold(db_session, confidence_threshold):
    """Validate all YOLO detections meet minimum confidence threshold."""
    query = text("""
        SELECT COUNT(*) as low_confidence_count
        FROM yolo_detections
        WHERE confidence_score < :threshold
    """)
    result = db_session.execute(query, {"threshold": confidence_threshold}).fetchone()
    assert result.low_confidence_count == 0

API Test

def test_visual_content_stats(client):
    """Test visual content statistics endpoint."""
    response = client.get("/api/reports/visual-content")
    assert response.status_code == 200
    data = response.json()
    assert "image_categories" in data

🐛 Troubleshooting

Database Connection Issues

Verify .env configuration
Check PostgreSQL is running
Test connection: python -c "from config.db_config import db_config; db_config.test_connection()"

API Tests Failing

Ensure FastAPI service is running
Verify API_BASE_URL in .env
Check API health: curl http://localhost:8000/health

Missing Dependencies

pip install -r requirements.txt --upgrade

Skip Tests for Missing Tables

Tests automatically skip if tables don't exist (using pytest.skip()).

📚 References

📄 License

MIT License - see LICENSE file for details

👤 Author

Software QA Test Architect - Senait Gebreab
Enterprise Data & ML Systems

🤝 Contributing

Contributions welcome! Please:

Fork the repository
Create a feature branch
Add tests for new functionality
Submit a pull request

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
api_tests		api_tests
config		config
data_quality		data_quality
ml_tests		ml_tests
monitoring		monitoring
tests		tests
.gitignore		.gitignore
CONTRIBUTING.md		CONTRIBUTING.md
QUICK_START.md		QUICK_START.md
README.md		README.md
pytest.ini		pytest.ini
requirements.txt		requirements.txt
setup.py		setup.py

Latest commit

History

Folders and files

Repository files navigation