43 - Advanced Testing Techniques

# File: de-onboarding/tests/test_sales_pipeline.py
from hypothesis import given, strategies as st
import pandas as pd
from typing import Dict, Any
import pytest
from unittest.mock import patch

# Define strategy for sales data
sale_strategy = st.dictionaries(
    keys=st.just("sales"),  # Fixed key for sales data
    values=st.lists(  # List of sale records
        st.dictionaries(
            keys=st.text(alphabet="Halalabcdefghijklmnopqrstuvwxyz", min_size=1),  # Halal product names
            values=st.one_of(
                st.floats(min_value=0.01, allow_nan=False),  # Positive prices
                st.integers(min_value=1)  # Positive quantities
            ),
            min_size=3  # Ensure product, price, quantity fields
        ),
        min_size=1  # At least one record
    )
)

@given(sale_strategy)
def test_sales_data_structure(sales_data: Dict[str, Any]) -> None:
    """Test sales data structure."""
    assert isinstance(sales_data, dict), "Sales data must be a dictionary"
    assert "sales" in sales_data, "Sales key must exist"
    assert isinstance(sales_data["sales"], list), "Sales value must be a list"
    print(f"Generated sales data: {sales_data}")  # Debug

def test_read_config_mocked() -> None:
    """Test read_config with mocked YAML loading."""
    mock_config = {"min_price": 10.0, "product_prefix": "Halal", "required_fields": ["product", "price", "quantity"]}
    with patch("pipeline.yaml.safe_load", return_value=mock_config):
        config = read_config("data/config.yaml")
        assert config["min_price"] == 10.0
        print(f"Mocked config: {config}")  # Debug

**Production Note**: Mocking is critical in production to test pipeline components without accessing external systems (e.g., cloud storage, APIs), ensuring fast, isolated tests. In Chapter 66, you’ll use mocking for CI/CD pipelines.

# File: de-onboarding/tests/test_sales_pipeline.py
import time
import pandas as pd
import pytest
from typing import Tuple

def process_sales(df: pd.DataFrame) -> Tuple[float, int]:
    """Dummy pipeline for performance testing."""
    df["amount"] = df["price"] * df["quantity"]
    total_sales = df["amount"].sum()
    return total_sales, len(df)

def test_performance() -> None:
    """Test pipeline performance over multiple runs."""
    data = {
        "product": ["Halal Laptop"] * 1000,
        "price": [999.99] * 1000,
        "quantity": [2] * 1000
    }
    df = pd.DataFrame(data)

    durations = []
    for _ in range(3):
        start_time = time.time()
        total_sales, count = process_sales(df)
        durations.append(time.time() - start_time)

    avg_duration = sum(durations) / len(durations)
    print(f"Processed {count} records in {avg_duration:.3f} seconds (avg of 3 runs)")  # Debug
    assert avg_duration < 1.0, f"Average processing took too long: {avg_duration:.3f}s"
    assert total_sales == 999.99 * 2 * 1000, "Incorrect total sales"

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2

# File: de-onboarding/pipeline.py
from typing import Dict, List, Tuple
import pandas as pd
import yaml
from pydantic import BaseModel, validator
import numpy as np

class Sale(BaseModel):
    """Pydantic model for a sale record."""
    product: str
    price: float
    quantity: int

    @validator("product")
    def product_must_be_halal(cls, v: str, values: Dict) -> str:
        """Ensure product starts with Halal."""
        prefix = "Halal"
        if not v.startswith(prefix):
            raise ValueError(f"Product must start with {prefix}")
        return v

    @validator("price")
    def price_must_be_positive(cls, v: float, values: Dict) -> float:
        """Ensure price is positive and has <=2 decimals."""
        if v <= 0:
            raise ValueError("Price must be positive")
        if round(v, 2) != v:
            raise ValueError("Price must have <=2 decimals")
        return v

    @validator("quantity")
    def quantity_must_be_valid(cls, v: int, values: Dict) -> int:
        """Ensure quantity is positive and <=100."""
        if v <= 0 or v > 100:
            raise ValueError("Quantity must be between 1 and 100")
        return v

def read_config(config_path: str) -> Dict:
    """Read YAML configuration."""
    print(f"Opening config: {config_path}")  # Debug
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    print(f"Loaded config: {config}")  # Debug
    return config

def load_and_validate_sales(csv_path: str, config: Dict) -> Tuple[pd.DataFrame, int, int]:
    """Load and validate sales CSV using Pandas filtering."""
    print(f"Loading CSV: {csv_path}")  # Debug
    df = pd.read_csv(csv_path)
    print("Initial DataFrame:")  # Debug
    print(df.head())

    required_fields = config["required_fields"]
    if not all(field in df.columns for field in required_fields):
        print(f"Missing required fields: {required_fields}")  # Debug
        return pd.DataFrame(), 0, len(df)

    df = df.dropna(subset=["product", "price", "quantity"])
    valid_df = df[
        (df["product"].str.startswith(config["product_prefix"])) &
        (df["price"].apply(lambda x: isinstance(x, (int, float)))) &
        (df["price"] >= config["min_price"]) &
        (df["price"] == df["price"].round(config["max_decimals"])) &
        (df["quantity"].apply(lambda x: isinstance(x, int))) &
        (df["quantity"] > 0) &
        (df["quantity"] <= config["max_quantity"])
    ]

    invalid_rows = df[~df.index.isin(valid_df.index)]
    for _, row in invalid_rows.iterrows():
        print(f"Invalid sale: {row.to_dict()}")  # Debug

    return valid_df, len(valid_df), len(df)

def process_sales(df: pd.DataFrame) -> Tuple[float, List[str]]:
    """Process sales data."""
    if df.empty:
        print("No valid sales data")  # Debug
        return 0.0, []

    df["amount"] = df["price"] * df["quantity"]
    total_sales = float(np.sum(df["amount"].values))
    unique_products = df["product"].unique().tolist()
    print(f"Processed DataFrame:")  # Debug
    print(df)
    return total_sales, unique_products

def main() -> None:
    """Main function."""
    csv_path = "data/sales.csv"
    config_path = "data/config.yaml"
    config = read_config(config_path)
    df, valid_sales, total_records = load_and_validate_sales(csv_path, config)
    total_sales, unique_products = process_sales(df)

    print("\nSales Report:")  # Debug
    print(f"Total Records: {total_records}")
    print(f"Valid Sales: {valid_sales}")
    print(f"Total Sales: ${total_sales:.2f}")
    print(f"Unique Products: {unique_products}")

if __name__ == "__main__":
    main()

# File: de-onboarding/tests/test_pipeline.py
from hypothesis import given, strategies as st
import pytest
import pandas as pd
import time
from typing import Dict, List, Tuple
from pipeline import read_config, load_and_validate_sales, process_sales, Sale
import numpy as np
from unittest.mock import patch

# Hypothesis strategies
sale_strategy = st.dictionaries(
    keys=st.just("sales"),  # Fixed key for sales data
    values=st.lists(  # List of sale records
        st.dictionaries(
            keys=st.text(alphabet="Halalabcdefghijklmnopqrstuvwxyz", min_size=1),  # Halal product names
            values=st.one_of(
                st.floats(min_value=0.01, allow_nan=False),  # Positive prices
                st.integers(min_value=1)  # Positive quantities
            ),
            min_size=3  # Ensure product, price, quantity fields
        ),
        min_size=1  # At least one record
    )
)

@pytest.fixture
def config() -> Dict:
    """Fixture for config."""
    return read_config("data/config.yaml")

def test_read_config() -> None:
    """Test config loading."""
    config = read_config("data/config.yaml")
    assert isinstance(config, dict), "Config must be a dictionary"
    assert "min_price" in config, "Config must have min_price"
    print(f"Config: {config}")  # Debug

def test_read_config_mocked() -> None:
    """Test read_config with mocked YAML loading."""
    mock_config = {"min_price": 10.0, "product_prefix": "Halal", "required_fields": ["product", "price", "quantity"]}
    with patch("pipeline.yaml.safe_load", return_value=mock_config):
        config = read_config("data/config.yaml")
        assert config["min_price"] == 10.0
        print(f"Mocked config: {config}")  # Debug

@given(st.lists(sale_strategy, min_size=1))
def test_load_and_validate_sales(sales_data: List[Dict], config: Dict) -> None:
    """Property-based test for sales validation."""
    df = pd.DataFrame(sales_data)
    valid_df, valid_count, total_count = load_and_validate_sales("data/sales.csv", config)
    assert isinstance(valid_df, pd.DataFrame), "Output must be a DataFrame"
    assert valid_count <= total_count, "Valid count cannot exceed total"
    print(f"Valid rows: {valid_count}, Total rows: {total_count}")  # Debug

def test_process_sales() -> None:
    """Unit test for processing sales."""
    data = {
        "product": ["Halal Laptop", "Halal Mouse"],
        "price": [999.99, 24.99],
        "quantity": [2, 10]
    }
    df = pd.DataFrame(data)
    total_sales, unique_products = process_sales(df)
    assert total_sales == 999.99 * 2 + 24.99 * 10, "Incorrect total sales"
    assert set(unique_products) == {"Halal Laptop", "Halal Mouse"}, "Incorrect products"
    print(f"Total sales: {total_sales}, Products: {unique_products}")  # Debug

def test_performance() -> None:
    """Performance test for pipeline over multiple runs."""
    data = {
        "product": ["Halal Laptop"] * 1000,
        "price": [999.99] * 1000,
        "quantity": [2] * 1000
    }
    df = pd.DataFrame(data)

    durations = []
    for _ in range(3):
        start_time = time.time()
        total_sales, _ = process_sales(df)
        durations.append(time.time() - start_time)

    avg_duration = sum(durations) / len(durations)
    assert avg_duration < 1.0, f"Average processing took too long: {avg_duration:.3f}s"
    assert total_sales == 999.99 * 2 * 1000, "Incorrect total sales"
    print(f"Processed 1000 records in {avg_duration:.3f} seconds (avg of 3 runs)")  # Debug

def test_edge_cases(config: Dict) -> None:
    """Test edge cases with empty DataFrame."""
    empty_df = pd.DataFrame(columns=["product", "price", "quantity"])
    total_sales, unique_products = process_sales(empty_df)
    assert total_sales == 0.0, "Empty DataFrame should return 0"
    assert unique_products == [], "Empty DataFrame should return empty list"
    print("Edge case: empty DataFrame passed")  # Debug

def test_edge_case_files(config: Dict) -> None:
    """Test edge case CSVs."""
    for csv_path in ["data/invalid.csv", "data/malformed.csv", "data/negative.csv"]:
        df, valid_count, total_count = load_and_validate_sales(csv_path, config)
        assert valid_count <= total_count, f"Invalid count for {csv_path}"
        assert isinstance(df, pd.DataFrame), f"Output must be a DataFrame for {csv_path}"
        print(f"Tested {csv_path}: {valid_count} valid, {total_count} total")  # Debug
        # Expected output for invalid.csv: 0 valid, 2 total (missing 'product' column)
        # Expected output for malformed.csv: 1 valid, 2 total (non-integer quantity filtered)
        # Expected output for negative.csv: 1 valid, 2 total (negative price filtered)

@given(st.floats(allow_nan=False))
def test_sale_price(price: float) -> None:
    sale = Sale(product="Halal Laptop", price=price, quantity=2)
    assert sale.price > 0

def slow_process_sales(df: pd.DataFrame) -> Tuple[float, List[str]]:
    total = 0.0
    products = set()
    for _, row in df.iterrows():
        total += row["price"] * row["quantity"]
        products.add(row["product"])
    return total, list(products)

from hypothesis import given, strategies as st
from pipeline import Sale

@given(
    st.text(min_size=1, alphabet="Halalabcdefghijklmnopqrstuvwxyz"),
    st.floats(min_value=10.0, max_value=1000.0, allow_nan=False),
    st.integers(min_value=1, max_value=100)
)
def test_sale_model(product: str, price: float, quantity: int) -> None:
    """Test Sale model with valid inputs."""
    sale = Sale(product=product, price=round(price, 2), quantity=quantity)
    assert sale.product.startswith("Halal")
    assert sale.price >= 10.0
    assert 1 <= sale.quantity <= 100
    print(f"Tested sale: {sale}")  # Debug

def test_large_performance() -> None:
    """Performance test for large dataset."""
    data = {
        "product": ["Halal Laptop"] * 10000,
        "price": [999.99] * 10000,
        "quantity": [2] * 10000
    }
    df = pd.DataFrame(data)

    durations = []
    for _ in range(3):
        start_time = time.time()
        total_sales, _ = process_sales(df)
        durations.append(time.time() - start_time)

    avg_duration = sum(durations) / len(durations)
    assert avg_duration < 2.0, f"Average processing took too long: {avg_duration:.3f}s"
    assert total_sales == 999.99 * 2 * 10000
    print(f"Processed 10000 records in {avg_duration:.3f} seconds (avg of 3 runs)")  # Debug

from hypothesis import given, strategies as st
from pipeline import Sale

@given(st.floats(min_value=0.01, allow_nan=False))
def test_sale_price(price: float) -> None:
    """Test Sale model with positive prices."""
    sale = Sale(product="Halal Laptop", price=round(price, 2), quantity=2)
    assert sale.price > 0
    print(f"Tested price: {sale.price}")  # Debug

import pytest
import pandas as pd
import time
from typing import Tuple, List

def slow_process_sales(df: pd.DataFrame) -> Tuple[float, List[str]]:
    """Buggy pipeline using slow loop."""
    total = 0.0
    products = set()
    for _, row in df.iterrows():
        total += row["price"] * row["quantity"]
        products.add(row["product"])
    return total, list(products)

def process_sales(df: pd.DataFrame) -> Tuple[float, List[str]]:
    """Fixed pipeline using vectorized operations."""
    df["amount"] = df["price"] * df["quantity"]
    total_sales = float(df["amount"].sum())
    unique_products = df["product"].unique().tolist()
    return total_sales, unique_products

def test_slow_pipeline_performance() -> None:
    """Test and fix slow pipeline performance."""
    data = {
        "product": ["Halal Laptop"] * 10000,
        "price": [999.99] * 10000,
        "quantity": [2] * 10000
    }
    df = pd.DataFrame(data)

    start_time = time.time()
    total_sales, products = slow_process_sales(df)
    slow_duration = time.time() - start_time
    print(f"Slow pipeline: {slow_duration:.3f} seconds")  # Debug

    start_time = time.time()
    total_sales_fixed, products_fixed = process_sales(df)
    fast_duration = time.time() - start_time
    print(f"Fixed pipeline: {fast_duration:.3f} seconds")  # Debug

    assert fast_duration < 0.5, f"Fixed pipeline too slow: {fast_duration:.3f}s"
    assert total_sales == total_sales_fixed, "Incorrect total sales"
    assert set(products) == set(products_fixed), "Incorrect products"