09 - Introduction to Testing in Python

# File: de-onboarding/processor.py
from typing import Dict, List, Tuple  # For type annotations
import pandas as pd  # For DataFrame operations
import numpy as np  # For numerical computations
import yaml  # For YAML parsing
import utils  # Import custom utils module

def read_config(config_path: str) -> Dict[str, any]:
    """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[str, any]) -> Tuple[pd.DataFrame, int, int]:
    """Load sales CSV and validate using Pandas."""
    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"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")  # Log error
        return pd.DataFrame(), 0, len(df)

    df = df.dropna(subset=["product"])
    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(utils.is_integer)]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(utils.is_numeric_value)]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    df = df[df["price"].apply(lambda x: utils.apply_valid_decimals(x, config["max_decimals"]))]

    total_records = len(df)
    print("Validated DataFrame:")  # Debug
    print(df)
    return df, len(df), total_records

def process_sales(df: pd.DataFrame, config: Dict[str, any]) -> Tuple[Dict[str, any], int]:
    """Process sales: compute total and top products."""
    if df.empty:
        print("No valid sales data")  # Log empty
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}, 0

    df["amount"] = df["price"] * df["quantity"]
    print("DataFrame with Amount:")  # Debug
    print(df)

    total_sales = np.sum(df["amount"].values)
    unique_products = df["product"].unique().tolist()
    sales_by_product = df.groupby("product")["amount"].sum()
    top_products = sales_by_product.sort_values(ascending=False).head(3).to_dict()

    valid_sales = len(df)
    print(f"Valid sales: {valid_sales} records")  # Log
    return {
        "total_sales": float(total_sales),
        "unique_products": unique_products,
        "top_products": top_products
    }, valid_sales

# File: de-onboarding/tests/test_processor.py
import unittest  # Import unittest
import pandas as pd  # For DataFrame operations
import processor  # Import processor module
from typing import Dict, Tuple  # For type annotations

class TestSalesProcessor(unittest.TestCase):
    def setUp(self) -> None:
        """Set up test fixtures."""
        self.config: Dict[str, any] = {
            "min_price": 10.0,
            "max_quantity": 100,
            "required_fields": ["product", "price", "quantity"],
            "product_prefix": "Halal",
            "max_decimals": 2
        }
        self.sales_data: pd.DataFrame = pd.DataFrame({
            "product": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
            "price": [999.99, 24.99, 49.99],
            "quantity": [2, 10, 5]
        })

    def test_process_sales_valid(self) -> None:
        """Test processing valid sales data."""
        results, valid_sales = processor.process_sales(self.sales_data, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 2499.83,
            "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
            "top_products": {
                "Halal Laptop": 1999.98,
                "Halal Mouse": 249.90,
                "Halal Keyboard": 249.95
            }
        }
        print(f"Valid results: {results}")  # Log
        self.assertAlmostEqual(results["total_sales"], expected_results["total_sales"], places=2)
        self.assertEqual(results["unique_products"], expected_results["unique_products"])
        self.assertEqual(results["top_products"], expected_results["top_products"])
        self.assertEqual(valid_sales, 3)

    def test_process_sales_empty(self) -> None:
        """Test processing empty DataFrame."""
        empty_df: pd.DataFrame = pd.DataFrame(columns=["product", "price", "quantity"])
        results, valid_sales = processor.process_sales(empty_df, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 0.0,
            "unique_products": [],
            "top_products": {}
        }
        print(f"Empty results: {results}")  # Log
        self.assertEqual(results, expected_results)
        self.assertEqual(valid_sales, 0)

if __name__ == "__main__":
    unittest.main()  # Run tests

# Expected Output:
# test_process_sales_empty (test_processor.TestSalesProcessor.test_process_sales_empty)
# Test processing empty DataFrame. ... ok
# test_process_sales_valid (test_processor.TestSalesProcessor.test_process_sales_valid)
# Test processing valid sales data. ... ok
#
# ----------------------------------------------------------------------
# Ran 2 tests in 0.XXXs
#
# OK

# File: de-onboarding/tests/test_processor_pytest.py
import pandas as pd  # For DataFrame operations
import pytest  # Import pytest
import processor  # Import processor module
from typing import Dict, any  # For type annotations

@pytest.fixture
def config() -> Dict[str, any]:
    """Fixture for config dictionary."""
    return {
        "min_price": 10.0,
        "max_quantity": 100,
        "required_fields": ["product", "price", "quantity"],
        "product_prefix": "Halal",
        "max_decimals": 2
    }

@pytest.fixture
def sales_data() -> pd.DataFrame:
    """Fixture for sales DataFrame."""
    return pd.DataFrame({
        "product": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
        "price": [999.99, 24.99, 49.99],
        "quantity": [2, 10, 5]
    })

def test_process_sales_valid(sales_data: pd.DataFrame, config: Dict[str, any]) -> None:
    """Test processing valid sales data."""
    results, valid_sales = processor.process_sales(sales_data, config)
    expected_results: Dict[str, any] = {
        "total_sales": 2499.83,
        "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
        "top_products": {
            "Halal Laptop": 1999.98,
            "Halal Mouse": 249.90,
            "Halal Keyboard": 249.95
        }
    }
    print(f"Valid results: {results}")  # Log
    print("Sharia compliance verified: all products start with 'Halal'")  # Log Sharia compliance
    assert results["total_sales"] == pytest.approx(expected_results["total_sales"], 0.01)
    assert results["unique_products"] == expected_results["unique_products"]
    assert results["top_products"] == expected_results["top_products"]
    assert valid_sales == 3

def test_process_sales_empty(config: Dict[str, any]) -> None:
    """Test processing empty DataFrame."""
    empty_df: pd.DataFrame = pd.DataFrame(columns=["product", "price", "quantity"])
    results, valid_sales = processor.process_sales(empty_df, config)
    expected_results: Dict[str, any] = {
        "total_sales": 0.0,
        "unique_products": [],
        "top_products": {}
    }
    print(f"Empty results: {results}")  # Log
    assert results == expected_results
    assert valid_sales == 0

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

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

# File: de-onboarding/utils.py (from Chapter 3, updated with type annotations)
def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

def clean_string(s: str) -> str:
    """Strip whitespace from string."""
    return s.strip()

def is_numeric_value(x: any) -> bool:
    """Check if value is an integer or float."""
    return isinstance(x, (int, float))

def has_valid_decimals(x: any, max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)

def apply_valid_decimals(x: any, max_decimals: int) -> bool:
    """Apply has_valid_decimals to a value."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: any) -> bool:
    """Check if value is an integer when converted to string."""
    return str(x).isdigit()

# File: de-onboarding/tests/test_processor.py (unittest)
import unittest
import pandas as pd
import processor
from typing import Dict, Tuple

class TestSalesProcessor(unittest.TestCase):
    def setUp(self) -> None:
        """Set up test fixtures."""
        self.config: Dict[str, any] = processor.read_config("data/config.yaml")
        self.sales_data: pd.DataFrame = pd.read_csv("data/sales.csv")

    def test_process_sales_valid(self) -> None:
        """Test processing valid sales data."""
        df, _, _ = processor.load_and_validate_sales("data/sales.csv", self.config)
        results, valid_sales = processor.process_sales(df, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 2499.83,
            "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
            "top_products": {
                "Halal Laptop": 1999.98,
                "Halal Mouse": 249.90,
                "Halal Keyboard": 249.95
            }
        }
        print(f"Valid results: {results}")  # Log
        self.assertAlmostEqual(results["total_sales"], expected_results["total_sales"], places=2)
        self.assertEqual(results["unique_products"], expected_results["unique_products"])
        self.assertEqual(results["top_products"], expected_results["top_products"])
        self.assertEqual(valid_sales, 3)

    def test_process_sales_empty(self) -> None:
        """Test processing empty CSV."""
        df, _, _ = processor.load_and_validate_sales("data/empty.csv", self.config)
        results, valid_sales = processor.process_sales(df, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 0.0,
            "unique_products": [],
            "top_products": {}
        }
        print(f"Empty results: {results}")  # Log
        self.assertEqual(results, expected_results)
        self.assertEqual(valid_sales, 0)

    def test_process_sales_invalid(self) -> None:
        """Test processing invalid CSV."""
        df, _, _ = processor.load_and_validate_sales("data/invalid.csv", self.config)
        results, valid_sales = processor.process_sales(df, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 0.0,
            "unique_products": [],
            "top_products": {}
        }
        print(f"Invalid results: {results}")  # Log
        self.assertEqual(results, expected_results)
        self.assertEqual(valid_sales, 0)

    def test_process_sales_malformed(self) -> None:
        """Test processing malformed CSV."""
        df, _, _ = processor.load_and_validate_sales("data/malformed.csv", self.config)
        results, valid_sales = processor.process_sales(df, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 249.90,
            "unique_products": ["Halal Mouse"],
            "top_products": {"Halal Mouse": 249.90}
        }
        print(f"Malformed results: {results}")  # Log
        self.assertAlmostEqual(results["total_sales"], expected_results["total_sales"], places=2)
        self.assertEqual(results["unique_products"], expected_results["unique_products"])
        self.assertEqual(results["top_products"], expected_results["top_products"])
        self.assertEqual(valid_sales, 1)

    def test_process_sales_negative(self) -> None:
        """Test processing negative CSV."""
        df, _, _ = processor.load_and_validate_sales("data/negative.csv", self.config)
        results, valid_sales = processor.process_sales(df, self.config)
        expected_results: Dict[str, any] = {
            "total_sales": 249.90,
            "unique_products": ["Halal Mouse"],
            "top_products": {"Halal Mouse": 249.90}
        }
        print(f"Negative results: {results}")  # Log
        self.assertAlmostEqual(results["total_sales"], expected_results["total_sales"], places=2)
        self.assertEqual(results["unique_products"], expected_results["unique_products"])
        self.assertEqual(results["top_products"], expected_results["top_products"])
        self.assertEqual(valid_sales, 1)

if __name__ == "__main__":
    unittest.main()

# File: de-onboarding/tests/test_processor_pytest.py
import pandas as pd
import pytest
import processor
from typing import Dict, any

@pytest.fixture
def config() -> Dict[str, any]:
    """Fixture for config dictionary."""
    return processor.read_config("data/config.yaml")

@pytest.fixture
def sales_data() -> pd.DataFrame:
    """Fixture for sales DataFrame."""
    return pd.read_csv("data/sales.csv")

def test_process_sales_valid(sales_data: pd.DataFrame, config: Dict[str, any]) -> None:
    """Test processing valid sales data."""
    df, _, _ = processor.load_and_validate_sales("data/sales.csv", config)
    results, valid_sales = processor.process_sales(df, config)
    expected_results: Dict[str, any] = {
        "total_sales": 2499.83,
        "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
        "top_products": {
            "Halal Laptop": 1999.98,
            "Halal Mouse": 249.90,
            "Halal Keyboard": 249.95
        }
    }
    print(f"Valid results: {results}")  # Log
    print("Sharia compliance verified: all products start with 'Halal'")  # Log Sharia compliance
    assert results["total_sales"] == pytest.approx(expected_results["total_sales"], 0.01)
    assert results["unique_products"] == expected_results["unique_products"]
    assert results["top_products"] == expected_results["top_products"]
    assert valid_sales == 3

def test_process_sales_empty(config: Dict[str, any]) -> None:
    """Test processing empty CSV."""
    df, _, _ = processor.load_and_validate_sales("data/empty.csv", config)
    results, valid_sales = processor.process_sales(df, config)
    expected_results: Dict[str, any] = {
        "total_sales": 0.0,
        "unique_products": [],
        "top_products": {}
    }
    print(f"Empty results: {results}")  # Log
    assert results == expected_results
    assert valid_sales == 0

def test_process_sales_invalid(config: Dict[str, any]) -> None:
    """Test processing invalid CSV."""
    df, _, _ = processor.load_and_validate_sales("data/invalid.csv", config)
    results, valid_sales = processor.process_sales(df, config)
    expected_results: Dict[str, any] = {
        "total_sales": 0.0,
        "unique_products": [],
        "top_products": {}
    }
    print(f"Invalid results: {results}")  # Log
    assert results == expected_results
    assert valid_sales == 0

def test_process_sales_malformed(config: Dict[str, any]) -> None:
    """Test processing malformed CSV."""
    df, _, _ = processor.load_and_validate_sales("data/malformed.csv", config)
    results, valid_sales = processor.process_sales(df, config)
    expected_results: Dict[str, any] = {
        "total_sales": 249.90,
        "unique_products": ["Halal Mouse"],
        "top_products": {"Halal Mouse": 249.90}
    }
    print(f"Malformed results: {results}")  # Log
    assert results["total_sales"] == pytest.approx(expected_results["total_sales"], 0.01)
    assert results["unique_products"] == expected_results["unique_products"]
    assert results["top_products"] == expected_results["top_products"]
    assert valid_sales == 1

def test_process_sales_negative(config: Dict[str, any]) -> None:
    """Test processing negative CSV."""
    df, _, _ = processor.load_and_validate_sales("data/negative.csv", config)
    results, valid_sales = processor.process_sales(df, config)
    expected_results: Dict[str, any] = {
        "total_sales": 249.90,
        "unique_products": ["Halal Mouse"],
        "top_products": {"Halal Mouse": 249.90}
    }
    print(f"Negative results: {results}")  # Log
    assert results["total_sales"] == pytest.approx(expected_results["total_sales"], 0.01)
    assert results["unique_products"] == expected_results["unique_products"]
    assert results["top_products"] == expected_results["top_products"]
    assert valid_sales == 1

# File: de-onboarding/write_comparison.py
def write_comparison(output_path: str) -> None:
    """Write unittest vs pytest comparison."""
    comparison: str = """
unittest vs pytest Comparison:

1. Syntax:
   - unittest: Class-based, uses methods like assertEqual, assertAlmostEqual.
   - pytest: Function-based, uses standard Python assert statements, with pytest.approx for floats.
   - Example:
     unittest: self.assertEqual(results["total_sales"], 2499.83)
     pytest: assert results["total_sales"] == 2499.83
   - Advantage: pytest is simpler, requiring less boilerplate code.

2. Setup:
   - unittest: Uses setUp method for fixtures, defined in test class.
   - pytest: Uses @pytest.fixture decorator for reusable fixtures, more flexible.
   - Advantage: pytest fixtures are easier to reuse across tests.

3. Execution:
   - unittest: Run with `python -m unittest test_file.py`.
   - pytest: Run with `pytest test_file.py -v` for verbose output.
   - Advantage: pytest auto-discovers tests, simplifying execution.

4. Features:
   - unittest: Built-in, good for structured, complex suites.
   - pytest: Rich ecosystem, supports plugins, fixtures, and parallel testing.
   - Advantage: pytest is more extensible for pipeline testing.

5. Use Case:
   - unittest: Suitable for large, structured projects with strict conventions.
   - pytest: Ideal for rapid development and data engineering pipelines.
   - Implication: pytest is preferred for Hijra Group's iterative analytics.

Conclusion: For data engineering, pytest's simplicity and flexibility make it ideal for testing pipelines, while unittest suits rigid, large-scale projects. Both ensure reliable analytics for Hijra Group's Sharia-compliant data.
"""
    print(f"Writing comparison to: {output_path}")  # Log
    with open(output_path, "w") as file:
        file.write(comparison.strip())
    print(f"Comparison written to {output_path}")

if __name__ == "__main__":
    write_comparison("data/test_comparison.txt")

# File: de-onboarding/tests/test_config_unittest.py
import unittest
import processor
from typing import Dict, any

class TestConfig(unittest.TestCase):
    def test_read_config(self) -> None:
        """Test reading config.yaml."""
        config: Dict[str, any] = processor.read_config("data/config.yaml")
        expected_config: Dict[str, any] = {
            "min_price": 10.0,
            "max_quantity": 100,
            "required_fields": ["product", "price", "quantity"],
            "product_prefix": "Halal",
            "max_decimals": 2
        }
        print(f"Config: {config}")  # Log
        self.assertEqual(config, expected_config)

if __name__ == "__main__":
    unittest.main()

# Output:
# Config: {'min_price': 10.0, 'max_quantity': 100, 'required_fields': ['product', 'price', 'quantity'], 'product_prefix': 'Halal', 'max_decimals': 2}
# ...
# Ran 1 test in 0.XXXs
# OK

# File: de-onboarding/tests/test_config_pytest.py
import pytest
import processor
from typing import Dict, any

def test_read_config() -> None:
    """Test reading config.yaml."""
    config: Dict[str, any] = processor.read_config("data/config.yaml")
    expected_config: Dict[str, any] = {
        "min_price": 10.0,
        "max_quantity": 100,
        "required_fields": ["product", "price", "quantity"],
        "product_prefix": "Halal",
        "max_decimals": 2
    }
    print(f"Config: {config}")  # Log
    assert config == expected_config

def test_read_invalid_config() -> None:
    """Test reading invalid_config.yaml, expecting failure."""
    with pytest.raises(KeyError) as exc_info:
        config: Dict[str, any] = processor.read_config("data/invalid_config.yaml")
        print(f"Config: {config}")  # Log (won't reach here)
    print(f"Expected KeyError: {exc_info.value}")  # Log
    assert "product_prefix" in str(exc_info.value)

# Output:
# Config: {'min_price': 10.0, 'max_quantity': 100, 'required_fields': ['product', 'price', 'quantity'], 'product_prefix': 'Halal', 'max_decimals': 2}
# ...
# Expected KeyError: 'product_prefix'
# ...
# 2 passed in 0.XXs

# File: de-onboarding/tests/test_validation_unittest.py
import unittest
import pandas as pd
import processor
from typing import Dict, Tuple

class TestValidation(unittest.TestCase):
    def setUp(self) -> None:
        """Set up test fixtures."""
        self.config: Dict[str, any] = processor.read_config("data/config.yaml")

    def test_load_and_validate_malformed(self) -> None:
        """Test validation with malformed.csv."""
        df, valid_sales, total_records = processor.load_and_validate_sales("data/malformed.csv", self.config)
        expected_df: pd.DataFrame = pd.DataFrame({
            "product": ["Halal Mouse"],
            "price": [24.99],
            "quantity": [10]
        })
        print(f"Validated DataFrame: {df}")  # Log
        pd.testing.assert_frame_equal(df.reset_index(drop=True), expected_df.reset_index(drop=True))
        self.assertEqual(valid_sales, 1)
        self.assertEqual(total_records, 1)

if __name__ == "__main__":
    unittest.main()

# Output:
# Validated DataFrame: ...
# ...
# Ran 1 test in 0.XXXs
# OK

# File: de-onboarding/tests/test_validation_pytest.py
import pytest
import pandas as pd
import processor
from typing import Dict, Tuple

@pytest.fixture
def config() -> Dict[str, any]:
    """Fixture for config dictionary."""
    return processor.read_config("data/config.yaml")

def test_load_and_validate_negative(config: Dict[str, any]) -> None:
    """Test validation with negative.csv, ensuring Sharia-compliant Halal products."""
    df, valid_sales, total_records = processor.load_and_validate_sales("data/negative.csv", config)
    expected_df: pd.DataFrame = pd.DataFrame({
        "product": ["Halal Mouse"],
        "price": [24.99],
        "quantity": [10]
    })
    print(f"Validated DataFrame: {df}")  # Log
    pd.testing.assert_frame_equal(df.reset_index(drop=True), expected_df.reset_index(drop=True))
    assert valid_sales == 1
    assert total_records == 1
    # Verify Sharia compliance: all products start with 'Halal'
    assert all(df["product"].str.startswith("Halal")), "Non-Halal products detected"

# Output:
# Validated DataFrame: ...
# ...
# 1 passed in 0.XXs

# File: de-onboarding/write_ex5_concepts.py
def write_concepts(output_path: str) -> None:
    """Write conceptual and practical analysis of unit testing and pytest fixtures."""
    explanation: str = """
Unit testing ensures that each function in Hijra Group’s sales pipelines, like calculating total sales, produces correct results, preventing costly errors in financial analytics. For example, a bug in `process_sales` could misreport sales totals, affecting strategic decisions. Unit tests catch these errors early, ensuring Sharia-compliant data (e.g., Halal products) is processed accurately. pytest’s fixtures improve test maintainability by reusing setup code, like loading `config.yaml`, across multiple tests, reducing duplication and making tests easier to update when pipeline logic changes. For instance, a unit test for `process_sales` could detect an incorrect total sales calculation due to a type mismatch, preventing inaccurate financial reports. Adding a test for an invalid `config.yaml` (missing `product_prefix`) fails because `load_and_validate_sales` expects a `product_prefix` key, raising a KeyError, which highlights the importance of config validation.
"""
    print(f"Writing concepts to: {output_path}")  # Log
    with open(output_path, "w") as file:
        file.write(explanation.strip())
    print(f"Concepts written to {output_path}")

if __name__ == "__main__":
    write_concepts("data/ex5_concepts.txt")

# Output:
# Writing concepts to: data/ex5_concepts.txt
# Concepts written to data/ex5_concepts.txt