Testing Strategy

This document describes the comprehensive testing strategy for the Graffold project, including test organization, execution, and best practices.

Table of Contents

• Overview
• Test Categories
• Running Tests
• Pytest Markers
• Mocking Strategy
• Writing Tests
• CI/CD Integration
• Performance Targets

Overview

The test suite is organized into multiple categories with different execution strategies:

• Unit tests (750 tests, ~80%): Fast, isolated tests with mocked dependencies
• Integration tests (49 tests, ~5%): Tests with real external services
• Property-based tests (121 tests, ~13%): Hypothesis-based correctness properties
• Slow tests (53 tests, ~6%): Long-running or report-only tests
• Known failures (2 tests, <1%): Expected failures that don't block CI

Total: 927 tests with 100% categorization

Test Categories

Unit Tests

Purpose: Fast, isolated tests with mocked external dependencies

Characteristics: - Complete in <3 minutes total - Use mocked LLM and database connections - No network calls or external service dependencies - Default test category (no explicit marker required)

Example:

def test_graph_tool_count_nodes(mock_neo4j_database):
    """Test counting nodes with mocked database."""
    mock_neo4j_database._execute_cypher.return_value = [{'count': 42}]

    tools = GraphTools(db=mock_neo4j_database)
    count_tool = tools.get_tools()[0]

    result = count_tool.invoke({'label': 'Protein'})
    assert json.loads(result)["count"] == 42

Integration Tests

Purpose: Validate functionality with real external services

Characteristics: - Require live Neo4j databases and LLM services - Run in separate CI job with longer timeout (10 minutes) - Marked with @pytest.mark.integration - Don't block PR merges (informational only)

Example:

@pytest.mark.integration
class TestEmbeddingPipelineIntegration:
    """Integration tests with real database and LLM services."""

    def test_full_pipeline(self):
        # Uses real services configured via environment variables
        db = DatabaseFactory.create_database(
            uri=os.getenv("NEO4J_URI"), ...
        )
        result = pipeline.run(db)
        assert result.success

Property-Based Tests

Purpose: Test correctness properties across many generated inputs

Characteristics: - Use Hypothesis for property-based testing - Files end with *_properties.py - Module-level marker: pytestmark = pytest.mark.properties - Configurable max_examples (50 in CI, 100 locally)

Example:

from __future__ import annotations

import pytest
from hypothesis import given, strategies as st

pytestmark = pytest.mark.properties

@given(st.text(min_size=1))
def test_entity_name_normalization(name):
    """Property: normalized names are always lowercase and trimmed."""
    normalized = normalize_entity_name(name)
    assert normalized == normalized.lower().strip()

Slow Tests

Purpose: Long-running tests or tests with >100 Hypothesis examples

Characteristics: - Marked with @pytest.mark.slow - Run in separate CI job with 10-minute timeout - Typically report-only or comprehensive validation tests

Example:

@pytest.mark.slow
@given(st.lists(st.text(), min_size=100, max_size=1000))
def test_large_batch_processing(entities):
    """Test processing large batches of entities."""
    result = process_entities(entities)
    assert len(result) == len(entities)

Known Failures

Purpose: Document expected failures without blocking CI

Characteristics: - Marked with @pytest.mark.xfail(reason="...") - Run in separate CI job with continue-on-error: true - Don't block PR merges - Should include clear documentation of the issue

Example:

@pytest.mark.xfail(reason="LangChain tool decorator requires real callable, not MagicMock")
def test_complex_tool_mocking(mock_llm):
    """Test that requires complex tool mocking.

    Known issue: LangChain's @tool decorator requires actual callable
    functions with __name__ attribute, not MagicMock objects.
    """
    # Test implementation

Running Tests

Local Development

# Run all unit tests (default, fast feedback)
pytest

# Run specific test category
pytest -m "integration"
pytest -m "properties"
pytest -m "slow"
pytest -m "xfail"

# Run all tests except slow and integration
pytest -m "not integration and not slow"

# Run specific test file
pytest tests/test_entity_resolver.py

# Run specific test function
pytest tests/test_entity_resolver.py::test_consolidate_by_uniprot_id

# Run with verbose output
pytest -v

# Run without parallel execution (for debugging)
pytest -n 0

# Run with coverage
pytest --cov=src --cov-report=html

CI Execution

Tests run automatically in GitHub Actions with parallel execution:

# Unit tests (3-minute timeout)
pytest -v -m "not integration and not slow and not properties" -n auto

# Integration tests (10-minute timeout, load-grouped)
pytest -v -m "integration" -n auto --dist loadgroup

# Property-based tests (10-minute timeout)
pytest -v -m "properties" -n auto

# Slow tests (10-minute timeout)
pytest -v -m "slow" -n auto

# Known failures (10-minute timeout, non-blocking)
pytest -v -m "xfail" -n auto

Pytest Markers

Markers are defined in pyproject.toml and used to categorize tests:

[tool.pytest.ini_options]
markers = [
    "unit: marks tests as unit tests (fast, isolated, mocked dependencies)",
    "integration: marks tests as integration tests requiring live services",
    "slow: marks tests as slow/report-only (deselect with '-m \"not slow\"')",
    "properties: marks tests as property-based tests using hypothesis",
    "xfail: marks tests as expected failures (doesn't block CI)",
]

When to Use Each Marker

Marker	Use When	Example
unit	Explicit unit test marking (optional, default)	Fast, isolated tests with mocks
integration	Test requires real external services	Database queries, LLM API calls
slow	Test takes >30 seconds or uses >100 Hypothesis examples	Large batch processing, comprehensive validation
properties	Property-based test using Hypothesis	Invariant testing across many inputs
xfail	Known failure that shouldn't block CI	Complex mocking issues, upstream bugs

Marker Usage Patterns

Module-level marker (applies to all tests in file):

import pytest

pytestmark = pytest.mark.properties

Class-level marker (applies to all tests in class):

@pytest.mark.integration
class TestDatabaseIntegration:
    def test_query(self):
        pass

Function-level marker (applies to single test):

@pytest.mark.slow
def test_large_dataset():
    pass

Multiple markers:

@pytest.mark.slow
@pytest.mark.integration
def test_full_pipeline():
    pass

Mocking Strategy

The test suite uses centralized pytest fixtures in tests/conftest.py for consistent mocking across all tests.

Available Fixtures

LLM Mocking

• mock_llm - Generic LLM with invoke() and generate() methods
• mock_async_llm - Async LLM with ainvoke() and agenerate() methods
• mock_ollama_llm - Ollama-specific response format
• mock_bedrock_llm - AWS Bedrock response format
• mock_sagemaker_llm - AWS SageMaker response format
• mock_openai_llm - OpenAI response format

Database Mocking

• mock_neo4j_database - Graph database with _execute_cypher() method (works for both Neo4j and Memgraph backends)
• mock_neo4j_driver - Database driver with session management

Other Mocking

• mock_embedder - Embedding model returning numpy arrays (384-dim)
• sample_graph_schema - Sample graph schema for testing
• sample_entities - Sample entity extraction results
• sample_relationships - Sample relationship extraction results

Mocking Best Practices

1. Use centralized fixtures: Prefer fixtures from conftest.py over creating local mocks
2. Configure return values: Set return_value or side_effect for specific test scenarios
3. Verify calls: Use assert_called_with() to verify correct parameters
4. Test error paths: Mock exceptions to test error handling
5. Keep mocks simple: Only mock what's necessary for the test

Mocking Examples

Basic mocking:

def test_with_mock(mock_neo4j_database):
    # Configure mock return value
    mock_neo4j_database._execute_cypher.return_value = [{'count': 42}]

    # Use in test
    result = my_function(mock_neo4j_database)

    # Verify
    assert result == 42
    mock_neo4j_database._execute_cypher.assert_called_once()

Custom mock behavior:

def test_custom_behavior(mock_llm):
    # Override default behavior
    custom_response = MagicMock()
    custom_response.content = json.dumps({"entities": [...]})
    mock_llm.invoke.return_value = custom_response

    result = extract_entities(mock_llm, "text")
    assert len(result["entities"]) > 0

Testing error handling:

def test_error_handling(mock_neo4j_database):
    # Mock exception
    mock_neo4j_database._execute_cypher.side_effect = Exception("Connection failed")

    # Verify error handling
    with pytest.raises(Exception, match="Connection failed"):
        my_function(mock_neo4j_database)

See also: - Test Mocking Guide - Comprehensive mocking guide - Test Mocking Quick Reference - Quick reference card

Writing Tests

Test File Organization

tests/
├── conftest.py                          # Centralized fixtures
├── test_entity_resolver.py              # Unit tests
├── test_entity_resolver_properties.py   # Property-based tests
├── test_embedding_pipeline.py           # Integration tests
└── test_large_batch_processing.py       # Slow tests

Test Naming Conventions

• Test files: test_*.py
• Test classes: Test*
• Test functions: test_*
• Property-based test files: test_*_properties.py

Python Import Order

Critical: from __future__ import annotations must be at the very top of the file.

Correct order: 1. Module docstring 2. Future imports (from __future__ import annotations) 3. Standard library imports 4. Third-party imports 5. Local imports

Example:

"""Test module for entity resolution."""

from __future__ import annotations

import json
import os
from typing import Any

import pytest
from hypothesis import given, strategies as st

from src.processors.entity_resolver import EntityResolver

Common mistake: Placing future imports after pytest imports causes SyntaxError.

Well-Structured Test Examples

Unit Test Example

"""Unit tests for entity resolver."""

from __future__ import annotations

import pytest

from src.processors.entity_resolver import EntityResolver


class TestEntityResolver:
    """Test suite for EntityResolver."""

    def test_consolidate_by_uniprot_id(self, mock_neo4j_database):
        """Test entity consolidation by UniProt ID.

        Given: Multiple protein nodes with same UniProt ID
        When: Consolidation is run
        Then: Nodes are merged into single canonical node
        """
        # Arrange
        mock_neo4j_database._execute_cypher.return_value = [
            {'merged': 5, 'canonical': 'P04637'}
        ]
        resolver = EntityResolver(db=mock_neo4j_database)

        # Act
        result = resolver.consolidate_by_uniprot_id()

        # Assert
        assert result['merged'] == 5
        assert result['canonical'] == 'P04637'
        mock_neo4j_database._execute_cypher.assert_called_once()

Property-Based Test Example

"""Property-based tests for entity resolver."""

from __future__ import annotations

import pytest
from hypothesis import given, strategies as st

from src.processors.entity_resolver import normalize_entity_name

pytestmark = pytest.mark.properties


@given(st.text(min_size=1))
def test_normalization_idempotent(name):
    """Property: normalizing twice gives same result as normalizing once."""
    normalized_once = normalize_entity_name(name)
    normalized_twice = normalize_entity_name(normalized_once)
    assert normalized_once == normalized_twice


@given(st.text(min_size=1))
def test_normalization_lowercase(name):
    """Property: normalized names are always lowercase."""
    normalized = normalize_entity_name(name)
    assert normalized == normalized.lower()

Integration Test Example

"""Integration tests for embedding pipeline."""

from __future__ import annotations

import os

import pytest

from src.core.database import Neo4jDatabase
from src.ingest_pipeline.embedding_pipeline import EmbeddingPipeline


@pytest.mark.integration
class TestEmbeddingPipelineIntegration:
    """Integration tests with real database and embedding model."""

    def test_generate_embeddings(self):
        """Test embedding generation with real database.

        Requires: Neo4j or Memgraph running on localhost:7687
        """
        # Arrange
        db = DatabaseFactory.create_database(
            uri=os.getenv("NEO4J_URI"),
            user=os.getenv("NEO4J_USER"),
            password=os.getenv("NEO4J_PASSWORD")
        )
        pipeline = EmbeddingPipeline(db=db)

        # Act
        result = pipeline.generate_embeddings(limit=10)

        # Assert
        assert result['embeddings_created'] > 0
        assert result['errors'] == 0

Hypothesis Configuration

Property-based tests use Hypothesis with profile-based configuration:

• CI profile: 50 examples for faster execution
• Dev profile: 100 examples for thorough local testing

Configuration is automatic based on environment detection in tests/conftest.py:

from hypothesis import settings

settings.register_profile("ci", max_examples=50)
settings.register_profile("dev", max_examples=100)

if os.getenv("CI") == "true" or os.getenv("GITHUB_ACTIONS") == "true":
    settings.load_profile("ci")
else:
    settings.load_profile("dev")

Mark tests with >100 examples as slow:

@pytest.mark.slow
@given(st.lists(st.text(), min_size=100, max_size=1000))
def test_large_dataset(data):
    """Test with many examples (>100)."""
    pass

CI/CD Integration

GitHub Actions Workflow

Tests run in parallel jobs using matrix strategy:

strategy:
  fail-fast: false
  matrix:
    test-category:
      - name: unit
        marker: "not integration and not slow and not properties"
        timeout: 3
      - name: integration
        marker: "integration"
        timeout: 10
      - name: property-based
        marker: "properties"
        timeout: 10
      - name: slow
        marker: "slow"
        timeout: 10

Parallel Execution

• pytest-xdist: Enables parallel test execution with -n auto
• Auto-detection: Automatically detects CPU cores (typically 10 workers in CI)
• Load distribution:
• Default: LoadScheduling distributes tests evenly
• Integration tests: --dist loadgroup keeps related tests together

Job Configuration

Job	Timeout	Blocking	Workers	Distribution
Unit	3 min	Yes	10	LoadScheduling
Integration	10 min	No	10	LoadGroupScheduling
Property-based	10 min	No	10	LoadScheduling
Slow	10 min	No	10	LoadScheduling
Known failures	10 min	No	10	LoadScheduling

Blocking: Only unit tests block PR merges. Other jobs provide informational results.

Environment Variables

CI sets up test environment variables (used for both Neo4j and Memgraph):

- name: Set up environment variables for testing
  run: |
    echo "NEO4J_URI=bolt://localhost:7687" >> $GITHUB_ENV
    echo "NEO4J_USER=neo4j" >> $GITHUB_ENV
    echo "NEO4J_PASSWORD=testpass" >> $GITHUB_ENV
    echo "ENTREZ_EMAIL=test@example.com" >> $GITHUB_ENV
    echo "ENTREZ_API_KEY=fake-key" >> $GITHUB_ENV
    # ... more environment variables

Performance Targets

Current Performance

Metric	Target	Actual	Status
Unit tests	<3 min	21.3s	✅ 7.5x faster
Total CI time	<10 min	~3-4 min	✅ 4-5x faster
Stretch goal	<5 min	~3-4 min	✅ Achieved
Test categorization	100%	100%	✅ Complete
Blocking failures	0	0	✅ Clean

Optimization Strategies

1. Parallel execution: -n auto uses all CPU cores (10 workers)
2. Test categorization: Separate fast/slow tests for optimal feedback
3. Mocked dependencies: Unit tests avoid network calls
4. Hypothesis profiles: Lower max_examples in CI (50 vs 100)
5. Load grouping: Integration tests use --dist loadgroup for shared setup

Developer Feedback Loop

• Local unit tests: <1 minute (15-20x faster than baseline)
• CI unit tests: ~21 seconds
• Full CI pipeline: ~3-4 minutes
• Baseline (before optimization): ~15-20 minutes

Troubleshooting

Common Issues

Fixture not found error:

fixture 'mock_db' not found

Solution: Use mock_neo4j_database (centralized fixture name).

Import order SyntaxError:

SyntaxError: from __future__ imports must occur at the beginning of the file

Solution: Move from __future__ import annotations to top of file, after docstring.

Hypothesis examples not using profile:

# Wrong: explicit max_examples overrides profile
@given(st.text())
@settings(max_examples=200)
def test_something(text):
    pass

# Right: use profile defaults
@given(st.text())
def test_something(text):
    pass

Test marked as slow but still runs in unit job:

# Wrong: missing pytest import
@pytest.mark.slow  # NameError: name 'pytest' is not defined
def test_something():
    pass

# Right: import pytest
import pytest

@pytest.mark.slow
def test_something():
    pass

Debugging Tests

# Run single test with verbose output
pytest tests/test_entity_resolver.py::test_consolidate -v

# Run without parallel execution (easier debugging)
pytest -n 0

# Show print statements
pytest -s

# Drop into debugger on failure
pytest --pdb

# Show available fixtures
pytest --fixtures

# Show available markers
pytest --markers

# Collect tests without running
pytest --collect-only

Summary

The Graffold testing strategy provides:

• ✅ Fast feedback: Unit tests complete in <1 minute locally
• ✅ Comprehensive coverage: 927 tests across 5 categories
• ✅ Parallel execution: 10 workers with automatic load distribution
• ✅ Reliable mocking: Centralized fixtures for consistent behavior
• ✅ CI optimization: 4-5x faster than baseline (~3-4 min total)
• ✅ Clear categorization: 100% test coverage with appropriate markers
• ✅ Non-blocking validation: Integration/slow tests don't block PRs

For more details, see: - Test Mocking Guide - Test Mocking Quick Reference - CI Configuration - Pytest Configuration