Testing Guide - Zyeta.io

This guide outlines the testing strategy, standards, and best practices for the Zyeta backend codebase. It serves as a blueprint for developers to ensure consistent quality and reliability through automated testing.

Overview

A robust testing approach is crucial for maintaining code quality and ensuring that our services function correctly after changes. Our testing strategy comprises multiple levels:

Test Pyramid

The test pyramid represents the ideal distribution of different test types:

This visualization demonstrates that:

Unit tests should form the majority of your test suite (fastest to run, easiest to write)
Integration tests should be fewer in number but cover critical component interactions
End-to-End tests should be the fewest but test complete user workflows

Test Directory Structure

Organize tests with a clear structure that mirrors the source code:

zyeta.backend/
├── src/
│   ├── models/
│   ├── services/
│   └── ...
└── tests/
    ├── unit/
    │   ├── models/
    │   └── services/
    ├── integration/
    │   └── services/
    ├── e2e/
    │   └── api/
    ├── conftest.py
    └── fixtures/
        ├── database.py
        ├── models.py
        └── services.py

Setting Up the Testing Environment

Prerequisites

The project uses pytest for running tests. Dependencies are managed through Poetry and include:

pytest
pytest-asyncio (for testing async code)
pytest-cov (for coverage reporting)

These are already included in pyproject.toml.

Test Configuration

Create a conftest.py file at the root of the tests directory to define shared fixtures:

import asyncio
import os
import pytest
from typing import AsyncGenerator, Generator

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from src.database import Base, get_db
from src.config import get_settings

# Test database URL - use an isolated test database
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/test_zyeta"

@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
    """Create an instance of the default event loop for each test session."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def db_engine():
    """Create a test database engine."""
    engine = create_async_engine(TEST_DATABASE_URL)
    
    # Create all tables
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    
    yield engine
    
    # Clean up
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    
    await engine.dispose()

@pytest.fixture
async def db_session(db_engine) -> AsyncGenerator[AsyncSession, None]:
    """Create a test database session."""
    async_session = sessionmaker(
        db_engine, expire_on_commit=False, class_=AsyncSession
    )
    
    async with async_session() as session:
        # Begin transaction
        async with session.begin():
            yield session
            # Rollback transaction after test completes
            await session.rollback()

@pytest.fixture
async def override_get_db(db_session: AsyncSession):
    """Override the get_db dependency for testing."""
    async def _override_get_db():
        yield db_session
    
    return _override_get_db

Test Fixture Dependencies

Unit Tests

Unit tests focus on testing individual components in isolation.

Writing Unit Tests for Models

# tests/unit/models/test_agent_model.py
import pytest
from uuid import uuid4

from src.models import AgentModel

@pytest.mark.asyncio
async def test_agent_model_creation(db_session):
    """Test creating an AgentModel instance."""
    # Arrange
    agent_data = {
        "organization_id": uuid4(),
        "user_id": uuid4(),
        "name": "Test Agent",
        "description": "Test agent description",
        "model_id": uuid4(),
        "is_active": True,
        "settings": {"key": "value"}
    }
    
    # Act
    agent = AgentModel(**agent_data)
    db_session.add(agent)
    await db_session.commit()
    await db_session.refresh(agent)
    
    # Assert
    assert agent.id is not None
    assert agent.name == "Test Agent"
    assert agent.is_active is True

Unit Test Flow

Writing Unit Tests for Services

# tests/unit/services/test_agent_service.py
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4

from src.services.agents.service import AgentService
from src.services.agents.schema import AgentCreate

@pytest.mark.asyncio
async def test_create_agent(db_session):
    """Test creating an agent with AgentService."""
    # Arrange
    org_id = uuid4()
    user_mock = {"id": uuid4(), "org_id": org_id}
    agent_data = AgentCreate(
        name="Test Agent",
        description="Test agent description",
        model_id=uuid4(),
        is_active=True,
        settings={"key": "value"}
    )
    
    acquire_mock = MagicMock()
    agent_service = AgentService(acquire_mock)
    
    # Act
    result = await agent_service.create(
        org_id=org_id,
        data=agent_data,
        session=db_session,
        user=user_mock
    )
    
    # Assert
    assert result.name == "Test Agent"
    assert result.organization_id == org_id

Integration Tests

Integration tests verify that different components work together correctly.

Testing Service Integration

# tests/integration/services/test_agent_kb_integration.py
import pytest
from uuid import uuid4

from src.services.agents.service import AgentService
from src.services.kb.service import KBService
from src.services.agents.schema import AgentCreate
from src.services.kb.schema import KBCreate

@pytest.mark.asyncio
async def test_agent_with_kb_integration(db_session):
    """Test integration between Agent and Knowledge Base services."""
    # Arrange
    org_id = uuid4()
    user_mock = {"id": uuid4(), "org_id": org_id}
    
    # Create a knowledge base
    kb_service = KBService(None)
    kb_data = KBCreate(
        name="Test KB",
        description="Test KB description",
        embedding_model="text-embedding-ada-002"
    )
    kb = await kb_service.create(
        org_id=org_id,
        data=kb_data,
        session=db_session,
        user=user_mock
    )
    
    # Create an agent that uses the knowledge base
    agent_service = AgentService(None)
    agent_data = AgentCreate(
        name="Test Agent with KB",
        description="Agent using KB",
        model_id=uuid4(),
        settings={"knowledge_base_id": str(kb.id)}
    )
    
    # Act
    agent = await agent_service.create(
        org_id=org_id,
        data=agent_data,
        session=db_session,
        user=user_mock
    )
    
    # Assert
    assert agent.settings.get("knowledge_base_id") == str(kb.id)
    # Additional assertions to verify integration

Integration Test Service Interaction

Testing Database Operations

# tests/integration/services/test_agent_db_operations.py
import pytest
from uuid import uuid4

from src.services.agents.service import AgentService
from src.services.agents.schema import AgentCreate, AgentUpdate

@pytest.mark.asyncio
async def test_agent_crud_operations(db_session):
    """Test CRUD operations for Agents."""
    # Arrange
    org_id = uuid4()
    user_mock = {"id": uuid4(), "org_id": org_id}
    agent_service = AgentService(None)
    
    # Act - Create
    agent_data = AgentCreate(
        name="Test Agent",
        description="Initial description",
        model_id=uuid4(),
        settings={}
    )
    agent = await agent_service.create(
        org_id=org_id,
        data=agent_data,
        session=db_session,
        user=user_mock
    )
    
    # Act - Read
    retrieved_agent = await agent_service.get(
        agent_id=agent.id,
        session=db_session,
        user=user_mock
    )
    
    # Act - Update
    update_data = AgentUpdate(
        description="Updated description"
    )
    updated_agent = await agent_service.update(
        agent_id=agent.id,
        data=update_data,
        session=db_session,
        user=user_mock
    )
    
    # Act - Delete
    delete_result = await agent_service.delete(
        agent_id=agent.id,
        session=db_session,
        user=user_mock
    )
    
    # Assert
    assert retrieved_agent.id == agent.id
    assert updated_agent.description == "Updated description"
    assert delete_result["success"] is True

CRUD Operations Sequence

End-to-End (E2E) Tests

E2E tests verify that the entire system works as expected.

Testing API Endpoints

# tests/e2e/api/test_agent_api.py
import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI
import json
from uuid import uuid4

from src.app import app
from src.database import get_db
from src.dependencies.security import JWTBearer, RBAC

# Mock dependencies
@pytest.fixture
def client(override_get_db):
    """Create a test client with overridden dependencies."""
    app.dependency_overrides[get_db] = override_get_db
    
    # Mock auth dependency
    async def override_jwt():
        return {"user_id": str(uuid4()), "org_id": str(uuid4())}
    
    async def override_rbac(*args, **kwargs):
        return {"user_id": str(uuid4()), "org_id": str(uuid4())}
    
    app.dependency_overrides[JWTBearer] = override_jwt
    app.dependency_overrides[RBAC] = override_rbac
    
    with TestClient(app) as test_client:
        yield test_client
    
    # Clear overrides
    app.dependency_overrides.clear()

def test_create_agent(client):
    """Test creating an agent via API endpoint."""
    # Arrange
    org_id = str(uuid4())
    agent_data = {
        "name": "API Test Agent",
        "description": "Created via API test",
        "model_id": str(uuid4()),
        "is_active": True,
        "settings": {"test_key": "test_value"}
    }
    
    # Act
    response = client.post(
        f"/api/agents/create?org_id={org_id}",
        json=agent_data
    )
    
    # Assert
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "API Test Agent"
    assert "id" in data

E2E Test Flow

Mocking

Proper use of mocks is essential for isolation and controlled testing.

Mocking External Services

# tests/unit/services/test_llm_service.py
import pytest
from unittest.mock import patch, MagicMock

from src.services.llm.service import LLMService

@pytest.mark.asyncio
async def test_generate_text_with_mocked_openai():
    """Test LLM service with mocked OpenAI client."""
    # Arrange
    mock_response = MagicMock()
    mock_response.choices = [MagicMock(message=MagicMock(content="Mocked response"))]
    
    # Mock the OpenAI client
    with patch("openai.AsyncClient.chat.completions.create", return_value=mock_response):
        llm_service = LLMService(None)
        
        # Act
        result = await llm_service.generate_text(
            model="gpt-4o",
            prompt="Hello, world!",
            max_tokens=100
        )
        
        # Assert
        assert result == "Mocked response"

Mocking Concept Visualization

Test Fixtures

Create reusable fixtures to reduce code duplication and simplify tests.

Model Fixtures

# tests/fixtures/models.py
import pytest
from uuid import uuid4

from src.models import AgentModel, UserModel, OrganizationModel

@pytest.fixture
async def test_organization(db_session):
    """Create a test organization."""
    org = OrganizationModel(
        name="Test Organization",
        slug="test-org",
        settings={},
        is_active=True
    )
    db_session.add(org)
    await db_session.commit()
    await db_session.refresh(org)
    return org

@pytest.fixture
async def test_user(db_session, test_organization):
    """Create a test user."""
    user = UserModel(
        email="test@example.com",
        password="hashed_password",
        first_name="Test",
        last_name="User",
        is_active=True
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user

@pytest.fixture
async def test_agent(db_session, test_organization, test_user):
    """Create a test agent."""
    agent = AgentModel(
        organization_id=test_organization.id,
        user_id=test_user.id,
        name="Test Agent",
        description="Test agent for unit tests",
        model_id=uuid4(),  # This would be a real model ID in a real test
        is_active=True,
        settings={}
    )
    db_session.add(agent)
    await db_session.commit()
    await db_session.refresh(agent)
    return agent

Test Coverage

Aim for high test coverage of the codebase.

Running Coverage Reports

# Run tests with coverage
pytest --cov=src tests/

# Generate HTML coverage report
pytest --cov=src --cov-report=html tests/

Coverage Targets

Unit Tests: Aim for 80%+ coverage of all functions and methods
Integration Tests: Cover all major service interactions
E2E Tests: Cover all API endpoints and primary user flows

Continuous Integration

Tests will be integrated into the CI/CD pipeline to ensure code quality.

CI/CD Workflow

GitHub Actions Configuration

Update the .github/workflows/pre-commit.yml file to include testing:

name: Tests and Linting
on:
  pull_request:
    branches: [main, dev]
  push:
    branches: [main, dev]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:14
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_USER: postgres
          POSTGRES_DB: test_zyeta
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.10'
          cache: 'pip'
      
      - name: Install Poetry
        run: |
          curl -sSL https://install.python-poetry.org | python3 -
          poetry config virtualenvs.create false
      
      - name: Install dependencies
        run: |
          poetry install --no-interaction
      
      - name: Run tests
        run: |
          pytest --cov=src tests/
          
      - name: Generate coverage report
        run: |
          pytest --cov=src --cov-report=xml tests/
      
      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v3
        with:
          file: ./coverage.xml

  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
      - run: pip install ruff
      - run: pip install mypy
      - run: pip install pre-commit
      - run: pre-commit run --all-files

Testing Strategies by Component

Models

Test model instantiation
Test relationships between models
Test constraints and defaults

Services

Test business logic
Test error handling
Test edge cases
Mock external dependencies

API Endpoints

Test request validation
Test authentication/authorization
Test response format
Test error responses

Component Testing Focus

Best Practices

General Guidelines

Test pyramid: Write more unit tests than integration tests, and more integration tests than E2E tests
Test isolation: Each test should be independent and not rely on the state from other tests
Descriptive naming: Use clear, descriptive names for test functions
Arrange-Act-Assert: Structure tests with clear sections for setup, execution, and verification
Don’t test implementation details: Focus on behavior, not implementation

Test Organization Pattern

Asynchronous Testing

Since our application uses async/await, follow these practices:

Use pytest.mark.asyncio decorator for async test functions
Use async fixtures where needed
Handle event loops properly

@pytest.mark.asyncio
async def test_async_function():
    # Test async code here
    result = await some_async_function()
    assert result == expected_value

Testing Database Operations

Use transactions to isolate tests
Roll back after each test
Use a separate test database

Common Pitfalls

Slow tests: Keep tests fast, especially unit tests
Flaky tests: Avoid tests that sometimes pass and sometimes fail
Over-mocking: Don’t mock everything; test real interactions where practical
Hardcoded test data: Use fixtures and factories for test data
Missing edge cases: Test error paths, not just happy paths

Testing Pitfalls and Solutions

Conclusion

A comprehensive testing strategy is essential for maintaining a healthy codebase. By following this guide, developers can write effective tests that ensure the reliability and quality of the Zyeta backend.

Remember that tests are an investment in the future of the codebase, making it safer to refactor and extend functionality while maintaining confidence in the system’s behavior.

Additional Resources

Was this page helpful?

Adding New Models Database Migration Guide

On this page

Overview
Test Pyramid
Test Directory Structure
Setting Up the Testing Environment
Prerequisites
Test Configuration
Test Fixture Dependencies
Unit Tests
Writing Unit Tests for Models
Unit Test Flow
Writing Unit Tests for Services
Integration Tests
Testing Service Integration
Integration Test Service Interaction
Testing Database Operations
CRUD Operations Sequence
End-to-End (E2E) Tests
Testing API Endpoints
E2E Test Flow
Mocking
Mocking External Services
Mocking Concept Visualization
Test Fixtures
Model Fixtures
Test Coverage
Running Coverage Reports
Coverage Targets
Continuous Integration
CI/CD Workflow
GitHub Actions Configuration
Testing Strategies by Component
Models
Services
API Endpoints
Component Testing Focus
Best Practices
General Guidelines
Test Organization Pattern
Asynchronous Testing
Testing Database Operations
Common Pitfalls
Testing Pitfalls and Solutions
Conclusion
Additional Resources

Introduction

Getting Started

Architecture

Authentication

API Reference

Development Guides

Troubleshooting

​Overview

​Test Pyramid

​Test Directory Structure

​Setting Up the Testing Environment

​Prerequisites

​Test Configuration

​Test Fixture Dependencies

​Unit Tests

​Writing Unit Tests for Models

​Unit Test Flow

​Writing Unit Tests for Services

​Integration Tests

​Testing Service Integration

​Integration Test Service Interaction

​Testing Database Operations

​CRUD Operations Sequence

​End-to-End (E2E) Tests

​Testing API Endpoints

​E2E Test Flow

​Mocking

​Mocking External Services

​Mocking Concept Visualization

​Test Fixtures

​Model Fixtures

​Test Coverage

​Running Coverage Reports

​Coverage Targets

​Continuous Integration

​CI/CD Workflow

​GitHub Actions Configuration

​Testing Strategies by Component

​Models

​Services

​API Endpoints

​Component Testing Focus

​Best Practices

​General Guidelines

​Test Organization Pattern

​Asynchronous Testing

​Testing Database Operations

​Common Pitfalls

​Testing Pitfalls and Solutions

​Conclusion

​Additional Resources

Overview

Test Pyramid

Test Directory Structure

Setting Up the Testing Environment

Prerequisites

Test Configuration

Test Fixture Dependencies

Unit Tests

Writing Unit Tests for Models

Unit Test Flow

Writing Unit Tests for Services

Integration Tests

Testing Service Integration

Integration Test Service Interaction

Testing Database Operations

CRUD Operations Sequence

End-to-End (E2E) Tests

Testing API Endpoints

E2E Test Flow

Mocking

Mocking External Services

Mocking Concept Visualization

Test Fixtures

Model Fixtures

Test Coverage

Running Coverage Reports

Coverage Targets

Continuous Integration

CI/CD Workflow

GitHub Actions Configuration

Testing Strategies by Component

Models

Services

API Endpoints

Component Testing Focus

Best Practices

General Guidelines

Test Organization Pattern

Asynchronous Testing

Testing Database Operations

Common Pitfalls

Testing Pitfalls and Solutions

Conclusion

Additional Resources