Testing

The AI Ingredient Scanner includes a comprehensive test suite covering unit tests, integration tests, end-to-end validation, and performance benchmarks.

Running Tests

Quick Start

# Activate virtual environment
source venv/bin/activate

# Run all tests with coverage
pytest tests/ -v --cov

# Run specific test file
pytest tests/test_agents.py -v

# Run with verbose output
pytest tests/ -v --tb=short

# Run only fast tests (exclude performance)
pytest tests/ -m "not slow"

Test Configuration

The project uses pytest.ini for configuration:

[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -v --cov=config --cov=agents --cov=tools --cov=state --cov=graph --cov=services --cov-report=term-missing --cov-fail-under=70
filterwarnings =
    ignore::DeprecationWarning

Test Categories

Unit Tests

Tests for individual agent functions and tools.

Agent Tests (test_agents.py)

class TestResearchAgent:
    """Tests for Research Agent."""

    def test_create_unknown_ingredient(self):
        """Test creation of unknown ingredient record."""
        result = _create_unknown_ingredient("mystery_ingredient")
        assert result["name"] == "mystery_ingredient"
        assert result["source"] == "unknown"
        assert result["confidence"] == 0.0

    def test_has_research_data_false(self, base_state):
        """Test has_research_data returns False when empty."""
        assert has_research_data(base_state) is False

    @patch("agents.research.lookup_ingredient")
    @patch("agents.research.grounded_ingredient_search")
    def test_research_ingredients_fallback(self, mock_search, mock_lookup):
        """Test research falls back to grounded search."""
        mock_lookup.return_value = create_ingredient(confidence=0.5)
        mock_search.return_value = create_ingredient(confidence=0.8)

        result = research_ingredients(state)
        assert result["ingredient_data"][0]["source"] == "google_search"

Tool Tests (test_tools.py)

class TestSafetyScorer:
    """Tests for safety scoring functions."""

    def test_calculate_risk_sensitive_skin_fragrance(self):
        """Test risk increases for fragrance with sensitive skin."""
        risk = calculate_risk_score(fragrance_ingredient, sensitive_profile)
        assert risk == 0.7  # 0.4 base + 0.3 modifier

    def test_classify_risk_level_boundaries(self):
        """Test risk classification at boundaries."""
        assert classify_risk_level(0.29) == RiskLevel.LOW
        assert classify_risk_level(0.30) == RiskLevel.MEDIUM
        assert classify_risk_level(0.60) == RiskLevel.HIGH


class TestAllergenMatcher:
    """Tests for allergen matching functions."""

    def test_check_allergen_match_positive(self):
        """Test positive allergen match."""
        ingredient = create_ingredient(
            name="whey protein",
            safety_notes="Derived from milk",
        )
        is_match, allergy = check_allergen_match(ingredient, profile)
        assert is_match is True
        assert allergy == "milk"

Integration Tests

API Tests (test_api.py)

class TestHealthEndpoints:
    """Tests for health check endpoints."""

    def test_root_endpoint(self, client):
        """Test root endpoint returns OK status."""
        response = client.get("/")
        assert response.status_code == 200
        assert response.json()["status"] == "ok"

    def test_health_endpoint(self, client):
        """Test health endpoint returns healthy status."""
        response = client.get("/health")
        assert response.status_code == 200


class TestAnalyzeEndpoint:
    """Tests for the /analyze endpoint."""

    def test_analyze_missing_ingredients(self, client):
        """Test that empty ingredients returns error."""
        response = client.post("/analyze", json={"ingredients": ""})
        assert response.status_code == 400

Workflow Tests (test_workflow.py)

class TestWorkflowExecution:
    """Tests for workflow execution."""

    @patch("agents.research.research_ingredients")
    @patch("agents.analysis.analyze_ingredients")
    @patch("agents.critic.validate_report")
    def test_happy_path(self, mock_critic, mock_analysis, mock_research):
        """Test successful workflow execution."""
        mock_research.return_value = state_with_ingredients
        mock_analysis.return_value = state_with_report
        mock_critic.return_value = state_approved

        result = run_analysis(
            session_id="test",
            product_name="Test",
            ingredients=["water"],
            allergies=[],
            skin_type="normal",
            expertise="beginner",
        )

        assert result["critic_feedback"]["result"] == ValidationResult.APPROVED

End-to-End Tests (test_e2e.py)

class TestEndToEndWorkflow:
    """E2E tests for the complete analysis workflow."""

    def test_complete_workflow_happy_path(self, mock_llm_responses):
        """Test complete workflow from start to finish."""
        result = run_analysis(
            session_id="e2e-test-001",
            product_name="Test Moisturizer",
            ingredients=["Water", "Glycerin", "Vitamin E"],
            allergies=[],
            skin_type="normal",
            expertise="beginner",
        )

        # Verify workflow completed successfully
        assert result.get("error") is None
        assert result.get("analysis_report") is not None
        assert result.get("critic_feedback") is not None

        # Verify routing history shows complete flow
        history = result.get("routing_history", [])
        assert "research" in history
        assert "analysis" in history
        assert "critic" in history

Performance Tests (test_performance.py)

class TestAPIPerformance:
    """Performance tests for API endpoints."""

    def test_health_endpoint_response_time(self, client):
        """Health endpoint should respond within 100ms."""
        start = time.time()
        response = client.get("/health")
        elapsed = time.time() - start

        assert response.status_code == 200
        assert elapsed < 0.1

    def test_concurrent_health_checks(self, client):
        """Multiple concurrent health checks should all succeed."""
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [executor.submit(client.get, "/health") for _ in range(10)]
            results = [f.result() for f in concurrent.futures.as_completed(futures)]

        assert all(r.status_code == 200 for r in results)

Coverage Report

Test Markers

Custom markers for test categorization:

# Run only E2E tests
pytest -m e2e

# Skip slow tests
pytest -m "not slow"

# Run only performance tests
pytest -m performance

# Run integration tests
pytest -m integration

Best Practices

@patch("agents.research.lookup_ingredient")
@patch("agents.research.grounded_ingredient_search")
def test_with_mocks(self, mock_search, mock_lookup):
    mock_lookup.return_value = create_test_ingredient("water")
    # Test logic here

@pytest.fixture
def base_state() -> WorkflowState:
    return WorkflowState(
        session_id="test",
        product_name="Test",
        raw_ingredients=["water"],
        user_profile=base_user_profile(),
        ingredient_data=[],
        retry_count=0,
        routing_history=[],
        error=None,
    )

Related Documentation